Promote enqueue lanes independently and share waiters across deferred states

CHANGELOG.md

@@ -31,6 +31,32 @@ 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 one dedicated task per lane, shared by scheduled and retryable
+  rows. The task alternates states within its 32-batch budget and waits up
+  to 30 s for each lock acquisition; the leader skips the lane meanwhile.
+  Several finite producer transactions queued ahead can exhaust that wait.
+  A timed-out waiter releases its queue position and retries on a later
+  tick, so sustained contention can still prevent promotion. 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 each state and 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 (migration v053).** A lane's enqueue and claim sequences belonged to
   the role that first enqueued into, claimed from or promoted into it, so with

awa-metrics/src/lib.rs

@@ -70,6 +70,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";
@@ -192,6 +193,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.
@@ -482,6 +488,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")
@@ -909,6 +920,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(
@@ -1462,6 +1490,65 @@ mod tests {
         assert!(saw_already_pruned);
     }
 
+    #[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),
+            ])
+        );
+    }
+
     #[test]
     fn prune_attempts_count_every_outcome_per_ring() {
         use awa_model::{PruneDurations, PruneOutcome, SkipReason};

awa-model/src/lib.rs

@@ -54,9 +54,9 @@ pub use partitioned_queue::{
     partition_for_ordering_key, partition_hash64, PartitionedQueue, PartitionedQueueError,
 };
 pub use queue_storage::{
-    ClaimedEntry, ClaimedRuntimeJob, PruneDurations, PruneOutcome, QueueCounts, QueueStorage,
-    QueueStorageConfig, RingLedgerFoldOutcome, RingRotationCadence, RotateOutcome, SkipReason,
-    TerminalDeltaRollupOutcome, TerminalRollupFoldOutcome,
+    ClaimedEntry, ClaimedRuntimeJob, PromoteOutcome, PromotionLane, PruneDurations, PruneOutcome,
+    QueueCounts, QueueStorage, QueueStorageConfig, RingLedgerFoldOutcome, RingRotationCadence,
+    RotateOutcome, SkipReason, TerminalDeltaRollupOutcome, TerminalRollupFoldOutcome,
 };
 pub use storage::StorageStatus;