From 320984b92e02d6b3759341a32afc02353998f4c3 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Thu, 25 Jun 2026 14:16:00 +0000 Subject: [PATCH] feat(debounce): claim-based exactly-once batch consumption Eliminates the rare duplicate/loss when two survivors land on one debounce batch (a narrow push/pull race), without locking the worker pull hot path. - migration: v2_job_debounce_batch gains consumed_at + consumed_by. - pull side (maybe_apply_debouncing): instead of deleting the batch on consume, a survivor atomically claims its own row + any unclaimed siblings (stamping consumed_by = itself) and accumulates exactly the rows it claimed. A second survivor of the same batch finds its row already consumed by another job and runs empty (no duplicate); a re-pulled survivor recognizes its own prior claim and keeps its accumulated args; a never-batched job (CE/legacy) keeps its own args. Non-accumulate debounce paths still hard-delete their batch rows. - complete_debounced_job (EE companion) never completes a running predecessor, so its in-flight run is not killed (no loss); the claim then prevents the duplicate the guard would otherwise allow. - monitor: GC sweep deletes consumed batch rows past a 1h grace. Together with the running-survivor guard this makes debounce accumulation exactly-once. Adds tests: batch_consumed_exactly_once, repull_keeps_accumulated. Co-Authored-By: Claude Opus 4.8 (1M context) --- ...acf623cd6e5fde2d075a4f1f0d4852efcedca.json | 23 ++ ...af401e6d800488ade04facb6af14578c30a89.json | 40 +++ ...9984da751f9ae10c8b15ada925bd371501e18.json | 20 ++ ...bfa8763c03dfe3f0c74edf958667035fb5135.json | 25 ++ ...bcb900ac113c14584fa933d2a94318e47a354.json | 15 ++ ...c2c8b1fc14a8de665b15a2f2dfd6e0b5b7fdf.json | 14 ++ backend/ee-repo-ref.txt | 2 +- ...625135355_debounce_batch_consumed.down.sql | 4 + ...60625135355_debounce_batch_consumed.up.sql | 17 ++ backend/src/monitor.rs | 28 +++ backend/windmill-queue/src/jobs.rs | 234 ++++++++++++------ backend/windmill-queue/tests/debounce_test.rs | 131 ++++++++++ 12 files changed, 472 insertions(+), 81 deletions(-) create mode 100644 backend/.sqlx/query-0d6c3399bc637e2c599534c2cafacf623cd6e5fde2d075a4f1f0d4852efcedca.json create mode 100644 backend/.sqlx/query-57f375e89d63ac118c5c6767487af401e6d800488ade04facb6af14578c30a89.json create mode 100644 backend/.sqlx/query-70f01b322765442de8888b6d9b79984da751f9ae10c8b15ada925bd371501e18.json create mode 100644 backend/.sqlx/query-b9ce07cda12e2540f428d4c8e5bbfa8763c03dfe3f0c74edf958667035fb5135.json create mode 100644 backend/.sqlx/query-bf5db4bba06a2af085577f36446bcb900ac113c14584fa933d2a94318e47a354.json create mode 100644 backend/.sqlx/query-e050734d7642b26f8859982c55ec2c8b1fc14a8de665b15a2f2dfd6e0b5b7fdf.json create mode 100644 backend/migrations/20260625135355_debounce_batch_consumed.down.sql create mode 100644 backend/migrations/20260625135355_debounce_batch_consumed.up.sql diff --git a/backend/.sqlx/query-0d6c3399bc637e2c599534c2cafacf623cd6e5fde2d075a4f1f0d4852efcedca.json b/backend/.sqlx/query-0d6c3399bc637e2c599534c2cafacf623cd6e5fde2d075a4f1f0d4852efcedca.json new file mode 100644 index 0000000000..ee50840b07 --- /dev/null +++ b/backend/.sqlx/query-0d6c3399bc637e2c599534c2cafacf623cd6e5fde2d075a4f1f0d4852efcedca.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT args->>$2 FROM v2_job WHERE id = ANY($1)", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "?column?", + "type_info": "Text" + } + ], + "parameters": { + "Left": [ + "UuidArray", + "Text" + ] + }, + "nullable": [ + null + ] + }, + "hash": "0d6c3399bc637e2c599534c2cafacf623cd6e5fde2d075a4f1f0d4852efcedca" +} diff --git a/backend/.sqlx/query-57f375e89d63ac118c5c6767487af401e6d800488ade04facb6af14578c30a89.json b/backend/.sqlx/query-57f375e89d63ac118c5c6767487af401e6d800488ade04facb6af14578c30a89.json new file mode 100644 index 0000000000..f634fc2d4b --- /dev/null +++ b/backend/.sqlx/query-57f375e89d63ac118c5c6767487af401e6d800488ade04facb6af14578c30a89.json @@ -0,0 +1,40 @@ +{ + "db_name": "PostgreSQL", + "query": "WITH mine AS (\n SELECT debounce_batch, consumed_by FROM v2_job_debounce_batch WHERE id = $1\n ), claim_self AS (\n UPDATE v2_job_debounce_batch SET consumed_at = now(), consumed_by = $1\n WHERE id = $1 AND consumed_at IS NULL\n RETURNING debounce_batch\n ), claim_rest AS (\n UPDATE v2_job_debounce_batch SET consumed_at = now(), consumed_by = $1\n WHERE debounce_batch = (SELECT debounce_batch FROM claim_self)\n AND id <> $1 AND consumed_at IS NULL\n RETURNING id\n )\n SELECT\n EXISTS (SELECT 1 FROM mine) AS \"had_row!\",\n (SELECT debounce_batch FROM claim_self) AS claimed_batch,\n (SELECT consumed_by FROM mine) AS prev_consumed_by,\n ARRAY(SELECT id FROM claim_rest) AS \"claimed_ids!\"\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "had_row!", + "type_info": "Bool" + }, + { + "ordinal": 1, + "name": "claimed_batch", + "type_info": "Int8" + }, + { + "ordinal": 2, + "name": "prev_consumed_by", + "type_info": "Uuid" + }, + { + "ordinal": 3, + "name": "claimed_ids!", + "type_info": "UuidArray" + } + ], + "parameters": { + "Left": [ + "Uuid" + ] + }, + "nullable": [ + null, + null, + null, + null + ] + }, + "hash": "57f375e89d63ac118c5c6767487af401e6d800488ade04facb6af14578c30a89" +} diff --git a/backend/.sqlx/query-70f01b322765442de8888b6d9b79984da751f9ae10c8b15ada925bd371501e18.json b/backend/.sqlx/query-70f01b322765442de8888b6d9b79984da751f9ae10c8b15ada925bd371501e18.json new file mode 100644 index 0000000000..e885c72039 --- /dev/null +++ b/backend/.sqlx/query-70f01b322765442de8888b6d9b79984da751f9ae10c8b15ada925bd371501e18.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 AND consumed_at < now() - interval '1 hour'\n RETURNING 1\n ) SELECT count(*) FROM del", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "count", + "type_info": "Int8" + } + ], + "parameters": { + "Left": [] + }, + "nullable": [ + null + ] + }, + "hash": "70f01b322765442de8888b6d9b79984da751f9ae10c8b15ada925bd371501e18" +} diff --git a/backend/.sqlx/query-b9ce07cda12e2540f428d4c8e5bbfa8763c03dfe3f0c74edf958667035fb5135.json b/backend/.sqlx/query-b9ce07cda12e2540f428d4c8e5bbfa8763c03dfe3f0c74edf958667035fb5135.json new file mode 100644 index 0000000000..fc002b7074 --- /dev/null +++ b/backend/.sqlx/query-b9ce07cda12e2540f428d4c8e5bbfa8763c03dfe3f0c74edf958667035fb5135.json @@ -0,0 +1,25 @@ +{ + "db_name": "PostgreSQL", + "query": "\n WITH target AS (\n SELECT id FROM v2_job_queue WHERE id = $1 AND NOT running FOR UPDATE\n ), completed AS (\n INSERT INTO v2_job_completed\n (workspace_id, id, started_at, duration_ms, result,\n flow_status, workflow_as_code_status, status, worker)\n SELECT\n q.workspace_id, q.id, q.started_at,\n (EXTRACT('epoch' FROM now()) - EXTRACT('epoch' FROM COALESCE(q.started_at, now()))) * 1000,\n $3::text::jsonb,\n s.flow_status,\n s.workflow_as_code_status,\n 'skipped'::job_status,\n q.worker\n FROM v2_job_queue q\n LEFT JOIN v2_job_status s ON s.id = q.id\n WHERE q.id IN (SELECT id FROM target)\n ON CONFLICT (id) DO UPDATE SET status = EXCLUDED.status, result = EXCLUDED.result\n RETURNING 1 AS x\n ), _deleted AS (\n DELETE FROM v2_job_queue WHERE id IN (SELECT id FROM target)\n ), _logged AS (\n INSERT INTO job_logs (logs, job_id, workspace_id)\n SELECT $4, $1, $2 WHERE EXISTS (SELECT 1 FROM target)\n ON CONFLICT (job_id) DO UPDATE SET logs = concat(job_logs.logs, EXCLUDED.logs)\n )\n SELECT x FROM completed\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "x", + "type_info": "Int4" + } + ], + "parameters": { + "Left": [ + "Uuid", + "Varchar", + "Text", + "Text" + ] + }, + "nullable": [ + null + ] + }, + "hash": "b9ce07cda12e2540f428d4c8e5bbfa8763c03dfe3f0c74edf958667035fb5135" +} diff --git a/backend/.sqlx/query-bf5db4bba06a2af085577f36446bcb900ac113c14584fa933d2a94318e47a354.json b/backend/.sqlx/query-bf5db4bba06a2af085577f36446bcb900ac113c14584fa933d2a94318e47a354.json new file mode 100644 index 0000000000..22618ad999 --- /dev/null +++ b/backend/.sqlx/query-bf5db4bba06a2af085577f36446bcb900ac113c14584fa933d2a94318e47a354.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO v2_job_debounce_batch (id, debounce_batch)\n SELECT $1, debounce_batch FROM v2_job_debounce_batch WHERE id = $2", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Uuid", + "Uuid" + ] + }, + "nullable": [] + }, + "hash": "bf5db4bba06a2af085577f36446bcb900ac113c14584fa933d2a94318e47a354" +} diff --git a/backend/.sqlx/query-e050734d7642b26f8859982c55ec2c8b1fc14a8de665b15a2f2dfd6e0b5b7fdf.json b/backend/.sqlx/query-e050734d7642b26f8859982c55ec2c8b1fc14a8de665b15a2f2dfd6e0b5b7fdf.json new file mode 100644 index 0000000000..87cd891113 --- /dev/null +++ b/backend/.sqlx/query-e050734d7642b26f8859982c55ec2c8b1fc14a8de665b15a2f2dfd6e0b5b7fdf.json @@ -0,0 +1,14 @@ +{ + "db_name": "PostgreSQL", + "query": "DELETE FROM v2_job_debounce_batch WHERE debounce_batch = (\n SELECT debounce_batch FROM v2_job_debounce_batch WHERE id = $1\n )", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Uuid" + ] + }, + "nullable": [] + }, + "hash": "e050734d7642b26f8859982c55ec2c8b1fc14a8de665b15a2f2dfd6e0b5b7fdf" +} diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 793a6f9440..8a25dcbd4e 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -aa82e3f706803380b36c7950ff82297a76a71e2e \ No newline at end of file +30d740e619fad219108ec4b4c6a9d67c1ab42d46 \ No newline at end of file diff --git a/backend/migrations/20260625135355_debounce_batch_consumed.down.sql b/backend/migrations/20260625135355_debounce_batch_consumed.down.sql new file mode 100644 index 0000000000..8ce64f87ac --- /dev/null +++ b/backend/migrations/20260625135355_debounce_batch_consumed.down.sql @@ -0,0 +1,4 @@ +DROP INDEX IF EXISTS idx_v2_job_debounce_batch_consumed_at; +ALTER TABLE v2_job_debounce_batch + DROP COLUMN IF EXISTS consumed_at, + DROP COLUMN IF EXISTS consumed_by; diff --git a/backend/migrations/20260625135355_debounce_batch_consumed.up.sql b/backend/migrations/20260625135355_debounce_batch_consumed.up.sql new file mode 100644 index 0000000000..2559cbb74e --- /dev/null +++ b/backend/migrations/20260625135355_debounce_batch_consumed.up.sql @@ -0,0 +1,17 @@ +-- Claim-based, exactly-once consumption of debounce batches. +-- A batch row is "claimed" by the survivor that accumulates its args. Stamping the +-- row consumed (instead of deleting it) lets a later-pulled survivor of the same +-- batch tell "my contribution was already processed" (consumed_at set -> no-op) +-- apart from "I was never batched" (no row at all; CE / legacy -> run my own args). +-- consumed_at doubles as the GC timestamp. NULL = not yet consumed. +-- consumed_by records which job claimed the row, so a job that is re-pulled (e.g. +-- crash recovery) can tell its own prior claim (keep its accumulated args) apart from +-- a sibling survivor having swept it in (run empty). +ALTER TABLE v2_job_debounce_batch + ADD COLUMN IF NOT EXISTS consumed_at TIMESTAMP WITH TIME ZONE, + ADD COLUMN IF NOT EXISTS consumed_by UUID; + +-- Keeps the GC sweep (delete consumed rows past a grace period) cheap. +CREATE INDEX IF NOT EXISTS idx_v2_job_debounce_batch_consumed_at + ON v2_job_debounce_batch (consumed_at) + WHERE consumed_at IS NOT NULL; diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index f3fe7f01ec..df71c1710d 100644 --- a/backend/src/monitor.rs +++ b/backend/src/monitor.rs @@ -2691,6 +2691,9 @@ pub async fn monitor_db( if let Err(e) = cleanup_debounce_orphaned_keys(&db).await { tracing::error!("Error cleaning up debounce keys: {:?}", e); } + if let Err(e) = cleanup_consumed_debounce_batches(&db).await { + tracing::error!("Error cleaning up consumed debounce batches: {:?}", e); + } } } }; @@ -4393,6 +4396,31 @@ RETURNING key,job_id Ok(()) } +/// GC for claim-based debounce batches: once a batch row has been consumed (its +/// args accumulated into some survivor's run), it only lingers to let a later-pulled +/// survivor of the same batch tell "already consumed" from "never batched". A generous +/// grace period (>> any debounce window) makes that decision safe; after it, the rows +/// are dead weight. A re-pulled survivor whose row was GC'd correctly falls back to its +/// own (already-accumulated, persisted) args, so the grace period is not correctness- +/// critical. +async fn cleanup_consumed_debounce_batches(db: &DB) -> error::Result<()> { + 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 '1 hour' + RETURNING 1 + ) SELECT count(*) FROM del" + ) + .fetch_one(db) + .await? + .unwrap_or(0); + + if deleted > 0 { + tracing::info!("Cleaned up {deleted} consumed debounce batch rows"); + } + Ok(()) +} + async fn cleanup_debounce_keys_for_completed_jobs(db: &DB) -> error::Result<()> { // If min version doesn't support runnable settings, clean up debounce keys for completed jobs if !windmill_common::min_version::MIN_VERSION_SUPPORTS_RUNNABLE_SETTINGS_V0 diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 5cb8911c7e..423b3357a8 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -3172,87 +3172,77 @@ impl PulledJobResult { if let Some(args) = &mut j.args { args.remove(field_name); } + + // No accumulation on this path: just clean up the batch rows. + sqlx::query!( + "DELETE FROM v2_job_debounce_batch WHERE debounce_batch = ( + SELECT debounce_batch FROM v2_job_debounce_batch WHERE id = $1 + )", + j_id, + ) + .execute(db) + .await?; } else if let Some(arg_name_to_accumulate) = // TODO: Maybe support multiple arguments in future debounce_args_to_accumulate.as_ref().and_then(|v| v.get(0)) { - tracing::debug!( - job_id = %j_id, - job_kind = ?kind, - arg_name = arg_name_to_accumulate, - "Accumulating debounced arguments from batch" - ); - let mut accumulated_arg: Vec> = vec![]; - for str_o in sqlx::query_scalar!( - "WITH ids AS ( - SELECT id as job_id FROM v2_job_debounce_batch WHERE debounce_batch = ( - SELECT debounce_batch FROM v2_job_debounce_batch WHERE id = $1 - ) - ) SELECT args->>$2 FROM ids LEFT JOIN v2_job ON v2_job.id = ids.job_id + // Claim this job's contribution to its debounce batch exactly once. + // Instead of deleting the batch rows, mark them consumed (stamping + // consumed_by = this job). A batch normally has a single survivor that + // sweeps every row; only a narrow push/pull race can leave two survivors + // on one batch. The claim lets the second survivor tell apart: + // - already swept in by the other survivor -> run empty (no duplicate), + // - 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. + let claim = sqlx::query!( + "WITH mine AS ( + SELECT debounce_batch, consumed_by FROM v2_job_debounce_batch WHERE id = $1 + ), claim_self AS ( + UPDATE v2_job_debounce_batch SET consumed_at = now(), consumed_by = $1 + WHERE id = $1 AND consumed_at IS NULL + RETURNING debounce_batch + ), claim_rest AS ( + UPDATE v2_job_debounce_batch SET consumed_at = now(), consumed_by = $1 + WHERE debounce_batch = (SELECT debounce_batch FROM claim_self) + AND id <> $1 AND consumed_at IS NULL + RETURNING id + ) + SELECT + EXISTS (SELECT 1 FROM mine) AS \"had_row!\", + (SELECT debounce_batch FROM claim_self) AS claimed_batch, + (SELECT consumed_by FROM mine) AS prev_consumed_by, + ARRAY(SELECT id FROM claim_rest) AS \"claimed_ids!\" ", j_id, - arg_name_to_accumulate, ) - .fetch_all(db) - .await? - .into_iter() - { - if let Some(s) = str_o.as_ref() { - match serde_json::from_str::>>(s) { - Ok(ref mut vec) => accumulated_arg.append(vec), - Err(_) => { - // Value is not an array — wrap the scalar into a - // single-element array. This supports union types - // like T | T[] where the caller may pass a bare T. - match RawValue::from_string(s.to_string()) { - Ok(raw) => accumulated_arg.push(raw), - Err(e) => { - return Err(error::Error::ArgumentErr(format!("cannot consolidate argument `{arg_name_to_accumulate}`: value is neither a valid list nor a valid JSON value\nUnwrapped Error: {e}"))); - } - } - } - } - } - } + .fetch_one(db) + .await?; - tracing::debug!( - job_id = %j_id, - arg_name = arg_name_to_accumulate, - accumulated_count = accumulated_arg.len(), - "Accumulated arguments from debounced jobs in batch" - ); - - // If the batch query returned no entries (e.g. CE where - // v2_job_debounce_batch is never populated), keep the - // original value unchanged instead of replacing it with []. - if !accumulated_arg.is_empty() { - 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!( - "Accumulating debounced argument `{arg_name_to_accumulate}`:\n original: {original_value}\n accumulated: {}\n\n", - &new_value - ), - &(db.into()), - ) - .await; + let consumed_by_other = + claim.claimed_batch.is_none() && claim.prev_consumed_by != Some(j_id); + if !claim.had_row { + // Never batched (CE / workers behind v2): keep the job's own args. + tracing::debug!( + job_id = %j_id, + "Debounce: no batch row, keeping original args" + ); + } else if consumed_by_other { + // Another survivor of this batch already accumulated this job's + // contribution; run as a no-op so its items are not reprocessed. + tracing::info!( + job_id = %j_id, + arg_name = arg_name_to_accumulate, + "Debounce: contribution already consumed by a concurrent survivor, running empty" + ); j.args .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 + .insert( + arg_name_to_accumulate.to_owned(), + to_raw_value(&Vec::>::new()), + ); if let Some(ref args) = j.args { sqlx::query!( "UPDATE v2_job SET args = $2 WHERE id = $1", @@ -3262,18 +3252,102 @@ impl PulledJobResult { .execute(db) .await?; } - } - } + } else if claim.claimed_batch.is_some() { + // We claimed our own row (+ any unclaimed siblings): accumulate the + // args of exactly the rows we own. + let mut ids: Vec = Vec::with_capacity(claim.claimed_ids.len() + 1); + ids.push(j_id); + ids.extend(claim.claimed_ids.iter().copied()); - // Clean up the debounce batch entries now that the job has been pulled - sqlx::query!( - "DELETE FROM v2_job_debounce_batch WHERE debounce_batch = ( - SELECT debounce_batch FROM v2_job_debounce_batch WHERE id = $1 - )", - j_id, - ) - .execute(db) - .await?; + tracing::debug!( + job_id = %j_id, + job_kind = ?kind, + arg_name = arg_name_to_accumulate, + claimed = ids.len(), + "Accumulating debounced arguments from claimed batch rows" + ); + + let mut accumulated_arg: Vec> = vec![]; + for str_o in sqlx::query_scalar!( + "SELECT args->>$2 FROM v2_job WHERE id = ANY($1)", + &ids, + arg_name_to_accumulate, + ) + .fetch_all(db) + .await? + .into_iter() + { + if let Some(s) = str_o.as_ref() { + match serde_json::from_str::>>(s) { + Ok(ref mut vec) => accumulated_arg.append(vec), + Err(_) => { + // Value is not an array — wrap the scalar into a + // single-element array. This supports union types + // like T | T[] where the caller may pass a bare T. + match RawValue::from_string(s.to_string()) { + Ok(raw) => accumulated_arg.push(raw), + Err(e) => { + return Err(error::Error::ArgumentErr(format!("cannot consolidate argument `{arg_name_to_accumulate}`: value is neither a valid list nor a valid JSON value\nUnwrapped Error: {e}"))); + } + } + } + } + } + } + + if !accumulated_arg.is_empty() { + 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!( + "Accumulating debounced argument `{arg_name_to_accumulate}`:\n original: {original_value}\n accumulated: {}\n\n", + &new_value + ), + &(db.into()), + ) + .await; + + j.args + .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>>, + ) + .execute(db) + .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. + } else { + // Debounced but no args to accumulate (plain debounce / dependency job): + // consume the batch by removing this job's rows. + sqlx::query!( + "DELETE FROM v2_job_debounce_batch WHERE debounce_batch = ( + SELECT debounce_batch FROM v2_job_debounce_batch WHERE id = $1 + )", + j_id, + ) + .execute(db) + .await?; + } // Handle dependency job debouncing cleanup when a job is pulled for execution if is_djob_to_debounce { diff --git a/backend/windmill-queue/tests/debounce_test.rs b/backend/windmill-queue/tests/debounce_test.rs index 23b1e9e98b..996dbd51cd 100644 --- a/backend/windmill-queue/tests/debounce_test.rs +++ b/backend/windmill-queue/tests/debounce_test.rs @@ -4575,6 +4575,137 @@ mod debounce { Ok(()) } + /// Claim-based exactly-once: if two survivors end up on the same batch (only + /// possible in a narrow push/pull race), the args of each member are accumulated + /// into exactly ONE run. The survivor that claims the batch first accumulates + /// everyone; the second survivor finds its contribution already consumed and runs + /// empty — no item is dropped and none is processed twice. + #[sqlx::test(migrations = "../migrations", fixtures("base"))] + async fn test_debounce_batch_consumed_exactly_once(db: Pool) -> anyhow::Result<()> { + let key = "exactly_once_key"; + let settings = DebouncingSettings { + debounce_delay_s: Some(5), + debounce_key: Some(key.to_string()), + debounce_args_to_accumulate: Some(vec!["items".to_string()]), + ..Default::default() + }; + let rs_handle = setup_debouncing_settings(&db, &settings).await; + + // J1 superseded, J2 the (first) survivor of batch B = {J1, J2}. + let j1 = Uuid::new_v4(); + let j2 = Uuid::new_v4(); + push_debounced_script(&db, j1, vec![1], &settings, rs_handle).await; + let j2_args = push_debounced_script(&db, j2, vec![2], &settings, rs_handle).await; + + // Simulate the race outcome: a second survivor J3 ended up on the SAME batch B. + let j3 = Uuid::new_v4(); + let j3_args = serde_json::json!({ "items": [3] }); + insert_script_job_with_args(&db, j3, "test-workspace", "f/test/script", &j3_args).await; + sqlx::query!( + "UPDATE v2_job_queue SET runnable_settings_handle = $1 WHERE id = $2", + rs_handle, + j3, + ) + .execute(&db) + .await?; + sqlx::query!( + "INSERT INTO v2_job_debounce_batch (id, debounce_batch) + SELECT $1, debounce_batch FROM v2_job_debounce_batch WHERE id = $2", + j3, + j2, + ) + .execute(&db) + .await?; + + // J2 pulled first: claims the whole batch, accumulates everyone's items. + let mut j2_res = make_pulled_job_result( + j2, + "test-workspace", + "f/test/script", + &j2_args, + JobKind::Script, + "deno", + rs_handle, + ); + j2_res.maybe_apply_debouncing(&db).await?; + assert!(j2_res.job.is_some(), "J2 runs"); + assert_accumulated_items(&j2_res, &[1, 2, 3], "items"); + + // J3 pulled next: its contribution was already consumed by J2 -> runs empty, + // so [3] is not processed a second time. + let mut j3_res = make_pulled_job_result( + j3, + "test-workspace", + "f/test/script", + &j3_args, + JobKind::Script, + "deno", + rs_handle, + ); + j3_res.maybe_apply_debouncing(&db).await?; + let job = j3_res.job.as_ref().expect("J3 still runs (empty)"); + let items: Vec = + serde_json::from_str(job.job.args.as_ref().unwrap().get("items").unwrap().get())?; + assert!( + items.is_empty(), + "J3's items must be empty (already consumed by J2), got {items:?}" + ); + + Ok(()) + } + + /// A survivor re-pulled (e.g. crash recovery) must NOT mistake its own earlier + /// claim for a sibling's and wipe its accumulated args. consumed_by = self is + /// distinguished from consumed_by = another job. + #[sqlx::test(migrations = "../migrations", fixtures("base"))] + async fn test_debounce_repull_keeps_accumulated(db: Pool) -> anyhow::Result<()> { + let key = "repull_key"; + let settings = DebouncingSettings { + debounce_delay_s: Some(5), + debounce_key: Some(key.to_string()), + debounce_args_to_accumulate: Some(vec!["items".to_string()]), + ..Default::default() + }; + let rs_handle = setup_debouncing_settings(&db, &settings).await; + + let j1 = Uuid::new_v4(); + let j2 = Uuid::new_v4(); + push_debounced_script(&db, j1, vec![1], &settings, rs_handle).await; + let j2_args = push_debounced_script(&db, j2, vec![2], &settings, rs_handle).await; + + // First pull: J2 claims its batch and accumulates [1, 2]. + let mut first = make_pulled_job_result( + j2, + "test-workspace", + "f/test/script", + &j2_args, + JobKind::Script, + "deno", + rs_handle, + ); + first.maybe_apply_debouncing(&db).await?; + assert_accumulated_items(&first, &[1, 2], "items"); + + // Re-pull with the args persisted by the first pull: J2 sees its OWN prior claim + // (consumed_by = j2), so it keeps the accumulated args rather than running empty. + let persisted = first.job.as_ref().unwrap().job.args.as_ref().unwrap(); + let persisted_json = serde_json::to_value(persisted).unwrap(); + let mut second = make_pulled_job_result( + j2, + "test-workspace", + "f/test/script", + &persisted_json, + JobKind::Script, + "deno", + rs_handle, + ); + second.maybe_apply_debouncing(&db).await?; + assert!(second.job.is_some(), "re-pulled J2 still runs"); + assert_accumulated_items(&second, &[1, 2], "items"); + + Ok(()) + } + /// Test: Push-time (script) debounce with max_total_debounces_amount=2. /// 5 calls, each sending {x: [i]}. Expected: /// Call 1: debounced (scheduled_for set)