From daab7e4e05574c30e1113ad23974018c0049ee70 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Wed, 15 Apr 2026 21:01:29 +0000 Subject: [PATCH] [ee] fix: ignore arg values in default debounce key MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The default debounce key baked all non-accumulated arg values into a colon-joined suffix, so any varying non-accumulated field (e.g. a timestamp, a kafka offset, or a msg id present in a preprocessor's output) silently broke debouncing — every call got a unique key and no jobs ever collapsed into a single debounced run. The default key is now just the runnable's fully-qualified path. The surviving job keeps its own (latest) non-accumulated args, and debounce_args_to_accumulate is still merged across the batch at pull time. Users who want per-arg debouncing can still set a custom debounce_key template (e.g. "pp_$args[region]"). Drops the now-unused args_to_ignore_if_default parameter from resolve_debounce_key. Companion: windmill-labs/windmill-ee-private# Co-Authored-By: Claude Opus 4.6 (1M context) --- backend/ee-repo-ref.txt | 2 +- backend/windmill-queue/src/jobs.rs | 36 +++++++------------ backend/windmill-queue/tests/debounce_test.rs | 20 ++++++----- 3 files changed, 25 insertions(+), 33 deletions(-) diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index dc01a40e29..8692040f51 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -e587df81d31638259efec13d0e68e54c890fdc2e \ No newline at end of file +0cfd32222ace79d10e4a8608f91d40e2eb3a9437 \ No newline at end of file diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index f832413be7..3a8c7682a0 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -3613,33 +3613,23 @@ pub fn resolve_debounce_key<'b>( workspace_id: &str, job_kind: JobKind, args: &PushArgs<'b>, - args_to_ignore_if_default: Option<&String>, ) -> String { + // Default key is just the runnable's fully-qualified path — debouncing batches + // all calls to the same runnable regardless of args. The surviving job keeps its + // own (latest) args, and debounce_args_to_accumulate is merged across the batch + // at pull time. Users who want per-arg debouncing must set an explicit + // `debounce_key` template (e.g. "pp_$args[region]"). let original_debounce_key = unresolved_debounce_key .map(|x| crate::interpolate_args(x, &args, workspace_id)) - .unwrap_or(format!( - "{}#args:{}", - crate::fullpath_with_workspace(workspace_id, runnable_path.as_ref(), &job_kind), - args.args - .iter() - .filter_map(|(k, v)| { - if args_to_ignore_if_default - .map(|name| name == k) - .unwrap_or_default() - { - None - } else { - Some(v.to_string()) - } - }) - // TODO: disable sorted? - .sorted() - .collect_vec() - .join(":"), - )); - - tracing::debug!("Original debounce key (len={}): {}", original_debounce_key.len(), original_debounce_key); + .unwrap_or_else(|| { + crate::fullpath_with_workspace(workspace_id, runnable_path.as_ref(), &job_kind) + }); + tracing::debug!( + "Original debounce key (len={}): {}", + original_debounce_key.len(), + original_debounce_key + ); // If debounce_key is not too long (< 255 chars), keep it as is, otherwise hash it. // On cloud, we prepend "{workspace_id}:" so we must reserve space for that prefix diff --git a/backend/windmill-queue/tests/debounce_test.rs b/backend/windmill-queue/tests/debounce_test.rs index a2c6971c1c..38d09c5594 100644 --- a/backend/windmill-queue/tests/debounce_test.rs +++ b/backend/windmill-queue/tests/debounce_test.rs @@ -3128,7 +3128,7 @@ mod debounce { ) -> anyhow::Result<()> { let settings = DebouncingSettings { debounce_delay_s: Some(5), - debounce_key: None, // default key (includes args minus accumulated ones) + debounce_key: None, // default key is runnable path only — arg values are ignored debounce_args_to_accumulate: Some(vec!["items".to_string()]), ..Default::default() }; @@ -3180,10 +3180,12 @@ mod debounce { Ok(()) } - /// Test: debounce_args_to_accumulate does NOT cause debouncing when non-accumulated - /// args differ — only the accumulated arg is excluded from the key. + /// Test: non-accumulated args do NOT affect the default debounce key — jobs on the + /// same runnable still debounce together even when other args differ. The surviving + /// job keeps its own (latest) non-accumulated args; the accumulated arg is merged + /// across the batch at pull time. #[sqlx::test(migrations = "../migrations", fixtures("base"))] - async fn test_post_preprocessing_args_to_accumulate_different_non_accumulated( + async fn test_post_preprocessing_args_to_accumulate_non_accumulated_ignored( db: Pool, ) -> anyhow::Result<()> { let settings = DebouncingSettings { @@ -3198,7 +3200,7 @@ mod debounce { let args1 = serde_json::json!({"items": ["a"], "other": "foo"}); insert_flow_job_with_args(&db, job1, "test-workspace", "f/test/flow", &args1).await; - // Job 2: other = "bar" (different non-accumulated arg) + // Job 2: other = "bar" (different non-accumulated arg — should still collapse) let job2 = Uuid::new_v4(); let args2 = serde_json::json!({"items": ["b"], "other": "bar"}); insert_flow_job_with_args(&db, job2, "test-workspace", "f/test/flow", &args2).await; @@ -3227,14 +3229,14 @@ mod debounce { ) .await?; - // Both should still be queued — different "other" arg means different keys + // job1 should be debounced (completed) — the default key ignores non-accumulated args assert!( - is_queued(&db, &job1).await, - "job1 should still be queued (different key due to 'other' arg)" + is_completed(&db, &job1).await, + "job1 should be debounced by job2 even though 'other' differs" ); assert!( is_queued(&db, &job2).await, - "job2 should still be queued (different key due to 'other' arg)" + "job2 should still be queued (survivor with its own 'other' value)" ); Ok(())