From 588d8463dfa1c8936b8753b5f0fd794ecb115203 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Thu, 8 Oct 2026 12:56:33 +0200 Subject: [PATCH 01/14] feat: let a job see that its worker is draining (#11612) * feat: let a job see that its worker is draining * fix: gate worker draining check on job visibility and export sdk helper * fix: keep draining across re-registration and ignore parked jobs * fix: reset draining only on worker startup registration * chore: update ee-repo-ref to 55feb2e84332179faa4e17aeb419f7f480df79a6 This commit updates the EE repository reference after PR #857 was merged in windmill-ee-private. Previous ee-repo-ref: e5010fa046a3aba93595d1bf44eb6a0cef2b5273 New ee-repo-ref: 55feb2e84332179faa4e17aeb419f7f480df79a6 Automated by sync-ee-ref workflow. --------- Co-authored-by: windmill-internal-app[bot] --- ...3cd4a3f9226a5e0c6b15c2580ddd1dba6199b.json | 23 +++++++++ ...f7d4b418a9e82be56dfe7dd65d160e0dcfd13.json | 24 +++++++++ ...f20b6971b993e0671b6ccb45db5ef028550bc.json | 33 ------------ ...123a98ec175f9d3dc72e40a1cb40e14dec919.json | 14 +++++ ...64360a5d5c783d6ca0a3c4123ea126ed0e60b.json | 34 +++++++++++++ ...ceec2b6f1ae1a32fcf375161d04419404d9c.json} | 10 +++- backend/ee-repo-ref.txt | 2 +- ...261008070335_worker_ping_draining.down.sql | 1 + ...20261008070335_worker_ping_draining.up.sql | 1 + backend/summarized_schema.txt | 2 +- backend/tests/worker_ping_ip.rs | 32 +++++++++++- backend/windmill-api-auth/src/scopes.rs | 1 + .../tests/jobs_authed.rs | 48 +++++++++++++++++ backend/windmill-api-workers/src/lib.rs | 3 +- backend/windmill-api/openapi.yaml | 24 +++++++++ backend/windmill-api/src/jobs.rs | 51 +++++++++++++++++++ backend/windmill-common/src/worker.rs | 24 ++++++++- backend/windmill-worker/src/common.rs | 1 + backend/windmill-worker/src/worker.rs | 31 +++++++++-- backend/windmill-worker/src/worker_utils.rs | 50 ++++++++++++++++-- cli/src/guidance/skills.gen.ts | 36 +++++++++++++ .../(root)/(logged)/workers/+page.svelte | 11 +++- python-client/wmill/wmill/client.py | 27 ++++++++++ system_prompts/auto-generated/prompts.ts | 20 ++++++++ system_prompts/auto-generated/script.md | 20 ++++++++ system_prompts/auto-generated/sdks/python.md | 12 +++++ .../auto-generated/sdks/typescript.md | 8 +++ .../skills/write-script-bun/SKILL.md | 8 +++ .../skills/write-script-bunnative/SKILL.md | 8 +++ .../skills/write-script-deno/SKILL.md | 8 +++ .../skills/write-script-python3/SKILL.md | 12 +++++ typescript-client/build.jsr.sh | 2 +- typescript-client/build.sh | 4 +- typescript-client/client.ts | 13 +++++ 34 files changed, 548 insertions(+), 50 deletions(-) create mode 100644 backend/.sqlx/query-1e3407ad2ec46bd965f2aeede0d3cd4a3f9226a5e0c6b15c2580ddd1dba6199b.json create mode 100644 backend/.sqlx/query-7963eff5f7109874b94c0940eaef7d4b418a9e82be56dfe7dd65d160e0dcfd13.json delete mode 100644 backend/.sqlx/query-c748617e060bc41b5922df4b433f20b6971b993e0671b6ccb45db5ef028550bc.json create mode 100644 backend/.sqlx/query-e4b9c0e8a46dbfe29bace8d8b12123a98ec175f9d3dc72e40a1cb40e14dec919.json create mode 100644 backend/.sqlx/query-e7a8c0ee68b8626eb45a7eacdd364360a5d5c783d6ca0a3c4123ea126ed0e60b.json rename backend/.sqlx/{query-68bca8f839e47705b11d312ee874eceaa3d1d24d9053ad4aea94b9f8465585ca.json => query-fe22c4609d8b2472c0f83fedb92eceec2b6f1ae1a32fcf375161d04419404d9c.json} (86%) create mode 100644 backend/migrations/20261008070335_worker_ping_draining.down.sql create mode 100644 backend/migrations/20261008070335_worker_ping_draining.up.sql diff --git a/backend/.sqlx/query-1e3407ad2ec46bd965f2aeede0d3cd4a3f9226a5e0c6b15c2580ddd1dba6199b.json b/backend/.sqlx/query-1e3407ad2ec46bd965f2aeede0d3cd4a3f9226a5e0c6b15c2580ddd1dba6199b.json new file mode 100644 index 0000000000..036610e08c --- /dev/null +++ b/backend/.sqlx/query-1e3407ad2ec46bd965f2aeede0d3cd4a3f9226a5e0c6b15c2580ddd1dba6199b.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT wp.draining FROM v2_job_queue q JOIN worker_ping wp ON wp.worker = q.worker\n WHERE q.id = $1 AND q.workspace_id = $2 AND q.running AND q.started_at IS NOT NULL", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "draining", + "type_info": "Bool" + } + ], + "parameters": { + "Left": [ + "Uuid", + "Text" + ] + }, + "nullable": [ + false + ] + }, + "hash": "1e3407ad2ec46bd965f2aeede0d3cd4a3f9226a5e0c6b15c2580ddd1dba6199b" +} diff --git a/backend/.sqlx/query-7963eff5f7109874b94c0940eaef7d4b418a9e82be56dfe7dd65d160e0dcfd13.json b/backend/.sqlx/query-7963eff5f7109874b94c0940eaef7d4b418a9e82be56dfe7dd65d160e0dcfd13.json new file mode 100644 index 0000000000..1c8dfcfa57 --- /dev/null +++ b/backend/.sqlx/query-7963eff5f7109874b94c0940eaef7d4b418a9e82be56dfe7dd65d160e0dcfd13.json @@ -0,0 +1,24 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT wp.draining FROM v2_job j\n JOIN v2_job_queue q ON q.id = j.id\n JOIN worker_ping wp ON wp.worker = q.worker\n WHERE j.id = $1 AND j.workspace_id = $2 AND q.running AND q.started_at IS NOT NULL\n AND ($3::text[] IS NULL OR j.tag = ANY($3))", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "draining", + "type_info": "Bool" + } + ], + "parameters": { + "Left": [ + "Uuid", + "Text", + "TextArray" + ] + }, + "nullable": [ + false + ] + }, + "hash": "7963eff5f7109874b94c0940eaef7d4b418a9e82be56dfe7dd65d160e0dcfd13" +} diff --git a/backend/.sqlx/query-c748617e060bc41b5922df4b433f20b6971b993e0671b6ccb45db5ef028550bc.json b/backend/.sqlx/query-c748617e060bc41b5922df4b433f20b6971b993e0671b6ccb45db5ef028550bc.json deleted file mode 100644 index 7f6fe7a81c..0000000000 --- a/backend/.sqlx/query-c748617e060bc41b5922df4b433f20b6971b993e0671b6ccb45db5ef028550bc.json +++ /dev/null @@ -1,33 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "INSERT INTO worker_ping (worker_instance, worker, ip, custom_tags, worker_group, dedicated_worker, dedicated_workers, wm_version, vcpus, memory, job_isolation, native_mode) VALUES ($1, $2, COALESCE($3, 'NO IP'), $4, $5, $6, $7, $8, $9, $10, $11, $12) ON CONFLICT (worker)\n DO UPDATE set ping_at = now(), worker_instance = EXCLUDED.worker_instance, ip = COALESCE($3, worker_ping.ip), custom_tags = EXCLUDED.custom_tags, worker_group = EXCLUDED.worker_group, dedicated_worker = EXCLUDED.dedicated_worker, dedicated_workers = EXCLUDED.dedicated_workers, wm_version = EXCLUDED.wm_version, vcpus = COALESCE(EXCLUDED.vcpus, worker_ping.vcpus), memory = COALESCE(EXCLUDED.memory, worker_ping.memory), job_isolation = EXCLUDED.job_isolation, native_mode = EXCLUDED.native_mode, current_job_id = NULL, current_job_workspace_id = NULL\n RETURNING jobs_executed", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "jobs_executed", - "type_info": "Int4" - } - ], - "parameters": { - "Left": [ - "Varchar", - "Varchar", - "Text", - "TextArray", - "Varchar", - "Varchar", - "TextArray", - "Varchar", - "Int8", - "Int8", - "Text", - "Bool" - ] - }, - "nullable": [ - false - ] - }, - "hash": "c748617e060bc41b5922df4b433f20b6971b993e0671b6ccb45db5ef028550bc" -} diff --git a/backend/.sqlx/query-e4b9c0e8a46dbfe29bace8d8b12123a98ec175f9d3dc72e40a1cb40e14dec919.json b/backend/.sqlx/query-e4b9c0e8a46dbfe29bace8d8b12123a98ec175f9d3dc72e40a1cb40e14dec919.json new file mode 100644 index 0000000000..9ebf9289c2 --- /dev/null +++ b/backend/.sqlx/query-e4b9c0e8a46dbfe29bace8d8b12123a98ec175f9d3dc72e40a1cb40e14dec919.json @@ -0,0 +1,14 @@ +{ + "db_name": "PostgreSQL", + "query": "UPDATE worker_ping SET draining = true WHERE worker = $1", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [] + }, + "hash": "e4b9c0e8a46dbfe29bace8d8b12123a98ec175f9d3dc72e40a1cb40e14dec919" +} diff --git a/backend/.sqlx/query-e7a8c0ee68b8626eb45a7eacdd364360a5d5c783d6ca0a3c4123ea126ed0e60b.json b/backend/.sqlx/query-e7a8c0ee68b8626eb45a7eacdd364360a5d5c783d6ca0a3c4123ea126ed0e60b.json new file mode 100644 index 0000000000..9f51c69616 --- /dev/null +++ b/backend/.sqlx/query-e7a8c0ee68b8626eb45a7eacdd364360a5d5c783d6ca0a3c4123ea126ed0e60b.json @@ -0,0 +1,34 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO worker_ping (worker_instance, worker, ip, custom_tags, worker_group, dedicated_worker, dedicated_workers, wm_version, vcpus, memory, job_isolation, native_mode, draining) VALUES ($1, $2, COALESCE($3, 'NO IP'), $4, $5, $6, $7, $8, $9, $10, $11, $12, COALESCE($13, false)) ON CONFLICT (worker)\n DO UPDATE set ping_at = now(), worker_instance = EXCLUDED.worker_instance, ip = COALESCE($3, worker_ping.ip), custom_tags = EXCLUDED.custom_tags, worker_group = EXCLUDED.worker_group, dedicated_worker = EXCLUDED.dedicated_worker, dedicated_workers = EXCLUDED.dedicated_workers, wm_version = EXCLUDED.wm_version, vcpus = COALESCE(EXCLUDED.vcpus, worker_ping.vcpus), memory = COALESCE(EXCLUDED.memory, worker_ping.memory), job_isolation = EXCLUDED.job_isolation, native_mode = EXCLUDED.native_mode, current_job_id = NULL, current_job_workspace_id = NULL, draining = COALESCE($13, worker_ping.draining)\n RETURNING jobs_executed", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "jobs_executed", + "type_info": "Int4" + } + ], + "parameters": { + "Left": [ + "Varchar", + "Varchar", + "Text", + "TextArray", + "Varchar", + "Varchar", + "TextArray", + "Varchar", + "Int8", + "Int8", + "Text", + "Bool", + "Bool" + ] + }, + "nullable": [ + false + ] + }, + "hash": "e7a8c0ee68b8626eb45a7eacdd364360a5d5c783d6ca0a3c4123ea126ed0e60b" +} diff --git a/backend/.sqlx/query-68bca8f839e47705b11d312ee874eceaa3d1d24d9053ad4aea94b9f8465585ca.json b/backend/.sqlx/query-fe22c4609d8b2472c0f83fedb92eceec2b6f1ae1a32fcf375161d04419404d9c.json similarity index 86% rename from backend/.sqlx/query-68bca8f839e47705b11d312ee874eceaa3d1d24d9053ad4aea94b9f8465585ca.json rename to backend/.sqlx/query-fe22c4609d8b2472c0f83fedb92eceec2b6f1ae1a32fcf375161d04419404d9c.json index 583d2bcc6b..5ee5c7571d 100644 --- a/backend/.sqlx/query-68bca8f839e47705b11d312ee874eceaa3d1d24d9053ad4aea94b9f8465585ca.json +++ b/backend/.sqlx/query-fe22c4609d8b2472c0f83fedb92eceec2b6f1ae1a32fcf375161d04419404d9c.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "SELECT worker, worker_instance, EXTRACT(EPOCH FROM (now() - ping_at))::integer as last_ping, started_at, ip, jobs_executed,\n CASE WHEN $4 IS TRUE THEN current_job_id ELSE NULL END as last_job_id, CASE WHEN $4 IS TRUE THEN current_job_workspace_id ELSE NULL END as last_job_workspace_id,\n custom_tags, worker_group, wm_version, occupancy_rate, occupancy_rate_15s, occupancy_rate_5m, occupancy_rate_30m, memory, vcpus, memory_usage, wm_memory_usage, job_isolation, native_mode\n FROM worker_ping\n WHERE ($1::integer IS NULL AND ping_at > now() - interval '5 minute') OR (ping_at > now() - ($1 || ' seconds')::interval)\n ORDER BY ping_at desc LIMIT $2 OFFSET $3", + "query": "SELECT worker, worker_instance, EXTRACT(EPOCH FROM (now() - ping_at))::integer as last_ping, started_at, ip, jobs_executed,\n CASE WHEN $4 IS TRUE THEN current_job_id ELSE NULL END as last_job_id, CASE WHEN $4 IS TRUE THEN current_job_workspace_id ELSE NULL END as last_job_workspace_id,\n custom_tags, worker_group, wm_version, occupancy_rate, occupancy_rate_15s, occupancy_rate_5m, occupancy_rate_30m, memory, vcpus, memory_usage, wm_memory_usage, job_isolation, native_mode, draining\n FROM worker_ping\n WHERE ($1::integer IS NULL AND ping_at > now() - interval '5 minute') OR (ping_at > now() - ($1 || ' seconds')::interval)\n ORDER BY ping_at desc LIMIT $2 OFFSET $3", "describe": { "columns": [ { @@ -107,6 +107,11 @@ "ordinal": 20, "name": "native_mode", "type_info": "Bool" + }, + { + "ordinal": 21, + "name": "draining", + "type_info": "Bool" } ], "parameters": { @@ -138,8 +143,9 @@ true, true, true, + false, false ] }, - "hash": "68bca8f839e47705b11d312ee874eceaa3d1d24d9053ad4aea94b9f8465585ca" + "hash": "fe22c4609d8b2472c0f83fedb92eceec2b6f1ae1a32fcf375161d04419404d9c" } diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 5a36bc5e61..ff44a56995 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -e5689bf58a125ff46aabb6b5d0f030a7e5c715d3 +55feb2e84332179faa4e17aeb419f7f480df79a6 diff --git a/backend/migrations/20261008070335_worker_ping_draining.down.sql b/backend/migrations/20261008070335_worker_ping_draining.down.sql new file mode 100644 index 0000000000..64a5f2949d --- /dev/null +++ b/backend/migrations/20261008070335_worker_ping_draining.down.sql @@ -0,0 +1 @@ +ALTER TABLE worker_ping DROP COLUMN IF EXISTS draining; diff --git a/backend/migrations/20261008070335_worker_ping_draining.up.sql b/backend/migrations/20261008070335_worker_ping_draining.up.sql new file mode 100644 index 0000000000..33b5ad5e3e --- /dev/null +++ b/backend/migrations/20261008070335_worker_ping_draining.up.sql @@ -0,0 +1 @@ +ALTER TABLE worker_ping ADD COLUMN IF NOT EXISTS draining BOOLEAN NOT NULL DEFAULT FALSE; diff --git a/backend/summarized_schema.txt b/backend/summarized_schema.txt index 2a518a7cc8..284a822b47 100644 --- a/backend/summarized_schema.txt +++ b/backend/summarized_schema.txt @@ -219,7 +219,7 @@ websocket_trigger: path(char), url(char), script_path(char), is_flow(bool), work windmill_migrations: name(text), created_at(ts) worker_group_job_stats: hour(bigint), worker_group(text), script_lang(char), workspace_id(char), job_count(int), total_duration_ms(bigint) FK: (workspace_id) -> workspace(id) -worker_ping: worker(char), worker_instance(char), ping_at(ts), started_at(ts), ip(char), jobs_executed(int), custom_tags(text[]), worker_group(char), dedicated_worker(char), wm_version(char), current_job_id(uuid), current_job_workspace_id(char), vcpus(bigint), memory(bigint), occupancy_rate(float), memory_usage(bigint), wm_memory_usage(bigint), occupancy_rate_15s(float), occupancy_rate_5m(float), occupancy_rate_30m(float), job_isolation(text), dedicated_workers(text[]) +worker_ping: worker(char), worker_instance(char), ping_at(ts), started_at(ts), ip(char), jobs_executed(int), custom_tags(text[]), worker_group(char), dedicated_worker(char), wm_version(char), current_job_id(uuid), current_job_workspace_id(char), vcpus(bigint), memory(bigint), occupancy_rate(float), memory_usage(bigint), wm_memory_usage(bigint), occupancy_rate_15s(float), occupancy_rate_5m(float), occupancy_rate_30m(float), job_isolation(text), dedicated_workers(text[]), draining(bool) workspace: id(char), name(char), owner(char), deleted(bool), premium(bool), parent_workspace_id(char), is_dev_workspace(bool) FK: (parent_workspace_id) -> workspace(id) workspace_dependencies: id(bigint), name(char), content(text), language(script_lang), description(text), archived(bool), workspace_id(char), created_at(ts) diff --git a/backend/tests/worker_ping_ip.rs b/backend/tests/worker_ping_ip.rs index 772f4193c3..2421e888ef 100644 --- a/backend/tests/worker_ping_ip.rs +++ b/backend/tests/worker_ping_ip.rs @@ -2,10 +2,19 @@ use sqlx::{Pool, Postgres}; use uuid::Uuid; use windmill_common::{ external_ip::UNKNOWN_IP, - worker::{insert_ping_query, update_worker_ping_main_loop_query}, + worker::{insert_ping_query, set_worker_draining_query, update_worker_ping_main_loop_query}, }; async fn insert_ping(db: &Pool, worker: &str, ip: Option<&str>) -> anyhow::Result<()> { + register(db, worker, ip, None).await +} + +async fn register( + db: &Pool, + worker: &str, + ip: Option<&str>, + draining: Option, +) -> anyhow::Result<()> { insert_ping_query( "test-instance", worker, @@ -19,12 +28,33 @@ async fn insert_ping(db: &Pool, worker: &str, ip: Option<&str>) -> any None, None, false, + draining, db, ) .await?; Ok(()) } +/// A worker re-registers mid-life once its external IP resolves. That registration must not +/// clear a draining flag the same process already published; only a worker starting up under +/// a reclaimed name resets it. +#[sqlx::test] +async fn only_a_starting_worker_resets_draining(db: Pool) -> anyhow::Result<()> { + let draining = || async { + sqlx::query_scalar::<_, bool>("SELECT draining FROM worker_ping WHERE worker = 'wk-drain'") + .fetch_one(&db) + .await + }; + register(&db, "wk-drain", None, Some(false)).await?; + set_worker_draining_query("wk-drain", &db).await?; + register(&db, "wk-drain", Some("1.2.3.4"), None).await?; + assert!(draining().await?); + + register(&db, "wk-drain", None, Some(false)).await?; + assert!(!draining().await?); + Ok(()) +} + /// The external IP resolves in the background, so the initial ping often has none yet. That must /// not blank the address a previous process wrote to the row this one reclaims — worker names are /// stable across restarts under EXIT_AFTER_N_JOBS. diff --git a/backend/windmill-api-auth/src/scopes.rs b/backend/windmill-api-auth/src/scopes.rs index dce6a0ff72..b636e11a1c 100644 --- a/backend/windmill-api-auth/src/scopes.rs +++ b/backend/windmill-api-auth/src/scopes.rs @@ -996,6 +996,7 @@ pub fn is_own_job_runtime_route(route_path: &str, http_method: &str, job_id: uui ["job_metrics", "set_progress", ..] => (http_method == "POST", 2), ["job_metrics", "get_progress", ..] => (http_method == "GET", 2), ["jobs" | "jobs_u", "resume_urls" | "wac_approval_urls", ..] => (http_method == "GET", 2), + ["jobs", "worker_is_draining", ..] => (http_method == "GET", 2), ["jobs", "wac", "inline_checkpoint", ..] => (http_method == "POST", 3), ["jobs", "run", "workflow_as_code", ..] => (http_method == "POST", 3), _ => return false, diff --git a/backend/windmill-api-integration-tests/tests/jobs_authed.rs b/backend/windmill-api-integration-tests/tests/jobs_authed.rs index c3da2f3fb9..2c5d2ee78e 100644 --- a/backend/windmill-api-integration-tests/tests/jobs_authed.rs +++ b/backend/windmill-api-integration-tests/tests/jobs_authed.rs @@ -702,3 +702,51 @@ async fn test_list_filtered_job_uuids_resolved_excludes_queue( Ok(()) } + +#[sqlx::test(migrations = "../migrations", fixtures("base"))] +async fn test_worker_is_draining_follows_the_job_worker(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + + let job = Uuid::new_v4(); + sqlx::query( + "INSERT INTO v2_job (id, workspace_id, created_by, permissioned_as, kind, tag, args) + VALUES ($1, 'test-workspace', 'test-user', 'u/test-user', 'script', 'deno', '{}'::jsonb)", + ) + .bind(job) + .execute(&db) + .await?; + sqlx::query( + "INSERT INTO v2_job_queue (id, workspace_id, scheduled_for, tag, running, started_at, worker) + VALUES ($1, 'test-workspace', now(), 'deno', true, now(), 'wk-drain-test')", + ) + .bind(job) + .execute(&db) + .await?; + sqlx::query( + "INSERT INTO worker_ping (worker, worker_instance, ip, worker_group, wm_version) + VALUES ('wk-drain-test', 'host', 'NO IP', 'default', 'v0')", + ) + .execute(&db) + .await?; + + let url = format!("http://localhost:{port}/api/w/test-workspace/jobs/worker_is_draining/{job}"); + let draining = || async { + let resp = authed(client().get(&url)).send().await?; + anyhow::Ok(resp.json::().await?) + }; + assert!(!draining().await?); + + windmill_common::worker::set_worker_draining_query("wk-drain-test", &db).await?; + assert!(draining().await?); + + // A parked job keeps its last worker on the queue row without running on it. + sqlx::query("UPDATE v2_job_queue SET started_at = NULL WHERE id = $1") + .bind(job) + .execute(&db) + .await?; + assert!(!draining().await?); + + Ok(()) +} diff --git a/backend/windmill-api-workers/src/lib.rs b/backend/windmill-api-workers/src/lib.rs index 0102f1b033..cf0b85609a 100644 --- a/backend/windmill-api-workers/src/lib.rs +++ b/backend/windmill-api-workers/src/lib.rs @@ -86,6 +86,7 @@ struct WorkerPing { job_isolation: Option, #[serde(skip_serializing_if = "Option::is_none")] native_mode: Option, + draining: bool, } // #[derive(Serialize, Deserialize)] @@ -118,7 +119,7 @@ async fn list_worker_pings( WorkerPing, "SELECT worker, worker_instance, EXTRACT(EPOCH FROM (now() - ping_at))::integer as last_ping, started_at, ip, jobs_executed, CASE WHEN $4 IS TRUE THEN current_job_id ELSE NULL END as last_job_id, CASE WHEN $4 IS TRUE THEN current_job_workspace_id ELSE NULL END as last_job_workspace_id, - custom_tags, worker_group, wm_version, occupancy_rate, occupancy_rate_15s, occupancy_rate_5m, occupancy_rate_30m, memory, vcpus, memory_usage, wm_memory_usage, job_isolation, native_mode + custom_tags, worker_group, wm_version, occupancy_rate, occupancy_rate_15s, occupancy_rate_5m, occupancy_rate_30m, memory, vcpus, memory_usage, wm_memory_usage, job_isolation, native_mode, draining FROM worker_ping WHERE ($1::integer IS NULL AND ping_at > now() - interval '5 minute') OR (ping_at > now() - ($1 || ' seconds')::interval) ORDER BY ping_at desc LIMIT $2 OFFSET $3", diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 9173eae28c..5e635e39bd 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -17322,6 +17322,27 @@ paths: schema: type: string + /w/{workspace}/jobs/worker_is_draining/{id}: + get: + summary: whether the worker running this job is draining + description: | + True once the worker running the job has received its shutdown signal. The worker + still waits for the job to finish, so a long-running job can poll this to exit early. + False when the job is not currently running on a worker. + operationId: workerIsDraining + tags: + - job + parameters: + - $ref: "#/components/parameters/WorkspaceId" + - $ref: "#/components/parameters/JobId" + responses: + "200": + description: whether the worker running this job is draining + content: + application/json: + schema: + type: boolean + /w/{workspace}/jobs_u/get_logs/{id}: get: summary: get job logs @@ -35901,6 +35922,9 @@ components: type: string native_mode: type: boolean + draining: + type: boolean + description: the worker received its shutdown signal and exits once its current job is done required: - worker - worker_instance diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index 7d700c4ab6..5b0da21c50 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -378,6 +378,10 @@ pub fn workspaced_service() -> Router { "/wac_approval_urls/{job_id}/{step_key}", get(get_wac_approval_urls).layer(cors.clone()), ) + .route( + "/worker_is_draining/{id}", + get(worker_is_draining).layer(cors.clone()), + ) .route( "/result_by_id/{job_id}/{node_id}", get(get_result_by_id).layer(cors.clone()), @@ -613,6 +617,53 @@ async fn get_root_job( Ok(Json(res)) } +/// Whether the worker running this job has received its shutdown signal. A draining worker +/// never interrupts its job, so a long-running script polls this to exit on its own terms. +async fn worker_is_draining( + authed: ApiAuthed, + Extension(db): Extension, + Extension(user_db): Extension, + Path((w_id, id)): Path<(String, Uuid)>, +) -> windmill_common::error::JsonResult { + // A job always may ask about itself. Anyone else must be able to see the job: the + // caller's row policies on v2_job decide, narrowed by the token's tag filter. + let draining = if authed.job_id == Some(id) { + sqlx::query_scalar!( + "SELECT wp.draining FROM v2_job_queue q JOIN worker_ping wp ON wp.worker = q.worker + WHERE q.id = $1 AND q.workspace_id = $2 AND q.running AND q.started_at IS NOT NULL", + id, + w_id + ) + .fetch_optional(&db) + .await? + } else { + let tags = get_scope_tags(&authed) + .map(|tags| tags.into_iter().map(str::to_string).collect::>()); + let mut tx = user_db.begin(&authed).await?; + let draining = sqlx::query_scalar!( + "SELECT wp.draining FROM v2_job j + JOIN v2_job_queue q ON q.id = j.id + JOIN worker_ping wp ON wp.worker = q.worker + WHERE j.id = $1 AND j.workspace_id = $2 AND q.running AND q.started_at IS NOT NULL + AND ($3::text[] IS NULL OR j.tag = ANY($3))", + id, + w_id, + tags.as_deref() + ) + .fetch_optional(&mut *tx) + .await?; + tx.commit().await?; + draining + } + .unwrap_or(false); + windmill_common::feature_usage::log_feature_usage( + "worker_draining", + "check", + if draining { "draining" } else { "not_draining" }, + ); + Ok(Json(draining)) +} + async fn compute_root_job_for_flow(db: &DB, w_id: &str, job_id: Uuid) -> error::Result { let root_job = sqlx::query_scalar!( r#"SELECT COALESCE(root_job, flow_innermost_root_job, parent_job, id) as "root_job!" FROM v2_job WHERE id = $1 AND workspace_id = $2"#, diff --git a/backend/windmill-common/src/worker.rs b/backend/windmill-common/src/worker.rs index e185d1d445..52fa2da905 100644 --- a/backend/windmill-common/src/worker.rs +++ b/backend/windmill-common/src/worker.rs @@ -1952,6 +1952,7 @@ pub enum PingType { MainLoop, Job, InitScript, + Draining, } #[derive(Serialize, Deserialize)] pub struct Ping { @@ -1974,6 +1975,8 @@ pub struct Ping { pub occupancy_rate_30m: Option, pub job_isolation: Option, pub native_mode: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub draining: Option, pub ping_type: PingType, } pub async fn update_ping_http( @@ -2026,6 +2029,7 @@ pub async fn update_ping_http( insert_ping.memory, insert_ping.job_isolation, insert_ping.native_mode.unwrap_or(false), + insert_ping.draining, db, ) .await?; @@ -2054,10 +2058,21 @@ pub async fn update_ping_http( ) .await? } + PingType::Draining => set_worker_draining_query(worker_name, db).await?, } Ok(()) } +pub async fn set_worker_draining_query(worker_name: &str, db: &DB) -> anyhow::Result<()> { + sqlx::query!( + "UPDATE worker_ping SET draining = true WHERE worker = $1", + worker_name + ) + .execute(db) + .await?; + Ok(()) +} + #[derive(Debug, Serialize, Deserialize)] pub struct JobCancelled { pub canceled_by: String, @@ -2156,6 +2171,9 @@ pub async fn fetch_raw_script_from_app_query( /// `started_at` and `jobs_executed` are the only two columns carried over, being the /// continuity itself — plus `ip` for as long as `ip` is `None`, which means the external IP /// lookup has not resolved yet and the predecessor's address is still the best guess. +/// +/// `draining` is `Some(false)` from a worker starting up and `None` from one re-registering +/// mid-life, which must leave alone a flag its own shutdown may have set in the meantime. pub async fn insert_ping_query( worker_instance: &str, worker_name: &str, @@ -2169,6 +2187,7 @@ pub async fn insert_ping_query( memory: Option, job_isolation: Option, native_mode: bool, + draining: Option, db: &DB, ) -> anyhow::Result { // A NULL `ip` means the external IP lookup is still in flight; a later ping fills it in, and @@ -2176,8 +2195,8 @@ pub async fn insert_ping_query( // lookup that has failed reports `external_ip::UNRETRIEVABLE_IP`, which does overwrite it. The // literal below must stay equal to `external_ip::UNKNOWN_IP`. let previous_jobs_executed = sqlx::query_scalar!( - "INSERT INTO worker_ping (worker_instance, worker, ip, custom_tags, worker_group, dedicated_worker, dedicated_workers, wm_version, vcpus, memory, job_isolation, native_mode) VALUES ($1, $2, COALESCE($3, 'NO IP'), $4, $5, $6, $7, $8, $9, $10, $11, $12) ON CONFLICT (worker) - DO UPDATE set ping_at = now(), worker_instance = EXCLUDED.worker_instance, ip = COALESCE($3, worker_ping.ip), custom_tags = EXCLUDED.custom_tags, worker_group = EXCLUDED.worker_group, dedicated_worker = EXCLUDED.dedicated_worker, dedicated_workers = EXCLUDED.dedicated_workers, wm_version = EXCLUDED.wm_version, vcpus = COALESCE(EXCLUDED.vcpus, worker_ping.vcpus), memory = COALESCE(EXCLUDED.memory, worker_ping.memory), job_isolation = EXCLUDED.job_isolation, native_mode = EXCLUDED.native_mode, current_job_id = NULL, current_job_workspace_id = NULL + "INSERT INTO worker_ping (worker_instance, worker, ip, custom_tags, worker_group, dedicated_worker, dedicated_workers, wm_version, vcpus, memory, job_isolation, native_mode, draining) VALUES ($1, $2, COALESCE($3, 'NO IP'), $4, $5, $6, $7, $8, $9, $10, $11, $12, COALESCE($13, false)) ON CONFLICT (worker) + DO UPDATE set ping_at = now(), worker_instance = EXCLUDED.worker_instance, ip = COALESCE($3, worker_ping.ip), custom_tags = EXCLUDED.custom_tags, worker_group = EXCLUDED.worker_group, dedicated_worker = EXCLUDED.dedicated_worker, dedicated_workers = EXCLUDED.dedicated_workers, wm_version = EXCLUDED.wm_version, vcpus = COALESCE(EXCLUDED.vcpus, worker_ping.vcpus), memory = COALESCE(EXCLUDED.memory, worker_ping.memory), job_isolation = EXCLUDED.job_isolation, native_mode = EXCLUDED.native_mode, current_job_id = NULL, current_job_workspace_id = NULL, draining = COALESCE($13, worker_ping.draining) RETURNING jobs_executed", worker_instance, worker_name, @@ -2191,6 +2210,7 @@ pub async fn insert_ping_query( memory, job_isolation.as_deref(), native_mode, + draining, ) .fetch_one(db) .await?; diff --git a/backend/windmill-worker/src/common.rs b/backend/windmill-worker/src/common.rs index 01523eba59..745f202826 100644 --- a/backend/windmill-worker/src/common.rs +++ b/backend/windmill-worker/src/common.rs @@ -643,6 +643,7 @@ pub async fn update_worker_ping_for_failed_init_script( wm_memory_usage: None, job_isolation: None, native_mode: None, + draining: None, ping_type: PingType::InitScript, }, ) diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 2acafa539f..d0d5d8b4e2 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -155,7 +155,7 @@ use crate::{ handle_app_dependency_job, handle_dependency_job, handle_flow_dependency_job, tally_unfinished_dependency_deploy, }, - worker_utils::{insert_ping, queue_vacuum, update_worker_ping_full}, + worker_utils::{insert_ping, mark_worker_draining, queue_vacuum, update_worker_ping_full}, }; #[cfg(feature = "rust")] @@ -2713,7 +2713,7 @@ pub async fn run_worker( let mut last_ping = Instant::now() - Duration::from_secs(NUM_SECS_PING + 1); let mut reported_ip = cached_ip(); - let previous_jobs_executed = insert_ping(hostname, &worker_name, reported_ip, conn) + let previous_jobs_executed = insert_ping(hostname, &worker_name, reported_ip, true, conn) .await .expect("initial ping could be sent"); @@ -3160,6 +3160,31 @@ pub async fn run_worker( let mut killpill_rx2 = killpill_rx.resubscribe(); + // The loop below only sees the killpill once the running job is done, which is too late + // for that job to learn its worker is going away. + { + let mut killpill_rx = killpill_rx.resubscribe(); + let conn = conn.clone(); + let worker_name = worker_name.clone(); + tokio::spawn(async move { + if matches!( + killpill_rx.recv().await, + Err(broadcast::error::RecvError::Closed) + ) { + return; + } + for attempt in 1..=3 { + match mark_worker_draining(&conn, &worker_name).await { + Ok(()) => return, + Err(e) => { + tracing::warn!(worker = %worker_name, "failed to mark worker as draining (attempt {attempt}/3): {e:#}"); + tokio::time::sleep(Duration::from_secs(2)).await; + } + } + } + }); + } + loop { let last_processing_duration_secs = last_processing_duration.load(Ordering::SeqCst); if last_processing_duration_secs > 5 { @@ -3235,7 +3260,7 @@ pub async fn run_worker( // which costs at most the last job's id here: no job of this worker is in flight at this // point in the loop, and the next one refills them. if ip_just_resolved && conn.as_sql().is_none() { - if let Err(e) = insert_ping(hostname, &worker_name, ip, &conn).await { + if let Err(e) = insert_ping(hostname, &worker_name, ip, false, &conn).await { tracing::warn!( worker = %worker_name, hostname = %hostname, "failed to re-register with the resolved external IP: {e}" diff --git a/backend/windmill-worker/src/worker_utils.rs b/backend/windmill-worker/src/worker_utils.rs index 3a313e9360..cabf004f4d 100644 --- a/backend/windmill-worker/src/worker_utils.rs +++ b/backend/windmill-worker/src/worker_utils.rs @@ -7,9 +7,9 @@ use windmill_common::{ external_ip::UNKNOWN_IP, worker::{ get_memory, get_vcpus, get_windmill_memory_usage, get_worker_memory_usage, - insert_ping_query, update_job_ping_query, update_worker_ping_from_job_query, - update_worker_ping_main_loop_query, Connection, Ping, PingType, NATIVE_MODE_RESOLVED, - WORKER_CONFIG, WORKER_GROUP, + insert_ping_query, set_worker_draining_query, update_job_ping_query, + update_worker_ping_from_job_query, update_worker_ping_main_loop_query, Connection, Ping, + PingType, NATIVE_MODE_RESOLVED, WORKER_CONFIG, WORKER_GROUP, }, KillpillSender, DB, }; @@ -164,6 +164,7 @@ async fn update_worker_ping_full_inner( wm_memory_usage: get_windmill_memory_usage(), job_isolation: None, native_mode: Some(native_mode), + draining: None, ping_type: PingType::MainLoop, }, ) @@ -179,6 +180,7 @@ pub async fn insert_ping( worker_instance: &str, worker_name: &str, ip: Option<&str>, + reset_draining: bool, db: &Connection, ) -> anyhow::Result { let (tags, dw, dws, native_mode) = { @@ -224,6 +226,7 @@ pub async fn insert_ping( memory, job_isolation, native_mode, + reset_draining.then_some(false), db, ) .await; @@ -256,6 +259,7 @@ pub async fn insert_ping( wm_memory_usage: get_windmill_memory_usage(), job_isolation, native_mode: Some(native_mode), + draining: reset_draining.then_some(false), ping_type: PingType::Initial, }, ) @@ -314,6 +318,7 @@ pub async fn update_worker_ping_from_job( &Ping { last_job_executed: Some(job_id.clone()), last_job_workspace_id: Some(w_id.to_string()), + draining: None, ping_type: PingType::Job, worker_instance: None, ip: None, @@ -342,6 +347,45 @@ pub async fn update_worker_ping_from_job( Ok(()) } +/// Publishes on the worker's ping row that it has received its killpill, so the job it is +/// still running can find out and exit early. +pub async fn mark_worker_draining(conn: &Connection, worker_name: &str) -> anyhow::Result<()> { + match conn { + Connection::Sql(db) => set_worker_draining_query(worker_name, db).await, + Connection::Http(client) => { + client + .post::( + UPDATE_PING_URL, + None, + &Ping { + last_job_executed: None, + last_job_workspace_id: None, + draining: None, + ping_type: PingType::Draining, + worker_instance: None, + ip: None, + tags: None, + dw: None, + dws: None, + version: None, + vcpus: None, + memory: None, + memory_usage: None, + wm_memory_usage: None, + jobs_executed: None, + occupancy_rate: None, + occupancy_rate_15s: None, + occupancy_rate_5m: None, + occupancy_rate_30m: None, + job_isolation: None, + native_mode: None, + }, + ) + .await + } + } +} + pub async fn ping_job_status( conn: &Connection, job_id: &Uuid, diff --git a/cli/src/guidance/skills.gen.ts b/cli/src/guidance/skills.gen.ts index dc9541b4c5..e5b3619a27 100644 --- a/cli/src/guidance/skills.gen.ts +++ b/cli/src/guidance/skills.gen.ts @@ -839,6 +839,14 @@ async setProgress(percent: number, jobId?: any): Promise */ async getProgress(jobId?: any): Promise +/** + * Whether the worker running this job has received its shutdown signal. + * A draining worker waits for its current job to finish before exiting, so a + * long-running script can poll this and return early to let the worker be replaced. + * @param jobId? Job to check, defaults to the current job + */ +async workerIsDraining(jobId?: string): Promise + /** * Set a flow user state * @param key key of the state @@ -1674,6 +1682,14 @@ async setProgress(percent: number, jobId?: any): Promise */ async getProgress(jobId?: any): Promise +/** + * Whether the worker running this job has received its shutdown signal. + * A draining worker waits for its current job to finish before exiting, so a + * long-running script can poll this and return early to let the worker be replaced. + * @param jobId? Job to check, defaults to the current job + */ +async workerIsDraining(jobId?: string): Promise + /** * Set a flow user state * @param key key of the state @@ -2615,6 +2631,14 @@ async setProgress(percent: number, jobId?: any): Promise */ async getProgress(jobId?: any): Promise +/** + * Whether the worker running this job has received its shutdown signal. + * A draining worker waits for its current job to finish before exiting, so a + * long-running script can poll this and return early to let the worker be replaced. + * @param jobId? Job to check, defaults to the current job + */ +async workerIsDraining(jobId?: string): Promise + /** * Set a flow user state * @param key key of the state @@ -4620,6 +4644,18 @@ def set_progress(value: int, job_id: Optional[str] = None) # Progress value (0-100) or None if not set def get_progress(job_id: Optional[str] = None) -> Any +# Whether the worker running the job has received its shutdown signal. +# +# A draining worker waits for its current job to finish before exiting. A long-running +# script can poll this and return early so the worker can be replaced. +# +# Args: +# job_id: Job ID (defaults to current WM_JOB_ID) +# +# Returns: +# True if the worker running the job is draining +def worker_is_draining(job_id: Optional[str] = None) -> bool + # Set the user state of a flow at a given key def set_flow_user_state(key: str, value: Any) -> None diff --git a/frontend/src/routes/(root)/(logged)/workers/+page.svelte b/frontend/src/routes/(root)/(logged)/workers/+page.svelte index f7e0f07a54..0fb9243276 100644 --- a/frontend/src/routes/(root)/(logged)/workers/+page.svelte +++ b/frontend/src/routes/(root)/(logged)/workers/+page.svelte @@ -1022,7 +1022,7 @@ {#if workers} - {#each workers as { worker, custom_tags, last_ping, started_at, jobs_executed, last_job_id, last_job_workspace_id, occupancy_rate_15s, occupancy_rate_5m, occupancy_rate_30m, occupancy_rate, wm_version, vcpus, memory, memory_usage, wm_memory_usage, native_mode }} + {#each workers as { worker, custom_tags, last_ping, started_at, jobs_executed, last_job_id, last_job_workspace_id, occupancy_rate_15s, occupancy_rate_5m, occupancy_rate_30m, occupancy_rate, wm_version, vcpus, memory, memory_usage, wm_memory_usage, native_mode, draining }} {@const isWorkerAlive = isWorkerMaybeAlive(last_ping)} {@const tagMismatchInfo = getTagMismatchInfo( custom_tags, @@ -1087,6 +1087,15 @@ {#if native_mode} Native {/if} + {#if draining} + + Draining + {#snippet text()} + This worker received a shutdown signal. It no longer pulls jobs + and exits once its current job is done. + {/snippet} + + {/if} {#if !config || !config.worker_tags || config.worker_tags.length === 0} diff --git a/python-client/wmill/wmill/client.py b/python-client/wmill/wmill/client.py index ffcb6a3fe1..97da60aadc 100644 --- a/python-client/wmill/wmill/client.py +++ b/python-client/wmill/wmill/client.py @@ -823,6 +823,22 @@ class Windmill: else: return r.json() + def worker_is_draining(self, job_id: Optional[str] = None) -> bool: + """Whether the worker running the job has received its shutdown signal. + + A draining worker waits for its current job to finish before exiting. A long-running + script can poll this and return early so the worker can be replaced. + + Args: + job_id: Job ID (defaults to current WM_JOB_ID) + + Returns: + True if the worker running the job is draining + """ + job_id = job_id or os.environ.get("WM_JOB_ID") + + return self.get(f"/w/{self.workspace}/jobs/worker_is_draining/{job_id}").json() + def set_flow_user_state(self, key: str, value: Any) -> None: """Set the user state of a flow at a given key""" flow_id = self.get_root_job_id() @@ -2027,6 +2043,17 @@ def get_progress(job_id: Optional[str] = None) -> Any: return _client.get_progress(job_id) +@init_global_client +def worker_is_draining(job_id: Optional[str] = None) -> bool: + """ + Whether the worker running this job has received its shutdown signal. + + A draining worker waits for its current job to finish before exiting. A long-running + script can poll this and return early so the worker can be replaced. + """ + return _client.worker_is_draining(job_id) + + def set_shared_state_pickle(value: Any, path="state.pickle") -> None: """ Set the state in the shared folder using pickle diff --git a/system_prompts/auto-generated/prompts.ts b/system_prompts/auto-generated/prompts.ts index 00e5655154..2a5e48d5fe 100644 --- a/system_prompts/auto-generated/prompts.ts +++ b/system_prompts/auto-generated/prompts.ts @@ -1724,6 +1724,14 @@ async setProgress(percent: number, jobId?: any): Promise */ async getProgress(jobId?: any): Promise +/** + * Whether the worker running this job has received its shutdown signal. + * A draining worker waits for its current job to finish before exiting, so a + * long-running script can poll this and return early to let the worker be replaced. + * @param jobId? Job to check, defaults to the current job + */ +async workerIsDraining(jobId?: string): Promise + /** * Set a flow user state * @param key key of the state @@ -2396,6 +2404,18 @@ def set_progress(value: int, job_id: Optional[str] = None) # Progress value (0-100) or None if not set def get_progress(job_id: Optional[str] = None) -> Any +# Whether the worker running the job has received its shutdown signal. +# +# A draining worker waits for its current job to finish before exiting. A long-running +# script can poll this and return early so the worker can be replaced. +# +# Args: +# job_id: Job ID (defaults to current WM_JOB_ID) +# +# Returns: +# True if the worker running the job is draining +def worker_is_draining(job_id: Optional[str] = None) -> bool + # Set the user state of a flow at a given key def set_flow_user_state(key: str, value: Any) -> None diff --git a/system_prompts/auto-generated/script.md b/system_prompts/auto-generated/script.md index cb5ae40490..14ca869180 100644 --- a/system_prompts/auto-generated/script.md +++ b/system_prompts/auto-generated/script.md @@ -1639,6 +1639,14 @@ async setProgress(percent: number, jobId?: any): Promise */ async getProgress(jobId?: any): Promise +/** + * Whether the worker running this job has received its shutdown signal. + * A draining worker waits for its current job to finish before exiting, so a + * long-running script can poll this and return early to let the worker be replaced. + * @param jobId? Job to check, defaults to the current job + */ +async workerIsDraining(jobId?: string): Promise + /** * Set a flow user state * @param key key of the state @@ -2311,6 +2319,18 @@ def set_progress(value: int, job_id: Optional[str] = None) # Progress value (0-100) or None if not set def get_progress(job_id: Optional[str] = None) -> Any +# Whether the worker running the job has received its shutdown signal. +# +# A draining worker waits for its current job to finish before exiting. A long-running +# script can poll this and return early so the worker can be replaced. +# +# Args: +# job_id: Job ID (defaults to current WM_JOB_ID) +# +# Returns: +# True if the worker running the job is draining +def worker_is_draining(job_id: Optional[str] = None) -> bool + # Set the user state of a flow at a given key def set_flow_user_state(key: str, value: Any) -> None diff --git a/system_prompts/auto-generated/sdks/python.md b/system_prompts/auto-generated/sdks/python.md index 4241d49efa..13a93ed063 100644 --- a/system_prompts/auto-generated/sdks/python.md +++ b/system_prompts/auto-generated/sdks/python.md @@ -240,6 +240,18 @@ def set_progress(value: int, job_id: Optional[str] = None) # Progress value (0-100) or None if not set def get_progress(job_id: Optional[str] = None) -> Any +# Whether the worker running the job has received its shutdown signal. +# +# A draining worker waits for its current job to finish before exiting. A long-running +# script can poll this and return early so the worker can be replaced. +# +# Args: +# job_id: Job ID (defaults to current WM_JOB_ID) +# +# Returns: +# True if the worker running the job is draining +def worker_is_draining(job_id: Optional[str] = None) -> bool + # Set the user state of a flow at a given key def set_flow_user_state(key: str, value: Any) -> None diff --git a/system_prompts/auto-generated/sdks/typescript.md b/system_prompts/auto-generated/sdks/typescript.md index 47b0be888d..55df35ccf7 100644 --- a/system_prompts/auto-generated/sdks/typescript.md +++ b/system_prompts/auto-generated/sdks/typescript.md @@ -196,6 +196,14 @@ async setProgress(percent: number, jobId?: any): Promise */ async getProgress(jobId?: any): Promise +/** + * Whether the worker running this job has received its shutdown signal. + * A draining worker waits for its current job to finish before exiting, so a + * long-running script can poll this and return early to let the worker be replaced. + * @param jobId? Job to check, defaults to the current job + */ +async workerIsDraining(jobId?: string): Promise + /** * Set a flow user state * @param key key of the state diff --git a/system_prompts/auto-generated/skills/write-script-bun/SKILL.md b/system_prompts/auto-generated/skills/write-script-bun/SKILL.md index 21b2f382c3..6ad44db068 100644 --- a/system_prompts/auto-generated/skills/write-script-bun/SKILL.md +++ b/system_prompts/auto-generated/skills/write-script-bun/SKILL.md @@ -392,6 +392,14 @@ async setProgress(percent: number, jobId?: any): Promise */ async getProgress(jobId?: any): Promise +/** + * Whether the worker running this job has received its shutdown signal. + * A draining worker waits for its current job to finish before exiting, so a + * long-running script can poll this and return early to let the worker be replaced. + * @param jobId? Job to check, defaults to the current job + */ +async workerIsDraining(jobId?: string): Promise + /** * Set a flow user state * @param key key of the state diff --git a/system_prompts/auto-generated/skills/write-script-bunnative/SKILL.md b/system_prompts/auto-generated/skills/write-script-bunnative/SKILL.md index f9e902fe78..a5f599a946 100644 --- a/system_prompts/auto-generated/skills/write-script-bunnative/SKILL.md +++ b/system_prompts/auto-generated/skills/write-script-bunnative/SKILL.md @@ -404,6 +404,14 @@ async setProgress(percent: number, jobId?: any): Promise */ async getProgress(jobId?: any): Promise +/** + * Whether the worker running this job has received its shutdown signal. + * A draining worker waits for its current job to finish before exiting, so a + * long-running script can poll this and return early to let the worker be replaced. + * @param jobId? Job to check, defaults to the current job + */ +async workerIsDraining(jobId?: string): Promise + /** * Set a flow user state * @param key key of the state diff --git a/system_prompts/auto-generated/skills/write-script-deno/SKILL.md b/system_prompts/auto-generated/skills/write-script-deno/SKILL.md index 71608cd16e..ce0c610da7 100644 --- a/system_prompts/auto-generated/skills/write-script-deno/SKILL.md +++ b/system_prompts/auto-generated/skills/write-script-deno/SKILL.md @@ -394,6 +394,14 @@ async setProgress(percent: number, jobId?: any): Promise */ async getProgress(jobId?: any): Promise +/** + * Whether the worker running this job has received its shutdown signal. + * A draining worker waits for its current job to finish before exiting, so a + * long-running script can poll this and return early to let the worker be replaced. + * @param jobId? Job to check, defaults to the current job + */ +async workerIsDraining(jobId?: string): Promise + /** * Set a flow user state * @param key key of the state diff --git a/system_prompts/auto-generated/skills/write-script-python3/SKILL.md b/system_prompts/auto-generated/skills/write-script-python3/SKILL.md index decb356ba6..4946ed0f6e 100644 --- a/system_prompts/auto-generated/skills/write-script-python3/SKILL.md +++ b/system_prompts/auto-generated/skills/write-script-python3/SKILL.md @@ -448,6 +448,18 @@ def set_progress(value: int, job_id: Optional[str] = None) # Progress value (0-100) or None if not set def get_progress(job_id: Optional[str] = None) -> Any +# Whether the worker running the job has received its shutdown signal. +# +# A draining worker waits for its current job to finish before exiting. A long-running +# script can poll this and return early so the worker can be replaced. +# +# Args: +# job_id: Job ID (defaults to current WM_JOB_ID) +# +# Returns: +# True if the worker running the job is draining +def worker_is_draining(job_id: Optional[str] = None) -> bool + # Set the user state of a flow at a given key def set_flow_user_state(key: str, value: Any) -> None diff --git a/typescript-client/build.jsr.sh b/typescript-client/build.jsr.sh index 92e11640f3..41a8eac37e 100755 --- a/typescript-client/build.jsr.sh +++ b/typescript-client/build.jsr.sh @@ -22,6 +22,6 @@ cp "${script_dirpath}/sqlUtils.ts" "${script_dirpath}/src/" echo "" >> "${script_dirpath}/src/index.ts" echo 'export type { DenoS3LightClientSettings } from "./s3Types";' >> "${script_dirpath}/src/index.ts" echo "" >> "${script_dirpath}/src/index.ts" -echo 'export { type Base64, setClient, getWorkspace, workerHasInternalServer, SHARED_FOLDER, getStatePath, getInternalState, setInternalState, getResumeEndpoints, getResult, getResultMaybe, resolveDefaultResource, databaseUrlFromResource, base64ToUint8Array, uint8ArrayToBase64, getVariable, setVariable, getResource, setResource, getResumeUrls, setState, setProgress, getProgress, getState, getIdToken, denoS3LightClientSettings, cancelJob, loadS3FileStream, loadS3File, writeS3File, deleteS3File, signS3Objects, signS3Object, getPresignedS3PublicUrls, getPresignedS3PublicUrl, task, runScript, runScriptAsync, runScriptByPath, runScriptByHash, runScriptByPathAsync, runScriptByHashAsync, type RunScriptAsyncOptions, runFlow, runFlowAsync, waitJob, getRootJobId, setFlowUserState, getFlowUserState, usernameToEmail, requestInteractiveSlackApproval, type Sql, requestInteractiveTeamsApproval, appendToResultStream, streamResult, datatable, ducklake, upsertPartition, appendPartition, type DucklakeMaterializeOptions, type SqlStatement, type DatatableSqlTemplateFunction, type SqlTemplateFunction, type S3Object, type S3ObjectRecord, type S3ObjectURI } from "./client";' >> "${script_dirpath}/src/index.ts" +echo 'export { type Base64, setClient, getWorkspace, workerHasInternalServer, SHARED_FOLDER, getStatePath, getInternalState, setInternalState, getResumeEndpoints, getResult, getResultMaybe, resolveDefaultResource, databaseUrlFromResource, base64ToUint8Array, uint8ArrayToBase64, getVariable, setVariable, getResource, setResource, getResumeUrls, setState, setProgress, getProgress, workerIsDraining, getState, getIdToken, denoS3LightClientSettings, cancelJob, loadS3FileStream, loadS3File, writeS3File, deleteS3File, signS3Objects, signS3Object, getPresignedS3PublicUrls, getPresignedS3PublicUrl, task, runScript, runScriptAsync, runScriptByPath, runScriptByHash, runScriptByPathAsync, runScriptByHashAsync, type RunScriptAsyncOptions, runFlow, runFlowAsync, waitJob, getRootJobId, setFlowUserState, getFlowUserState, usernameToEmail, requestInteractiveSlackApproval, type Sql, requestInteractiveTeamsApproval, appendToResultStream, streamResult, datatable, ducklake, upsertPartition, appendPartition, type DucklakeMaterializeOptions, type SqlStatement, type DatatableSqlTemplateFunction, type SqlTemplateFunction, type S3Object, type S3ObjectRecord, type S3ObjectURI } from "./client";' >> "${script_dirpath}/src/index.ts" diff --git a/typescript-client/build.sh b/typescript-client/build.sh index 127afe2154..f927acd26e 100755 --- a/typescript-client/build.sh +++ b/typescript-client/build.sh @@ -46,7 +46,7 @@ cp "${script_dirpath}/sqlUtils.ts" "${script_dirpath}/src/" echo "" >> "${script_dirpath}/src/index.ts" echo 'export type { DenoS3LightClientSettings } from "./s3Types";' >> "${script_dirpath}/src/index.ts" echo "" >> "${script_dirpath}/src/index.ts" -echo 'export { type Base64, setClient, getWorkspace, workerHasInternalServer, SHARED_FOLDER, getStatePath, getInternalState, setInternalState, getResumeEndpoints, getResult, getResultMaybe, resolveDefaultResource, databaseUrlFromResource, base64ToUint8Array, uint8ArrayToBase64, getVariable, setVariable, getResource, setResource, getResumeUrls, setState, setProgress, getProgress, getState, getIdToken, denoS3LightClientSettings, cancelJob, loadS3FileStream, loadS3File, writeS3File, deleteS3File, signS3Objects, signS3Object, getPresignedS3PublicUrls, getPresignedS3PublicUrl, task, taskScript, taskFlow, workflow, step, sleep, parallel, waitForApproval, getApprovalUrls, type TaskOptions, type TaskRetry, type Jsonified, type JsonifiedFn, WorkflowCtx, _workflowCtx, setWorkflowCtx, StepSuspend, runScript, runScriptAsync, runScriptByPath, runScriptByHash, runScriptByPathAsync, runScriptByHashAsync, type RunScriptAsyncOptions, runFlow, runFlowAsync, waitJob, getRootJobId, setFlowUserState, getFlowUserState, usernameToEmail, requestInteractiveSlackApproval, type Sql, requestInteractiveTeamsApproval, appendToResultStream, streamResult, datatable, ducklake, upsertPartition, appendPartition, type DucklakeMaterializeOptions, type SqlStatement, type DatatableSqlTemplateFunction, type SqlTemplateFunction, type S3Object, type S3ObjectRecord, type S3ObjectURI, commitKafkaOffsets } from "./client";' >> "${script_dirpath}/src/index.ts" +echo 'export { type Base64, setClient, getWorkspace, workerHasInternalServer, SHARED_FOLDER, getStatePath, getInternalState, setInternalState, getResumeEndpoints, getResult, getResultMaybe, resolveDefaultResource, databaseUrlFromResource, base64ToUint8Array, uint8ArrayToBase64, getVariable, setVariable, getResource, setResource, getResumeUrls, setState, setProgress, getProgress, workerIsDraining, getState, getIdToken, denoS3LightClientSettings, cancelJob, loadS3FileStream, loadS3File, writeS3File, deleteS3File, signS3Objects, signS3Object, getPresignedS3PublicUrls, getPresignedS3PublicUrl, task, taskScript, taskFlow, workflow, step, sleep, parallel, waitForApproval, getApprovalUrls, type TaskOptions, type TaskRetry, type Jsonified, type JsonifiedFn, WorkflowCtx, _workflowCtx, setWorkflowCtx, StepSuspend, runScript, runScriptAsync, runScriptByPath, runScriptByHash, runScriptByPathAsync, runScriptByHashAsync, type RunScriptAsyncOptions, runFlow, runFlowAsync, waitJob, getRootJobId, setFlowUserState, getFlowUserState, usernameToEmail, requestInteractiveSlackApproval, type Sql, requestInteractiveTeamsApproval, appendToResultStream, streamResult, datatable, ducklake, upsertPartition, appendPartition, type DucklakeMaterializeOptions, type SqlStatement, type DatatableSqlTemplateFunction, type SqlTemplateFunction, type S3Object, type S3ObjectRecord, type S3ObjectURI, commitKafkaOffsets } from "./client";' >> "${script_dirpath}/src/index.ts" # Build default export by combining client utilities + services # This preserves backward compatibility for `import wmill from "windmill-client"` @@ -63,6 +63,7 @@ import { setState, setProgress, getProgress, + workerIsDraining, getState, getIdToken, denoS3LightClientSettings, @@ -153,6 +154,7 @@ const wmill = { setState, setProgress, getProgress, + workerIsDraining, getState, getIdToken, denoS3LightClientSettings, diff --git a/typescript-client/client.ts b/typescript-client/client.ts index 4ec07d6fa6..3f7b03117a 100644 --- a/typescript-client/client.ts +++ b/typescript-client/client.ts @@ -724,6 +724,19 @@ export async function getProgress(jobId?: any): Promise { }); } +/** + * Whether the worker running this job has received its shutdown signal. + * A draining worker waits for its current job to finish before exiting, so a + * long-running script can poll this and return early to let the worker be replaced. + * @param jobId? Job to check, defaults to the current job + */ +export async function workerIsDraining(jobId?: string): Promise { + return await JobService.workerIsDraining({ + id: jobId ?? getEnv("WM_JOB_ID") ?? "NO_JOB_ID", + workspace: getWorkspace(), + }); +} + /** * Set a flow user state * @param key key of the state From 2758920f1182ddd12e6473a644a4a966a435522e Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Thu, 8 Oct 2026 13:02:26 +0200 Subject: [PATCH 02/14] fix: sync back volume files rewritten to the same size (#11620) * fix: upload volume files rewritten to the same size Co-Authored-By: Claude Opus 5.5 * fix: never match a volume file stamp with an unreadable change time Co-Authored-By: Claude Opus 5.5 * chore: bump ee ref Co-Authored-By: Claude Opus 5.5 * chore: update ee-repo-ref to 127575148c84bb6132d2946c0a95c6dea0f58bb2 This commit updates the EE repository reference after PR #860 was merged in windmill-ee-private. Previous ee-repo-ref: e06755ddd48c834e75537cf44da017fdc3a9585b New ee-repo-ref: 127575148c84bb6132d2946c0a95c6dea0f58bb2 Automated by sync-ee-ref workflow. --------- Co-authored-by: Claude Opus 5.5 Co-authored-by: windmill-internal-app[bot] --- backend/ee-repo-ref.txt | 2 +- backend/windmill-worker-volumes/src/lib.rs | 34 ++++++++++++++++++++++ 2 files changed, 35 insertions(+), 1 deletion(-) diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index ff44a56995..14bd926cf5 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -55feb2e84332179faa4e17aeb419f7f480df79a6 +127575148c84bb6132d2946c0a95c6dea0f58bb2 diff --git a/backend/windmill-worker-volumes/src/lib.rs b/backend/windmill-worker-volumes/src/lib.rs index e2e4ab866b..364795510a 100644 --- a/backend/windmill-worker-volumes/src/lib.rs +++ b/backend/windmill-worker-volumes/src/lib.rs @@ -54,11 +54,45 @@ pub struct VolumeMount { pub target: String, } +/// A file's size and last change time. Sync-back trusts a pulled file's stored hash only while +/// its stamp is the one taken at hand-off: size alone misses a rewrite of the same length. A +/// write within the filesystem's timestamp granularity of the hand-off is not seen. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct FileStamp { + size: u64, + changed: Option<(i64, i64)>, +} + +impl FileStamp { + /// Whether the file is as it was at `earlier`. An unreadable change time never matches. + pub fn unchanged_since(&self, earlier: &FileStamp) -> bool { + self.changed.is_some() && self == earlier + } + + pub fn of(meta: &std::fs::Metadata) -> Self { + // ctime rather than mtime: a tool that restores mtime (cp -p, tar, rsync -t) cannot set it. + #[cfg(unix)] + let changed = { + use std::os::unix::fs::MetadataExt; + Some((meta.ctime(), meta.ctime_nsec())) + }; + #[cfg(not(unix))] + let changed = meta + .modified() + .ok() + .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok()) + .map(|d| (d.as_secs() as i64, d.subsec_nanos() as i64)); + Self { size: meta.len(), changed } + } +} + pub struct VolumeState { pub mount: VolumeMount, pub local_dir: PathBuf, pub manifest: HashMap, pub symlinks: HashMap, + /// Stamp of each pulled file as the job received it, keyed like `manifest`. + pub stamps: HashMap, } pub struct DownloadStats { From a85d0b76cc494aaa4a484a273ec534bdc2958102 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Thu, 8 Oct 2026 13:03:47 +0200 Subject: [PATCH 03/14] feat: add Pi template to the AI sandbox picker (#11619) Co-authored-by: Claude Opus 5.5 --- frontend/src/lib/components/EditorBar.svelte | 1 + .../src/lib/components/ScriptBuilder.svelte | 22 +++++- .../src/lib/components/ScriptEditor.svelte | 1 + .../common/languageIcons/LanguageIcon.svelte | 33 ++++++++- .../flows/content/FlowInputs.svelte | 12 +++ .../flows/content/FlowInputsQuick.svelte | 73 +++++++++--------- .../components/flows/flowStateUtils.svelte.ts | 2 +- .../flows/map/InsertModuleInner.svelte | 28 ++++--- .../flows/pickers/FlowScriptPicker.svelte | 2 +- .../pickers/FlowScriptPickerQuick.svelte | 2 +- .../flows/pickers/TopLevelNode.svelte | 1 + .../components/graph/graphBuilder.svelte.ts | 2 +- frontend/src/lib/components/script_builder.ts | 1 + frontend/src/lib/script_helpers.ts | 8 ++ .../src/lib/templates/pi_sandbox.ts.template | 74 +++++++++++++++++++ 15 files changed, 208 insertions(+), 54 deletions(-) create mode 100644 frontend/src/lib/templates/pi_sandbox.ts.template diff --git a/frontend/src/lib/components/EditorBar.svelte b/frontend/src/lib/components/EditorBar.svelte index 81ef6c1c96..2db471e2c8 100644 --- a/frontend/src/lib/components/EditorBar.svelte +++ b/frontend/src/lib/components/EditorBar.svelte @@ -105,6 +105,7 @@ | 'powershell' | 'bunnative' | 'claudesandbox' + | 'pisandbox' | 'wac_python' | 'wac_typescript' | 'ci_test_bun' diff --git a/frontend/src/lib/components/ScriptBuilder.svelte b/frontend/src/lib/components/ScriptBuilder.svelte index ad869ed492..f69cf37f1d 100644 --- a/frontend/src/lib/components/ScriptBuilder.svelte +++ b/frontend/src/lib/components/ScriptBuilder.svelte @@ -596,6 +596,7 @@ | 'powershell' | 'bunnative' | 'claudesandbox' + | 'pisandbox' | 'wac_python' | 'wac_typescript' | 'ci_test_bun' @@ -1543,10 +1544,12 @@ (lang == script.language && template != 'bunnative' && template != 'docker' && - template != 'claudesandbox') || + template != 'claudesandbox' && + template != 'pisandbox') || (template == 'bunnative' && lang == 'bunnative') || (template == 'docker' && lang == 'docker') || - (template == 'claudesandbox' && lang == 'bun')} + ((template == 'claudesandbox' || template == 'pisandbox') && + lang == 'bun')} @@ -1597,6 +1600,21 @@ > Claude Sandbox +