fix(queue-storage): promote each enqueue lane in its own transaction and hand busy lanes to a waiter (0.6)

CHANGELOG.md

@@ -6,6 +6,75 @@ Notable changes between releases. Detailed migration notes for storage transitio
 
 ### 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()
 }
 

awa-metrics/src/lib.rs

@@ -64,6 +64,7 @@ pub mod names {
     pub const MAINTENANCE_PROMOTE_BATCHES: &str = "awa.maintenance.promote_batches";
     pub const MAINTENANCE_PROMOTE_BATCH_SIZE: &str = "awa.maintenance.promote_batch_size";
     pub const MAINTENANCE_PROMOTE_DURATION: &str = "awa.maintenance.promote_duration";
+    pub const MAINTENANCE_PROMOTE_LANE_CONTENTION: &str = "awa.maintenance.promote.lane_contention";
     pub const MAINTENANCE_BRANCH_DURATION: &str = "awa.maintenance.branch.duration";
     pub const MAINTENANCE_BRANCH_OVERRUN: &str = "awa.maintenance.branch.overrun";
     pub const MAINTENANCE_ROTATE_ATTEMPTS: &str = "awa.maintenance.rotate.attempts";
@@ -162,6 +163,11 @@ pub struct AwaMetrics {
     pub promotion_batch_size: Histogram<u64>,
     /// Promotion query duration.
     pub promotion_duration_seconds: Histogram<f64>,
+    /// Promotion lanes whose locks were busy past the leader's wait budget,
+    /// attributed to (state, queue, outcome): `handed_off` to a dedicated
+    /// waiter, `waiter_limit` when every waiter was busy, `waiter_timed_out`
+    /// when a waiter's own budget ran out.
+    pub promotion_lane_contention: Counter<u64>,
     /// Current in-flight jobs (can go up and down).
     pub jobs_in_flight: UpDownCounter<i64>,
     /// Total heartbeat batches sent.
@@ -404,6 +410,11 @@ impl AwaMetrics {
                 .with_unit("s")
                 .with_boundaries(LATENCY_BUCKETS_SECONDS.to_vec())
                 .build(),
+            promotion_lane_contention: meter
+                .u64_counter(names::MAINTENANCE_PROMOTE_LANE_CONTENTION)
+                .with_description("Promotion lanes whose locks were busy, by state/queue/outcome")
+                .with_unit("{lane}")
+                .build(),
             jobs_in_flight: meter
                 .i64_up_down_counter(names::JOB_IN_FLIGHT)
                 .with_description("Current number of in-flight jobs")
@@ -749,6 +760,23 @@ impl AwaMetrics {
             .record(duration.as_secs_f64(), &attrs);
     }
 
+    /// Record a promotion lane whose locks were busy past the leader's wait
+    /// budget, with what became of it (`handed_off`, `waiter_limit`,
+    /// `waiter_timed_out`).
+    pub fn record_promotion_lane_contention(
+        &self,
+        state: &str,
+        queue: &str,
+        outcome: &'static str,
+    ) {
+        let attrs = [
+            opentelemetry::KeyValue::new("awa.job.state", state.to_string()),
+            opentelemetry::KeyValue::new("awa.job.queue", queue.to_string()),
+            opentelemetry::KeyValue::new("awa.promote.outcome", outcome),
+        ];
+        self.promotion_lane_contention.add(1, &attrs);
+    }
+
     /// Record in-flight change.
     pub fn record_in_flight_change(&self, queue: &str, delta: i64) {
         let attrs = [opentelemetry::KeyValue::new(
@@ -1162,6 +1190,65 @@ mod tests {
         metrics.record_job_kind_info("minimal", None, None, None, None, &[]);
     }
 
+    #[test]
+    fn promotion_lane_contention_counts_each_outcome_per_state_and_queue() {
+        let exporter = InMemoryMetricExporter::default();
+        let provider = SdkMeterProvider::builder()
+            .with_periodic_exporter(exporter.clone())
+            .build();
+        let metrics = AwaMetrics::new(&provider.meter("awa-metrics-test"));
+
+        metrics.record_promotion_lane_contention("scheduled", "emails", "handed_off");
+        metrics.record_promotion_lane_contention("scheduled", "emails", "handed_off");
+        metrics.record_promotion_lane_contention("retryable", "emails", "waiter_limit");
+        metrics.record_promotion_lane_contention("scheduled", "billing", "waiter_timed_out");
+        provider.force_flush().expect("flush metrics");
+        let resource_metrics = exporter.get_finished_metrics().expect("read metrics");
+
+        let mut counts = BTreeMap::new();
+        for resource in &resource_metrics {
+            for scope in resource.scope_metrics() {
+                for metric in scope.metrics() {
+                    if let (
+                        names::MAINTENANCE_PROMOTE_LANE_CONTENTION,
+                        AggregatedMetrics::U64(MetricData::Sum(sum)),
+                    ) = (metric.name(), metric.data())
+                    {
+                        for point in sum.data_points() {
+                            let label = |key: &str| {
+                                point
+                                    .attributes()
+                                    .find(|kv| kv.key.as_str() == key)
+                                    .map(|kv| kv.value.as_str().into_owned())
+                                    .unwrap_or_default()
+                            };
+                            counts.insert(
+                                (
+                                    label("awa.job.state"),
+                                    label("awa.job.queue"),
+                                    label("awa.promote.outcome"),
+                                ),
+                                point.value(),
+                            );
+                        }
+                    }
+                }
+            }
+        }
+
+        let key = |state: &str, queue: &str, outcome: &str| {
+            (state.to_string(), queue.to_string(), outcome.to_string())
+        };
+        assert_eq!(
+            counts,
+            BTreeMap::from([
+                (key("retryable", "emails", "waiter_limit"), 1),
+                (key("scheduled", "billing", "waiter_timed_out"), 1),
+                (key("scheduled", "emails", "handed_off"), 2),
+            ])
+        );
+    }
+
     /// A second-scale observation must not fall in the SDK's default
     /// millisecond-scale buckets (upper bounds 5, 10, 25, ...), and every
     /// histogram must export the intended explicit boundaries.