Forward-port 0.6.14 to main

.github/workflows/ci.yml

@@ -285,6 +285,7 @@ jobs:
         ports:
           - 5432:5432
         options: >-
+          --shm-size=256m
           --health-cmd pg_isready
           --health-interval 10s
           --health-timeout 5s
@@ -328,6 +329,7 @@ jobs:
         ports:
           - 5432:5432
         options: >-
+          --shm-size=256m
           --health-cmd pg_isready
           --health-interval 10s
           --health-timeout 5s
@@ -368,6 +370,7 @@ jobs:
         ports:
           - 5432:5432
         options: >-
+          --shm-size=256m
           --health-cmd pg_isready
           --health-interval 10s
           --health-timeout 5s
@@ -402,6 +405,7 @@ jobs:
         ports:
           - 5432:5432
         options: >-
+          --shm-size=256m
           --health-cmd pg_isready
           --health-interval 10s
           --health-timeout 5s
@@ -449,6 +453,7 @@ jobs:
         ports:
           - 5432:5432
         options: >-
+          --shm-size=256m
           --health-cmd pg_isready
           --health-interval 10s
           --health-timeout 5s
@@ -507,7 +512,7 @@ jobs:
           POSTGRES_DB: awa_test
         ports:
           - 5432:5432
-        options: --health-cmd pg_isready --health-interval 10s --health-timeout 5s --health-retries 5
+        options: --shm-size=256m --health-cmd pg_isready --health-interval 10s --health-timeout 5s --health-retries 5
     defaults:
       run:
         working-directory: awa-python
@@ -579,6 +584,7 @@ jobs:
         ports:
           - 5432:5432
         options: >-
+          --shm-size=256m
           --health-cmd pg_isready
           --health-interval 10s
           --health-timeout 5s
@@ -685,6 +691,7 @@ jobs:
         ports:
           - 5432:5432
         options: >-
+          --shm-size=256m
           --health-cmd pg_isready
           --health-interval 10s
           --health-timeout 5s
@@ -790,6 +797,7 @@ jobs:
         ports:
           - 5432:5432
         options: >-
+          --shm-size=256m
           --health-cmd pg_isready
           --health-interval 10s
           --health-timeout 5s

CHANGELOG.md

@@ -4,6 +4,10 @@ Notable changes between releases. Detailed migration notes for storage transitio
 
 ## [Unreleased]
 
+### Forward-ported from 0.6.14
+
+- Everything [0.6.14](https://github.com/hardbyte/awa/releases/tag/v0.6.14) shipped that this branch did not already have; see its release notes for the details. Promotion runs each enqueue lane in its own transaction, hands a busy lane to a dedicated waiter, and wakes claimers after the lanes commit; `awa.maintenance.promote.lane_contention` counts the hand-offs. Rescue skips an attempt whose retained ready bodies are ambiguous instead of stopping rescue of every other attempt. Lifecycle events no longer re-read a completed job, and a failed re-read no longer fails a committed callback wait. Admin job lists and kind counts find their candidates in read phases that release partition locks between them, and probe receipt closures more cheaply; a job list still loads each candidate with a point read, which holds `ACCESS SHARE` on every ready, lease and claim partition for the length of that read. `awa migrate` rejects `--sql` combined with `--extract-to`. The persona soak suite (`awa/tests/persona_soak_test.rs`, `scripts/persona-soak.sh`) is added.
+
 ### Added
 
 - **The queue-storage prune `lock_timeout` is configurable, and prune

awa-cli/src/main.rs

@@ -43,7 +43,7 @@ enum Commands {
         #[arg(long)]
         extract_to: Option<String>,
         /// Print migration SQL to stdout instead of applying
-        #[arg(long)]
+        #[arg(long, conflicts_with = "extract_to")]
         sql: bool,
         /// Only include migrations after this version (exclusive)
         #[arg(long)]

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};