diff --git a/backend/.sqlx/query-9b781d92eba2f6eabe16f99cf908f566e4b42b1f9fa445f2719ac79ad535277d.json b/backend/.sqlx/query-9b781d92eba2f6eabe16f99cf908f566e4b42b1f9fa445f2719ac79ad535277d.json new file mode 100644 index 0000000000..e2132d7737 --- /dev/null +++ b/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" +} diff --git a/backend/.sqlx/query-9f28b636e96aa3461d84de37815a04628af9dcfb6baa2c25fb7b32152227dc77.json b/backend/.sqlx/query-9f28b636e96aa3461d84de37815a04628af9dcfb6baa2c25fb7b32152227dc77.json new file mode 100644 index 0000000000..6a39c40970 --- /dev/null +++ b/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" +} diff --git a/backend/.sqlx/query-ea277c48e7966ab01d9b18e3c0e357033b999d8f3c23241e4c4053cc573f6ca2.json b/backend/.sqlx/query-ea277c48e7966ab01d9b18e3c0e357033b999d8f3c23241e4c4053cc573f6ca2.json new file mode 100644 index 0000000000..f236c2aae0 --- /dev/null +++ b/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" +} diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index 4b8e1b678b..6f44d24721 100644 --- a/backend/src/monitor.rs +++ b/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" ) diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 423b3357a8..f6fe92c148 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/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>>, ) - .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>>, ) - .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. diff --git a/backend/windmill-queue/tests/debounce_test.rs b/backend/windmill-queue/tests/debounce_test.rs index 6bfa6549c8..df98e5dd7c 100644 --- a/backend/windmill-queue/tests/debounce_test.rs +++ b/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| 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