starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-09-04 08:01:54 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(debounce): claim whole batch in one UPDATE (no deadlock); assert test setup

Browse Source 
Both Codex (P1) and Claude (P2) flagged a deadlock: the claim used two writable
CTEs (claim_self then claim_rest), locking the self row before siblings, so two
survivors of the same batch pulled concurrently acquired row locks in opposite
order and PostgreSQL aborted one with deadlock_detected (a transient pull error
on exactly the two-survivors race this path handles).

Replace with a single `UPDATE ... WHERE debounce_batch = (...) AND consumed_at IS
NULL RETURNING id` that claims the whole batch: both transactions lock rows in
the same scan order, so one simply waits and re-evaluates under EvalPlanQual.
A `claimed_self` flag (EXISTS id = self in the claimed set) plus the `mine`
snapshot still distinguishes fresh-claim / consumed-by-other / own-re-pull.

Also assert add_survivor_to_batch_of actually inserts a row (rows_affected == 1)
so a mis-set-up test can't pass vacuously.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Ruben Fiszel
2026-06-25 16:24:26 +00:00
parent d961c870c3
commit be475a7947
3 changed files with 90 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-a565b2d34b3ca1d513d3fed2e23e8687718879f7e746a78675f121eb9511f6be.json
Generated
+40
View File
@@ -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 ), claimed AS (\n -- Claim the whole batch in ONE update so concurrent same-batch\n -- survivors lock rows in identical scan order (no lock-ordering\n -- deadlock); each re-evaluates `consumed_at IS NULL` under EvalPlanQual\n -- and skips rows the other already took. A claim therefore consumes\n -- every still-unclaimed row of the batch atomically.\n UPDATE v2_job_debounce_batch SET consumed_at = now(), consumed_by = $1\n WHERE debounce_batch = (SELECT debounce_batch FROM mine)\n AND consumed_at IS NULL\n RETURNING id\n )\n SELECT\n EXISTS (SELECT 1 FROM mine) AS \"had_row!\",\n (SELECT consumed_by FROM mine) AS prev_consumed_by,\n ARRAY(SELECT id FROM claimed) AS \"claimed_ids!\",\n EXISTS (SELECT 1 FROM claimed WHERE id = $1) AS \"claimed_self!\"\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "had_row!",
"type_info": "Bool"
},
{
"ordinal": 1,
"name": "prev_consumed_by",
"type_info": "Uuid"
},
{
"ordinal": 2,
"name": "claimed_ids!",
"type_info": "UuidArray"
},
{
"ordinal": 3,
"name": "claimed_self!",
"type_info": "Bool"
}
],
"parameters": {
"Left": [
"Uuid"
]
},
"nullable": [
null,
null,
null,
null
]
},
"hash": "a565b2d34b3ca1d513d3fed2e23e8687718879f7e746a78675f121eb9511f6be"
}
backend/windmill-queue/src/jobs.rs
+42 -44
View File
@@ -3514,66 +3514,38 @@ impl PulledJobResult {
let claim = sqlx::query!(
"WITH mine AS (
SELECT debounce_batch, consumed_by FROM v2_job_debounce_batch WHERE id = $1
), claim_self AS (
), claimed AS (
-- Claim the whole batch in ONE update so concurrent same-batch
-- survivors lock rows in identical scan order (no lock-ordering
-- deadlock); each re-evaluates `consumed_at IS NULL` under EvalPlanQual
-- and skips rows the other already took. A claim therefore consumes
-- every still-unclaimed row of the batch atomically.
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
WHERE debounce_batch = (SELECT debounce_batch FROM mine)
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!\"
ARRAY(SELECT id FROM claimed) AS \"claimed_ids!\",
EXISTS (SELECT 1 FROM claimed WHERE id = $1) AS \"claimed_self!\"
",
j_id,
)
.fetch_one(&mut *tx)
.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(),
to_raw_value(&Vec::<Box<RawValue>>::new()),
);
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(&mut *tx)
.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<Uuid> = Vec::with_capacity(claim.claimed_ids.len() + 1);
ids.push(j_id);
ids.extend(claim.claimed_ids.iter().copied());
} else if claim.claimed_self {
// We claimed our own row; since a claim takes the whole batch, this also
// swept any not-yet-claimed siblings. Accumulate exactly the rows we own.
let ids = claim.claimed_ids;
tracing::debug!(
job_id = %j_id,
@@ -3643,9 +3615,35 @@ impl PulledJobResult {
.await?;
}
}
} else if claim.prev_consumed_by == Some(j_id) {
// Our own prior claim seen again on a re-pull (e.g. crash recovery):
// keep the args we already persisted on the first pull.
} else {
// Another survivor already accumulated this job's contribution
// (consumed_by a different job); 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(),
to_raw_value(&Vec::<Box<RawValue>>::new()),
);
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(&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?;
if let Some(msg) = accumulation_log {
backend/windmill-queue/tests/debounce_test.rs
+8 -1
View File
@@ -4727,7 +4727,7 @@ mod debounce {
.execute(db)
.await
.unwrap();
sqlx::query!(
let inserted = sqlx::query!(
"INSERT INTO v2_job_debounce_batch (id, debounce_batch)
SELECT $1, debounce_batch FROM v2_job_debounce_batch WHERE id = $2",
id,
@@ -4736,6 +4736,13 @@ mod debounce {
.execute(db)
.await
.unwrap();
// `of_job` must already have a batch row, else this no-ops and the test would
// pass vacuously (the job would end up never-batched, keeping its own args).
assert_eq!(
inserted.rows_affected(),
1,
"add_survivor_to_batch_of: {of_job} has no batch row to share"
);
args
}