fix(relay): align outbox pending index with pgschema-representable shape

The idx_relay_admin_outbox_pending index declared retry_after NULLS FIRST in
both migration 0034 and schema.sql, but pgschema 1.7.4 (the real CI/test-relay
bootstrap path) silently discards per-key NULLS FIRST when it re-emits the
index, producing catalog indoption 0 0 while a fully-migrated database keeps
2 0. The desired-state bootstrap therefore diverged from the migration
contract, and the prior parity regression missed it because it applied
schema.sql via sqlx::raw_sql (which preserves NULLS FIRST) rather than through
bin/pgschema.

Drop NULLS FIRST from the index in both migration 0034 and schema.sql so both
paths converge on plain-ascending (retry_after, created_at). The claim query's
own ORDER BY retry_after NULLS FIRST, created_at ASC keeps the never-retried-
first semantics; Postgres applies that ordering to the small pending candidate
set regardless of the index's stored null ordering, and the partial predicate
is what makes the index selective. Migrations 0032-0035 are branch-local and
unshipped, so editing 0034 carries no checksum/brownfield risk.

Rewrite the parity regression to bootstrap the desired state through the real
bin/pgschema apply binary and assert per-key indoption (pg_index) alongside the
rendered indexdef, so a construct pgschema cannot represent can no longer pass.

Co-authored-by: Will Pfleger <pfleger.will@gmail.com>
Signed-off-by: Will Pfleger <pfleger.will@gmail.com>

crates/buzz-db/src/migration.rs

@@ -159,6 +159,42 @@ mod tests {
 
     const TEST_DB_URL: &str = "postgres://buzz:buzz_dev@localhost:5432/buzz"; // sadscan:disable np.postgres.1
 
+    /// Connection parameters parsed out of a `postgres://user:pass@host:port/db`
+    /// URL so the parity test can pass them to the `bin/pgschema` binary, which
+    /// takes discrete `--host/--port/--user/--password/--db` flags rather than a
+    /// URL. Only the shapes this test emits (`BUZZ_TEST_DATABASE_URL` /
+    /// `DATABASE_URL` / `TEST_DB_URL`) are supported.
+    struct PgConn {
+        host: String,
+        port: u16,
+        user: String,
+        password: String,
+    }
+
+    fn parse_pg_url(url: &str) -> PgConn {
+        let opts: sqlx::postgres::PgConnectOptions =
+            url.parse().expect("parse postgres connection url");
+        PgConn {
+            host: opts.get_host().to_owned(),
+            port: opts.get_port(),
+            user: opts.get_username().to_owned(),
+            password: parse_pg_password(url),
+        }
+    }
+
+    /// `PgConnectOptions` intentionally does not expose the password via a
+    /// getter, so read it straight out of the URL authority. Falls back to the
+    /// `PGPASSWORD` env var, then empty.
+    fn parse_pg_password(url: &str) -> String {
+        url.split_once("://")
+            .and_then(|(_, rest)| rest.split_once('@'))
+            .map(|(authority, _)| authority)
+            .and_then(|authority| authority.split_once(':'))
+            .map(|(_, pass)| pass.to_owned())
+            .or_else(|| std::env::var("PGPASSWORD").ok())
+            .unwrap_or_default()
+    }
+
     #[derive(Debug, Clone, Copy, PartialEq, Eq)]
     enum ConstraintKind {
         ForeignKey,
@@ -1839,12 +1875,23 @@ mod tests {
     /// match — nothing enforces that automatically, and the lease/claim-token
     /// migrations (0034/0035) once drifted for exactly this reason.
     ///
-    /// This bootstraps one probe database from `schema.sql` and migrates
-    /// another through 1–35, then asserts the two admin tables have identical
-    /// column definitions (name, type, nullability, default) and identical
-    /// catalog-normalized index definitions. Columns are keyed by name, not
-    /// ordinal, because migrations append via `ALTER TABLE` while `schema.sql`
-    /// declares them inline — positions legitimately differ, shapes must not.
+    /// This bootstraps one probe database from `schema.sql` **through the real
+    /// `bin/pgschema apply` binary** — the exact path CI (`ci.yml`) and both
+    /// test-relay launchers take — and migrates another through 1–35, then
+    /// asserts the two admin tables have identical column definitions (name,
+    /// type, nullability, default) and identical index shapes, including each
+    /// key's catalog sort/null options (`pg_index.indoption`). Columns are keyed
+    /// by name, not ordinal, because migrations append via `ALTER TABLE` while
+    /// `schema.sql` declares them inline — positions legitimately differ, shapes
+    /// must not.
+    ///
+    /// Driving the real binary is load-bearing: `pgschema` 1.7.4 discards
+    /// per-key `NULLS FIRST`/`NULLS LAST` when it re-emits an index, so a naive
+    /// `sqlx::raw_sql(schema.sql)` bootstrap would preserve ordering the actual
+    /// deployment path silently drops — the same false-confidence class as the
+    /// drift this test guards against. `indoption` (not just `indexdef` text) is
+    /// asserted so a resurrected `NULLS FIRST` in a migration that `pgschema`
+    /// cannot represent is caught.
     #[tokio::test]
     #[ignore = "requires Postgres"]
     async fn admin_schema_parity_between_desired_state_and_migrations() {
@@ -1866,21 +1913,30 @@ mod tests {
             .expect("read column definitions")
         }
 
-        async fn indexes(pool: &PgPool, table: &str) -> Vec<(String, String)> {
+        // Index name + rendered definition + per-key sort/null options. indoption
+        // is a int2vector rendered as text (e.g. `{2,0}` = NULLS FIRST ASC on key
+        // 0, plain ASC on key 1) so ordering divergences that `indexdef` text may
+        // still show but `pgschema` cannot reproduce are compared structurally.
+        async fn index_shapes(pool: &PgPool, table: &str) -> Vec<(String, String, String)> {
             sqlx::query_as(
-                "SELECT indexname, indexdef FROM pg_indexes \
-                 WHERE schemaname = 'public' AND tablename = $1 \
-                 ORDER BY indexname",
+                "SELECT c.relname, pg_get_indexdef(i.indexrelid), i.indoption::int2[]::text \
+                 FROM pg_class c \
+                 JOIN pg_index i ON i.indexrelid = c.oid \
+                 JOIN pg_class t ON t.oid = i.indrelid \
+                 JOIN pg_namespace n ON n.oid = t.relnamespace \
+                 WHERE n.nspname = 'public' AND t.relname = $1 \
+                 ORDER BY c.relname",
             )
             .bind(table)
             .fetch_all(pool)
             .await
-            .expect("read index definitions")
+            .expect("read index shapes")
         }
 
         let base_url = std::env::var("BUZZ_TEST_DATABASE_URL")
             .or_else(|_| std::env::var("DATABASE_URL"))
             .unwrap_or_else(|_| TEST_DB_URL.to_owned());
+        let conn = parse_pg_url(&base_url);
         let admin = PgPool::connect(&base_url)
             .await
             .expect("connect admin database");
@@ -1897,18 +1953,53 @@ mod tests {
             .await
             .expect("create migrated probe database");
 
-        let schema_sql = std::fs::read_to_string(
-            std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("../../schema/schema.sql"),
-        )
-        .expect("read schema/schema.sql");
+        // Bootstrap the desired-state probe through the real pgschema binary, the
+        // same invocation the test-relay launchers use. The freshly-created probe
+        // db doubles as pgschema's plan database (--plan-*), which avoids the
+        // embedded-Postgres download and matches start-relay-for-tests.sh.
+        let pgschema = std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("../../bin/pgschema");
+        let schema_file =
+            std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("../../schema/schema.sql");
+        let port = conn.port.to_string();
+        let apply = std::process::Command::new(&pgschema)
+            .args([
+                "apply",
+                "--auto-approve",
+                "--file",
+                schema_file.to_str().expect("schema path utf-8"),
+                "--host",
+                &conn.host,
+                "--port",
+                &port,
+                "--user",
+                &conn.user,
+                "--password",
+                &conn.password,
+                "--db",
+                &desired_db,
+                "--plan-host",
+                &conn.host,
+                "--plan-port",
+                &port,
+                "--plan-user",
+                &conn.user,
+                "--plan-password",
+                &conn.password,
+                "--plan-db",
+                &desired_db,
+            ])
+            .output()
+            .expect("run bin/pgschema apply (hermit env required)");
+        assert!(
+            apply.status.success(),
+            "pgschema apply failed: {}\n{}",
+            String::from_utf8_lossy(&apply.stdout),
+            String::from_utf8_lossy(&apply.stderr),
+        );
+
         let desired = PgPool::connect(&format!("{base_prefix}/{desired_db}"))
             .await
             .expect("connect desired-state probe database");
-        sqlx::raw_sql(AssertSqlSafe(schema_sql))
-            .execute(&desired)
-            .await
-            .expect("apply desired-state schema");
-
         let migrated = PgPool::connect(&format!("{base_prefix}/{migrated_db}"))
             .await
             .expect("connect migrated probe database");
@@ -1925,10 +2016,12 @@ mod tests {
                  from the migrations; update schema/schema.sql to match"
             );
             assert_eq!(
-                indexes(&desired, table).await,
-                indexes(&migrated, table).await,
-                "index parity mismatch for {table}: schema.sql desired state has drifted \
-                 from the migrations; update schema/schema.sql to match"
+                index_shapes(&desired, table).await,
+                index_shapes(&migrated, table).await,
+                "index-shape parity mismatch for {table}: the pgschema-bootstrapped desired \
+                 state (including per-key indoption) has drifted from the migrations. If a \
+                 migration uses a construct pgschema cannot represent (e.g. NULLS FIRST), the \
+                 migration and schema.sql must both use a representable shape."
             );
         }
 

crates/buzz-db/src/relay_admin_actions.rs

@@ -1322,6 +1322,11 @@ pub async fn claim_pending_outbox_batch(
         WHERE state = 'pending'
           AND (lease_expires_at IS NULL OR lease_expires_at < now())
           AND (retry_after IS NULL OR retry_after <= now())
+        -- NULLS FIRST is carried by this ORDER BY, not by the supporting index
+        -- (idx_relay_admin_outbox_pending is plain ascending so the desired-state
+        -- schema can match it via pgschema). Never-retried rows (retry_after IS
+        -- NULL) are claimed before rescheduled ones; Postgres applies this
+        -- ordering to the small pending candidate set regardless of index shape.
         ORDER BY retry_after NULLS FIRST, created_at ASC
         LIMIT $1
         FOR UPDATE SKIP LOCKED

migrations/0034_relay_admin_action_lease.sql

@@ -23,7 +23,16 @@ ALTER TABLE relay_admin_outbox
 
 -- Replace the pending-delivery index to include retry_after so the worker
 -- claim query can filter efficiently.
+--
+-- The index column order is plain ascending (retry_after, created_at); the
+-- claim query's own ORDER BY carries the NULLS FIRST semantics (never-retried
+-- rows, retry_after IS NULL, claim first). Postgres sorts the pending candidate
+-- set correctly regardless of the index's stored null ordering, and the partial
+-- WHERE state = 'pending' predicate is what makes this index selective. Keeping
+-- the declaration free of NULLS FIRST lets the desired-state schema (applied via
+-- pgschema, which does not emit per-key null ordering) match this catalog shape
+-- exactly — see admin_schema_parity_between_desired_state_and_migrations.
 DROP INDEX IF EXISTS idx_relay_admin_outbox_pending;
 CREATE INDEX idx_relay_admin_outbox_pending
-    ON relay_admin_outbox (retry_after NULLS FIRST, created_at)
+    ON relay_admin_outbox (retry_after, created_at)
     WHERE state = 'pending';

schema/schema.sql

@@ -1776,7 +1776,7 @@ CREATE TABLE relay_admin_outbox (
 CREATE INDEX idx_relay_admin_outbox_action
     ON relay_admin_outbox (action_id);
 CREATE INDEX idx_relay_admin_outbox_pending
-    ON relay_admin_outbox (retry_after NULLS FIRST, created_at)
+    ON relay_admin_outbox (retry_after, created_at)
     WHERE state = 'pending';
 
 INSERT INTO _operator_global_tables (table_name, reason) VALUES