fix(debounce): atomic claim+persist, GC only non-queued rows; reword comment

Address review findings:

- [P1] Claim and accumulated-args persist are now in one transaction. Before,
  a crash between stamping batch rows consumed_by=self and the `UPDATE v2_job
  SET args` could let a zombie re-pull see its own prior claim and keep only its
  own args (dropping the siblings it had claimed). Wrapping claim + accumulate +
  persist in a tx makes them commit together or roll back together (re-pull then
  re-claims cleanly).

- [P1] GC of consumed batch rows now also requires the job to no longer be in
  v2_job_queue. A consumed sibling can stay queued well past any time grace under
  a concurrency limit / backlog; reclaiming its marker by age alone let its
  eventual pull treat it as never-batched and re-run its item (a duplicate).
  Keeping the row until the job leaves the queue preserves the "already consumed"
  signal. Test extended with a still-queued consumed row that must survive GC.

- [P2] Drop "Customer" attribution from a test doc comment (AGENTS.md).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

backend/.sqlx/query-9b781d92eba2f6eabe16f99cf908f566e4b42b1f9fa445f2719ac79ad535277d.json

@@ -0,0 +1,20 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "WITH del AS (\n                DELETE FROM v2_job_debounce_batch\n                WHERE consumed_at IS NOT NULL\n                  AND consumed_at < now() - interval '10 minutes'\n                  AND id NOT IN (SELECT id FROM v2_job_queue)\n                RETURNING 1\n            ) SELECT count(*) as \"c!\" FROM del",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "c!",
+        "type_info": "Int8"
+      }
+    ],
+    "parameters": {
+      "Left": []
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "9b781d92eba2f6eabe16f99cf908f566e4b42b1f9fa445f2719ac79ad535277d"
+}

backend/.sqlx/query-9f28b636e96aa3461d84de37815a04628af9dcfb6baa2c25fb7b32152227dc77.json

@@ -0,0 +1,20 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "WITH del AS (\n            DELETE FROM v2_job_debounce_batch\n            WHERE consumed_at IS NOT NULL\n              AND consumed_at < now() - interval '10 minutes'\n              AND id NOT IN (SELECT id FROM v2_job_queue)\n            RETURNING 1\n        ) SELECT count(*) FROM del",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "count",
+        "type_info": "Int8"
+      }
+    ],
+    "parameters": {
+      "Left": []
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "9f28b636e96aa3461d84de37815a04628af9dcfb6baa2c25fb7b32152227dc77"
+}

backend/.sqlx/query-ea277c48e7966ab01d9b18e3c0e357033b999d8f3c23241e4c4053cc573f6ca2.json

@@ -0,0 +1,17 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "INSERT INTO v2_job_debounce_batch (id, debounce_batch, consumed_at) VALUES\n                ($1, nextval('debounce_batch_seq'), now() - interval '20 minutes'),\n                ($2, nextval('debounce_batch_seq'), now() - interval '1 minute'),\n                ($3, nextval('debounce_batch_seq'), NULL),\n                ($4, nextval('debounce_batch_seq'), now() - interval '20 minutes')",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Uuid",
+        "Uuid",
+        "Uuid",
+        "Uuid"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "ea277c48e7966ab01d9b18e3c0e357033b999d8f3c23241e4c4053cc573f6ca2"
+}

backend/src/monitor.rs

@@ -4404,13 +4404,19 @@ RETURNING key,job_id
 /// own (already-accumulated, persisted) args, so the grace period is not correctness-
 /// critical.
 async fn cleanup_consumed_debounce_batches(db: &DB) -> error::Result<()> {
-    // 10 minutes is far longer than any debounce window, so any survivor that could
-    // still reference a consumed row has been pulled well before then; keeping it short
-    // bounds table growth under high-throughput debounce.
+    // Only reclaim a consumed row once its job has LEFT the queue. A consumed sibling
+    // (its contribution already accumulated by another survivor) can sit queued well
+    // past any time-based grace under a concurrency limit / worker backlog; removing its
+    // row while still queued would make its eventual pull treat it as never-batched and
+    // re-run its item (a duplicate). Keeping the row until the job is no longer queued
+    // guarantees that pull still sees "already consumed" and runs empty. The age floor
+    // is just a safety margin on top.
     let deleted = sqlx::query_scalar!(
         "WITH del AS (
             DELETE FROM v2_job_debounce_batch
-            WHERE consumed_at IS NOT NULL AND consumed_at < now() - interval '10 minutes'
+            WHERE consumed_at IS NOT NULL
+              AND consumed_at < now() - interval '10 minutes'
+              AND id NOT IN (SELECT id FROM v2_job_queue)
             RETURNING 1
         ) SELECT count(*) FROM del"
     )

backend/windmill-queue/src/jobs.rs

@@ -3195,6 +3195,12 @@ impl PulledJobResult {
                 //   - its own earlier claim on re-pull -> keep its accumulated args,
                 //   - never batched (CE / workers behind v2) -> keep its own args.
                 // Consumed rows are GC'd by the monitor.
+                // Claim + accumulate + persist atomically: a crash between stamping the
+                // batch rows consumed_by=self and persisting the merged args would
+                // otherwise let a zombie re-pull see its own prior claim and keep only
+                // its own args (dropping the siblings it had claimed). One transaction
+                // makes the claim and the merged-args write commit together (or neither).
+                let mut tx = db.begin().await?;
                 let claim = sqlx::query!(
                     "WITH mine AS (
                         SELECT debounce_batch, consumed_by FROM v2_job_debounce_batch WHERE id = $1
@@ -3216,7 +3222,7 @@ impl PulledJobResult {
                     ",
                     j_id,
                 )
-                .fetch_one(db)
+                .fetch_one(&mut *tx)
                 .await?;
 
                 let consumed_by_other =
@@ -3249,7 +3255,7 @@ impl PulledJobResult {
                             j_id,
                             args as &Json<HashMap<String, Box<RawValue>>>,
                         )
-                        .execute(db)
+                        .execute(&mut *tx)
                         .await?;
                     }
                 } else if claim.claimed_batch.is_some() {
@@ -3273,7 +3279,7 @@ impl PulledJobResult {
                         &ids,
                         arg_name_to_accumulate,
                     )
-                    .fetch_all(db)
+                    .fetch_all(&mut *tx)
                     .await?
                     .into_iter()
                     {
@@ -3329,13 +3335,14 @@ impl PulledJobResult {
                                 j_id,
                                 args as &Json<HashMap<String, Box<RawValue>>>,
                             )
-                            .execute(db)
+                            .execute(&mut *tx)
                             .await?;
                         }
                     }
                 }
                 // else: claimed_batch is None but it was our own prior claim (re-pull) —
                 // keep the args we already persisted on the first pull.
+                tx.commit().await?;
             } else {
                 // Debounced but no args to accumulate (plain debounce / dependency job):
                 // consume the batch by removing this job's rows.

backend/windmill-queue/tests/debounce_test.rs

@@ -4952,29 +4952,48 @@ mod debounce {
         let old = Uuid::new_v4();
         let recent = Uuid::new_v4();
         let unconsumed = Uuid::new_v4();
+        // A consumed-long-ago sibling that is STILL QUEUED (e.g. stuck behind a
+        // concurrency limit): its marker must survive GC so its eventual pull still sees
+        // "already consumed" and runs empty (no duplicate).
+        let queued_old = Uuid::new_v4();
+        insert_script_job_with_args(
+            &db,
+            queued_old,
+            "test-workspace",
+            "f/test/script",
+            &serde_json::json!({ "items": [9] }),
+        )
+        .await;
         sqlx::query!(
             "INSERT INTO v2_job_debounce_batch (id, debounce_batch, consumed_at) VALUES
                 ($1, nextval('debounce_batch_seq'), now() - interval '20 minutes'),
                 ($2, nextval('debounce_batch_seq'), now() - interval '1 minute'),
-                ($3, nextval('debounce_batch_seq'), NULL)",
+                ($3, nextval('debounce_batch_seq'), NULL),
+                ($4, nextval('debounce_batch_seq'), now() - interval '20 minutes')",
             old,
             recent,
             unconsumed,
+            queued_old,
         )
         .execute(&db)
         .await?;
 
-        // Mirror the monitor GC sweep.
+        // Mirror the monitor GC sweep (age floor + only-if-no-longer-queued).
         let deleted = sqlx::query_scalar!(
             "WITH del AS (
                 DELETE FROM v2_job_debounce_batch
-                WHERE consumed_at IS NOT NULL AND consumed_at < now() - interval '10 minutes'
+                WHERE consumed_at IS NOT NULL
+                  AND consumed_at < now() - interval '10 minutes'
+                  AND id NOT IN (SELECT id FROM v2_job_queue)
                 RETURNING 1
             ) SELECT count(*) as \"c!\" FROM del"
         )
         .fetch_one(&db)
         .await?;
-        assert_eq!(deleted, 1, "only the old consumed row is GC'd");
+        assert_eq!(
+            deleted, 1,
+            "only the old, no-longer-queued consumed row is GC'd"
+        );
 
         let exists = |id: Uuid, db: Pool<Postgres>| async move {
             sqlx::query_scalar!(
@@ -4991,6 +5010,10 @@ mod debounce {
             "recently consumed row kept"
         );
         assert!(exists(unconsumed, db.clone()).await, "unconsumed row kept");
+        assert!(
+            exists(queued_old, db.clone()).await,
+            "old consumed row whose job is still queued must be kept"
+        );
         Ok(())
     }
 
@@ -5006,7 +5029,7 @@ mod debounce {
         .expect("set running");
     }
 
-    /// Customer regression (ported from #9781): post-preprocessing debounce with
+    /// Regression (ref #9781): post-preprocessing debounce with
     /// `debounce_args_to_accumulate` under a concurrency limit. A survivor accumulates
     /// its own element and starts running; a later same-key message must start a NEW
     /// batch (survive) rather than be folded into the running survivor and silently