diff --git a/backend/.sqlx/query-a565b2d34b3ca1d513d3fed2e23e8687718879f7e746a78675f121eb9511f6be.json b/backend/.sqlx/query-a565b2d34b3ca1d513d3fed2e23e8687718879f7e746a78675f121eb9511f6be.json new file mode 100644 index 0000000000..a8dc380f40 --- /dev/null +++ b/backend/.sqlx/query-a565b2d34b3ca1d513d3fed2e23e8687718879f7e746a78675f121eb9511f6be.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 ), 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" +} diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 7927dae6de..1e2605e742 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -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::>::new()), - ); - if let Some(ref args) = j.args { - sqlx::query!( - "UPDATE v2_job SET args = $2 WHERE id = $1", - j_id, - args as &Json>>, - ) - .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 = 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::>::new()), + ); + if let Some(ref args) = j.args { + sqlx::query!( + "UPDATE v2_job SET args = $2 WHERE id = $1", + j_id, + args as &Json>>, + ) + .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 { diff --git a/backend/windmill-queue/tests/debounce_test.rs b/backend/windmill-queue/tests/debounce_test.rs index cee509a46b..d74f20ff3f 100644 --- a/backend/windmill-queue/tests/debounce_test.rs +++ b/backend/windmill-queue/tests/debounce_test.rs @@ -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 }