From afa9ced84cfec2b45a33657812836908cb32775a Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Wed, 7 Oct 2026 18:44:57 +0200 Subject: [PATCH 01/90] [ee] feat: job cancel signals (oidc refusal, agent 5s ping, job.canceled) (#11592) * feat: faster agent cancel detection, job.canceled event, worker_group span attribute Co-Authored-By: Claude Opus 5.5 (1M context) * fix: report no unix signal in job.canceled where none is sent Co-Authored-By: Claude Opus 5.5 (1M context) * refactor: agent workers ping every 5s instead of a 2s read-only cancel check Co-Authored-By: Claude Opus 5.5 (1M context) * fix: emit job.canceled on windows when the process exits unsignaled Co-Authored-By: Claude Opus 5.5 (1M context) * chore: pin ee ref with the queue delay gauge main requires Co-Authored-By: Claude Opus 5.5 (1M context) --------- Co-authored-by: Claude Opus 5.5 (1M context) --- ...42cba3a5532c64a9594d2e0ccdf21dee90d0.json} | 6 +- backend/ee-repo-ref.txt | 2 +- backend/tests/job_token_scopes.rs | 17 ++ backend/windmill-common/src/log_context.rs | 1 + backend/windmill-runtime-nativets/src/lib.rs | 4 +- backend/windmill-worker/src/handle_child.rs | 168 ++++++++++++------ backend/windmill-worker/src/worker.rs | 2 + 7 files changed, 143 insertions(+), 57 deletions(-) rename backend/.sqlx/{query-ee769c78ec0a782de18911cdcb655aad93cf23321cd0f42f7ffa5348bfbea26e.json => query-58ea7e2c6b07ffc7ab7387b2ebb542cba3a5532c64a9594d2e0ccdf21dee90d0.json} (54%) diff --git a/backend/.sqlx/query-ee769c78ec0a782de18911cdcb655aad93cf23321cd0f42f7ffa5348bfbea26e.json b/backend/.sqlx/query-58ea7e2c6b07ffc7ab7387b2ebb542cba3a5532c64a9594d2e0ccdf21dee90d0.json similarity index 54% rename from backend/.sqlx/query-ee769c78ec0a782de18911cdcb655aad93cf23321cd0f42f7ffa5348bfbea26e.json rename to backend/.sqlx/query-58ea7e2c6b07ffc7ab7387b2ebb542cba3a5532c64a9594d2e0ccdf21dee90d0.json index bb7e4fa36b..e03d995f15 100644 --- a/backend/.sqlx/query-ee769c78ec0a782de18911cdcb655aad93cf23321cd0f42f7ffa5348bfbea26e.json +++ b/backend/.sqlx/query-58ea7e2c6b07ffc7ab7387b2ebb542cba3a5532c64a9594d2e0ccdf21dee90d0.json @@ -1,11 +1,11 @@ { "db_name": "PostgreSQL", - "query": "SELECT EXISTS (SELECT 1 FROM v2_job_queue WHERE id = $1 AND workspace_id = $2 AND running)", + "query": "SELECT canceled_by IS NOT NULL FROM v2_job_queue WHERE id = $1 AND workspace_id = $2 AND running", "describe": { "columns": [ { "ordinal": 0, - "name": "exists", + "name": "?column?", "type_info": "Bool" } ], @@ -19,5 +19,5 @@ null ] }, - "hash": "ee769c78ec0a782de18911cdcb655aad93cf23321cd0f42f7ffa5348bfbea26e" + "hash": "58ea7e2c6b07ffc7ab7387b2ebb542cba3a5532c64a9594d2e0ccdf21dee90d0" } diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 5f22bddd84..dbcf14cb3c 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -acb254ac509aff244849c09e50a8d39aae185138 +0b78e2c6fefcbad3c465bf75ebfd6ebd6912ad19 \ No newline at end of file diff --git a/backend/tests/job_token_scopes.rs b/backend/tests/job_token_scopes.rs index 6af694bff0..f805168bfa 100644 --- a/backend/tests/job_token_scopes.rs +++ b/backend/tests/job_token_scopes.rs @@ -410,5 +410,22 @@ async fn test_oidc_claim_carries_the_presented_scopes(db: Pool) -> any )?; assert_eq!(claims["job_token_scopes"], expected, "{job}"); } + + // A soft-canceled job is still `running` until its worker kills it. + sqlx::query("UPDATE v2_job_queue SET canceled_by = 'test-user' WHERE id = $1") + .bind(Uuid::parse_str(OIDC_JOB)?) + .execute(&db) + .await?; + let resp = client + .post(format!("{base}/oidc/token/test-audience")) + .bearer_auth(job_token(&db, OIDC_JOB).await?) + .send() + .await?; + assert_eq!( + resp.status(), + StatusCode::FORBIDDEN, + "{}", + resp.text().await? + ); Ok(()) } diff --git a/backend/windmill-common/src/log_context.rs b/backend/windmill-common/src/log_context.rs index cdb1e27aea..c5cd191da8 100644 --- a/backend/windmill-common/src/log_context.rs +++ b/backend/windmill-common/src/log_context.rs @@ -49,6 +49,7 @@ pub struct LogContext { // Worker / job span (windmill-worker/src/worker.rs) pub worker: Option, + pub worker_group: Option, pub hostname: Option, pub tag: Option, pub job_id: Option, diff --git a/backend/windmill-runtime-nativets/src/lib.rs b/backend/windmill-runtime-nativets/src/lib.rs index 1c2a76cda5..75aacbcc64 100644 --- a/backend/windmill-runtime-nativets/src/lib.rs +++ b/backend/windmill-runtime-nativets/src/lib.rs @@ -929,8 +929,8 @@ pub async fn eval_fetch_timeout( for line in logged.as_deref().unwrap_or(&log).lines() { tracing::info!( target: "windmill:job_log", - job_id = ?job_id, - workspace_id = ?w_id_for_tracing, + job_id = %job_id, + workspace_id = %w_id_for_tracing, "{line}" ); if *OTEL_JOB_LOGS { diff --git a/backend/windmill-worker/src/handle_child.rs b/backend/windmill-worker/src/handle_child.rs index 22a2bf4e8b..fa06edf6bf 100644 --- a/backend/windmill-worker/src/handle_child.rs +++ b/backend/windmill-worker/src/handle_child.rs @@ -210,6 +210,53 @@ pub async fn handle_child( } } + impl KillReason { + /// Reports a kill once the process is gone: a span event on the enclosing span and, + /// through the logs bridge, a log record carrying the job's log context. + fn emit_canceled_event( + &self, + job_id: &Uuid, + timeout_reason: &str, + signal: &'static str, + detected_at: chrono::DateTime, + ) { + // `force` is only known for a soft cancel: a job completed under the worker + // may have been force canceled or completed by anything else server-side. + let (kill_reason, canceled_by, canceled_reason, force) = match self { + KillReason::TooManyLogs => return, + KillReason::Timeout { .. } => { + ("timeout", Some("timeout"), Some(timeout_reason), None) + } + KillReason::Cancelled(by) => ( + "cancelled", + by.as_ref().and_then(|x| x.username.as_deref()), + by.as_ref().and_then(|x| x.reason.as_deref()), + Some(false), + ), + KillReason::AlreadyCompleted => ("already_completed", None, None, None), + }; + let exited_at = chrono::Utc::now(); + let root_job = windmill_common::log_context::current_log_context() + .and_then(|c| c.root_job.clone()); + let rfc3339 = |t: chrono::DateTime| { + t.to_rfc3339_opts(chrono::SecondsFormat::Millis, true) + }; + tracing::info!( + job_id = %job_id, + root_job, + kill_reason, + canceled_by, + canceled_reason, + force, + kill_signal = signal, + cancel_detected_at = %rfc3339(detected_at), + process_exited_at = %rfc3339(exited_at), + kill_duration_ms = (exited_at - detected_at).num_milliseconds(), + "job.canceled" + ); + } + } + let (timeout_duration, timeout_warn_msg, timeout_source) = resolve_job_timeout(&conn, w_id, job_id, custom_timeout).await; if let Some(msg) = timeout_warn_msg { @@ -232,6 +279,7 @@ pub async fn handle_child( UpdateJobPollingExit::AlreadyCompleted => KillReason::AlreadyCompleted, }, }; + let detected_at = chrono::Utc::now(); tx.send(()).expect("rx should never be dropped"); drop(tx); @@ -268,59 +316,78 @@ pub async fn handle_child( } }; - #[allow(unused_variables)] - if let Some(id) = child.id() { - if *MAX_WAIT_FOR_SIGINT > 0 { - #[cfg(any(target_os = "linux", target_os = "macos"))] - signal::kill(Pid::from_raw(id as i32), Signal::SIGINT).unwrap(); + // Resolves once the process is reaped, to the last signal it was sent, or to `None` + // when the kill could not be confirmed. SIGINT and SIGTERM are only sent where the + // sends below are compiled in: elsewhere an exit during their wait is unsignaled. + let unix_signal = |s: &'static str| { + if cfg!(any(target_os = "linux", target_os = "macos")) { s } else { "none" } + }; + let kill = async { + #[allow(unused_variables)] + if let Some(id) = child.id() { + if *MAX_WAIT_FOR_SIGINT > 0 { + #[cfg(any(target_os = "linux", target_os = "macos"))] + signal::kill(Pid::from_raw(id as i32), Signal::SIGINT).unwrap(); - for _ in 0..*MAX_WAIT_FOR_SIGINT { + for _ in 0..*MAX_WAIT_FOR_SIGINT { + if child.try_wait().is_ok_and(|x| x.is_some()) { + break; + } + sleep(Duration::from_secs(1)).await; + } if child.try_wait().is_ok_and(|x| x.is_some()) { - break; + set_reason.await; + return Ok(Some(unix_signal("SIGINT"))); } - sleep(Duration::from_secs(1)).await; } - if child.try_wait().is_ok_and(|x| x.is_some()) { - set_reason.await; - return Ok(Err(kill_reason)); + if sigterm { + #[cfg(any(target_os = "linux", target_os = "macos"))] + signal::kill(Pid::from_raw(id as i32), Signal::SIGTERM).unwrap(); + + for _ in 0..*MAX_WAIT_FOR_SIGTERM { + if child.try_wait().is_ok_and(|x| x.is_some()) { + break; + } + sleep(Duration::from_secs(1)).await; + } + if child.try_wait().is_ok_and(|x| x.is_some()) { + set_reason.await; + return Ok(Some(unix_signal("SIGTERM"))); + } } } - if sigterm { - #[cfg(any(target_os = "linux", target_os = "macos"))] - signal::kill(Pid::from_raw(id as i32), Signal::SIGTERM).unwrap(); - - for _ in 0..*MAX_WAIT_FOR_SIGTERM { - if child.try_wait().is_ok_and(|x| x.is_some()) { - break; + #[cfg(windows)] + { + let pid_to_kill = child.id(); + let killed = match kill_process_tree(pid_to_kill).await { + Ok(_) => { + tracing::debug!( + "successfully killed process tree with PID: {:?}", + pid_to_kill + ); + true } - sleep(Duration::from_secs(1)).await; - } - if child.try_wait().is_ok_and(|x| x.is_some()) { - set_reason.await; - return Ok(Err(kill_reason)); - } + Err(e) => { + tracing::error!("failed to kill process tree: {:?}", e); + false + } + }; + set_reason.await; + return Ok(killed.then_some("taskkill")); } - } - #[cfg(windows)] - { - let pid_to_kill = child.id(); - match kill_process_tree(pid_to_kill).await { - Ok(_) => tracing::debug!( - "successfully killed process tree with PID: {:?}", - pid_to_kill - ), - Err(e) => tracing::error!("failed to kill process tree: {:?}", e), - }; - set_reason.await; - return Ok(Err(kill_reason)); - } - #[cfg(unix)] - { - /* send SIGKILL and reap child process */ - let (_, kill) = future::join(set_reason, Box::into_pin(child.kill())).await; - kill.map(|()| Err(kill_reason)) + #[cfg(unix)] + { + /* send SIGKILL and reap child process */ + let (_, kill) = future::join(set_reason, Box::into_pin(child.kill())).await; + kill.map(|()| Some("SIGKILL")) + } + }; + let signal: io::Result> = kill.await; + if let Some(signal) = signal? { + kill_reason.emit_canceled_event(&job_id, &timeout_reason, signal, detected_at); } + Ok(Err(kill_reason)) }; let mut stream_result = Vec::new(); @@ -957,9 +1024,9 @@ where let conn = conn.clone(); // No tick at t=0: a job that finishes within the first interval issues none of the - // ping/metric statements below. Cancels are picked up by the first job ping (500 ms in, - // 2 s on agent workers). Memory is still sampled at start, without SQL: it is the only - // `mem_peak` reading a short job gets, and the completed job row reports it. + // ping/metric statements below. Cancels are picked up by the next job ping. Memory is still + // sampled at start, without SQL: it is the only `mem_peak` reading a short job gets, + // and the completed job row reports it. let mut interval = interval_at(Instant::now() + update_job_interval, update_job_interval); interval.set_missed_tick_behavior(MissedTickBehavior::Skip); *mem_peak = (*mem_peak).max(get_mem.next().await.unwrap_or(0)); @@ -1023,11 +1090,10 @@ where } } } - if matches!(conn, Connection::Http(_)) { - if i % 4 != 0 { - // only ping every 4th time (2s) on http agent mode - continue; - } + // An agent worker pays an HTTP round trip per ping, so it pings every 2 s for + // the first 10 s, then every 5 s. + if matches!(conn, Connection::Http(_)) && i % if i < 20 { 4 } else { 10 } != 0 { + continue; } let ping_job_status = ping_job_status(&conn, &job_id, Some(*mem_peak), if current_mem > 0 { Some(current_mem) } else { None }).await.unwrap_or_else(|e| { tracing::error!("Unable to ping job status for job {job_id}. Error was: {:?}", e); diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 80ca37acb8..96d8619988 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -2005,6 +2005,7 @@ pub fn create_span_with_name( root_job = field::Empty, workspace_id = %arc_job.workspace_id, worker = %worker_name, + worker_group = %*WORKER_GROUP, hostname = field::Empty, tag = %arc_job.tag, language = field::Empty, @@ -2156,6 +2157,7 @@ pub fn log_context_for_job( job_id: Some(arc_job.id.to_string()), workspace_id: Some(arc_job.workspace_id.clone()), worker: Some(worker_name.to_string()), + worker_group: Some(WORKER_GROUP.clone()), tag: Some(arc_job.tag.clone()), job_kind: Some(arc_job.kind.as_str().to_string()), created_by: Some(arc_job.created_by.clone()), From 14381cefee6fb2b0ece52fbf0001c2ce61bde1a1 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Wed, 7 Oct 2026 19:02:34 +0200 Subject: [PATCH 02/90] fix: scope the python wheel cache to workspaces with their own registry (#11597) * feat: pin python lockfile hashes and key the wheel cache by them Co-Authored-By: Claude Opus 5.5 (1M context) * fix: keep pinned python cache entries out of reach of unpinned installs Co-Authored-By: Claude Opus 5.5 (1M context) * fix: scope python cache entries of index-overriding workspaces by hashed id Co-Authored-By: Claude Opus 5.5 (1M context) * fix: keep the python cache scope consistent with the index urls read Co-Authored-By: Claude Opus 5.5 (1M context) * fix: read python index urls and their cache scope from one snapshot Co-Authored-By: Claude Opus 5.5 (1M context) * fix: drop hash pinning, keep only the workspace scope of the wheel cache Co-Authored-By: Claude Opus 5.5 (1M context) * fix: resolve pip index urls only after a resolution cache miss Co-Authored-By: Claude Opus 5.5 (1M context) --------- Co-authored-by: Claude Opus 5.5 (1M context) --- .../windmill-worker/src/python_executor.rs | 122 +++++++++++------- backend/windmill-worker/src/worker.rs | 16 +++ 2 files changed, 93 insertions(+), 45 deletions(-) diff --git a/backend/windmill-worker/src/python_executor.rs b/backend/windmill-worker/src/python_executor.rs index 3fce0fb938..f1814ac3fd 100644 --- a/backend/windmill-worker/src/python_executor.rs +++ b/backend/windmill-worker/src/python_executor.rs @@ -156,7 +156,7 @@ use crate::{ }, get_proxy_envs_for_lang, handle_child::handle_child, - is_sandboxing_enabled, read_ee_registry_with_workspace_override, + is_sandboxing_enabled, read_ee_registry, worker_utils::ping_job_status, PyV, DISABLE_NUSER, HOME_ENV, INDEX_CERT, NATIVE_CERT, NSJAIL_AVAILABLE, NSJAIL_PATH, NSJAIL_PY_RLIMIT_AS_MB, PATH_ENV, PIP_EXTRA_INDEX_URL, PIP_INDEX_URL, PROXY_ENVS, @@ -279,7 +279,12 @@ pub async fn uv_pip_compile( #[cfg(feature = "enterprise")] let requirements = replace_pip_secret(conn, w_id, &requirements, worker_name, job_id).await?; - let ws_suffix = crate::workspace_registry_cache_suffix(w_id).await; + let pip_indexes = PipIndexes::snapshot(w_id).await; + let ws_suffix = if pip_indexes.ws_scoped { + format!(":ws:{w_id}") + } else { + String::new() + }; let exclude_newer_suffix = uv_exclude_newer .as_deref() .map(|v| format!("-en{}", v)) @@ -343,31 +348,12 @@ pub async fn uv_pip_compile( if no_cache { args.extend(["--no-cache"]); } - let pip_extra_index_url = read_ee_registry_with_workspace_override( - PIP_EXTRA_INDEX_URL.read().await.clone(), - "pip_extra_index_url", - "pip extra index url", - job_id, - w_id, - conn, - ) - .await - .map(handle_ephemeral_token); + let (pip_extra_index_url, pip_index_url) = pip_indexes.urls(job_id, w_id, conn).await; if let Some(url) = pip_extra_index_url.as_ref() { url.split(",").for_each(|url| { args.extend(["--extra-index-url", url]); }); } - let pip_index_url = read_ee_registry_with_workspace_override( - PIP_INDEX_URL.read().await.clone(), - "pip_index_url", - "pip index url", - job_id, - w_id, - conn, - ) - .await - .map(handle_ephemeral_token); if let Some(url) = pip_index_url.as_ref() { args.extend(["--index-url", url]); } @@ -2068,6 +2054,62 @@ Returned from server: py_version - {:?}, py_version_v2 - {:?} Ok((pyv, additional_python_paths)) } +/// The pip index settings a job resolves and installs from, and whether its workspace +/// overrides any registry. +/// +/// Both come from one snapshot of the workspace's overrides: read separately, a settings +/// reload in between could file an install from an overridden index under a shared cache name. +struct PipIndexes { + extra_index_url: Option, + index_url: Option, + /// Caches filled through these URLs must not be shared with other workspaces. + ws_scoped: bool, +} + +impl PipIndexes { + async fn snapshot(w_id: &str) -> Self { + let overrides = crate::workspace_registry_overrides(w_id).await; + let ws_value = |key: &str| match overrides.get(key) { + Some(serde_json::Value::String(s)) => Some(s.clone()), + _ => None, + }; + // An empty value means unset, and an empty workspace override unsets the global one. + Self { + extra_index_url: ws_value("pip_extra_index_url") + .or(PIP_EXTRA_INDEX_URL.read().await.clone()) + .filter(|s| !s.trim().is_empty()), + index_url: ws_value("pip_index_url") + .or(PIP_INDEX_URL.read().await.clone()) + .filter(|s| !s.trim().is_empty()), + ws_scoped: !overrides.is_empty(), + } + } + + /// The `(extra index, index)` URLs to hand uv. Kept apart from [`Self::snapshot`]: it runs + /// `EPHEMERAL_TOKEN_CMD`, which a caller served from a cache has no reason to pay for. + async fn urls( + self, + job_id: &Uuid, + w_id: &str, + conn: &Connection, + ) -> (Option, Option) { + ( + read_ee_registry( + self.extra_index_url, + "pip extra index url", + job_id, + w_id, + conn, + ) + .await + .map(handle_ephemeral_token), + read_ee_registry(self.index_url, "pip index url", job_id, w_id, conn) + .await + .map(handle_ephemeral_token), + ) + } +} + lazy_static::lazy_static! { static ref PIP_SECRET_VARIABLE: Regex = Regex::new(r"\$\{PIP_SECRET:([^\s\}]+)\}").unwrap(); @@ -2527,28 +2569,18 @@ pub async fn handle_python_reqs( job_id ); - let pip_indexes = ( - read_ee_registry_with_workspace_override( - PIP_EXTRA_INDEX_URL.read().await.clone(), - "pip_extra_index_url", - "pip extra index url", - job_id, - w_id, - conn, - ) - .await - .map(handle_ephemeral_token), - read_ee_registry_with_workspace_override( - PIP_INDEX_URL.read().await.clone(), - "pip_index_url", - "pip index url", - job_id, - w_id, - conn, - ) - .await - .map(handle_ephemeral_token), - ); + let pip_indexes = PipIndexes::snapshot(w_id).await; + let ws_scoped = pip_indexes.ws_scoped; + let pip_indexes = pip_indexes.urls(job_id, w_id, conn).await; + // The wheel dir is shared by every workspace on the worker, and by every worker through + // the object store, under a name that says nothing of the index it was filled from. A + // workspace with its own registries gets its own, as its resolution cache does. The id + // is hashed: it is not restricted to characters safe in a file name. + let ws_dir_suffix = if ws_scoped { + format!("-ws-{}", &calculate_hash(w_id)[..16]) + } else { + String::new() + }; // Cached paths let mut req_with_penv: Vec<(String, String)> = vec![]; @@ -2577,7 +2609,7 @@ pub async fn handle_python_reqs( let py_prefix = &py_version.to_cache_dir(false); let venv_p = format!( - "{py_prefix}/{}", + "{py_prefix}/{}{ws_dir_suffix}", req.replace(' ', "").replace('/', "").replace(':', "") ); if metadata(venv_p.clone() + "/.valid.windmill").await.is_ok() { diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 96d8619988..28d226ea3f 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -949,6 +949,22 @@ pub async fn workspace_registry_cache_suffix(w_id: &str) -> String { } } +/// The registry overrides of a workspace, empty when it has none. +/// +/// One snapshot: registry settings reload under running jobs, so a caller whose cache key +/// has to agree with the registry values it used reads both from here rather than calling +/// [`workspace_registry_cache_suffix`] and a `read_ee_registry_*` helper separately. +pub async fn workspace_registry_overrides( + w_id: &str, +) -> std::collections::HashMap { + let registries = WORKSPACE_REGISTRIES.read().await; + registries + .as_ref() + .and_then(|m| m.get(w_id)) + .cloned() + .unwrap_or_default() +} + /// The name a build artifact is cached under, derived from `base` — the runnable's own /// cache-key input — and its inline modules. /// From 95c4b3096cf10c1fe59de7a1a6876dfb5f02c39f Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Wed, 7 Oct 2026 19:28:29 +0200 Subject: [PATCH 03/90] feat: pause a worker group without stopping its workers (#11596) * feat: pause a worker group without stopping its workers * chore: bump ee ref * fix: keep dedicated workers running across a pause toggle * fix: edit a copy of the worker group config in the drawer * fix: stable priority tag order, keep pause-only configs in the drawer * fix: keep inherited worker tags when saving from the drawer * fix: keep an explicit empty tag list when leaving dedicated mode * fix: keep inherited tags when creating a config from the drawer Co-Authored-By: Claude Opus 5.5 (1M context) * chore: update ee-repo-ref to 6f23819e947b18878000e580ddefccfd01a73166 This commit updates the EE repository reference after PR #853 was merged in windmill-ee-private. Previous ee-repo-ref: 6cf802bdbe17b4a66757016515044b8a9ef9ef19 New ee-repo-ref: 6f23819e947b18878000e580ddefccfd01a73166 Automated by sync-ee-ref workflow. * feat: confirm before applying a worker config change that restarts workers Co-Authored-By: Claude Opus 5.5 (1M context) * chore: pin the ee commit that compiles against main Co-Authored-By: Claude Opus 5.5 (1M context) * chore: retrigger ci after the github outage dropped the push event Co-Authored-By: Claude Opus 5.5 (1M context) --------- Co-authored-by: Claude Opus 5.5 (1M context) Co-authored-by: windmill-internal-app[bot] --- ...9573aefaeb5dbc6ac07f5e664bc8fe2cb2683.json | 22 +++ ...c8f780ab4c9d3d373d3cd0bb489e0d173040c.json | 22 +++ backend/ee-repo-ref.txt | 2 +- backend/src/monitor.rs | 9 +- backend/tests/worker_group_pause.rs | 71 ++++++++++ backend/windmill-api-configs/src/lib.rs | 28 +++- .../windmill-common/src/instance_config.rs | 3 + backend/windmill-common/src/worker.rs | 66 ++++++++- backend/windmill-worker/src/worker.rs | 4 + .../src/lib/components/WorkerGroup.svelte | 133 ++++++++++++++++-- 10 files changed, 340 insertions(+), 20 deletions(-) create mode 100644 backend/.sqlx/query-91fdd565bd210f8aa4c6d5f5e309573aefaeb5dbc6ac07f5e664bc8fe2cb2683.json create mode 100644 backend/.sqlx/query-a6d355b3121f199f3827c05aaa7c8f780ab4c9d3d373d3cd0bb489e0d173040c.json create mode 100644 backend/tests/worker_group_pause.rs diff --git a/backend/.sqlx/query-91fdd565bd210f8aa4c6d5f5e309573aefaeb5dbc6ac07f5e664bc8fe2cb2683.json b/backend/.sqlx/query-91fdd565bd210f8aa4c6d5f5e309573aefaeb5dbc6ac07f5e664bc8fe2cb2683.json new file mode 100644 index 0000000000..ba917c82db --- /dev/null +++ b/backend/.sqlx/query-91fdd565bd210f8aa4c6d5f5e309573aefaeb5dbc6ac07f5e664bc8fe2cb2683.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT COALESCE(config->'paused' = 'true'::jsonb, false) FROM config WHERE name = $1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "coalesce", + "type_info": "Bool" + } + ], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [ + null + ] + }, + "hash": "91fdd565bd210f8aa4c6d5f5e309573aefaeb5dbc6ac07f5e664bc8fe2cb2683" +} diff --git a/backend/.sqlx/query-a6d355b3121f199f3827c05aaa7c8f780ab4c9d3d373d3cd0bb489e0d173040c.json b/backend/.sqlx/query-a6d355b3121f199f3827c05aaa7c8f780ab4c9d3d373d3cd0bb489e0d173040c.json new file mode 100644 index 0000000000..e1657e156b --- /dev/null +++ b/backend/.sqlx/query-a6d355b3121f199f3827c05aaa7c8f780ab4c9d3d373d3cd0bb489e0d173040c.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT COALESCE(config->'paused' = 'true'::jsonb, false) FROM config WHERE name = $1 FOR UPDATE", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "coalesce", + "type_info": "Bool" + } + ], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [ + null + ] + }, + "hash": "a6d355b3121f199f3827c05aaa7c8f780ab4c9d3d373d3cd0bb489e0d173040c" +} diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index dbcf14cb3c..fffcc99883 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -0b78e2c6fefcbad3c465bf75ebfd6ebd6912ad19 \ No newline at end of file +1600d49ad5513af8b54ad65688d3d1c7edc977d9 \ No newline at end of file diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index 677ecd11a2..8d0af4915d 100644 --- a/backend/src/monitor.rs +++ b/backend/src/monitor.rs @@ -427,6 +427,8 @@ pub async fn initial_load( native_mode, // an agent worker never reads its group's config, only its token object_store_cache_config: None, + // nor whether it is paused: the server refuses it jobs instead + paused: false, })); } } @@ -5689,8 +5691,13 @@ pub async fn reload_worker_config(db: &DB, tx: KillpillSender, kill_if_change: b .as_ref() .is_some_and(|dws| !dws.is_empty()); + // Pausing or resuming must not restart the workers, dedicated ones included: a reload + // whose only difference is the flag skips every restart below. + let only_pause_changed = wc.paused != config.paused + && WorkerConfig { paused: wc.paused, ..config.clone() } == **wc; + if **wc != config || has_dedicated { - if kill_if_change { + if kill_if_change && !only_pause_changed { if has_dedicated || wc.dedicated_worker != config.dedicated_worker || wc.dedicated_workers != config.dedicated_workers diff --git a/backend/tests/worker_group_pause.rs b/backend/tests/worker_group_pause.rs new file mode 100644 index 0000000000..b528bbfd77 --- /dev/null +++ b/backend/tests/worker_group_pause.rs @@ -0,0 +1,71 @@ +//! In its own test binary: it flips the process-wide `WORKER_CONFIG`, which would stall every +//! other worker test sharing the process. + +use futures::StreamExt; +use sqlx::{Pool, Postgres}; +use std::time::Duration; +use windmill_common::{ + jobs::{JobPayload, RawCode}, + scripts::ScriptLang, + worker::{Connection, WORKER_CONFIG}, +}; +use windmill_test_utils::*; + +fn set_paused(paused: bool) { + let mut wc = (**WORKER_CONFIG.load()).clone(); + wc.paused = paused; + WORKER_CONFIG.store(std::sync::Arc::new(wc)); +} + +#[sqlx::test(fixtures("base"))] +async fn paused_worker_leaves_jobs_queued_until_resumed(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + + set_paused(true); + let id = RunJob::from(JobPayload::Code(RawCode { + hash: None, + content: "echo done".to_string(), + path: None, + lock: None, + language: ScriptLang::Bash, + cache_ttl: None, + cache_ignore_s3_path: None, + dedicated_worker: None, + concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default() + .into(), + debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(), + modules: None, + tag: None, + })) + .push(&db) + .await; + + let mut completed = listen_for_completed_jobs(&db).await; + in_test_worker( + Connection::Sql(db.clone()), + async { + tokio::time::sleep(Duration::from_secs(3)).await; + let running = + sqlx::query_scalar::<_, bool>("SELECT running FROM v2_job_queue WHERE id = $1") + .bind(id) + .fetch_optional(&db) + .await + .unwrap(); + assert_eq!( + running, + Some(false), + "a paused worker must not pull the job" + ); + + set_paused(false); + while completed.next().await != Some(id) {} + }, + port, + ) + .await; + + assert!(completed_job(id, &db).await.success); + Ok(()) +} diff --git a/backend/windmill-api-configs/src/lib.rs b/backend/windmill-api-configs/src/lib.rs index 7598021f00..f1f4792e71 100644 --- a/backend/windmill-api-configs/src/lib.rs +++ b/backend/windmill-api-configs/src/lib.rs @@ -224,12 +224,13 @@ async fn update_config( #[cfg(not(feature = "enterprise"))] let config = if name.starts_with("worker__") { - // In CE, only allow setting worker_tags, cache_clear, init_bash, and native_mode + // In CE, only allow setting worker_tags, cache_clear, init_bash, native_mode and paused serde_json::json!({ "worker_tags": config.get("worker_tags"), "cache_clear": config.get("cache_clear"), "init_bash": config.get("init_bash"), - "native_mode": config.get("native_mode") + "native_mode": config.get("native_mode"), + "paused": config.get("paused") }) } else { config @@ -272,6 +273,18 @@ async fn update_config( } let mut tx = db.begin().await?; + let was_paused = if name.starts_with("worker__") { + sqlx::query_scalar!( + "SELECT COALESCE(config->'paused' = 'true'::jsonb, false) FROM config WHERE name = $1 FOR UPDATE", + &name + ) + .fetch_optional(&mut *tx) + .await? + .flatten() + .unwrap_or(false) + } else { + false + }; sqlx::query!( "INSERT INTO config (name, config) VALUES ($1, $2) ON CONFLICT (name) DO UPDATE SET config = EXCLUDED.config", &name, @@ -291,6 +304,17 @@ async fn update_config( ) .await?; tx.commit().await?; + + if name.starts_with("worker__") { + let paused = config.get("paused").and_then(|v| v.as_bool()) == Some(true); + if paused != was_paused { + windmill_common::feature_usage::log_feature_usage( + "worker_group", + "pause_toggle", + if paused { "pause" } else { "resume" }, + ); + } + } Ok(format!("Updated config {name}")) } diff --git a/backend/windmill-common/src/instance_config.rs b/backend/windmill-common/src/instance_config.rs index a4164cd2c5..700ba71df5 100644 --- a/backend/windmill-common/src/instance_config.rs +++ b/backend/windmill-common/src/instance_config.rs @@ -963,6 +963,9 @@ pub struct WorkerGroupConfig { pub autoscaling: Option, #[serde(skip_serializing_if = "Option::is_none")] pub native_mode: Option, + /// Workers of the group pull no new job while this is set; queued jobs wait. + #[serde(skip_serializing_if = "Option::is_none")] + pub paused: Option, /// Object store this group's dependency cache uses instead of the instance one. Same shape /// as the instance `object_store_cache_config`. #[serde(skip_serializing_if = "Option::is_none")] diff --git a/backend/windmill-common/src/worker.rs b/backend/windmill-common/src/worker.rs index e68b81e4de..e185d1d445 100644 --- a/backend/windmill-common/src/worker.rs +++ b/backend/windmill-common/src/worker.rs @@ -477,6 +477,7 @@ lazy_static::lazy_static! { env_vars: Default::default(), native_mode: false, object_store_cache_config: Default::default(), + paused: false, }); pub static ref WORKER_PULL_QUERIES: arc_swap::ArcSwap> = arc_swap::ArcSwap::from_pointee(vec![]); @@ -844,11 +845,14 @@ pub fn make_suspended_pull_query(tags: &[String]) -> String { } // pub async fn make_suspended pub async fn store_suspended_pull_query(wc: &WorkerConfig) { - if wc.worker_tags.len() == 0 { - tracing::error!("Empty tags in worker tags, skipping"); - return; - } - let query = make_suspended_pull_query(&wc.worker_tags); + // An empty tag list must clear the query: keeping the previous one would leave a reloaded + // worker pulling suspended jobs for tags it no longer listens to. + let query = if wc.worker_tags.is_empty() { + tracing::error!("Empty tags in worker tags, pulling no suspended jobs"); + String::new() + } else { + make_suspended_pull_query(&wc.worker_tags) + }; WORKER_SUSPENDED_PULL_QUERY.store(std::sync::Arc::new(query)); } @@ -922,6 +926,42 @@ pub async fn store_pull_query(wc: &WorkerConfig) { WORKER_PULL_QUERIES_FAIRNESS.store(std::sync::Arc::new(fairness_queries)); } +const WORKER_GROUP_PAUSED_CACHE_TTL: Duration = Duration::from_secs(10); + +lazy_static::lazy_static! { + static ref WORKER_GROUP_PAUSED_CACHE: quick_cache::sync::Cache = + quick_cache::sync::Cache::new(1000); +} + +/// Whether the config of `worker_group` is paused, as the server sees it. A worker that reads +/// its own config knows from `WORKER_CONFIG`; this is for the server deciding on behalf of +/// workers that do not, and is cached since they ask on every pull. +pub async fn is_worker_group_paused(db: &DB, worker_group: &str) -> bool { + if let Some((paused, at)) = WORKER_GROUP_PAUSED_CACHE.get(worker_group) { + if at.elapsed() < WORKER_GROUP_PAUSED_CACHE_TTL { + return paused; + } + } + let paused = match sqlx::query_scalar!( + "SELECT COALESCE(config->'paused' = 'true'::jsonb, false) FROM config WHERE name = $1", + format!("worker__{worker_group}") + ) + .fetch_optional(db) + .await + { + Ok(paused) => paused.flatten().unwrap_or(false), + Err(e) => { + tracing::error!("Could not read whether worker group {worker_group} is paused: {e:#}"); + return false; + } + }; + WORKER_GROUP_PAUSED_CACHE.insert( + worker_group.to_string(), + (paused, std::time::Instant::now()), + ); + paused +} + lazy_static::lazy_static! { pub static ref WINDMILL_DIR: String = { let dir = std::env::var("WINDMILL_DIR") @@ -2470,6 +2510,12 @@ pub async fn load_worker_config( .collect(); priority_tags_sorted.push(PriorityTags { priority: 0, tags: Vec::from_iter(all_tags_set) }); // push the tags that were not listed as high priority with a priority = 0 priority_tags_sorted.sort_by_key(|elt| Reverse(elt.priority)); // sort by priority DESC + + // The buckets come out of hash maps: without a fixed order two loads of the same + // config compare unequal, and a reload reads that as a config change. + for bucket in priority_tags_sorted.iter_mut() { + bucket.tags.sort(); + } } else { // if no priority is used, push all tags with a priority to 0 priority_tags_sorted.push(PriorityTags { priority: 0, tags: worker_tags.clone() }); @@ -2539,6 +2585,7 @@ pub async fn load_worker_config( env_vars: resolved_env_vars, native_mode, object_store_cache_config: config.object_store_cache_config, + paused: config.paused.unwrap_or(false), }) } @@ -2629,6 +2676,7 @@ pub struct WorkerConfigOpt { pub env_vars_allowlist: Option>, pub native_mode: Option, pub object_store_cache_config: Option, + pub paused: Option, } impl Default for WorkerConfigOpt { @@ -2648,6 +2696,7 @@ impl Default for WorkerConfigOpt { env_vars_allowlist: Default::default(), native_mode: Default::default(), object_store_cache_config: Default::default(), + paused: Default::default(), } } } @@ -2670,12 +2719,15 @@ pub struct WorkerConfig { /// in the group config. Raw JSON: `windmill-common` cannot depend on the object store crate /// that parses it, and comparing the raw value is what tells a reload the store changed. pub object_store_cache_config: Option, + /// The group pulls no job from the queue: its workers finish what they hold, keep pinging + /// and leave queued jobs for when it is resumed. + pub paused: bool, } impl std::fmt::Debug for WorkerConfig { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - write!(f, "WorkerConfig {{ worker_tags: {:?}, priority_tags_sorted: {:?}, dedicated_worker: {:?}, dedicated_workers: {:?}, init_bash: {:?}, periodic_script_bash: {:?}, periodic_script_interval_seconds: {:?}, cache_clear: {:?}, additional_python_paths: {:?}, pip_local_dependencies: {:?}, env_vars: {:?}, native_mode: {:?}, object_store_cache_config: {} }}", - self.worker_tags, self.priority_tags_sorted, self.dedicated_worker, self.dedicated_workers, self.init_bash, self.periodic_script_bash, self.periodic_script_interval_seconds, self.cache_clear, self.additional_python_paths, self.pip_local_dependencies, self.env_vars.iter().map(|(k, v)| format!("{}: {}{} ({} chars)", k, &v[..3.min(v.len())], "***", v.len())).collect::>().join(", "), self.native_mode, + write!(f, "WorkerConfig {{ worker_tags: {:?}, priority_tags_sorted: {:?}, dedicated_worker: {:?}, dedicated_workers: {:?}, init_bash: {:?}, periodic_script_bash: {:?}, periodic_script_interval_seconds: {:?}, cache_clear: {:?}, additional_python_paths: {:?}, pip_local_dependencies: {:?}, env_vars: {:?}, native_mode: {:?}, paused: {:?}, object_store_cache_config: {} }}", + self.worker_tags, self.priority_tags_sorted, self.dedicated_worker, self.dedicated_workers, self.init_bash, self.periodic_script_bash, self.periodic_script_interval_seconds, self.cache_clear, self.additional_python_paths, self.pip_local_dependencies, self.env_vars.iter().map(|(k, v)| format!("{}: {}{} ({} chars)", k, &v[..3.min(v.len())], "***", v.len())).collect::>().join(", "), self.native_mode, self.paused, // holds bucket credentials self.object_store_cache_config.as_ref().map(|_| "***").unwrap_or("None")) } diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 28d226ea3f..2acafa539f 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -3424,6 +3424,10 @@ pub async fn run_worker( tokio::time::sleep(Duration::from_millis(200)).await; continue; } + } else if WORKER_CONFIG.load().paused { + // Checked after the same-worker channel: a flow this worker already runs + // still needs it for its remaining same-worker steps. + Ok(None) } else { match &conn { Connection::Sql(db) => { diff --git a/frontend/src/lib/components/WorkerGroup.svelte b/frontend/src/lib/components/WorkerGroup.svelte index 65d00f9ba6..3b399a5a1e 100644 --- a/frontend/src/lib/components/WorkerGroup.svelte +++ b/frontend/src/lib/components/WorkerGroup.svelte @@ -104,6 +104,7 @@ autoscaling?: AutoscalingConfig native_mode?: boolean object_store_cache_config?: ObjectStoreConfig + paused?: boolean } = $state({}) function loadNConfig() { @@ -111,10 +112,9 @@ ? config.worker_tags != undefined || config.dedicated_worker != undefined || config.dedicated_workers != undefined - ? config - : { - worker_tags: [] - } + ? // a copy: editing the draft must not touch the saved config it is compared against + $state.snapshot(config) + : { ...$state.snapshot(config), worker_tags: [] } : { worker_tags: [] } @@ -212,6 +212,7 @@ periodic_script_interval_seconds?: number native_mode?: boolean object_store_cache_config?: ObjectStoreConfig + paused?: boolean } activeWorkers: number customTags: string[] | undefined @@ -268,8 +269,10 @@ if (!config && !nconfig) return false if (!config || !nconfig) return true - const cleaned1 = cleanValueProperties(config) - const cleaned2 = cleanValueProperties(nconfig) + // Compared without empty values: the draft is filled with empty defaults a saved + // config may lack, which are not an edit. + const cleaned1 = cleanWorkerGroupConfig(cleanValueProperties(config)) + const cleaned2 = cleanWorkerGroupConfig(cleanValueProperties(nconfig)) return ( orderedJsonStringify(replaceFalseWithUndefined(cleaned1)) !== @@ -298,6 +301,47 @@ let openDelete = $state(false) let openClean = $state(false) let openRestart = $state(false) + let pendingConfig: typeof nconfig | undefined = $state() + + // Mirrors `reload_worker_config` in backend/src/monitor.rs, which decides the restart. It + // compares the body that is sent rather than the draft, and only the fields a worker loads. + const RESTART_FIELDS = [ + 'dedicated_worker', + 'dedicated_workers', + 'init_bash', + 'periodic_script_bash', + 'periodic_script_interval_seconds', + 'native_mode' + ] as const + const OTHER_WORKER_FIELDS = [ + 'worker_tags', + 'priority_tags', + 'additional_python_paths', + 'pip_local_dependencies', + 'object_store_cache_config' + ] as const + function restartsWorkers(body: typeof nconfig): boolean { + if (activeWorkers === 0) return false + const saved: typeof nconfig = config ?? {} + const comparable = (c: typeof nconfig, k: keyof typeof nconfig) => + orderedJsonStringify(replaceFalseWithUndefined(cleanWorkerGroupConfig({ v: c[k] }))) + const changed = (k: keyof typeof nconfig) => comparable(saved, k) !== comparable(body, k) + const restartFieldChanged = RESTART_FIELDS.some(changed) + const onlyPauseChanged = + changed('paused') && + !restartFieldChanged && + !OTHER_WORKER_FIELDS.some(changed) && + !hasEnvVarChanges + const dedicated = + body.dedicated_worker != undefined || (body.dedicated_workers?.length ?? 0) > 0 + return !onlyPauseChanged && (dedicated || restartFieldChanged) + } + + async function saveConfig(body: typeof nconfig) { + await ConfigService.updateConfig({ name: 'worker__' + name, requestBody: body }) + sendUserToast('Configuration set') + dispatch('reload') + } // Compute hashed tags for display (actual tags used by the worker) let hashedDedicatedTags: Map = $state(new Map()) @@ -399,6 +443,31 @@ + { + pendingConfig = undefined + }} + on:confirmed={async () => { + const body = pendingConfig + pendingConfig = undefined + if (body) { + await saveConfig(body) + } + }} +> +
+ Applying these changes will restart the {pluralize(activeWorkers, 'worker')} of worker group '{name}'. + Workers finish their running job, shut down gracefully and are expected to be restarted by + their supervisor. + Pausing or resuming a worker group alone does not restart its workers. +
+ + {/if} + {#if nconfig !== undefined} +
+ + {/if} +