fix(worker): keep planner statistics fresh on tables autovacuum never analyses

CHANGELOG.md

@@ -4,8 +4,79 @@ Notable changes between releases. Detailed migration notes for storage transitio
 
 ## [Unreleased]
 
+### Added
+
+- **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.
+
 ### Fixed
 
+- Statistics-refresh permission warnings distinguish caller grants from
+  helper/table ownership and ANALYZE privileges.
+- **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.
 - **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

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,