perf(insert): shorten lane lock holds for COPY and skip-conflict batches

.github/workflows/nightly-chaos.yml

@@ -197,6 +197,17 @@ jobs:
           cargo test --release --package awa --test concurrent_lifecycle_test \
             test_concurrent_queue_sweep -- --exact --ignored --nocapture \
             | tee artifacts/nightly/rust-queue-sweep.txt
+      - name: Run per-lane enqueue contention benchmark
+        if: always()
+        timeout-minutes: 5
+        run: |
+          set -o pipefail
+          # Many producers batch-inserting into one queue lane. Gates the
+          # per-lane rate and p99 through benchmarks/baseline.json, and the
+          # lane-lock hold time relative to the per-row insert function.
+          cargo test --release --package awa --test enqueue_lane_contention_test \
+            test_enqueue_lane_contention_gate -- --exact --ignored --nocapture \
+            | tee artifacts/nightly/rust-enqueue-lane.txt
       - name: Run MVCC horizon overlap benchmark
         if: always()
         timeout-minutes: 10

CHANGELOG.md

@@ -4,8 +4,110 @@ Notable changes between releases. Detailed migration notes for storage transitio
 
 ## [Unreleased]
 
+### Added
+
+- **`insert_many_skip_unique_conflicts` inserts a batch and skips the jobs
+  whose unique key is already claimed.** `insert_many` fails the whole batch
+  with a unique violation, so a producer that replays work had to insert the
+  batch again one job at a time, taking the lane lock once per job. The new
+  function returns the inserted rows in input order, with skipped jobs absent.
+  It needs migration v055 and returns a validation error on an older schema;
+  `insert_many_copy` (and Python's `insert_many_copy`) skips unique conflicts on
+  every schema.
+
+- **The maintenance leader keeps planner statistics fresh on the queue-storage
+  tables autovacuum never analyses.** Autovacuum does not analyse partitioned
+  tables, so the thirteen ring parents (`ready_entries`, `done_entries`,
+  `leases`, ...) had no inherited statistics, and the two small head tables
+  (`queue_claim_heads`, `queue_enqueue_heads`) never reach the autoanalyze
+  threshold, so the planner priced lane joins against them at its defaults
+  while they drove joins over hundreds of thousands of retained rows. The
+  leader now runs `ANALYZE` on them when it is elected and then every five
+  minutes. `ClientBuilder::statistics_refresh_interval` (Python
+  `statistics_refresh_interval_ms`) changes the cadence; zero disables it. One
+  pass takes under a second on a small ring and a few seconds on a large one.
+  It analyses one relation per transaction with a 2 s `lock_timeout` (a call that
+  times out waiting for a lock is retried next interval and recorded as
+  `skipped_lock`) and never makes prune wait: a prune attempt that
+  lands while ANALYZE samples the oldest slot's tables returns `blocked` and
+  retries on its normal backoff. Results are in
+  `awa.maintenance.analyze.attempts` and `awa.maintenance.analyze.duration`
+  (scope `heads`, `ring_parent` or `pass`; outcome `ok`, `skipped_privilege`,
+  `skipped_lock`, `unavailable` or `failed`).
+- **Migration v054 adds `awa.analyze_queue_storage_relation(schema, relation)`,
+  so a runtime role that does not own the tables can refresh statistics.**
+  `ANALYZE` skips a table its caller may not analyse, and the documented
+  runtime grants do not include `MAINTAIN`, so the refresh runs through this
+  `SECURITY DEFINER` function owned by the migration role. It analyses only a
+  head table or ring parent of a schema that `awa.queue_storage_schemas()`
+  reports as trusted. It is not executable by `PUBLIC`, because the lock it
+  takes lasts until the caller's transaction ends and would block ring prune:
+  the migration grants `EXECUTE` to the roles that already hold INSERT, UPDATE
+  and DELETE on the head tables (the runtime grantees), and the function
+  requires the calling session user to hold them. A role created later needs
+  the documented `GRANT EXECUTE ON ALL FUNCTIONS IN SCHEMA awa`. Run
+  `awa migrate` as the schema owner (see
+  [database roles](docs/security/database-roles.md)). A binary running ahead of
+  the migration probes for the function once per pass and, like one whose
+  function's owner may not analyse the tables, logs one warning naming what is
+  missing and skips the refresh without making failing calls or affecting job
+  processing. Earlier binaries never call the function.
+  Missing EXECUTE or head-table privileges are checked before each pass,
+  avoiding a failing helper call for every relation.
+
+### Changed
+
+- **COPY and opt-in skip-conflict batch enqueues hold lane locks only for reservation, ready-row writes and commit.** Migration v055 adds `awa.insert_jobs_compat_batch`, which allocates job ids, claims unique keys in key order, registers lanes and prepares rows before reserving each lane's sequence range in lane order. `insert_many_copy` probes its own connection on every call and uses the function when installed, falling back to per-row inserts on older schemas. `insert_many_skip_unique_conflicts` requires the function. Generic `insert_many` keeps its single-statement per-row path on every supported schema, preserving caller transactions when one process writes to databases at different migration levels. All paths return rows in input order and share the same per-lane lock with older binaries.
+
 ### Fixed
 
+- **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
+  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. Lanes are now created 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. After v053 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 migration. 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 ahead of v053 detects that `provision_lane` is missing and creates
+  lanes as before, and older binaries create correctly owned lanes once v053
+  has refreshed the helpers they call. See "Lane ownership and v053" in the
+  0.6 to 0.7 upgrade guide.
+
+- **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. 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
@@ -15,10 +117,7 @@ Notable changes between releases. Detailed migration notes for storage transitio
   a `QueueStorage` enqueue could deadlock with it (`40P01`). The two insert
   paths now sort rows by `(queue, priority, shard)`, queues compared bytewise,
   before they reach `awa.insert_job_compat`, the order `QueueStorage` already
-  used, and still return rows in input order. Job ids within a batch that
-  spans several lanes are now allocated lane by lane rather than in input
-  order, and when an `insert_many_copy` batch carries the same unique key in
-  more than one lane, the row in the lowest lane is the one inserted. A
+  used, and still return rows in input order. A
   transaction that enqueues with separate statements still takes locks in
   statement order; see "`deadlock detected` while enqueueing" in the
   troubleshooting guide.

awa-cli/src/main.rs

@@ -273,6 +273,23 @@ fn sanitize_filename(description: &str) -> String {
 /// mutating commands require an explicit `--context` or `--database-url`
 /// once more than one context is defined, and are confirmation-gated on
 /// contexts marked `production = true`.
+/// After a migration, say which lane sequences still need repair and how to
+/// finish it. Read-only.
+async fn print_lane_sequence_leftovers(
+    pool: &sqlx::PgPool,
+) -> Result<(), Box<dyn std::error::Error>> {
+    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}");
+    }
+    Ok(())
+}
+
 fn access_for(command: &Commands) -> context::Access {
     use context::Access::{Mutating, ReadOnly};
     match command {
@@ -328,11 +345,13 @@ fn access_for(command: &Commands) -> context::Access {
         },
         Commands::Storage { command } => match command {
             StorageCommands::Status => ReadOnly,
+            StorageCommands::LaneSequences { repair: false, .. } => ReadOnly,
             StorageCommands::Finalize { check: true, .. } => ReadOnly,
             StorageCommands::FlipRingAuthority { check: true, .. } => ReadOnly,
             StorageCommands::Prepare { .. }
             | StorageCommands::PrepareQueueStorageSchema { .. }
             | StorageCommands::PrepareQueue { .. }
+            | StorageCommands::LaneSequences { .. }
             | StorageCommands::Abort
             | StorageCommands::EnterMixedTransition { .. }
             | StorageCommands::Finalize { .. }
@@ -811,7 +830,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")]
@@ -826,6 +845,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
@@ -1182,6 +1212,7 @@ async fn run() -> Result<(), Box<dyn std::error::Error>> {
                 };
                 awa_model::migrations::run_with_options(&pool, options).await?;
                 println!("Migrations applied successfully.");
+                print_lane_sequence_leftovers(&pool).await?;
             }
         }
 
@@ -1933,6 +1964,39 @@ async fn run() -> 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?;

awa-metrics/src/lib.rs

@@ -77,6 +77,8 @@ pub mod names {
     pub const MAINTENANCE_PRUNE_ATTEMPTS: &str = "awa.maintenance.prune.attempts";
     pub const MAINTENANCE_PRUNE_SKIPPED_ROWS: &str = "awa.maintenance.prune.skipped_rows";
     pub const MAINTENANCE_PRUNE_DURATION: &str = "awa.maintenance.prune.duration";
+    pub const MAINTENANCE_ANALYZE_ATTEMPTS: &str = "awa.maintenance.analyze.attempts";
+    pub const MAINTENANCE_ANALYZE_DURATION: &str = "awa.maintenance.analyze.duration";
     pub const STORAGE_TRANSITION_READY: &str = "awa.storage.transition_ready";
     pub const STORAGE_CANONICAL_LIVE_BACKLOG: &str = "awa.storage.canonical_live_backlog";
     pub const STORAGE_LIVE_RUNTIME_CAPABILITY: &str = "awa.storage.live_runtime_capability";
@@ -277,6 +279,13 @@ pub struct AwaMetrics {
     /// rather than one per tick. Fleets alert on this rather than
     /// scraping the matching `tracing::warn!` line. Issue #242.
     pub maintenance_branch_overrun_total: Counter<u64>,
+    /// Planner-statistics `ANALYZE` calls by scope (`heads`, `ring_parent`, `pass`)
+    /// and outcome (`ok`, `skipped_privilege`, `skipped_lock`,
+    /// `unavailable`, `failed`). One increment per relation per refresh pass.
+    pub maintenance_analyze_attempts: Counter<u64>,
+    /// Wall-clock duration of one planner-statistics `ANALYZE` call, by scope
+    /// and outcome.
+    pub maintenance_analyze_duration_seconds: Histogram<f64>,
     /// Current ring `current_slot` per ring, sampled from each rotate call.
     /// The slot number itself isn't meaningful but the rate of advance is —
     /// dashboards plot `rate(slot_changes)` to see whether rotation is
@@ -620,6 +629,21 @@ impl AwaMetrics {
                 )
                 .with_unit("{episode}")
                 .build(),
+            maintenance_analyze_attempts: meter
+                .u64_counter(names::MAINTENANCE_ANALYZE_ATTEMPTS)
+                .with_description(
+                    "Planner-statistics ANALYZE calls by scope (heads/ring_parent) and outcome",
+                )
+                .with_unit("{attempt}")
+                .build(),
+            maintenance_analyze_duration_seconds: meter
+                .f64_histogram(names::MAINTENANCE_ANALYZE_DURATION)
+                .with_description(
+                    "Duration of one planner-statistics ANALYZE call by scope and outcome",
+                )
+                .with_unit("s")
+                .with_boundaries(WAIT_DURATION_BUCKETS_SECONDS.to_vec())
+                .build(),
             ring_current_slot: meter
                 .i64_gauge(names::RING_CURRENT_SLOT)
                 .with_description("Current slot index per ring (queue/lease/claim)")
@@ -1280,6 +1304,24 @@ impl AwaMetrics {
         self.maintenance_branch_overrun_total.add(1, &attrs);
     }
 
+    /// Record one planner-statistics `ANALYZE` call. `scope` is `"heads"`,
+    /// `"ring_parent"` or `"pass"` (a pass-level probe or failure); `outcome` is `"ok"`, `"skipped_privilege"`,
+    /// `"skipped_lock"`, `"unavailable"` or `"failed"`.
+    pub fn record_statistics_analyze(
+        &self,
+        scope: &'static str,
+        outcome: &'static str,
+        duration: Duration,
+    ) {
+        let attrs = [
+            opentelemetry::KeyValue::new("awa.analyze.scope", scope),
+            opentelemetry::KeyValue::new("awa.analyze.outcome", outcome),
+        ];
+        self.maintenance_analyze_attempts.add(1, &attrs);
+        self.maintenance_analyze_duration_seconds
+            .record(duration.as_secs_f64(), &attrs);
+    }
+
     /// Record a ring prune outcome.
     pub fn record_prune_outcome(&self, ring: &'static str, outcome: &awa_model::PruneOutcome) {
         let (label, reason, count) = match outcome {
@@ -1457,7 +1499,7 @@ mod tests {
             .build();
         let metrics = AwaMetrics::new(&provider.meter("awa-metrics-test"));
 
-        let seconds: [(&str, &Histogram<f64>, &[f64]); 9] = [
+        let seconds: [(&str, &Histogram<f64>, &[f64]); 10] = [
             (
                 names::ENQUEUE_DURATION,
                 &metrics.enqueue_duration_seconds,
@@ -1478,6 +1520,11 @@ mod tests {
                 &metrics.maintenance_branch_duration_seconds,
                 &WAIT_DURATION_BUCKETS_SECONDS,
             ),
+            (
+                names::MAINTENANCE_ANALYZE_DURATION,
+                &metrics.maintenance_analyze_duration_seconds,
+                &WAIT_DURATION_BUCKETS_SECONDS,
+            ),
             (
                 names::DISPATCH_WAKE_TO_CLAIM_DURATION,
                 &metrics.dispatch_wake_to_claim_seconds,