From 8eb36ce008b4efe2be9a9bfebc91af68070f7a6c Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Thu, 23 Jul 2026 19:05:09 +0200 Subject: [PATCH] fix: treat concurrent_limit/timeout <= 0 as unset instead of a zero cap (#10288) * fix: treat concurrent_limit/timeout <= 0 as unset instead of a zero cap Co-Authored-By: Claude Opus 4.8 (1M context) * fix: flow-step timeout <= 0 inherits the script timeout, not the global default Co-Authored-By: Claude Opus 4.8 (1M context) --------- Co-authored-by: Claude Opus 4.8 (1M context) --- backend/windmill-api-flows/src/flows.rs | 9 +- backend/windmill-api-scripts/src/scripts.rs | 7 + backend/windmill-queue/src/jobs.rs | 42 ++-- .../tests/concurrency_limit_zero_test.rs | 21 ++ .../windmill-types/src/runnable_settings.rs | 190 +++++++++++++++++- backend/windmill-worker/src/common.rs | 4 +- backend/windmill-worker/src/worker.rs | 19 +- backend/windmill-worker/src/worker_flow.rs | 64 ++++-- cli/src/commands/script/script.ts | 51 ++++- 9 files changed, 355 insertions(+), 52 deletions(-) create mode 100644 backend/windmill-queue/tests/concurrency_limit_zero_test.rs diff --git a/backend/windmill-api-flows/src/flows.rs b/backend/windmill-api-flows/src/flows.rs index 7f39e61f21..97e6999e23 100644 --- a/backend/windmill-api-flows/src/flows.rs +++ b/backend/windmill-api-flows/src/flows.rs @@ -527,6 +527,11 @@ async fn create_flow( } check_scopes(&authed, || format!("flows:write:{}", nf.path))?; + // A `<= 0` flow timeout is "unset", not a 0-second limit that kills every run instantly. + // (The concurrency settings inside the flow value are normalized on deserialization; see + // ConcurrencySettings.) Runtime guards also protect already-stored rows. + nf.timeout = windmill_common::runnable_settings::none_if_non_positive(nf.timeout); + if let RuleCheckResult::Blocked(msg) = check_deploy_rules( &w_id, AuditAuthorable::username(&authed), @@ -1012,7 +1017,9 @@ async fn update_flow( } let flow_path = flow_path.to_path(); // The URL identifies the flow being updated; the body path is only needed to rename. - let nf = ef.into_new_flow(flow_path); + let mut nf = ef.into_new_flow(flow_path); + // A `<= 0` flow timeout is "unset", not a 0-second limit (see create_flow). + nf.timeout = windmill_common::runnable_settings::none_if_non_positive(nf.timeout); check_scopes(&authed, || format!("flows:write:{}", flow_path))?; if let RuleCheckResult::Blocked(msg) = check_deploy_rules( diff --git a/backend/windmill-api-scripts/src/scripts.rs b/backend/windmill-api-scripts/src/scripts.rs index 20aad98864..c2f023ce5a 100644 --- a/backend/windmill-api-scripts/src/scripts.rs +++ b/backend/windmill-api-scripts/src/scripts.rs @@ -914,6 +914,13 @@ async fn create_script_internal<'c>( } check_scopes(&authed, || format!("scripts:write:{}", ns.path))?; + // Normalize positive-only settings so a `<= 0` value (e.g. a CLI-pushed `0`) persists as + // disabled rather than as a zero-slot concurrency cap or a 0-second timeout. Deserialization + // already normalizes the concurrency fields; re-applying here also covers `timeout` and any + // NewScript built in-process rather than from a request body. + ns.timeout = windmill_common::runnable_settings::none_if_non_positive(ns.timeout); + ns.concurrency_settings = ns.concurrency_settings.normalized(); + guard_script_from_debounce_data(&ns).await?; let codebase = ns.codebase.as_ref(); diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 4911b08421..433d2e4ed5 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -1107,15 +1107,16 @@ async fn commit_completed_job( // Resolve the concurrency-limit settings on the pool *before* opening the // completion transaction: doing it inside the tx would hold a second // simultaneous connection from the small per-worker pool. - let has_concurrent_limit = completed_job.concurrent_limit.is_some() - || windmill_common::runnable_settings::prefetch_cached_from_handle( - completed_job.runnable_settings_handle, - db, - ) - .await? - .1 - .concurrent_limit - .is_some(); + let has_concurrent_limit = has_active_concurrency_limit(completed_job.concurrent_limit) + || has_active_concurrency_limit( + windmill_common::runnable_settings::prefetch_cached_from_handle( + completed_job.runnable_settings_handle, + db, + ) + .await? + .1 + .concurrent_limit, + ); // A genuine NUL (U+0000) in the result serializes to a `\u0000` escape that // the jsonb `result` column rejects with 22P05 ("unsupported Unicode escape @@ -3921,7 +3922,7 @@ pub async fn pull( let pulled_job_result = match job { #[cfg(feature = "private")] Some(job) - if concurrency_settings.concurrent_limit.is_some() + if has_active_concurrency_limit(concurrency_settings.concurrent_limit) // Concurrency limit is available for either enterprise job or dependency job && (cfg!(feature = "enterprise") || (job.is_dependency() && !*WMDEBUG_NO_DEBOUNCING)) => { @@ -3985,7 +3986,8 @@ pub async fn pull( .1 .maybe_fallback(None, job.concurrent_limit, job.concurrency_time_window_s); - let has_concurent_limit = concurrency_settings.concurrent_limit.is_some(); + let has_concurent_limit = + has_active_concurrency_limit(concurrency_settings.concurrent_limit); #[cfg(not(feature = "enterprise"))] if has_concurent_limit && !job.is_dependency() { @@ -3994,7 +3996,7 @@ pub async fn pull( #[cfg(not(feature = "enterprise"))] let has_concurent_limit = job.is_dependency() - && job.concurrent_limit.is_some() + && has_active_concurrency_limit(job.concurrent_limit) && cfg!(feature = "private") && !*WMDEBUG_NO_DEBOUNCING; // if we don't have private flag, we don't have concurrency limit @@ -4162,6 +4164,13 @@ async fn pull_single_job_and_mark_as_running_no_concurrency_limit<'c>( Ok(job_and_suspended) } +/// A concurrency limit is only active when it caps at 1+ slots. `Some(0)` (or negative) is +/// a disabled limit, not a zero-slot one — see [`ConcurrencySettings::normalized`]. The gate +/// checks must use this instead of `.is_some()` so a legacy stored `0` behaves as disabled. +pub fn has_active_concurrency_limit(concurrent_limit: Option) -> bool { + concurrent_limit.is_some_and(|n| n > 0) +} + pub async fn custom_concurrency_key( db: &Pool, job_id: &Uuid, @@ -6141,6 +6150,11 @@ async fn push_inner<'c, 'd>( }, }; + // Guard against an already-stored `concurrent_limit <= 0` reaching the queue: it would + // register a zero-slot concurrency key and permanently block the job. Coerce it to + // disabled before it is persisted onto the job row / concurrency key here. + concurrency_settings = concurrency_settings.normalized(); + // Enforce concurrency limit on all dependency jobs. // TODO: We can ignore this for scripts djobs. The main reason we need all djobs to be sequential is because we have // nodes_to_relock and we need all locks whose corresponding steps aren't in nodes_to_relock be already present. @@ -6328,7 +6342,7 @@ async fn push_inner<'c, 'd>( check_workspace_queue_cap(&mut *tx, workspace_id).await?; } - if concurrency_settings.concurrent_limit.is_some() { + if has_active_concurrency_limit(concurrency_settings.concurrent_limit) { let concurrency_key = resolve_concurrency_key( workspace_id, &args, @@ -6732,7 +6746,7 @@ pub async fn insert_concurrency_key_capped<'d, 'c, E: PgExecutor<'c> + Copy>( custom_concurrency_key, ); #[cfg(feature = "cloud")] - if *CLOUD_HOSTED && concurrent_limit.is_some() { + if *CLOUD_HOSTED && has_active_concurrency_limit(concurrent_limit) { check_concurrency_key_queue_cap(db, &concurrency_key).await?; } #[cfg(not(feature = "cloud"))] diff --git a/backend/windmill-queue/tests/concurrency_limit_zero_test.rs b/backend/windmill-queue/tests/concurrency_limit_zero_test.rs new file mode 100644 index 0000000000..bd72571bcd --- /dev/null +++ b/backend/windmill-queue/tests/concurrency_limit_zero_test.rs @@ -0,0 +1,21 @@ +//! Runtime gate for the `Some(0)` concurrency footgun: a stored `concurrent_limit <= 0` +//! must read as "disabled", never as a zero-slot cap that permanently blocks the job at the +//! concurrency gate (the re-queue storm the zombie monitor eventually fails as a fake OOM). +//! +//! Run with: +//! cargo test -p windmill-queue --test concurrency_limit_zero_test + +use windmill_queue::jobs::has_active_concurrency_limit; + +#[test] +fn zero_and_negative_are_not_active_limits() { + assert!(!has_active_concurrency_limit(None)); + assert!(!has_active_concurrency_limit(Some(0))); + assert!(!has_active_concurrency_limit(Some(-1))); +} + +#[test] +fn positive_limit_is_active() { + assert!(has_active_concurrency_limit(Some(1))); + assert!(has_active_concurrency_limit(Some(i32::MAX))); +} diff --git a/backend/windmill-types/src/runnable_settings.rs b/backend/windmill-types/src/runnable_settings.rs index ea0fd30dc0..038246cbe6 100644 --- a/backend/windmill-types/src/runnable_settings.rs +++ b/backend/windmill-types/src/runnable_settings.rs @@ -93,9 +93,7 @@ pub struct DebouncingSettings { pub debounce_args_to_accumulate: Option>, } -#[derive( - Debug, Default, Clone, Serialize, Deserialize, Hash, PartialEq, sqlx::FromRow, sqlx::Decode, -)] +#[derive(Debug, Default, Clone, Serialize, Hash, PartialEq, sqlx::FromRow, sqlx::Decode)] pub struct ConcurrencySettings { #[serde(skip_serializing_if = "Option::is_none")] pub concurrency_key: Option, @@ -105,7 +103,65 @@ pub struct ConcurrencySettings { pub concurrency_time_window_s: Option, } -#[derive(Debug, Clone, Serialize, Deserialize, sqlx::FromRow, Default)] +/// Shared normalization for the positive-only `Option` runnable settings +/// (`concurrent_limit`, `timeout`, ...): a `<= 0` value is never meaningful — zero +/// concurrent slots permanently blocks a runnable at the concurrency gate (a re-queue +/// storm the zombie monitor eventually fails with a misleading OOM error), and a +/// 0-second timeout kills every job on the spot. The frontend already treats `0` as +/// "disabled", so `<= 0` maps to `None` (unset) everywhere. Idempotent. +pub fn none_if_non_positive(v: Option) -> Option { + v.filter(|n| *n > 0) +} + +/// Coerce a `concurrent_limit <= 0` to disabled, dropping the now-meaningless time window +/// alongside it. Idempotent. +fn normalize_concurrency( + concurrent_limit: &mut Option, + concurrency_time_window_s: &mut Option, +) { + if none_if_non_positive(*concurrent_limit).is_none() { + *concurrent_limit = None; + *concurrency_time_window_s = None; + } +} + +impl ConcurrencySettings { + pub fn normalized(mut self) -> Self { + normalize_concurrency( + &mut self.concurrent_limit, + &mut self.concurrency_time_window_s, + ); + self + } +} + +// Manual `Deserialize` so every ingestion path (script/flow create & update, app and +// http-trigger payloads, and read-back of already-stored settings) normalizes a `<= 0` +// limit uniformly, without each call site remembering to call `normalized()`. +impl<'de> Deserialize<'de> for ConcurrencySettings { + fn deserialize(deserializer: D) -> Result + where + D: serde::Deserializer<'de>, + { + #[derive(Deserialize)] + struct Raw { + #[serde(default)] + concurrency_key: Option, + #[serde(default)] + concurrent_limit: Option, + #[serde(default)] + concurrency_time_window_s: Option, + } + let Raw { concurrency_key, concurrent_limit, concurrency_time_window_s } = + Raw::deserialize(deserializer)?; + Ok( + ConcurrencySettings { concurrency_key, concurrent_limit, concurrency_time_window_s } + .normalized(), + ) + } +} + +#[derive(Debug, Clone, Serialize, sqlx::FromRow, Default)] pub struct ConcurrencySettingsWithCustom { #[serde(skip_serializing_if = "Option::is_none")] pub custom_concurrency_key: Option, @@ -115,6 +171,41 @@ pub struct ConcurrencySettingsWithCustom { pub concurrency_time_window_s: Option, } +impl ConcurrencySettingsWithCustom { + pub fn normalized(mut self) -> Self { + normalize_concurrency( + &mut self.concurrent_limit, + &mut self.concurrency_time_window_s, + ); + self + } +} + +impl<'de> Deserialize<'de> for ConcurrencySettingsWithCustom { + fn deserialize(deserializer: D) -> Result + where + D: serde::Deserializer<'de>, + { + #[derive(Deserialize)] + struct Raw { + #[serde(default)] + custom_concurrency_key: Option, + #[serde(default)] + concurrent_limit: Option, + #[serde(default)] + concurrency_time_window_s: Option, + } + let Raw { custom_concurrency_key, concurrent_limit, concurrency_time_window_s } = + Raw::deserialize(deserializer)?; + Ok(ConcurrencySettingsWithCustom { + custom_concurrency_key, + concurrent_limit, + concurrency_time_window_s, + } + .normalized()) + } +} + impl DebouncingSettings { pub fn maybe_fallback( self, @@ -142,11 +233,15 @@ impl ConcurrencySettings { concurrent_limit: Option, concurrency_time_window_s: Option, ) -> Self { + // Legacy columns can still hold a stored `0` that predates ingestion normalization, + // so re-normalize here: this is the single load boundary for every DB-backed read + // (script/schedule read, flow value, and the worker pull path). Self { concurrency_key: self.concurrency_key.or(concurrency_key), concurrent_limit: self.concurrent_limit.or(concurrent_limit), concurrency_time_window_s: self.concurrency_time_window_s.or(concurrency_time_window_s), } + .normalized() } } @@ -229,4 +324,91 @@ mod tests { assert_eq!(r, Retry::default()); assert_eq!(r.exponential.multiplier, 1); } + + // The positive-only settings share one rule: `<= 0` means "unset". This is what keeps a + // stored `0` from being enforced as a zero-slot cap or a 0-second timeout. + #[test] + fn none_if_non_positive_coerces_zero_and_negative() { + assert_eq!(none_if_non_positive(Some(0)), None); + assert_eq!(none_if_non_positive(Some(-3)), None); + assert_eq!(none_if_non_positive(Some(1)), Some(1)); + assert_eq!(none_if_non_positive(Some(i32::MAX)), Some(i32::MAX)); + assert_eq!(none_if_non_positive(None), None); + } + + // Ingestion path (scripts flatten this on `NewScript`, flows on `FlowModule`): a `0` + // concurrent_limit deserializes to disabled and drops the now-meaningless time window, + // while a real limit and its window survive untouched. + #[test] + fn concurrency_settings_deserialize_normalizes_non_positive_limit() { + let zero: ConcurrencySettings = serde_json::from_value( + serde_json::json!({"concurrent_limit": 0, "concurrency_time_window_s": 30}), + ) + .unwrap(); + assert_eq!(zero.concurrent_limit, None); + assert_eq!(zero.concurrency_time_window_s, None); + + let negative: ConcurrencySettings = + serde_json::from_value(serde_json::json!({"concurrent_limit": -1})).unwrap(); + assert_eq!(negative.concurrent_limit, None); + + let real: ConcurrencySettings = serde_json::from_value( + serde_json::json!({"concurrent_limit": 2, "concurrency_time_window_s": 30}), + ) + .unwrap(); + assert_eq!(real.concurrent_limit, Some(2)); + assert_eq!(real.concurrency_time_window_s, Some(30)); + } + + // Per-flow-step overrides use the `custom_concurrency_key` variant; same rule. + #[test] + fn concurrency_settings_with_custom_deserialize_normalizes() { + let zero: ConcurrencySettingsWithCustom = serde_json::from_value( + serde_json::json!({"concurrent_limit": 0, "concurrency_time_window_s": 5}), + ) + .unwrap(); + assert_eq!(zero.concurrent_limit, None); + assert_eq!(zero.concurrency_time_window_s, None); + } + + // A normalized value serializes with the limit omitted (skip_serializing_if), matching the + // frontend's "disabled" representation instead of re-emitting a `0`. + #[test] + fn normalized_disabled_limit_serializes_as_omitted() { + let s = + ConcurrencySettings { concurrent_limit: Some(0), ..Default::default() }.normalized(); + let json = serde_json::to_value(&s).unwrap(); + assert!(json.get("concurrent_limit").is_none()); + } + + // Runtime load boundary: legacy rows still hold a raw `0` in the fallback columns. The + // fallback must not resurrect it as an active limit. + #[test] + fn maybe_fallback_normalizes_legacy_zero_column() { + let merged = ConcurrencySettings::default().maybe_fallback(None, Some(0), Some(30)); + assert_eq!(merged.concurrent_limit, None); + assert_eq!(merged.concurrency_time_window_s, None); + } + + // `NewScript`/`FlowModule` embed the settings via `#[serde(flatten)]`, which drives the + // manual Deserialize through a content-buffer deserializer rather than a plain map. Guard + // that path: normalization must still fire and sibling fields must still parse. + #[test] + fn flattened_concurrency_normalizes_and_preserves_siblings() { + #[derive(Deserialize)] + struct Wrapper { + name: String, + #[serde(flatten)] + concurrency: ConcurrencySettings, + } + let w: Wrapper = serde_json::from_value(serde_json::json!({ + "name": "s", + "concurrent_limit": 0, + "concurrency_time_window_s": 42, + })) + .unwrap(); + assert_eq!(w.name, "s"); + assert_eq!(w.concurrency.concurrent_limit, None); + assert_eq!(w.concurrency.concurrency_time_window_s, None); + } } diff --git a/backend/windmill-worker/src/common.rs b/backend/windmill-worker/src/common.rs index 6618ac7906..a406f3610c 100644 --- a/backend/windmill-worker/src/common.rs +++ b/backend/windmill-worker/src/common.rs @@ -1081,7 +1081,9 @@ pub async fn resolve_job_timeout( *MAX_TIMEOUT_DURATION }; - match custom_timeout_secs { + // A `custom_timeout_secs <= 0` is not a 0-second limit but "unset": fall through to the + // default/global-max timeout instead of killing the job immediately. + match windmill_common::runnable_settings::none_if_non_positive(custom_timeout_secs) { Some(timeout_secs) if Duration::from_secs(timeout_secs as u64) < global_max_timeout_duration => { diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 1b11d735c3..003ddd3876 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -3892,15 +3892,16 @@ pub async fn handle_queued_job( #[cfg(not(feature = "enterprise"))] if let Connection::Sql(db) = conn { - if (job.concurrent_limit.is_some() - || windmill_common::runnable_settings::prefetch_cached_from_handle( - job.runnable_settings_handle, - db, - ) - .await? - .1 - .concurrent_limit - .is_some()) + if (windmill_queue::jobs::has_active_concurrency_limit(job.concurrent_limit) + || windmill_queue::jobs::has_active_concurrency_limit( + windmill_common::runnable_settings::prefetch_cached_from_handle( + job.runnable_settings_handle, + db, + ) + .await? + .1 + .concurrent_limit, + )) && !job.kind.is_dependency() { logs.push_str("---\n"); diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index d2e5b20a1d..0793aff053 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -1525,9 +1525,14 @@ pub async fn update_flow_status_after_job_completion_internal( let concurrency_key = tag_and_concurrency_key .as_ref() .and_then(|x| x.concurrency_key.clone()); - let concurrent_limit = tag_and_concurrency_key - .as_ref() - .and_then(|x| x.concurrent_limit); + // `concurrent_limit` here can come straight from the raw flow JSON (see + // get_tag_and_concurrency), bypassing the ConcurrencySettings deserialization guard, + // so a stored `0` must still be coerced to disabled before we register a key for it. + let concurrent_limit = windmill_common::runnable_settings::none_if_non_positive( + tag_and_concurrency_key + .as_ref() + .and_then(|x| x.concurrent_limit), + ); let concurrency_time_window_s = tag_and_concurrency_key .as_ref() .and_then(|x| x.concurrency_time_window_s); @@ -4389,13 +4394,10 @@ async fn push_next_flow_job( ) .await?; - if timeout_value < 0 { - return Err(Error::ExecutionErr( - "Timeout value cannot be negative".to_string(), - )); - } - - Some(timeout_value) + // A `<= 0` step timeout (including a negative eval) means "no override": fall back + // to the referenced runnable's own timeout rather than a 0-second/negative timeout + // that would kill the step instantly. + effective_flow_step_timeout(Some(timeout_value), payload_tag.timeout) } else { payload_tag.timeout }; @@ -6048,6 +6050,18 @@ async fn flow_to_payload( }) } +/// Effective timeout for a flow step given the module's (already-evaluated) timeout override and +/// the timeout inherited from the referenced runnable. A `<= 0` override — or none — means "no +/// override": fall back to the inherited value (which is itself `None` when unset, i.e. the +/// instance default). A positive override wins. This keeps a step `timeout: 0` equivalent to an +/// omitted one rather than a 0-second, instant-kill timeout. +pub(crate) fn effective_flow_step_timeout( + module_override: Option, + inherited: Option, +) -> Option { + windmill_common::runnable_settings::none_if_non_positive(module_override).or(inherited) +} + pub async fn script_to_payload( script_hash: Option, script_path: String, @@ -6137,11 +6151,13 @@ pub async fn script_to_payload( module.delete_after_use.unwrap_or(false) || delete_after_use.unwrap_or(false); let final_delete_after_secs = module.delete_after_secs.or(delete_after_secs); - let flow_step_timeout = if module.timeout.is_some() { - None - } else { - script_timeout - }; + // Always carry the referenced script's own timeout as the inherited fallback. The module's + // timeout override (if any) is selected at the push site, where a `<= 0` override is treated + // as "no override" and falls back to this value — so `timeout: 0` on a step means "use the + // script's timeout", not a 0-second (immediate-kill) timeout. Normalize the inherited value + // too, so a legacy `0` script timeout resolves to the default rather than a zero-second kill. + let flow_step_timeout = + windmill_common::runnable_settings::none_if_non_positive(script_timeout); Ok(JobPayloadWithTag { payload, tag, @@ -6266,9 +6282,25 @@ pub async fn get_previous_job_result( #[cfg(test)] mod tests { - use super::extract_chat_message_from_flow_result; + use super::{effective_flow_step_timeout, extract_chat_message_from_flow_result}; use serde_json::{json, value::to_raw_value}; + // A `<= 0` step timeout override must behave as "no override" and inherit the referenced + // script's timeout, not collapse to a 0-second (instant-kill) timeout. A positive override + // still wins. Guards the flow-step timeout footgun. + #[test] + fn flow_step_timeout_zero_or_negative_inherits_script_timeout() { + // zero / negative override -> inherited script timeout + assert_eq!(effective_flow_step_timeout(Some(0), Some(300)), Some(300)); + assert_eq!(effective_flow_step_timeout(Some(-5), Some(300)), Some(300)); + // no inherited timeout either -> None (falls through to the instance default) + assert_eq!(effective_flow_step_timeout(Some(0), None), None); + // positive override wins over the inherited value + assert_eq!(effective_flow_step_timeout(Some(120), Some(300)), Some(120)); + // no override -> inherited + assert_eq!(effective_flow_step_timeout(None, Some(300)), Some(300)); + } + #[test] fn pretty_prints_full_result_when_no_override_is_present() { let value = json!({ diff --git a/cli/src/commands/script/script.ts b/cli/src/commands/script/script.ts index 3db9c7b53e..82fe63719b 100644 --- a/cli/src/commands/script/script.ts +++ b/cli/src/commands/script/script.ts @@ -100,6 +100,30 @@ export function isRawAppBackendPath(filePath: string): boolean { return isRawAppBackendPathInternal(filePath); } +/** + * The positive-only runnable settings (concurrent_limit, timeout, ...) treat any `<= 0` + * value as "unset": the backend coerces it to null (a 0-slot concurrency limit bricks the + * runnable, a 0s timeout kills every run). Coerce to undefined so it is serialized as + * omitted, never as 0, and redeploys don't churn against the backend-normalized value. + */ +export function nonePositiveInt( + v: number | undefined | null +): number | undefined { + return v != null && v > 0 ? v : undefined; +} + +/** + * Normalize a concurrent_limit + its time window together: when the limit is disabled + * (<= 0) the window is dropped too. Returns [concurrent_limit, concurrency_time_window_s]. + */ +export function normalizeConcurrency( + concurrentLimit: number | undefined | null, + concurrencyTimeWindowS?: number | undefined | null +): [number | undefined, number | undefined] { + const limit = nonePositiveInt(concurrentLimit); + return limit === undefined ? [undefined, undefined] : [limit, concurrencyTimeWindowS ?? undefined]; +} + /** * Checks if a path is inside a normal app folder (inline script). * Matches patterns like: .../myApp.app/... or .../myApp__app/... @@ -469,6 +493,15 @@ export async function handleFile( const moduleFolderPath = scriptBasePath + getModuleFolderSuffix(); const modules = await readModulesFromDisk(moduleFolderPath, opts?.defaultTs, moduleEntryPoint); + // A concurrent_limit of <= 0 means "concurrency disabled", not "zero slots" (which + // would brick the runnable at the queue's concurrency gate). Emit it as omitted rather + // than 0 so a redeploy never re-persists a zero-slot limit, and drop the now-meaningless + // time window alongside it. Mirrors the backend's ConcurrencySettings::normalized. + const [normConcurrentLimit, normConcurrencyTimeWindowS] = normalizeConcurrency( + typed?.concurrent_limit, + typed?.concurrency_time_window_s + ); + const requestBodyCommon: NewScript = { content, description: typed?.description ?? "", @@ -482,8 +515,8 @@ export async function handleFile( ws_error_handler_muted: typed?.ws_error_handler_muted, dedicated_worker: typed?.dedicated_worker, cache_ttl: typed?.cache_ttl, - concurrency_time_window_s: typed?.concurrency_time_window_s, - concurrent_limit: typed?.concurrent_limit, + concurrency_time_window_s: normConcurrencyTimeWindowS, + concurrent_limit: normConcurrentLimit, deployment_message: message, restart_unless_cancelled: typed?.restart_unless_cancelled, visible_to_runner_only: typed?.visible_to_runner_only, @@ -493,7 +526,7 @@ export async function handleFile( debounce_key: typed?.debounce_key, debounce_delay_s: typed?.debounce_delay_s, codebase: await codebase?.getDigest(forceTar), - timeout: typed?.timeout, + timeout: nonePositiveInt(typed?.timeout), on_behalf_of_email: typed?.on_behalf_of_email, envs: typed?.envs, modules: modules, @@ -530,9 +563,13 @@ export async function handleFile( remote.ws_error_handler_muted && typed.dedicated_worker == remote.dedicated_worker && typed.cache_ttl == remote.cache_ttl && - typed.concurrency_time_window_s == - remote.concurrency_time_window_s && - typed.concurrent_limit == remote.concurrent_limit && + normConcurrencyTimeWindowS == + normalizeConcurrency( + remote.concurrent_limit, + remote.concurrency_time_window_s + )[1] && + normConcurrentLimit == + normalizeConcurrency(remote.concurrent_limit)[0] && Boolean(typed.restart_unless_cancelled) == Boolean(remote.restart_unless_cancelled) && Boolean(typed.visible_to_runner_only) == @@ -540,7 +577,7 @@ export async function handleFile( Boolean(typed.has_preprocessor) == Boolean(remote.has_preprocessor) && typed.priority == Boolean(remote.priority) && - typed.timeout == remote.timeout && + nonePositiveInt(typed.timeout) == nonePositiveInt(remote.timeout) && //@ts-ignore typed.concurrency_key == remote["concurrency_key"] && typed.debounce_key == remote["debounce_key"] &&