From 0c4ba0b788c5697541580bccfc8b82edbbfae974 Mon Sep 17 00:00:00 2001 From: Brian Thorne Date: Fri, 6 Nov 2026 15:23:01 +1300 Subject: [PATCH 1/2] fix(queue-storage): report Blocked when any prune statement times out on a lock Prune and terminal-count rollup set a transaction-local lock_timeout before taking ACCESS EXCLUSIVE on a slot's partitions, but only the LOCK TABLE and TRUNCATE steps mapped a timeout to Blocked. The timeout also applies to the statements after them: the post-lock liveness checks over the lease and claim partitions, the failed-row carry into the current slot's done_entries and terminal-count deltas, the rollup delta append and the live-counter cleanup. A wait there surfaced as a 55P03 database error, was logged as a prune error and skipped the prune backoff. Run everything after the timeout is set as one block and settle it in one place: commit on success, and on a lock timeout from any statement roll back and report Blocked. Lease and claim prune use the same helper so the rule holds for every ring. The commit duration in PruneDurations is still measured around the commit. The new tests hold a SHARE lock on each relation the queue prune writes after its child locks, and on queue_terminal_live_counts for the rollup; both fail with 55P03 before this change. --- CHANGELOG.md | 17 + awa-model/src/queue_storage.rs | 1384 +++++++++++------------ awa/tests/queue_storage_runtime_test.rs | 169 +++ 3 files changed, 864 insertions(+), 706 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 065d50db1..bd61055e0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,23 @@ Notable changes between releases. Detailed migration notes for storage transitio ### Fixed +- **Ring prune and terminal-count rollup report `Blocked` when any of their + statements times out on a lock.** Prune sets a short `lock_timeout` (10 ms + by default) before it takes `ACCESS EXCLUSIVE` on a slot's partitions, but + only those `LOCK TABLE` statements and the final `TRUNCATE` turned a + timeout into `PruneOutcome::Blocked`. The timeout covers every later + statement in the transaction too: the post-lock liveness checks read the + lease and claim partitions, the failed-row carry inserts into the current + slot's `done_entries` and terminal-count deltas, and the fold appends + `queue_terminal_rollup_deltas` and deletes `queue_terminal_live_counts` + rows. When one of those waited out the timeout, for example behind + autovacuum truncating a relation or a hosted Postgres's own background + workers, prune failed with a `lock_not_available` (`55P03`) database error, + was logged as an error and skipped the prune backoff. A lock timeout + anywhere in a queue, lease or claim prune, or in a terminal-count rollup + slot, now rolls the whole transaction back and reports `Blocked` + (`blocked_slots` for rollup), so maintenance backs off and retries; nothing + the attempt wrote survives the rollback. - **Batch enqueues take their lane locks in one order, so they no longer deadlock with each other.** On queue storage every enqueue holds a transaction-scoped advisory lock on its `(queue, priority, shard)` lane diff --git a/awa-model/src/queue_storage.rs b/awa-model/src/queue_storage.rs index 339001705..bbe2e0fe9 100644 --- a/awa-model/src/queue_storage.rs +++ b/awa-model/src/queue_storage.rs @@ -713,6 +713,35 @@ async fn set_prune_lock_timeout_tx( Ok(()) } +/// Commit or roll back a maintenance transaction that has set the prune +/// `lock_timeout`. +/// +/// The timeout applies to every later statement in the transaction, not only +/// the explicit `LOCK TABLE`: an insert, update or read of another relation can +/// also queue behind a conflicting lock (autovacuum's heap truncation takes +/// `ACCESS EXCLUSIVE`, for example). A lock timeout anywhere therefore rolls the +/// whole transaction back and reports `blocked`, so the caller backs off and +/// retries instead of surfacing an error. +async fn settle_prune_tx( + tx: sqlx::Transaction<'_, Postgres>, + result: Result, + blocked: T, +) -> Result { + match result { + Ok(outcome) => { + tx.commit().await.map_err(map_sqlx_error)?; + Ok(outcome) + } + Err(err) => { + let _ = tx.rollback().await; + match err { + AwaError::Database(ref db_err) if is_lock_contention_error(db_err) => Ok(blocked), + err => Err(err), + } + } + } +} + fn validate_ident(ident: &str) -> Result<(), AwaError> { let mut chars = ident.chars(); match chars.next() { @@ -18350,237 +18379,215 @@ impl QueueStorage { return Ok(TerminalDeltaSlotRollup::SkippedActive); } - set_prune_lock_timeout_tx(&mut tx, self.prune_lock_timeout).await?; - - let lock_delta = sqlx::query(audited_sql(format!( - "LOCK TABLE {delta_child} IN ACCESS EXCLUSIVE MODE" - ))) - .execute(tx.as_mut()) - .await; + let locked: Result = async { + set_prune_lock_timeout_tx(&mut tx, self.prune_lock_timeout).await?; - match lock_delta { - Ok(_) => {} - Err(err) if is_lock_contention_error(&err) => { - let _ = tx.rollback().await; - return Ok(TerminalDeltaSlotRollup::Blocked); - } - Err(err) => { - let _ = tx.rollback().await; - return Err(map_sqlx_error(err)); - } - } - - let active_leases = - queue_prune_has_active_leases_tx(&mut tx, schema, slot, generation).await?; - - if active_leases { - tx.commit().await.map_err(map_sqlx_error)?; - return Ok(TerminalDeltaSlotRollup::SkippedActive); - } + sqlx::query(audited_sql(format!( + "LOCK TABLE {delta_child} IN ACCESS EXCLUSIVE MODE" + ))) + .execute(tx.as_mut()) + .await + .map_err(map_sqlx_error)?; - let unclosed_claim_refs = - queue_prune_has_unclosed_claim_refs_tx(&mut tx, schema, slot, generation).await?; + let active_leases = + queue_prune_has_active_leases_tx(&mut tx, schema, slot, generation).await?; - if unclosed_claim_refs { - tx.commit().await.map_err(map_sqlx_error)?; - return Ok(TerminalDeltaSlotRollup::SkippedActive); - } + if active_leases { + return Ok(TerminalDeltaSlotRollup::SkippedActive); + } - let delta_rows: i64 = sqlx::query_scalar(audited_sql(format!( - r#" - SELECT count(*)::bigint - FROM {delta_child} - WHERE ready_generation = $1 - "# - ))) - .bind(generation) - .fetch_one(tx.as_mut()) - .await - .map_err(map_sqlx_error)?; + let unclosed_claim_refs = + queue_prune_has_unclosed_claim_refs_tx(&mut tx, schema, slot, generation).await?; - if delta_rows == 0 { - tx.commit().await.map_err(map_sqlx_error)?; - return Ok(TerminalDeltaSlotRollup::Empty); - } + if unclosed_claim_refs { + return Ok(TerminalDeltaSlotRollup::SkippedActive); + } - // In ledger authority each touched lane collapses to one bucket-0 row - // holding its old total (summed over every bucket) plus the signed - // delta, which keeps a negative delta exact when the matching positive - // count sits in another bucket. Before the flip, N-1 maintenance can - // still win the election and folds per job-id bucket, so the fold - // keeps that shape until the flip fences it (ADR-041). - let grouped_keys: i64 = match authority { - RingAuthority::Ledger => sqlx::query_scalar(audited_sql(format!( + let delta_rows: i64 = sqlx::query_scalar(audited_sql(format!( r#" - WITH grouped AS MATERIALIZED ( - SELECT - ready_slot, - queue, - priority, - enqueue_shard, - SUM(terminal_delta)::bigint AS delta + SELECT count(*)::bigint FROM {delta_child} WHERE ready_generation = $1 - GROUP BY ready_slot, queue, priority, enqueue_shard - HAVING SUM(terminal_delta) <> 0 - ), - collapsed AS ( - DELETE FROM {schema}.queue_terminal_live_counts AS counts - USING grouped - WHERE counts.ready_slot = grouped.ready_slot - AND counts.queue = grouped.queue - AND counts.priority = grouped.priority - AND counts.enqueue_shard = grouped.enqueue_shard - RETURNING - counts.ready_slot, - counts.queue, - counts.priority, - counts.enqueue_shard, - counts.live_terminal_count - ), - folded AS ( - SELECT - grouped.ready_slot, - grouped.queue, - grouped.priority, - grouped.enqueue_shard, - (grouped.delta + COALESCE(SUM(collapsed.live_terminal_count), 0))::bigint - AS live_terminal_count - FROM grouped - LEFT JOIN collapsed - ON collapsed.ready_slot = grouped.ready_slot - AND collapsed.queue = grouped.queue - AND collapsed.priority = grouped.priority - AND collapsed.enqueue_shard = grouped.enqueue_shard - GROUP BY - grouped.ready_slot, - grouped.queue, - grouped.priority, - grouped.enqueue_shard, - grouped.delta - ), - inserted AS ( - INSERT INTO {schema}.queue_terminal_live_counts ( - ready_slot, - queue, - priority, - enqueue_shard, - counter_bucket, - live_terminal_count - ) - SELECT - ready_slot, - queue, - priority, - enqueue_shard, - 0::smallint, - live_terminal_count - FROM folded - WHERE live_terminal_count <> 0 - ORDER BY ready_slot, queue, priority, enqueue_shard - RETURNING 1 - ) - SELECT count(*)::bigint FROM grouped - "# - ))) - .bind(generation) - .fetch_one(tx.as_mut()) - .await - .map_err(map_sqlx_error)?, - RingAuthority::Columns => sqlx::query_scalar(audited_sql(format!( - r#" - WITH grouped AS MATERIALIZED ( - SELECT - ready_slot, - queue, - priority, - enqueue_shard, - counter_bucket, - SUM(terminal_delta)::bigint AS delta - FROM {delta_child} - WHERE ready_generation = $1 - GROUP BY ready_slot, queue, priority, enqueue_shard, counter_bucket - HAVING SUM(terminal_delta) <> 0 - ), - updated AS ( - UPDATE {schema}.queue_terminal_live_counts AS counts - SET live_terminal_count = GREATEST(0, counts.live_terminal_count + grouped.delta) - FROM grouped - WHERE counts.ready_slot = grouped.ready_slot - AND counts.queue = grouped.queue - AND counts.priority = grouped.priority - AND counts.enqueue_shard = grouped.enqueue_shard - AND counts.counter_bucket = grouped.counter_bucket - RETURNING - counts.ready_slot, - counts.queue, - counts.priority, - counts.enqueue_shard, - counts.counter_bucket - ), - inserted AS ( - INSERT INTO {schema}.queue_terminal_live_counts AS counts ( - ready_slot, - queue, - priority, - enqueue_shard, - counter_bucket, - live_terminal_count - ) - SELECT - grouped.ready_slot, - grouped.queue, - grouped.priority, - grouped.enqueue_shard, - grouped.counter_bucket, - grouped.delta - FROM grouped - WHERE grouped.delta > 0 - AND NOT EXISTS ( - SELECT 1 - FROM updated - WHERE updated.ready_slot = grouped.ready_slot - AND updated.queue = grouped.queue - AND updated.priority = grouped.priority - AND updated.enqueue_shard = grouped.enqueue_shard - AND updated.counter_bucket = grouped.counter_bucket - ) - ORDER BY ready_slot, queue, priority, enqueue_shard, counter_bucket - ON CONFLICT (ready_slot, queue, priority, enqueue_shard, counter_bucket) DO UPDATE - SET live_terminal_count = - GREATEST(0, counts.live_terminal_count + EXCLUDED.live_terminal_count) - RETURNING 1 - ) - SELECT count(*)::bigint FROM grouped - "# + "# ))) .bind(generation) .fetch_one(tx.as_mut()) .await - .map_err(map_sqlx_error)?, - }; - - let truncate_delta = sqlx::query(audited_sql(format!("TRUNCATE TABLE {delta_child}"))) - .execute(tx.as_mut()) - .await; + .map_err(map_sqlx_error)?; - match truncate_delta { - Ok(_) => { - tx.commit().await.map_err(map_sqlx_error)?; - Ok(TerminalDeltaSlotRollup::Rolled { - delta_rows, - grouped_keys, - }) - } - Err(err) if is_lock_contention_error(&err) => { - let _ = tx.rollback().await; - Ok(TerminalDeltaSlotRollup::Blocked) - } - Err(err) => { - let _ = tx.rollback().await; - Err(map_sqlx_error(err)) + if delta_rows == 0 { + return Ok(TerminalDeltaSlotRollup::Empty); } + + // In ledger authority each touched lane collapses to one bucket-0 row + // holding its old total (summed over every bucket) plus the signed + // delta, which keeps a negative delta exact when the matching positive + // count sits in another bucket. Before the flip, N-1 maintenance can + // still win the election and folds per job-id bucket, so the fold + // keeps that shape until the flip fences it (ADR-041). + let grouped_keys: i64 = match authority { + RingAuthority::Ledger => sqlx::query_scalar(audited_sql(format!( + r#" + WITH grouped AS MATERIALIZED ( + SELECT + ready_slot, + queue, + priority, + enqueue_shard, + SUM(terminal_delta)::bigint AS delta + FROM {delta_child} + WHERE ready_generation = $1 + GROUP BY ready_slot, queue, priority, enqueue_shard + HAVING SUM(terminal_delta) <> 0 + ), + collapsed AS ( + DELETE FROM {schema}.queue_terminal_live_counts AS counts + USING grouped + WHERE counts.ready_slot = grouped.ready_slot + AND counts.queue = grouped.queue + AND counts.priority = grouped.priority + AND counts.enqueue_shard = grouped.enqueue_shard + RETURNING + counts.ready_slot, + counts.queue, + counts.priority, + counts.enqueue_shard, + counts.live_terminal_count + ), + folded AS ( + SELECT + grouped.ready_slot, + grouped.queue, + grouped.priority, + grouped.enqueue_shard, + (grouped.delta + COALESCE(SUM(collapsed.live_terminal_count), 0))::bigint + AS live_terminal_count + FROM grouped + LEFT JOIN collapsed + ON collapsed.ready_slot = grouped.ready_slot + AND collapsed.queue = grouped.queue + AND collapsed.priority = grouped.priority + AND collapsed.enqueue_shard = grouped.enqueue_shard + GROUP BY + grouped.ready_slot, + grouped.queue, + grouped.priority, + grouped.enqueue_shard, + grouped.delta + ), + inserted AS ( + INSERT INTO {schema}.queue_terminal_live_counts ( + ready_slot, + queue, + priority, + enqueue_shard, + counter_bucket, + live_terminal_count + ) + SELECT + ready_slot, + queue, + priority, + enqueue_shard, + 0::smallint, + live_terminal_count + FROM folded + WHERE live_terminal_count <> 0 + ORDER BY ready_slot, queue, priority, enqueue_shard + RETURNING 1 + ) + SELECT count(*)::bigint FROM grouped + "# + ))) + .bind(generation) + .fetch_one(tx.as_mut()) + .await + .map_err(map_sqlx_error)?, + RingAuthority::Columns => sqlx::query_scalar(audited_sql(format!( + r#" + WITH grouped AS MATERIALIZED ( + SELECT + ready_slot, + queue, + priority, + enqueue_shard, + counter_bucket, + SUM(terminal_delta)::bigint AS delta + FROM {delta_child} + WHERE ready_generation = $1 + GROUP BY ready_slot, queue, priority, enqueue_shard, counter_bucket + HAVING SUM(terminal_delta) <> 0 + ), + updated AS ( + UPDATE {schema}.queue_terminal_live_counts AS counts + SET live_terminal_count = GREATEST(0, counts.live_terminal_count + grouped.delta) + FROM grouped + WHERE counts.ready_slot = grouped.ready_slot + AND counts.queue = grouped.queue + AND counts.priority = grouped.priority + AND counts.enqueue_shard = grouped.enqueue_shard + AND counts.counter_bucket = grouped.counter_bucket + RETURNING + counts.ready_slot, + counts.queue, + counts.priority, + counts.enqueue_shard, + counts.counter_bucket + ), + inserted AS ( + INSERT INTO {schema}.queue_terminal_live_counts AS counts ( + ready_slot, + queue, + priority, + enqueue_shard, + counter_bucket, + live_terminal_count + ) + SELECT + grouped.ready_slot, + grouped.queue, + grouped.priority, + grouped.enqueue_shard, + grouped.counter_bucket, + grouped.delta + FROM grouped + WHERE grouped.delta > 0 + AND NOT EXISTS ( + SELECT 1 + FROM updated + WHERE updated.ready_slot = grouped.ready_slot + AND updated.queue = grouped.queue + AND updated.priority = grouped.priority + AND updated.enqueue_shard = grouped.enqueue_shard + AND updated.counter_bucket = grouped.counter_bucket + ) + ORDER BY ready_slot, queue, priority, enqueue_shard, counter_bucket + ON CONFLICT (ready_slot, queue, priority, enqueue_shard, counter_bucket) DO UPDATE + SET live_terminal_count = + GREATEST(0, counts.live_terminal_count + EXCLUDED.live_terminal_count) + RETURNING 1 + ) + SELECT count(*)::bigint FROM grouped + "# + ))) + .bind(generation) + .fetch_one(tx.as_mut()) + .await + .map_err(map_sqlx_error)?, + }; + + sqlx::query(audited_sql(format!("TRUNCATE TABLE {delta_child}"))) + .execute(tx.as_mut()) + .await + .map_err(map_sqlx_error)?; + Ok(TerminalDeltaSlotRollup::Rolled { + delta_rows, + grouped_keys, + }) } + .await; + + settle_prune_tx(tx, locked, TerminalDeltaSlotRollup::Blocked).await } /// Fold pending `queue_terminal_rollup_deltas` rows into the @@ -18830,172 +18837,119 @@ impl QueueStorage { }); } - set_prune_lock_timeout_tx(&mut tx, self.prune_lock_timeout).await?; + let locked: Result = async { + set_prune_lock_timeout_tx(&mut tx, self.prune_lock_timeout).await?; - let lock_started = Instant::now(); - // Claims read segments before ready rows, while producers write in - // the opposite order. Queue behind segment readers, then abandon - // the transaction rather than wait with an exclusive child held. - let lock_tables = async { - sqlx::query(audited_sql(format!( - "LOCK TABLE {segment_child} IN ACCESS EXCLUSIVE MODE" - ))) - .execute(tx.as_mut()) - .await?; - sqlx::query(audited_sql(format!( - "LOCK TABLE {ready_child}, {claim_attempt_child}, {done_child}, {tomb_child}, {receipt_batch_child}, {receipt_tomb_child}, {delta_child} IN ACCESS EXCLUSIVE MODE NOWAIT" - ))) - .execute(tx.as_mut()) + let lock_started = Instant::now(); + // Claims read segments before ready rows, while producers write in + // the opposite order. Queue behind segment readers, then abandon + // the transaction rather than wait with an exclusive child held. + async { + sqlx::query(audited_sql(format!( + "LOCK TABLE {segment_child} IN ACCESS EXCLUSIVE MODE" + ))) + .execute(tx.as_mut()) + .await?; + sqlx::query(audited_sql(format!( + "LOCK TABLE {ready_child}, {claim_attempt_child}, {done_child}, {tomb_child}, {receipt_batch_child}, {receipt_tomb_child}, {delta_child} IN ACCESS EXCLUSIVE MODE NOWAIT" + ))) + .execute(tx.as_mut()) + .await + } .await - } - .await; - let lock_duration = lock_started.elapsed(); - - if let Err(err) = lock_tables { - let _ = tx.rollback().await; - if is_lock_contention_error(&err) { - return Ok(PruneOutcome::Blocked { slot }); + .map_err(map_sqlx_error)?; + let lock_duration = lock_started.elapsed(); + + // The pre-lock gates are only an MVCC snapshot. If an in-flight + // producer or claimer committed while the exclusive partition + // lock waited, these post-lock gates see it before TRUNCATE. + let active_leases = + queue_prune_has_active_leases_tx(&mut tx, schema, slot, generation).await?; + if active_leases { + return Ok(PruneOutcome::SkippedActive { + slot, + reason: SkipReason::QueueActiveLeases, + count: busy_indicator(active_leases), + }); } - return Err(map_sqlx_error(err)); - } - // The pre-lock gates are only an MVCC snapshot. If an in-flight - // producer or claimer committed while the exclusive partition - // lock waited, these post-lock gates see it before TRUNCATE. - let active_leases = - queue_prune_has_active_leases_tx(&mut tx, schema, slot, generation).await?; - if active_leases { - tx.commit().await.map_err(map_sqlx_error)?; - return Ok(PruneOutcome::SkippedActive { - slot, - reason: SkipReason::QueueActiveLeases, - count: busy_indicator(active_leases), - }); - } - - let pending = queue_prune_has_pending_ready_tx(&mut tx, schema, slot, generation).await?; - if pending { - tx.commit().await.map_err(map_sqlx_error)?; - return Ok(PruneOutcome::SkippedActive { - slot, - reason: SkipReason::QueuePendingReady, - count: busy_indicator(pending), - }); - } + let pending = queue_prune_has_pending_ready_tx(&mut tx, schema, slot, generation).await?; + if pending { + return Ok(PruneOutcome::SkippedActive { + slot, + reason: SkipReason::QueuePendingReady, + count: busy_indicator(pending), + }); + } - let unclosed_claim_refs = - queue_prune_has_unclosed_claim_refs_tx(&mut tx, schema, slot, generation).await?; - if unclosed_claim_refs { - tx.commit().await.map_err(map_sqlx_error)?; - return Ok(PruneOutcome::SkippedActive { - slot, - reason: SkipReason::QueueUnclosedClaimRefs, - count: busy_indicator(unclosed_claim_refs), - }); - } + let unclosed_claim_refs = + queue_prune_has_unclosed_claim_refs_tx(&mut tx, schema, slot, generation).await?; + if unclosed_claim_refs { + return Ok(PruneOutcome::SkippedActive { + slot, + reason: SkipReason::QueueUnclosedClaimRefs, + count: busy_indicator(unclosed_claim_refs), + }); + } - let failed_retention_secs = i64::try_from(failed_retention.as_secs()).unwrap_or(i64::MAX); - let retention_floor = failed_retention_secs > 0; - - // #337: failed terminal rows still inside the retention floor - // are re-homed into the live done segment before the TRUNCATE - // so they stay retryable. The ring rotation lock is held above, - // which serializes against rotate — the current slot (and its - // generation, read from the ledger cursor) cannot move while - // carried rows are re-inserted into it. - let carried_failed_rows = if retention_floor { - // Carried rows are widened to the self-contained synthetic - // done-row shape (the COALESCE chain mirrors - // `done_row_projection`) because the ready rows they would - // otherwise hydrate from are truncated with this slot. - // Deliberately no ON CONFLICT: lane_seq comes from - // never-resetting sequences, so a PK collision means - // corrupted terminal state and must abort the prune loudly - // rather than silently drop a terminal fact. - let carried = sqlx::query(audited_sql(format!( - r#" - INSERT INTO {schema}.done_entries ( - ready_slot, ready_generation, job_id, kind, queue, - args, state, priority, attempt, run_lease, - max_attempts, lane_seq, enqueue_shard, run_at, - attempted_at, finalized_at, created_at, unique_key, - unique_states, payload - ) - SELECT - $2, - $3, - done.job_id, - done.kind, - done.queue, - COALESCE(done.args, ready.args, '{{}}'::jsonb), - done.state, - done.priority, - done.attempt, - done.run_lease, - COALESCE(done.max_attempts, ready.max_attempts, 25::smallint), - done.lane_seq, - done.enqueue_shard, - COALESCE(done.run_at, ready.run_at, done.finalized_at), - COALESCE(done.attempted_at, ready.attempted_at), - done.finalized_at, - COALESCE(done.created_at, ready.created_at, done.finalized_at), - COALESCE(done.unique_key, ready.unique_key), - COALESCE(done.unique_states, ready.unique_states), - COALESCE(done.payload, ready.payload, '{{}}'::jsonb) - FROM {done_child} AS done - LEFT JOIN {ready_child} AS ready - ON ready.ready_slot = done.ready_slot - AND ready.ready_generation = done.ready_generation - AND ready.queue = done.queue - AND ready.priority = done.priority - AND ready.enqueue_shard = done.enqueue_shard - AND ready.lane_seq = done.lane_seq - WHERE done.state = 'failed' - AND done.finalized_at >= now() - make_interval(secs => $1::bigint) - "# - ))) - .bind(failed_retention_secs) - .bind(state.0) - .bind(current_generation) - .execute(tx.as_mut()) - .await - .map_err(map_sqlx_error)? - .rows_affected(); - - if carried > 0 { - // This slot's live counter rows and delta segment are - // wiped below, so the carried rows' exact-count - // evidence moves with them: re-append positive deltas - // under the current slot, grouped exactly like the - // completion path's delta append. - sqlx::query(audited_sql(format!( + let failed_retention_secs = i64::try_from(failed_retention.as_secs()).unwrap_or(i64::MAX); + let retention_floor = failed_retention_secs > 0; + + // #337: failed terminal rows still inside the retention floor + // are re-homed into the live done segment before the TRUNCATE + // so they stay retryable. The ring rotation lock is held above, + // which serializes against rotate — the current slot (and its + // generation, read from the ledger cursor) cannot move while + // carried rows are re-inserted into it. + let carried_failed_rows = if retention_floor { + // Carried rows are widened to the self-contained synthetic + // done-row shape (the COALESCE chain mirrors + // `done_row_projection`) because the ready rows they would + // otherwise hydrate from are truncated with this slot. + // Deliberately no ON CONFLICT: lane_seq comes from + // never-resetting sequences, so a PK collision means + // corrupted terminal state and must abort the prune loudly + // rather than silently drop a terminal fact. + let carried = sqlx::query(audited_sql(format!( r#" - INSERT INTO {schema}.queue_terminal_count_deltas ( - ready_slot, - ready_generation, - queue, - priority, - enqueue_shard, - counter_bucket, - terminal_delta + INSERT INTO {schema}.done_entries ( + ready_slot, ready_generation, job_id, kind, queue, + args, state, priority, attempt, run_lease, + max_attempts, lane_seq, enqueue_shard, run_at, + attempted_at, finalized_at, created_at, unique_key, + unique_states, payload ) SELECT $2, $3, - queue, - priority, - enqueue_shard, - mod( - mod(job_id, {TERMINAL_COUNTER_BUCKETS}::bigint) - + {TERMINAL_COUNTER_BUCKETS}::bigint, - {TERMINAL_COUNTER_BUCKETS}::bigint - )::smallint AS counter_bucket, - count(*)::bigint - FROM {done_child} - WHERE state = 'failed' - AND finalized_at >= now() - make_interval(secs => $1::bigint) - GROUP BY queue, priority, enqueue_shard, counter_bucket - ORDER BY queue, priority, enqueue_shard, counter_bucket + done.job_id, + done.kind, + done.queue, + COALESCE(done.args, ready.args, '{{}}'::jsonb), + done.state, + done.priority, + done.attempt, + done.run_lease, + COALESCE(done.max_attempts, ready.max_attempts, 25::smallint), + done.lane_seq, + done.enqueue_shard, + COALESCE(done.run_at, ready.run_at, done.finalized_at), + COALESCE(done.attempted_at, ready.attempted_at), + done.finalized_at, + COALESCE(done.created_at, ready.created_at, done.finalized_at), + COALESCE(done.unique_key, ready.unique_key), + COALESCE(done.unique_states, ready.unique_states), + COALESCE(done.payload, ready.payload, '{{}}'::jsonb) + FROM {done_child} AS done + LEFT JOIN {ready_child} AS ready + ON ready.ready_slot = done.ready_slot + AND ready.ready_generation = done.ready_generation + AND ready.queue = done.queue + AND ready.priority = done.priority + AND ready.enqueue_shard = done.enqueue_shard + AND ready.lane_seq = done.lane_seq + WHERE done.state = 'failed' + AND done.finalized_at >= now() - make_interval(secs => $1::bigint) "# ))) .bind(failed_retention_secs) @@ -19003,187 +18957,232 @@ impl QueueStorage { .bind(current_generation) .execute(tx.as_mut()) .await - .map_err(map_sqlx_error)?; - } - carried - } else { - 0 - }; - - // #290: scan the about-to-be-truncated partition for the rollup - // fold. The rollup column is *permanent* state, so we MUST fold - // from ground truth (the `{done_child}` partition itself), not - // from `queue_terminal_live_counts` — the counter can drift - // briefly during a rolling upgrade from a pre-counter binary - // and folding drift into the rollup would bake it in forever. - // The counter rows for this slot are still cleaned up after the - // truncate; reads from the counter (queue_counts_exact in #305) - // may transiently disagree with the rollup until the operator - // runs `awa storage rebuild-terminal-counters`, but the rollup - // itself stays authoritative. See PR #304 reviewer finding - // "High: A1 can persist stale counter state before the - // read-switch PR" for the trade-off. - // - // Carried failed rows are excluded from both columns: they are - // still live in `done_entries`, so folding them into the - // permanent rollup would double-count them. - let pruned_terminal_counts: Vec<(String, i16, i64, i64)> = - sqlx::query_as(audited_sql(format!( - r#" - WITH done_counts AS ( - SELECT - queue, - priority, - (count(*) FILTER (WHERE state <> 'failed'))::bigint AS completed, - (count(*) FILTER ( + .map_err(map_sqlx_error)? + .rows_affected(); + + if carried > 0 { + // This slot's live counter rows and delta segment are + // wiped below, so the carried rows' exact-count + // evidence moves with them: re-append positive deltas + // under the current slot, grouped exactly like the + // completion path's delta append. + sqlx::query(audited_sql(format!( + r#" + INSERT INTO {schema}.queue_terminal_count_deltas ( + ready_slot, + ready_generation, + queue, + priority, + enqueue_shard, + counter_bucket, + terminal_delta + ) + SELECT + $2, + $3, + queue, + priority, + enqueue_shard, + mod( + mod(job_id, {TERMINAL_COUNTER_BUCKETS}::bigint) + + {TERMINAL_COUNTER_BUCKETS}::bigint, + {TERMINAL_COUNTER_BUCKETS}::bigint + )::smallint AS counter_bucket, + count(*)::bigint + FROM {done_child} WHERE state = 'failed' - AND ( - $2::boolean IS FALSE - OR finalized_at < now() - make_interval(secs => $1::bigint) - ) - ))::bigint AS failed - FROM {done_child} - GROUP BY queue, priority - ), - -- Scan the entire partition being truncated (all generations), - -- mirroring done_counts' whole-child scan above. The rotate guard - -- keeps a partition to one generation at prune time, so this is the - -- same set today; counting the whole child keeps the rollup fold - -- conservative even if that invariant is ever weakened (a stray - -- generation's batches would otherwise be truncated without being - -- folded into the rollup -> terminal-count undercount). - batch_rows AS ( - SELECT - batch.ready_generation, - batch.queue, - batch.priority, - item.job_id, - item.run_lease - FROM {receipt_batch_child} AS batch - CROSS JOIN LATERAL unnest( - batch.job_ids, - batch.run_leases - ) AS item(job_id, run_lease) - ), - batch_counts AS ( + AND finalized_at >= now() - make_interval(secs => $1::bigint) + GROUP BY queue, priority, enqueue_shard, counter_bucket + ORDER BY queue, priority, enqueue_shard, counter_bucket + "# + ))) + .bind(failed_retention_secs) + .bind(state.0) + .bind(current_generation) + .execute(tx.as_mut()) + .await + .map_err(map_sqlx_error)?; + } + carried + } else { + 0 + }; + + // #290: scan the about-to-be-truncated partition for the rollup + // fold. The rollup column is *permanent* state, so we MUST fold + // from ground truth (the `{done_child}` partition itself), not + // from `queue_terminal_live_counts` — the counter can drift + // briefly during a rolling upgrade from a pre-counter binary + // and folding drift into the rollup would bake it in forever. + // The counter rows for this slot are still cleaned up after the + // truncate; reads from the counter (queue_counts_exact in #305) + // may transiently disagree with the rollup until the operator + // runs `awa storage rebuild-terminal-counters`, but the rollup + // itself stays authoritative. See PR #304 reviewer finding + // "High: A1 can persist stale counter state before the + // read-switch PR" for the trade-off. + // + // Carried failed rows are excluded from both columns: they are + // still live in `done_entries`, so folding them into the + // permanent rollup would double-count them. + let pruned_terminal_counts: Vec<(String, i16, i64, i64)> = + sqlx::query_as(audited_sql(format!( + r#" + WITH done_counts AS ( + SELECT + queue, + priority, + (count(*) FILTER (WHERE state <> 'failed'))::bigint AS completed, + (count(*) FILTER ( + WHERE state = 'failed' + AND ( + $2::boolean IS FALSE + OR finalized_at < now() - make_interval(secs => $1::bigint) + ) + ))::bigint AS failed + FROM {done_child} + GROUP BY queue, priority + ), + -- Scan the entire partition being truncated (all generations), + -- mirroring done_counts' whole-child scan above. The rotate guard + -- keeps a partition to one generation at prune time, so this is the + -- same set today; counting the whole child keeps the rollup fold + -- conservative even if that invariant is ever weakened (a stray + -- generation's batches would otherwise be truncated without being + -- folded into the rollup -> terminal-count undercount). + batch_rows AS ( + SELECT + batch.ready_generation, + batch.queue, + batch.priority, + item.job_id, + item.run_lease + FROM {receipt_batch_child} AS batch + CROSS JOIN LATERAL unnest( + batch.job_ids, + batch.run_leases + ) AS item(job_id, run_lease) + ), + batch_counts AS ( + SELECT + batch_rows.queue, + batch_rows.priority, + count(*)::bigint AS completed + FROM batch_rows + WHERE NOT EXISTS ( + SELECT 1 + FROM {receipt_tomb_child} AS tomb + WHERE tomb.ready_generation = batch_rows.ready_generation + AND tomb.job_id = batch_rows.job_id + AND tomb.run_lease = batch_rows.run_lease + ) + GROUP BY batch_rows.queue, batch_rows.priority + ), + keys AS ( + SELECT queue, priority FROM done_counts + UNION + SELECT queue, priority FROM batch_counts + ) SELECT - batch_rows.queue, - batch_rows.priority, - count(*)::bigint AS completed - FROM batch_rows - WHERE NOT EXISTS ( - SELECT 1 - FROM {receipt_tomb_child} AS tomb - WHERE tomb.ready_generation = batch_rows.ready_generation - AND tomb.job_id = batch_rows.job_id - AND tomb.run_lease = batch_rows.run_lease + keys.queue, + keys.priority, + ( + COALESCE(done_counts.completed, 0) + + COALESCE(batch_counts.completed, 0) + )::bigint AS pruned_completed_count, + COALESCE(done_counts.failed, 0)::bigint AS pruned_failed_count + FROM keys + LEFT JOIN done_counts USING (queue, priority) + LEFT JOIN batch_counts USING (queue, priority) + "# + ))) + .bind(failed_retention_secs) + .bind(retention_floor) + .fetch_all(tx.as_mut()) + .await + .map_err(map_sqlx_error)?; + + let truncate_started = Instant::now(); + self + .truncate_used_children_tx( + &mut tx, + &[ + &ready_child, + &claim_attempt_child, + &done_child, + &tomb_child, + &segment_child, + &receipt_batch_child, + &receipt_tomb_child, + &delta_child, + ], ) - GROUP BY batch_rows.queue, batch_rows.priority - ), - keys AS ( - SELECT queue, priority FROM done_counts - UNION - SELECT queue, priority FROM batch_counts - ) - SELECT - keys.queue, - keys.priority, - ( - COALESCE(done_counts.completed, 0) - + COALESCE(batch_counts.completed, 0) - )::bigint AS pruned_completed_count, - COALESCE(done_counts.failed, 0)::bigint AS pruned_failed_count - FROM keys - LEFT JOIN done_counts USING (queue, priority) - LEFT JOIN batch_counts USING (queue, priority) - "# + .await + .map_err(map_sqlx_error)?; + let truncate_duration = truncate_started.elapsed(); + if !pruned_terminal_counts.is_empty() { + // #371: append the pruned counts as delta rows instead + // of upserting `queue_terminal_rollups` here — the + // upsert under a pinned MVCC horizon left one + // unreclaimable dead rollup version per prune. The + // maintenance leader folds the deltas + // (`fold_terminal_rollup_deltas`) when the horizon is + // clear; exact readers add unfolded delta sums to + // their rollup reads. + self.append_terminal_rollup_deltas_tx( + &mut tx, + pruned_terminal_counts.into_iter(), + ) + .await?; + } + + // #290: the live counter rows for this slot are about to + // be orphans (their underlying done_entries rows just got + // truncated). Delete them in the same transaction. The + // rollup fold above already captured ground-truth from + // the partition scan; this just cleans up the counter + // index entries so a future insert into a re-rotated + // slot starts from zero. + sqlx::query(audited_sql(format!( + "DELETE FROM {schema}.queue_terminal_live_counts WHERE ready_slot = $1" ))) - .bind(failed_retention_secs) - .bind(retention_floor) - .fetch_all(tx.as_mut()) + .bind(slot) + .execute(tx.as_mut()) .await .map_err(map_sqlx_error)?; + Ok(PruneOutcome::Pruned { + slot, + generation, + carried_failed_rows, + durations: PruneDurations { + lock: lock_duration, + truncate: truncate_duration, + // Set once the transaction commits. + commit: Duration::ZERO, + }, + }) + } + .await; - let truncate_started = Instant::now(); - let truncate = self - .truncate_used_children_tx( - &mut tx, - &[ - &ready_child, - &claim_attempt_child, - &done_child, - &tomb_child, - &segment_child, - &receipt_batch_child, - &receipt_tomb_child, - &delta_child, - ], - ) - .await; - let truncate_duration = truncate_started.elapsed(); - - match truncate { - Ok(_) => { - if !pruned_terminal_counts.is_empty() { - // #371: append the pruned counts as delta rows instead - // of upserting `queue_terminal_rollups` here — the - // upsert under a pinned MVCC horizon left one - // unreclaimable dead rollup version per prune. The - // maintenance leader folds the deltas - // (`fold_terminal_rollup_deltas`) when the horizon is - // clear; exact readers add unfolded delta sums to - // their rollup reads. - self.append_terminal_rollup_deltas_tx( - &mut tx, - pruned_terminal_counts.into_iter(), - ) - .await?; - } - - // #290: the live counter rows for this slot are about to - // be orphans (their underlying done_entries rows just got - // truncated). Delete them in the same transaction. The - // rollup fold above already captured ground-truth from - // the partition scan; this just cleans up the counter - // index entries so a future insert into a re-rotated - // slot starts from zero. - sqlx::query(audited_sql(format!( - "DELETE FROM {schema}.queue_terminal_live_counts WHERE ready_slot = $1" - ))) - .bind(slot) - .execute(tx.as_mut()) - .await - .map_err(map_sqlx_error)?; - let commit_started = Instant::now(); - tx.commit().await.map_err(map_sqlx_error)?; - let commit_duration = commit_started.elapsed(); - if carried_failed_rows > 0 { - tracing::info!( - slot, - carried_failed_rows, - "Carried failed terminal rows inside the retention floor to the live done segment" - ); - } - Ok(PruneOutcome::Pruned { + let commit_started = Instant::now(); + let mut outcome = settle_prune_tx(tx, locked, PruneOutcome::Blocked { slot }).await?; + if let PruneOutcome::Pruned { + carried_failed_rows, + durations, + .. + } = &mut outcome + { + durations.commit = commit_started.elapsed(); + let carried_failed_rows = *carried_failed_rows; + if carried_failed_rows > 0 { + tracing::info!( slot, - generation, carried_failed_rows, - durations: PruneDurations { - lock: lock_duration, - truncate: truncate_duration, - commit: commit_duration, - }, - }) - } - Err(err) if is_lock_contention_error(&err) => { - let _ = tx.rollback().await; - Ok(PruneOutcome::Blocked { slot }) - } - Err(err) => { - let _ = tx.rollback().await; - Err(map_sqlx_error(err)) + "Carried failed terminal rows inside the retention floor to the live done segment" + ); } } + Ok(outcome) } #[tracing::instrument( @@ -19259,82 +19258,68 @@ impl QueueStorage { let lease_child = lease_child_name(schema, slot as usize); - set_prune_lock_timeout_tx(&mut tx, self.prune_lock_timeout).await?; + let locked: Result = async { + set_prune_lock_timeout_tx(&mut tx, self.prune_lock_timeout).await?; - let lock_started = Instant::now(); - let lock_table = sqlx::query(audited_sql(format!( - "LOCK TABLE {lease_child} IN ACCESS EXCLUSIVE MODE" - ))) - .execute(tx.as_mut()) - .await; - let lock_duration = lock_started.elapsed(); + let lock_started = Instant::now(); + sqlx::query(audited_sql(format!( + "LOCK TABLE {lease_child} IN ACCESS EXCLUSIVE MODE" + ))) + .execute(tx.as_mut()) + .await + .map_err(map_sqlx_error)?; + let lock_duration = lock_started.elapsed(); - if let Err(err) = lock_table { - let _ = tx.rollback().await; - if is_lock_contention_error(&err) { - return Ok(PruneOutcome::Blocked { slot }); + // The ring rotation serialization held above keeps the cursor + // stable, but keep the explicit recheck to document the truncate + // precondition. + let (current_slot, _) = self + .ring_cursor_tx(&mut tx, RingFamily::Lease, authority) + .await?; + + if current_slot == slot { + return Ok(PruneOutcome::SkippedActive { + slot, + reason: SkipReason::LeaseCurrent, + count: 0, + }); } - return Err(map_sqlx_error(err)); - } - // The ring rotation serialization held above keeps the cursor - // stable, but keep the explicit recheck to document the truncate - // precondition. - let (current_slot, _) = self - .ring_cursor_tx(&mut tx, RingFamily::Lease, authority) - .await?; + let active_leases = Self::relation_has_rows_tx(&mut tx, &lease_child).await?; - if current_slot == slot { - tx.commit().await.map_err(map_sqlx_error)?; - return Ok(PruneOutcome::SkippedActive { - slot, - reason: SkipReason::LeaseCurrent, - count: 0, - }); - } - - let active_leases = Self::relation_has_rows_tx(&mut tx, &lease_child).await?; + if active_leases { + return Ok(PruneOutcome::SkippedActive { + slot, + reason: SkipReason::LeaseActive, + count: busy_indicator(active_leases), + }); + } - if active_leases { - tx.commit().await.map_err(map_sqlx_error)?; - return Ok(PruneOutcome::SkippedActive { + let truncate_started = Instant::now(); + self.truncate_used_children_tx(&mut tx, &[&lease_child]) + .await + .map_err(map_sqlx_error)?; + let truncate_duration = truncate_started.elapsed(); + Ok(PruneOutcome::Pruned { slot, - reason: SkipReason::LeaseActive, - count: busy_indicator(active_leases), - }); + generation, + carried_failed_rows: 0, + durations: PruneDurations { + lock: lock_duration, + truncate: truncate_duration, + // Set once the transaction commits. + commit: Duration::ZERO, + }, + }) } + .await; - let truncate_started = Instant::now(); - let truncate = self - .truncate_used_children_tx(&mut tx, &[&lease_child]) - .await; - let truncate_duration = truncate_started.elapsed(); - - match truncate { - Ok(_) => { - let commit_started = Instant::now(); - tx.commit().await.map_err(map_sqlx_error)?; - let commit_duration = commit_started.elapsed(); - Ok(PruneOutcome::Pruned { - slot, - generation, - carried_failed_rows: 0, - durations: PruneDurations { - lock: lock_duration, - truncate: truncate_duration, - commit: commit_duration, - }, - }) - } - Err(err) if is_lock_contention_error(&err) => { - let _ = tx.rollback().await; - Ok(PruneOutcome::Blocked { slot }) - } - Err(err) => { - let _ = tx.rollback().await; - Err(map_sqlx_error(err)) - } + let commit_started = Instant::now(); + let mut outcome = settle_prune_tx(tx, locked, PruneOutcome::Blocked { slot }).await?; + if let PruneOutcome::Pruned { durations, .. } = &mut outcome { + durations.commit = commit_started.elapsed(); } + Ok(outcome) } pub async fn vacuum_leases(&self, pool: &PgPool) -> Result<(), AwaError> { @@ -19638,116 +19623,103 @@ impl QueueStorage { }); } - set_prune_lock_timeout_tx(&mut tx, self.prune_lock_timeout).await?; - - let lock_started = Instant::now(); - let lock_tables = sqlx::query(audited_sql(format!( - "LOCK TABLE {claim_child}, {claim_batch_child}, {closure_child}, {closure_batch_child} IN ACCESS EXCLUSIVE MODE" - ))) - .execute(tx.as_mut()) - .await; - let lock_duration = lock_started.elapsed(); - - if let Err(err) = lock_tables { - let _ = tx.rollback().await; - if is_lock_contention_error(&err) { - return Ok(PruneOutcome::Blocked { slot }); - } - return Err(map_sqlx_error(err)); - } + let locked: Result = async { + set_prune_lock_timeout_tx(&mut tx, self.prune_lock_timeout).await?; - // After taking ACCESS EXCLUSIVE, recheck that the slot is still - // not the current one. The ring rotation serialization should - // already make this stable, but keeping the explicit gate documents - // the truncate precondition. - let (current_slot, _) = self - .ring_cursor_tx(&mut tx, RingFamily::Claim, authority) - .await?; + let lock_started = Instant::now(); + sqlx::query(audited_sql(format!( + "LOCK TABLE {claim_child}, {claim_batch_child}, {closure_child}, {closure_batch_child} IN ACCESS EXCLUSIVE MODE" + ))) + .execute(tx.as_mut()) + .await + .map_err(map_sqlx_error)?; + let lock_duration = lock_started.elapsed(); - if current_slot == slot { - tx.commit().await.map_err(map_sqlx_error)?; - return Ok(PruneOutcome::SkippedActive { - slot, - reason: SkipReason::ClaimCurrent, - count: 0, - }); - } + // After taking ACCESS EXCLUSIVE, recheck that the slot is still + // not the current one. The ring rotation serialization should + // already make this stable, but keeping the explicit gate documents + // the truncate precondition. + let (current_slot, _) = self + .ring_cursor_tx(&mut tx, RingFamily::Claim, authority) + .await?; - let open_claims = claim_prune_has_open_claims_tx( - &mut tx, - schema, - &claim_child, - &claim_batch_child, - &closure_child, - &closure_batch_child, - ) - .await?; - if open_claims { - tx.commit().await.map_err(map_sqlx_error)?; - return Ok(PruneOutcome::SkippedActive { - slot, - reason: SkipReason::ClaimOpen, - count: busy_indicator(open_claims), - }); - } + if current_slot == slot { + return Ok(PruneOutcome::SkippedActive { + slot, + reason: SkipReason::ClaimCurrent, + count: 0, + }); + } - let truncate_started = Instant::now(); - let truncate = self - .truncate_used_children_tx( + let open_claims = claim_prune_has_open_claims_tx( &mut tx, - &[ - &claim_child, - &claim_batch_child, - &closure_child, - &closure_batch_child, - ], + schema, + &claim_child, + &claim_batch_child, + &closure_child, + &closure_batch_child, ) - .await; - let truncate_duration = truncate_started.elapsed(); + .await?; + if open_claims { + return Ok(PruneOutcome::SkippedActive { + slot, + reason: SkipReason::ClaimOpen, + count: busy_indicator(open_claims), + }); + } - match truncate { - Ok(_) => { - sqlx::query(audited_sql(format!( - r#" - UPDATE {schema}.claim_ring_slots - SET rescue_cursor_claimed_at = '-infinity'::timestamptz, - rescue_cursor_job_id = 0, - rescue_cursor_run_lease = 0, - deadline_cursor_deadline_at = '-infinity'::timestamptz, - deadline_cursor_job_id = 0, - deadline_cursor_run_lease = 0, - batch_deadline_cursor_deadline_at = '-infinity'::timestamptz, - batch_deadline_cursor_batch_id = 0 - WHERE slot = $1 - "# - ))) - .bind(slot) - .execute(tx.as_mut()) + let truncate_started = Instant::now(); + self + .truncate_used_children_tx( + &mut tx, + &[ + &claim_child, + &claim_batch_child, + &closure_child, + &closure_batch_child, + ], + ) .await .map_err(map_sqlx_error)?; - let commit_started = Instant::now(); - tx.commit().await.map_err(map_sqlx_error)?; - let commit_duration = commit_started.elapsed(); - Ok(PruneOutcome::Pruned { - slot, - generation, - carried_failed_rows: 0, - durations: PruneDurations { - lock: lock_duration, - truncate: truncate_duration, - commit: commit_duration, - }, - }) - } - Err(err) if is_lock_contention_error(&err) => { - let _ = tx.rollback().await; - Ok(PruneOutcome::Blocked { slot }) - } - Err(err) => { - let _ = tx.rollback().await; - Err(map_sqlx_error(err)) - } + let truncate_duration = truncate_started.elapsed(); + sqlx::query(audited_sql(format!( + r#" + UPDATE {schema}.claim_ring_slots + SET rescue_cursor_claimed_at = '-infinity'::timestamptz, + rescue_cursor_job_id = 0, + rescue_cursor_run_lease = 0, + deadline_cursor_deadline_at = '-infinity'::timestamptz, + deadline_cursor_job_id = 0, + deadline_cursor_run_lease = 0, + batch_deadline_cursor_deadline_at = '-infinity'::timestamptz, + batch_deadline_cursor_batch_id = 0 + WHERE slot = $1 + "# + ))) + .bind(slot) + .execute(tx.as_mut()) + .await + .map_err(map_sqlx_error)?; + Ok(PruneOutcome::Pruned { + slot, + generation, + carried_failed_rows: 0, + durations: PruneDurations { + lock: lock_duration, + truncate: truncate_duration, + // Set once the transaction commits. + commit: Duration::ZERO, + }, + }) } + .await; + + let commit_started = Instant::now(); + let mut outcome = settle_prune_tx(tx, locked, PruneOutcome::Blocked { slot }).await?; + if let PruneOutcome::Pruned { durations, .. } = &mut outcome { + durations.commit = commit_started.elapsed(); + } + Ok(outcome) } fn job_id_sequence(&self) -> String { diff --git a/awa/tests/queue_storage_runtime_test.rs b/awa/tests/queue_storage_runtime_test.rs index bf51546fe..2a8438a7d 100644 --- a/awa/tests/queue_storage_runtime_test.rs +++ b/awa/tests/queue_storage_runtime_test.rs @@ -15632,6 +15632,175 @@ async fn test_queue_storage_prune_oldest_blocks_on_reader_lock() { ); } +/// The prune `lock_timeout` covers every statement after it is set, not +/// only the `LOCK TABLE` on the slot's children. A statement that waits on +/// another table's lock — here the failed-row carry, its count deltas, the +/// rollup delta append and the live-counter cleanup — must roll the whole prune back +/// and report `Blocked`, leaving the slot to the next attempt. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn test_queue_storage_prune_lock_timeout_after_child_locks_reports_blocked() { + let (_db_guard, pool) = setup_pool(10).await; + let queue = "qs_prune_late_lock_timeout"; + let schema = "awa_qs_prune_late_lock_timeout"; + let store = create_store(&pool, schema).await; + + store + .enqueue_batch(&pool, queue, 1, 1) + .await + .expect("enqueue completed job"); + let claimed = store + .claim_batch(&pool, queue, 1) + .await + .expect("claim completed job"); + store + .complete_batch(&pool, &claimed) + .await + .expect("complete job"); + + let failed_job_id = enqueue_job( + &pool, + &store, + &CompleteJob { id: 1 }, + InsertOpts { + queue: queue.to_string(), + ..Default::default() + }, + ) + .await; + let claimed = store + .claim_runtime_batch(&pool, queue, 1, Duration::from_secs(30)) + .await + .expect("claim failing job") + .into_iter() + .next() + .expect("missing claimed job"); + store + .fail_terminal(&pool, failed_job_id, claimed.job.run_lease, "boom", None) + .await + .expect("fail job") + .expect("running job should fail"); + + let rotated = store.rotate(&pool).await.expect("rotate"); + assert!( + matches!(rotated, RotateOutcome::Rotated { slot: 1, .. }), + "unexpected rotate outcome: {rotated:?}" + ); + + let retention = Duration::from_secs(3600); + for held in [ + format!("{schema}.done_entries_1"), + format!("{schema}.queue_terminal_count_deltas_1"), + format!("{schema}.queue_terminal_rollup_deltas"), + format!("{schema}.queue_terminal_live_counts"), + ] { + let mut holder = pool.begin().await.expect("begin lock holder"); + sqlx::query(audited_sql(format!("LOCK TABLE {held} IN SHARE MODE"))) + .execute(holder.as_mut()) + .await + .expect("hold share lock"); + + let outcome = store.prune_oldest(&pool, retention).await; + assert!( + matches!(outcome, Ok(PruneOutcome::Blocked { slot: 0 })), + "prune waiting on {held} should report Blocked, got {outcome:?}" + ); + + holder.rollback().await.expect("release share lock"); + + assert_eq!( + done_entries_count(&pool, schema, queue).await, + 2, + "blocked prune on {held} must not truncate or carry rows" + ); + let rollup_rows: i64 = sqlx::query_scalar(audited_sql(format!( + "SELECT count(*)::bigint FROM {schema}.queue_terminal_rollup_deltas" + ))) + .fetch_one(&pool) + .await + .expect("count rollups"); + assert_eq!( + rollup_rows, 0, + "blocked prune on {held} must not fold rollups" + ); + assert_invariant_holds( + &pool, + schema, + queue, + &format!("after prune blocked on {held}"), + ) + .await; + } + + let pruned = store + .prune_oldest(&pool, retention) + .await + .expect("prune after locks released"); + assert!( + matches!( + pruned, + PruneOutcome::Pruned { + slot: 0, + carried_failed_rows: 1, + .. + } + ), + "unexpected prune outcome after locks released: {pruned:?}" + ); + assert_invariant_holds(&pool, schema, queue, "after prune").await; + let counts = store.queue_counts(&pool, queue).await.expect("counts"); + assert_eq!(counts.terminal, 2, "rollup (1) + carried live row (1)"); +} + +/// Terminal-delta rollup shares prune's lock timeout. A timeout on the +/// live-counter fold, after the delta child is locked, must report a blocked +/// slot rather than fail the whole rollup pass. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn test_queue_terminal_count_delta_rollup_lock_timeout_after_child_lock_reports_blocked() { + let (_db_guard, pool) = setup_pool(10).await; + let queue = "qs_terminal_delta_rollup_late_lock"; + let schema = "awa_qs_terminal_delta_rollup_late_lock"; + let store = create_store(&pool, schema).await; + + store + .enqueue_batch(&pool, queue, 1, 3) + .await + .expect("enqueue"); + let claimed = store.claim_batch(&pool, queue, 3).await.expect("claim"); + store + .complete_batch(&pool, &claimed) + .await + .expect("complete"); + match store.rotate(&pool).await.expect("rotate") { + RotateOutcome::Rotated { .. } => {} + other => panic!("expected Rotated, got {other:?}"), + } + + let mut holder = pool.begin().await.expect("begin lock holder"); + sqlx::query(audited_sql(format!( + "LOCK TABLE {schema}.queue_terminal_live_counts IN SHARE MODE" + ))) + .execute(holder.as_mut()) + .await + .expect("hold share lock"); + + let outcome = store.rollup_terminal_count_deltas(&pool, 4).await; + holder.rollback().await.expect("release share lock"); + let outcome = outcome.expect("rollup waiting on live counts should not error"); + assert_eq!(outcome.rolled_slots, 0); + assert_eq!(outcome.blocked_slots, 1); + assert_eq!(live_count_sum(&pool, schema, queue).await, 0); + assert_eq!(terminal_delta_sum(&pool, schema, queue).await, 3); + + let outcome = store + .rollup_terminal_count_deltas(&pool, 4) + .await + .expect("rollup after lock released"); + assert_eq!(outcome.rolled_slots, 1); + assert_eq!(outcome.blocked_slots, 0); + assert_eq!(live_count_sum(&pool, schema, queue).await, 3); + assert_eq!(terminal_delta_sum(&pool, schema, queue).await, 0); +} + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn test_queue_prune_allows_segment_reader_to_finish_claim_lock_order() { let (_db_guard, pool) = setup_pool(8).await; From 44fca8d8f1fabae00b7f69cb81ba0e266df62f72 Mon Sep 17 00:00:00 2001 From: Brian Thorne Date: Fri, 6 Nov 2026 18:07:42 +1300 Subject: [PATCH 2/2] fix(queue-storage): stop a terminal-count rollup pass at its first blocked slot The remaining sealed slots would most likely wait out the same lock timeout, and with a raised prune timeout that holds the maintenance leader for one timeout per slot. The rollup test now seals two slots with pending deltas and expects one blocked slot, then both rolled once the lock is released. --- CHANGELOG.md | 4 ++- awa-model/src/queue_storage.rs | 3 ++ awa/tests/queue_storage_runtime_test.rs | 40 ++++++++++++++----------- 3 files changed, 29 insertions(+), 18 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index bd61055e0..9beeac27e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -22,7 +22,9 @@ Notable changes between releases. Detailed migration notes for storage transitio anywhere in a queue, lease or claim prune, or in a terminal-count rollup slot, now rolls the whole transaction back and reports `Blocked` (`blocked_slots` for rollup), so maintenance backs off and retries; nothing - the attempt wrote survives the rollback. + the attempt wrote survives the rollback. A rollup pass stops at its first + blocked slot rather than waiting out the timeout again on each remaining + one. - **Batch enqueues take their lane locks in one order, so they no longer deadlock with each other.** On queue storage every enqueue holds a transaction-scoped advisory lock on its `(queue, priority, shard)` lane diff --git a/awa-model/src/queue_storage.rs b/awa-model/src/queue_storage.rs index bbe2e0fe9..c9e70ada8 100644 --- a/awa-model/src/queue_storage.rs +++ b/awa-model/src/queue_storage.rs @@ -18192,7 +18192,10 @@ impl QueueStorage { break; } TerminalDeltaSlotRollup::Blocked => { + // The next slot would most likely wait out the same lock + // timeout, and the leader loop is blocked while it does. outcome.blocked_slots += 1; + break; } } } diff --git a/awa/tests/queue_storage_runtime_test.rs b/awa/tests/queue_storage_runtime_test.rs index 2a8438a7d..3b57f3370 100644 --- a/awa/tests/queue_storage_runtime_test.rs +++ b/awa/tests/queue_storage_runtime_test.rs @@ -15753,7 +15753,8 @@ async fn test_queue_storage_prune_lock_timeout_after_child_locks_reports_blocked /// Terminal-delta rollup shares prune's lock timeout. A timeout on the /// live-counter fold, after the delta child is locked, must report a blocked -/// slot rather than fail the whole rollup pass. +/// slot rather than fail the rollup pass, and the pass stops there instead of +/// waiting out the same timeout on every remaining slot. #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn test_queue_terminal_count_delta_rollup_lock_timeout_after_child_lock_reports_blocked() { let (_db_guard, pool) = setup_pool(10).await; @@ -15761,18 +15762,20 @@ async fn test_queue_terminal_count_delta_rollup_lock_timeout_after_child_lock_re let schema = "awa_qs_terminal_delta_rollup_late_lock"; let store = create_store(&pool, schema).await; - store - .enqueue_batch(&pool, queue, 1, 3) - .await - .expect("enqueue"); - let claimed = store.claim_batch(&pool, queue, 3).await.expect("claim"); - store - .complete_batch(&pool, &claimed) - .await - .expect("complete"); - match store.rotate(&pool).await.expect("rotate") { - RotateOutcome::Rotated { .. } => {} - other => panic!("expected Rotated, got {other:?}"), + for _ in 0..2 { + store + .enqueue_batch(&pool, queue, 1, 3) + .await + .expect("enqueue"); + let claimed = store.claim_batch(&pool, queue, 3).await.expect("claim"); + store + .complete_batch(&pool, &claimed) + .await + .expect("complete"); + match store.rotate(&pool).await.expect("rotate") { + RotateOutcome::Rotated { .. } => {} + other => panic!("expected Rotated, got {other:?}"), + } } let mut holder = pool.begin().await.expect("begin lock holder"); @@ -15787,17 +15790,20 @@ async fn test_queue_terminal_count_delta_rollup_lock_timeout_after_child_lock_re holder.rollback().await.expect("release share lock"); let outcome = outcome.expect("rollup waiting on live counts should not error"); assert_eq!(outcome.rolled_slots, 0); - assert_eq!(outcome.blocked_slots, 1); + assert_eq!( + outcome.blocked_slots, 1, + "the pass stops at the first blocked slot" + ); assert_eq!(live_count_sum(&pool, schema, queue).await, 0); - assert_eq!(terminal_delta_sum(&pool, schema, queue).await, 3); + assert_eq!(terminal_delta_sum(&pool, schema, queue).await, 6); let outcome = store .rollup_terminal_count_deltas(&pool, 4) .await .expect("rollup after lock released"); - assert_eq!(outcome.rolled_slots, 1); + assert_eq!(outcome.rolled_slots, 2); assert_eq!(outcome.blocked_slots, 0); - assert_eq!(live_count_sum(&pool, schema, queue).await, 3); + assert_eq!(live_count_sum(&pool, schema, queue).await, 6); assert_eq!(terminal_delta_sum(&pool, schema, queue).await, 0); }