fix(queue-storage): create queue lanes as the schema owner (0.6)

CHANGELOG.md

@@ -6,6 +6,52 @@ Notable changes between releases. Detailed migration notes for storage transitio
 
 ### Fixed
 
+- **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-model/migrations/patches/definer_lane_provisioning.sql

@@ -0,0 +1,74 @@
+-- Schema patch definer_lane_provisioning: create queue lanes as the schema
+-- owner, whichever role touches them first (awa 0.7 migration v053, minus the
+-- schema_version row).
+--
+-- Applied after the v023 definitions, whose awa.refresh_lane_sequence_helpers
+-- now installs two SECURITY DEFINER functions in each queue-storage schema,
+-- provision_lane and provision_lane_sequences, owned by the schema owner. The
+-- lane helpers create a missing sequence through them, and each new sequence
+-- gets the grants of the head table it backs, so a lane is usable by every
+-- runtime role however its creator's default privileges are set. The block
+-- below refreshes the helpers on each installed schema and counts existing
+-- lane sequences that need repair. It does not reassign them: ALTER SEQUENCE
+-- OWNER takes the sequence's ACCESS EXCLUSIVE lock until commit. After the
+-- patch commits, `awa migrate` repairs each one in its own short transaction
+-- with a bounded lock wait (awa.repair_lane_sequence); another runner can do
+-- the same with `awa storage lane-sequences --repair`.
+--
+-- Live load: function replacement only. Reads the catalog; no table DDL, row
+-- scans or sequence locks.
+--
+-- 0.6 delivery: the 0.6 series cannot add a migration version, so this ships as
+-- an idempotent schema patch. `awa migrate` applies it when the schema is
+-- exactly at v040 and the probe in migrations.rs reports it missing;
+-- `awa migrate --sql` prints it as a repeatable R__ script for external
+-- runners. Released 0.6 binaries call the same helpers with the same arguments
+-- and cursor semantics, and lanes they create are owned by the schema owner
+-- too. Run it as the schema owner, or as a migrator that can SET ROLE to it.
+
+DO $$
+DECLARE
+    v_schema TEXT;
+    v_count BIGINT;
+BEGIN
+    FOR v_schema IN
+        SELECT n.nspname
+        FROM pg_namespace AS n
+        WHERE has_schema_privilege(current_user, n.oid, 'USAGE')
+          AND EXISTS (
+              SELECT 1 FROM pg_proc AS awa_p
+              WHERE awa_p.pronamespace = n.oid
+                AND awa_p.proname = 'claim_ready_runtime'
+                AND oidvectortypes(awa_p.proargtypes)
+                    = 'text, bigint, double precision, double precision'
+          )
+        ORDER BY n.nspname
+    LOOP
+        IF to_regclass(format('%I.queue_enqueue_heads', v_schema)) IS NULL
+           OR to_regclass(format('%I.queue_claim_heads', v_schema)) IS NULL
+           OR to_regprocedure(format('%I.queue_lane_sequence_name(text,text,smallint,smallint)', v_schema)) IS NULL
+           OR to_regprocedure(format('%I.set_sequence_next(text,bigint)', v_schema)) IS NULL THEN
+            CONTINUE;
+        END IF;
+
+        -- Serialize with a concurrent prepare_schema() of the same schema. This
+        -- is an advisory lock, not a table lock: claims and completions do not
+        -- take it.
+        PERFORM pg_advisory_xact_lock(
+            hashtextextended(format('awa.queue_storage.install:%I', v_schema), 0)
+        );
+        PERFORM awa.refresh_lane_sequence_helpers(v_schema);
+
+        SELECT count(*) INTO v_count
+        FROM awa.lane_sequence_problems(v_schema) AS problem;
+
+        IF v_count > 0 THEN
+            RAISE NOTICE '% lane sequence(s) in schema % are not owned by the schema owner or lack runtime grants', v_count, v_schema
+                USING HINT = format(
+                    'awa migrate attempts repair after this commits; review its warnings for any sequences left unrepaired. To retry, run awa storage lane-sequences --schema %s --repair as the migration role itself, without SET ROLE.',
+                    v_schema
+                );
+        END IF;
+    END LOOP;
+END;
+$$;