feat(worker): configurable queue-storage prune lock timeout and prune outcome metrics (0.6)

CHANGELOG.md

@@ -4,8 +4,101 @@ Notable changes between releases. Detailed migration notes for storage transitio
 
 ## [Unreleased]
 
+### Added
+
+- **The queue-storage prune `lock_timeout` is configurable, and prune
+  attempts are counted by outcome for every ring.** Prune (queue, lease and
+  claim rings) and terminal-delta rollup take their exclusive locks under a
+  10 ms `lock_timeout` that could only be changed through the hidden
+  `QueueStorage::with_prune_lock_timeout`. When short readers of the ring
+  partitions held a conflicting lock at that instant, prune reported
+  `blocked` every tick and the rings stopped recycling. Set it with
+  `ClientBuilder::prune_lock_timeout` (Rust) or
+  `queue_storage_prune_lock_timeout_ms` on `client.start()` (Python); the
+  timeout applies separately to each lock acquisition, not to the whole prune;
+  queue prune can wait on up to eight partition locks. The default stays 10 ms and zero or a value above PostgreSQL's `lock_timeout`
+  maximum is rejected. Start logs a warning when the value is at or above the
+  connection's `deadlock_timeout` (prune waiting that long can make deadlock
+  detection cancel a claim instead) or `statement_timeout`. `with_prune_lock_timeout` is now
+  public, with `set_prune_lock_timeout` and `prune_lock_timeout()`.
+  `awa_maintenance_prune_attempts_total{awa_ring,awa_ring_outcome,awa_ring_reason}`
+  gains `backed_off` (prune skipped by the backoff that follows consecutive
+  `skipped_active` or `blocked` results) and
+  `error` (prune failed with a database error) alongside the existing
+  `pruned`, `noop`, `blocked` and `skipped_active`. See
+  [Configuration](docs/configuration.md#what-the-knobs-mean).
+
 ### Fixed
 
+- **On queue storage, a queue whose enqueue lane is busy no longer holds
+  back promotion of other queues, nor the maintenance leader's rescue,
+  rotation and prune.** Promotion re-enqueues due scheduled and retryable
+  rows into their lane, which takes the lane's reservation lock that
+  producers hold until they commit. A promotion batch was one transaction
+  across every queue it touched, and the leader runs its branches one after
+  another, so a lane kept busy by producers delayed due jobs of unrelated
+  queues in the same batch, and rescue started seconds late. Promotion now
+  gives each `(queue, priority)` lane its own transaction under a 50 ms
+  `lock_timeout`. A lane that times out keeps its rows deferred and is
+  handed to a dedicated task that waits up to 30 s for the lock, queued
+  behind the producers already waiting, so the lane is still promoted in
+  turn however busy it stays; the leader skips it meanwhile. Each waiter
+  holds a pool connection, so the leader runs one per eight pool
+  connections, at least one and at most four; a further contended lane is
+  retried on the next tick. Batch selection skips rows another transaction
+  has locked, so locked due rows cannot fill every batch of a tick. `awa.maintenance.promote.lane_contention`
+  (`awa_maintenance_promote_lane_contention_total{awa_job_state,awa_job_queue,awa_promote_outcome}`)
+  counts `handed_off`, `waiter_limit` and `waiter_timed_out`. Promoted rows
+  now take lane sequence numbers in due order within their lane.
+  `QueueStorage::promote_due` keeps its signature and promotes lane by lane
+  without a lock timeout; `promote_due_lanes` and `promote_due_lane` expose
+  the per-lane form.
+- **A queue lane is usable by every runtime role, whichever role creates it
+  first.** A lane's enqueue and claim sequences belonged to the role that
+  first enqueued into, claimed from or promoted into it, so with several
+  runtime roles sharing a group role, any role other than the creator whose
+  default privileges did not cover it failed with `permission denied for
+  sequence`. That included the maintenance leader promoting into a shard
+  created after `enqueue_shards` was raised. The new schema patch
+  `definer_lane_provisioning` creates lanes through two `SECURITY DEFINER`
+  functions per queue-storage schema, `provision_lane` and
+  `provision_lane_sequences`, owned by the schema owner. Each new sequence gets
+  the privileges of the head table it backs, explicitly rather than through
+  default privileges. A runtime without schema `CREATE` can therefore create
+  lanes itself, and `awa storage prepare-queue` is optional and may run as any
+  runtime role. Once the patch commits, `awa migrate` gives existing lane
+  sequences to the schema owner and adds their missing grants, one sequence
+  per transaction with a 300 ms lock wait, and only where it is allowed to,
+  trying the effective role and then the login role. A sequence an open
+  transaction holds is skipped and reported as busy; one it may not change is
+  reported with the roles that can repair it. Neither fails the run. The new read-only `awa storage lane-sequences` (or
+  `awa.lane_sequence_problems(schema)`) lists what remains, with the
+  statements to run and the roles to run them as, and exits 2 while any
+  remain; `--repair` (or `awa.repair_lane_sequence(schema, sequence)` per
+  sequence) repairs what the connected role may. `awa migrate` prints a summary of the leftovers. A binary
+  running before the patch detects that `provision_lane` is missing and
+  creates lanes as before, and older binaries create correctly owned lanes once
+  the patch has refreshed the helpers they call. `awa migrate` applies the
+  patch on any v040 database; external runners apply the exported
+  `R__definer_lane_provisioning.sql` after `V40`. See
+  [`docs/upgrade-0.6.13-to-0.6.14.md`](docs/upgrade-0.6.13-to-0.6.14.md).
+- **A graceful shutdown no longer leaves a finished attempt for stale-heartbeat
+  rescue to run again.** Once a handler returned, its finalize (completion,
+  retry, cancel or failure, and any ADR-029 follow-up inserts) ran in a task
+  that `Client::shutdown` did not wait for. When the last handler returned
+  just as the drain ended, shutdown stopped the completion batcher before
+  that finalize reached it, and the direct fallback raced the process exit.
+  The attempt had already left the heartbeat set, so it went stale and was
+  rescued and executed again `heartbeat_staleness` later; a deploy could
+  re-run a few jobs on every rolling restart. Shutdown now waits for
+  pending finalization, within the same drain timeout, before it stops the
+  completion batcher, heartbeat and maintenance services. A finalize that
+  outlives the timeout is still left to rescue, as before.
+- **Point reads show the live queue and priority after a lane move.** Retained
+  ready rows from before a priority or queue change could win a read's tie
+  after the ring wrapped, returning the old fields even though the batch
+  operation succeeded. Point reads now exclude tombstoned source rows.
+
 - **Callback completion, `get_job`, lifecycle events and the admin reads
   find running jobs that priority aging promoted on queue storage** (backport
   of #600). Claims and leases record a job's aged effective priority while its

awa-cli/src/main.rs

@@ -417,7 +417,7 @@ enum StorageCommands {
         #[arg(long)]
         reset: bool,
     },
-    /// Provision lane sequences as the migrator, before starting a runtime without DDL
+    /// Provision every priority and shard of a queue ahead of its first enqueue
     PrepareQueue {
         /// Queue-storage schema to provision
         #[arg(long, default_value = "awa")]
@@ -432,6 +432,17 @@ enum StorageCommands {
         #[arg(long, default_value_t = 1)]
         queue_stripe_count: usize,
     },
+    /// List lane sequences not owned by the schema owner or missing runtime
+    /// grants, with the statements that fix them. Exits 2 while any remain.
+    LaneSequences {
+        /// Queue-storage schema to check
+        #[arg(long, default_value = "awa")]
+        schema: String,
+        /// Reassign and grant first, as far as the connected role (or, failing
+        /// that, its login role) is allowed to
+        #[arg(long)]
+        repair: bool,
+    },
     /// Abort a prepared or mixed-transition storage rollout before final activation
     Abort,
     /// Enter mixed transition and begin routing new writes to the prepared engine
@@ -669,6 +680,26 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
                 if !report.complete() {
                     std::process::exit(1);
                 }
+                if report.patches.iter().any(|patch| {
+                    patch.name == "definer_lane_provisioning"
+                        && patch.outcome == awa_model::migrations::PatchOutcome::AlreadyPresent
+                }) {
+                    // The patch was already in place, so this run repaired
+                    // nothing: report what an earlier repair left behind.
+                    let mut problems = Vec::new();
+                    for (schema, _) in
+                        awa_model::storage::lane_sequence_problem_counts(&pool).await?
+                    {
+                        problems.extend(
+                            awa_model::storage::lane_sequence_problems(&pool, &schema).await?,
+                        );
+                    }
+                    for next_step in awa_model::storage::lane_sequence_next_steps(
+                        problems.iter().map(|problem| (problem, false)),
+                    ) {
+                        eprintln!("warning: {next_step}");
+                    }
+                }
             }
         }
 
@@ -1245,6 +1276,39 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
                             }))?
                         );
                     }
+                    StorageCommands::LaneSequences { schema, repair } => {
+                        let next_steps = if repair {
+                            let mut conn = pool.acquire().await?;
+                            let repairs =
+                                awa_model::storage::repair_lane_sequences(&mut conn, &schema)
+                                    .await?;
+                            println!(
+                                "{}",
+                                serde_json::to_string_pretty(&serde_json::json!({
+                                    "schema": schema, "results": repairs,
+                                }))?
+                            );
+                            awa_model::storage::lane_sequence_repair_next_steps(&repairs)
+                        } else {
+                            let problems =
+                                awa_model::storage::lane_sequence_problems(&pool, &schema).await?;
+                            println!(
+                                "{}",
+                                serde_json::to_string_pretty(&serde_json::json!({
+                                    "schema": schema, "problems": problems,
+                                }))?
+                            );
+                            awa_model::storage::lane_sequence_next_steps(
+                                problems.iter().map(|problem| (problem, false)),
+                            )
+                        };
+                        if !next_steps.is_empty() {
+                            for next_step in next_steps {
+                                eprintln!("{next_step}");
+                            }
+                            std::process::exit(2);
+                        }
+                    }
                     StorageCommands::Abort => {
                         awa_model::storage::abort(&pool).await?;
                         let report = awa_model::storage::status_report(&pool).await?;
@@ -1420,6 +1484,18 @@ fn print_migration_report(report: &awa_model::migrations::MigrationReport) {
             );
         }
     }
+    if !report.lane_sequences.is_empty() {
+        let repaired = report
+            .lane_sequences
+            .iter()
+            .filter(|repair| repair.outcome == awa_model::storage::LaneSequenceOutcome::Repaired)
+            .count();
+        // Sequences left behind are logged by the migration runner.
+        println!(
+            "Lane sequences repaired: {repaired} of {}",
+            report.lane_sequences.len()
+        );
+    }
     match (
         &report.storage_routing_before,
         &report.storage_routing_after,

awa-cli/tests/migrate_patch_cli_test.rs

@@ -321,12 +321,11 @@ fn psql(database_url: &str, sql: &str, lock_timeout: Option<&str>) -> Output {
     let mut child = command
         .spawn()
         .expect("psql must be on PATH for the exported-SQL test");
-    child
-        .stdin
-        .take()
-        .unwrap()
-        .write_all(sql.as_bytes())
-        .unwrap();
+    // psql stops reading once ON_ERROR_STOP fires, which can be before the
+    // whole script is written.
+    if let Err(err) = child.stdin.take().unwrap().write_all(sql.as_bytes()) {
+        assert_eq!(err.kind(), std::io::ErrorKind::BrokenPipe, "{err}");
+    }
     child.wait_with_output().unwrap()
 }