fix: delete debounce_key on post-preprocessing limit exceeded (#8299)

* fix: delete debounce_key entry when post-preprocessing limits exceeded

For preprocessor flows, the runnable_settings_handle has
debounce_delay_s = None, so maybe_apply_debouncing at pull-time
won't clean up stale debounce_key entries. Previously we only
reset the entry (UPDATE), but since the flow executes immediately
without rescheduling, a stale entry would cause the next incoming
flow to incorrectly try to debounce against an already-executing job.

Change from UPDATE (reset) to DELETE so the entry is fully removed.
Update tests to expect deletion instead of reset.

Companion EE PR: https://github.com/windmill-labs/windmill-ee-private/pull/448

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix: persist accumulated debounce args to v2_job for flows

The in-memory arg accumulation in maybe_apply_debouncing was not
persisted back to v2_job. For scripts this is fine (single execution),
but for flows, subsequent steps re-read args from the DB via
get_mini_pulled_job and would see the original (non-accumulated) value.

Also improve the job log message to show both original and accumulated
argument values for clarity.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* chore: update ee-repo-ref to d1c14355026151ecdd31adda8e2c60ecd1b5ad65

This commit updates the EE repository reference after PR #448 was merged in windmill-ee-private.

Previous ee-repo-ref: bff784002a3335af7c10982599c8f03e536d5abf

New ee-repo-ref: d1c14355026151ecdd31adda8e2c60ecd1b5ad65

Automated by sync-ee-ref workflow.

* test: assert accumulated debounce args are persisted to v2_job

Add DB persistence assertions to accumulation tests to prevent
regressions on the fix that writes accumulated args back to v2_job.
Without this, flow steps re-reading args from the DB would see
the original (non-accumulated) value.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* sqlx

* chore: update ee-repo-ref.txt to ee-private main

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
Co-authored-by: windmill-internal-app[bot] <windmill-internal-app[bot]@users.noreply.github.com>

backend/.sqlx/query-12631fecee6aa11a45cf5c8d101c0dd8de50ac9b57e68198f637038d344fdd46.json

@@ -0,0 +1,28 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "\n            UPDATE kafka_trigger\n            SET\n                kafka_resource_path = $1,\n                group_id = $2,\n                topics = $3,\n                filters = $4,\n                auto_offset_reset = $5,\n                script_path = $6,\n                path = $7,\n                is_flow = $8,\n                edited_by = $9,\n                email = $10,\n                edited_at = now(),\n                server_id = NULL,\n                error = NULL,\n                error_handler_path = $13,\n                error_handler_args = $14,\n                retry = $15\n            WHERE\n                workspace_id = $11 AND path = $12\n            ",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Varchar",
+        "VarcharArray",
+        "JsonbArray",
+        "Varchar",
+        "Varchar",
+        "Varchar",
+        "Bool",
+        "Varchar",
+        "Varchar",
+        "Text",
+        "Text",
+        "Varchar",
+        "Jsonb",
+        "Jsonb"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "12631fecee6aa11a45cf5c8d101c0dd8de50ac9b57e68198f637038d344fdd46"
+}

backend/.sqlx/query-4b5a711986017654bdd495893a16ddc6ab09c98cd8723865cbd341404bc6a02f.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "\n            INSERT INTO kafka_trigger (\n                workspace_id,\n                path,\n                kafka_resource_path,\n                group_id,\n                topics,\n                filters,\n                script_path,\n                is_flow,\n                mode,\n                edited_by,\n                email,\n                edited_at,\n                error_handler_path,\n                error_handler_args,\n                retry\n            ) VALUES (\n                $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, now(), $12, $13, $14\n            )\n            ",
+  "query": "\n            INSERT INTO kafka_trigger (\n                workspace_id,\n                path,\n                kafka_resource_path,\n                group_id,\n                topics,\n                filters,\n                auto_offset_reset,\n                script_path,\n                is_flow,\n                mode,\n                edited_by,\n                email,\n                edited_at,\n                error_handler_path,\n                error_handler_args,\n                retry\n            ) VALUES (\n                $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, now(), $13, $14, $15\n            )\n            ",
   "describe": {
     "columns": [],
     "parameters": {
@@ -12,6 +12,7 @@
         "VarcharArray",
         "JsonbArray",
         "Varchar",
+        "Varchar",
         "Bool",
         {
           "Custom": {
@@ -34,5 +35,5 @@
     },
     "nullable": []
   },
-  "hash": "aed5439aa6dad950e505f9f8f6914fa5ca21319c501b2822c9bc751ddfc9a0a4"
+  "hash": "4b5a711986017654bdd495893a16ddc6ab09c98cd8723865cbd341404bc6a02f"
 }

backend/.sqlx/query-b1af20cbb737a0b98621a406e38831278517b0bb3715bbc5a3c0f2a57f5ff8c2.json

@@ -0,0 +1,14 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "\n                WITH _ AS (\n                    DELETE FROM debounce_key\n                    WHERE job_id = $1\n                )\n                UPDATE v2_job_debounce_batch\n                SET debounce_batch = nextval('debounce_batch_seq')\n                WHERE id = $1\n                    ",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Uuid"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "b1af20cbb737a0b98621a406e38831278517b0bb3715bbc5a3c0f2a57f5ff8c2"
+}

backend/.sqlx/query-c944e384c4b4c6455b431978adc54d176f294b661675392cf92561c0e6e02e6e.json

@@ -0,0 +1,22 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT args FROM v2_job WHERE id = $1",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "args",
+        "type_info": "Jsonb"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Uuid"
+      ]
+    },
+    "nullable": [
+      true
+    ]
+  },
+  "hash": "c944e384c4b4c6455b431978adc54d176f294b661675392cf92561c0e6e02e6e"
+}

backend/.sqlx/query-c9530931f670eab1208c4a284a55afdc3fcbb0eb5f98fd63e2ec89442becbfaa.json

@@ -1,14 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "\n                WITH _ AS (\n                    UPDATE debounce_key\n                    SET debounced_times = 0,\n                        first_started_at = now(),\n                        previous_job_id = NULL\n                    WHERE job_id = $1\n                )\n                UPDATE v2_job_debounce_batch\n                SET debounce_batch = nextval('debounce_batch_seq')\n                WHERE id = $1\n                    ",
-  "describe": {
-    "columns": [],
-    "parameters": {
-      "Left": [
-        "Uuid"
-      ]
-    },
-    "nullable": []
-  },
-  "hash": "c9530931f670eab1208c4a284a55afdc3fcbb0eb5f98fd63e2ec89442becbfaa"
-}

backend/.sqlx/query-e2921e44c70cf6c76c55177f2b56985e84c59ecb3e1a13fcf27d5f7ae5f8d84c.json

@@ -1,27 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "\n            UPDATE kafka_trigger\n            SET\n                kafka_resource_path = $1,\n                group_id = $2,\n                topics = $3,\n                filters = $4,\n                script_path = $5,\n                path = $6,\n                is_flow = $7,\n                edited_by = $8,\n                email = $9,\n                edited_at = now(),\n                server_id = NULL,\n                error = NULL,\n                error_handler_path = $12,\n                error_handler_args = $13,\n                retry = $14\n            WHERE\n                workspace_id = $10 AND path = $11\n            ",
-  "describe": {
-    "columns": [],
-    "parameters": {
-      "Left": [
-        "Varchar",
-        "Varchar",
-        "VarcharArray",
-        "JsonbArray",
-        "Varchar",
-        "Varchar",
-        "Bool",
-        "Varchar",
-        "Varchar",
-        "Text",
-        "Text",
-        "Varchar",
-        "Jsonb",
-        "Jsonb"
-      ]
-    },
-    "nullable": []
-  },
-  "hash": "e2921e44c70cf6c76c55177f2b56985e84c59ecb3e1a13fcf27d5f7ae5f8d84c"
-}

backend/ee-repo-ref.txt

@@ -1 +1 @@
-a70d7db187aa78a7fbfd3bfaf92372160cff320a
+d1c14355026151ecdd31adda8e2c60ecd1b5ad65

backend/windmill-queue/src/jobs.rs

@@ -3040,11 +3040,18 @@ impl PulledJobResult {
 
                 let new_value = to_raw_value(&accumulated_arg);
 
+                let original_value = j
+                    .args
+                    .as_ref()
+                    .and_then(|a| a.get(arg_name_to_accumulate))
+                    .map(|v| v.get().to_string())
+                    .unwrap_or_else(|| "null".to_string());
+
                 append_logs(
                     &j_id,
                     &j.workspace_id,
                     format!(
-                        "Substituting `{arg_name_to_accumulate}` with: {}\n\n",
+                        "Accumulating debounced argument `{arg_name_to_accumulate}`:\n  original: {original_value}\n  accumulated: {}\n\n",
                         &new_value
                     ),
                     &(db.into()),
@@ -3055,6 +3062,18 @@ impl PulledJobResult {
                     .get_or_insert(Json(Default::default()))
                     .as_mut()
                     .insert(arg_name_to_accumulate.to_owned(), new_value);
+
+                // Persist accumulated args to v2_job so that flow steps
+                // re-reading from the DB (via get_mini_pulled_job) see them
+                if let Some(ref args) = j.args {
+                    sqlx::query!(
+                        "UPDATE v2_job SET args = $2 WHERE id = $1",
+                        j_id,
+                        args as &Json<HashMap<String, Box<RawValue>>>,
+                    )
+                    .execute(db)
+                    .await?;
+                }
             }
 
             // Handle dependency job debouncing cleanup when a job is pulled for execution

backend/windmill-queue/tests/debounce_test.rs

@@ -975,7 +975,7 @@ mod debounce {
         Ok(())
     }
 
-    /// Test: Post-preprocessing debounce with max count limit resets the batch.
+    /// Test: Post-preprocessing debounce with max count limit deletes the debounce_key entry.
     #[sqlx::test(migrations = "../migrations", fixtures("base"))]
     async fn test_post_preprocessing_debounce_max_count_resets(
         db: Pool<Postgres>,
@@ -1018,13 +1018,14 @@ mod debounce {
         assert!(results[1].is_some(), "second job should get scheduled_for");
         assert!(results[2].is_some(), "third job should get scheduled_for");
 
-        // Job 4 (the one that exceeds the limit): when limit is exceeded,
-        // the batch resets and the job executes immediately (no scheduled_for delay)
-        // The exact behavior depends on whether the limit check happens before or after the
-        // new job is counted. Let's just verify the debounce_key is reset.
-        let dk = get_debounce_key(&db, "pp_max_count_key").await.unwrap();
-        // debounced_times should have been reset at some point
-        assert!(dk.0 == jobs[3], "debounce_key should point to last job");
+        // Job 4 exceeds the limit: the debounce_key entry should be DELETED
+        // (not just reset) so that a stale entry doesn't cause the next incoming
+        // flow to incorrectly debounce this already-executing job.
+        let dk = get_debounce_key(&db, "pp_max_count_key").await;
+        assert!(
+            dk.is_none(),
+            "debounce_key entry should be deleted when limits exceeded"
+        );
 
         Ok(())
     }
@@ -1526,16 +1527,18 @@ mod debounce {
             "job 5 should return None (limit exceeded, batch reset)"
         );
 
-        // After reset, debounced_times should be 0
-        let dk = get_debounce_key(&db, "pp_count_boundary_key")
-            .await
-            .unwrap();
-        assert_eq!(dk.2, 0, "debounced_times should be reset to 0 after limit");
+        // After limit exceeded, the debounce_key entry should be deleted
+        let dk = get_debounce_key(&db, "pp_count_boundary_key").await;
+        assert!(
+            dk.is_none(),
+            "debounce_key entry should be deleted when limits exceeded"
+        );
 
         Ok(())
     }
 
-    /// Test: after a max_count reset, a new batch starts fresh and debouncing works again.
+    /// Test: after a max_count limit exceeded, a new batch starts completely fresh.
+    /// The debounce_key entry is deleted, so the next cycle starts with a fresh INSERT.
     #[sqlx::test(migrations = "../migrations", fixtures("base"))]
     async fn test_post_preprocessing_max_count_reset_new_batch(
         db: Pool<Postgres>,
@@ -1549,8 +1552,8 @@ mod debounce {
         let args_hm = empty_args();
 
         // Limit check is `debounced_times > max`, so with max=2 we need 4 jobs
-        // to trigger reset (debounced_times=3 on the 4th job, 3>2=true).
-        // Cycle 1: jobs 1-4 (job 4 exceeds limit → reset)
+        // to trigger limit (debounced_times=3 on the 4th job, 3>2=true).
+        // Cycle 1: jobs 1-4 (job 4 exceeds limit → entry deleted)
         let mut cycle1 = Vec::new();
         for _ in 0..4 {
             let id = Uuid::new_v4();
@@ -1579,14 +1582,17 @@ mod debounce {
         );
         assert!(
             cycle1_results[3].is_none(),
-            "cycle1 job4 should reset (over limit)"
+            "cycle1 job4 should execute immediately (over limit)"
         );
 
-        // Verify debounced_times is reset to 0
-        let dk = get_debounce_key(&db, "pp_reset_cycle_key").await.unwrap();
-        assert_eq!(dk.2, 0, "debounced_times should be 0 after reset");
+        // Verify debounce_key entry is deleted after limit exceeded
+        let dk = get_debounce_key(&db, "pp_reset_cycle_key").await;
+        assert!(
+            dk.is_none(),
+            "debounce_key entry should be deleted after limit exceeded"
+        );
 
-        // Cycle 2: jobs 5-8 (new batch, should debounce independently)
+        // Cycle 2: jobs 5-8 (completely fresh batch since entry was deleted)
         let mut cycle2 = Vec::new();
         for _ in 0..4 {
             let id = Uuid::new_v4();
@@ -1607,19 +1613,17 @@ mod debounce {
             .await?;
             cycle2_results.push(r);
         }
-        // After cycle 1 reset, debounced_times=0. Cycle 2's first job hits ON CONFLICT
-        // and increments to 1 (unlike cycle 1's first job which was a fresh insert at 0).
-        // So cycle 2 reaches the limit one job sooner:
-        //   job5: dt=1, job6: dt=2, job7: dt=3 (>2 → reset), job8: dt=1
-        assert!(cycle2_results[0].is_some(), "cycle2 job1 scheduled (dt=1)");
-        assert!(cycle2_results[1].is_some(), "cycle2 job2 scheduled (dt=2)");
+        // Cycle 2 starts fresh (INSERT, not CONFLICT), so it behaves identically to cycle 1:
+        //   job5: dt=0 (INSERT), job6: dt=1, job7: dt=2, job8: dt=3 (>2 → limit)
         assert!(
-            cycle2_results[2].is_none(),
-            "cycle2 job3 should reset (dt=3 > 2)"
+            cycle2_results[0].is_some(),
+            "cycle2 job1 scheduled (dt=0, fresh INSERT)"
         );
+        assert!(cycle2_results[1].is_some(), "cycle2 job2 scheduled (dt=1)");
+        assert!(cycle2_results[2].is_some(), "cycle2 job3 scheduled (dt=2)");
         assert!(
-            cycle2_results[3].is_some(),
-            "cycle2 job4 scheduled (fresh after reset, dt=1)"
+            cycle2_results[3].is_none(),
+            "cycle2 job4 should execute immediately (dt=3 > 2)"
         );
 
         Ok(())
@@ -3074,6 +3078,29 @@ mod debounce {
         let other: String = serde_json::from_str(other_raw.get())?;
         assert_eq!(other, "x", "non-accumulated arg should be unchanged");
 
+        // Verify accumulated args were persisted to v2_job (needed for flows
+        // where subsequent steps re-read args from the DB)
+        let db_args: Option<serde_json::Value> =
+            sqlx::query_scalar!("SELECT args FROM v2_job WHERE id = $1", survivor_id,)
+                .fetch_one(&db)
+                .await?;
+        let db_args = db_args.expect("v2_job args should not be null after accumulation");
+        let db_items = db_args
+            .get("items")
+            .expect("persisted args should contain 'items'");
+        let db_items: Vec<i64> =
+            serde_json::from_value::<Vec<serde_json::Value>>(db_items.clone())?
+                .iter()
+                .map(|v| v.as_i64().unwrap())
+                .collect();
+        let mut db_items_sorted = db_items.clone();
+        db_items_sorted.sort();
+        assert_eq!(
+            db_items_sorted,
+            vec![1, 2, 3, 4, 5, 6],
+            "persisted args in v2_job should contain all accumulated items"
+        );
+
         Ok(())
     }
 
@@ -3504,6 +3531,27 @@ mod debounce {
 
         assert_accumulated_items(&result, &[10, 20, 30, 40, 50], "items");
 
+        // Verify accumulated args were persisted to v2_job
+        let db_args: Option<serde_json::Value> =
+            sqlx::query_scalar!("SELECT args FROM v2_job WHERE id = $1", survivor_id,)
+                .fetch_one(&db)
+                .await?;
+        let db_items = db_args
+            .expect("v2_job args should not be null")
+            .get("items")
+            .expect("persisted args should contain 'items'")
+            .clone();
+        let mut db_items: Vec<i64> = serde_json::from_value::<Vec<serde_json::Value>>(db_items)?
+            .iter()
+            .map(|v| v.as_i64().unwrap())
+            .collect();
+        db_items.sort();
+        assert_eq!(
+            db_items,
+            vec![10, 20, 30, 40, 50],
+            "persisted args in v2_job should contain all accumulated items"
+        );
+
         Ok(())
     }
 
@@ -3604,6 +3652,33 @@ mod debounce {
         let extra: String = serde_json::from_str(extra_raw.get())?;
         assert_eq!(extra, "v", "non-accumulated arg should be unchanged");
 
+        // Verify accumulated args were persisted to v2_job
+        let db_args: Option<serde_json::Value> =
+            sqlx::query_scalar!("SELECT args FROM v2_job WHERE id = $1", survivor_id,)
+                .fetch_one(&db)
+                .await?;
+        let db_args = db_args.expect("v2_job args should not be null");
+        let db_items = db_args
+            .get("items")
+            .expect("persisted args should contain 'items'")
+            .clone();
+        let mut db_items: Vec<i64> = serde_json::from_value::<Vec<serde_json::Value>>(db_items)?
+            .iter()
+            .map(|v| v.as_i64().unwrap())
+            .collect();
+        db_items.sort();
+        assert_eq!(
+            db_items,
+            vec![100, 200, 300, 400, 500, 600],
+            "persisted args in v2_job should contain all accumulated items"
+        );
+        // "extra" should also be persisted unchanged
+        let db_extra = db_args.get("extra").unwrap().as_str().unwrap();
+        assert_eq!(
+            db_extra, "v",
+            "persisted non-accumulated arg should be unchanged"
+        );
+
         Ok(())
     }