reliar_store_postgres/

migrate.rs

//! Explicit migration entry point (ADR 0018).
//!
//! **Never invoked implicitly** — no constructor, `Default`, or `acquire` runs a migration.
//! Reliar's bookkeeping lives in its own schema's `_migrations` table, never the shared,
//! one-per-database `_sqlx_migrations` sqlx would otherwise write to, so this can be added to a
//! database a host already migrates with its own tooling without either side noticing the other.

use core::fmt;
use std::time::Duration;

use sqlx::migrate::Migrator;
use sqlx::postgres::PgConnection;
use sqlx::{Connection, Executor, PgPool};

/// The first retry delay [`migrate`]'s lock poll waits between failed `pg_try_advisory_lock`
/// attempts (ADR 0040 amendment A).
const LOCK_POLL_FIRST_RETRY: Duration = Duration::from_millis(50);
/// The retry delay [`migrate`]'s lock poll backs off to and caps at.
const LOCK_POLL_MAX_RETRY: Duration = Duration::from_secs(1);

/// The crate's migrations, embedded at compile time from `migrations/` — the single source of
/// truth (ADR 0018): `cargo publish` packages only files under the crate's own directory, and
/// `sqlx::migrate!` resolves relative to `CARGO_MANIFEST_DIR` at compile time, so the SQL must
/// live here rather than at the repository root.
static MIGRATOR: Migrator = sqlx::migrate!("./migrations");

/// Where [`migrate`] creates Reliar's schema and its bookkeeping table.
///
/// ```
/// use reliar_store_postgres::MigrateOptions;
///
/// let options = MigrateOptions::default().schema("orders");
/// assert_eq!(options.schema, "orders");
/// ```
#[derive(Clone, Copy, Debug)]
#[non_exhaustive]
pub struct MigrateOptions<'a> {
    /// The schema to create (`CREATE SCHEMA IF NOT EXISTS`) and use for both the data tables
    /// and the `_migrations` bookkeeping table. Put this same value first on the connection's
    /// `search_path` — in the connection URL (`options=-c search_path=<schema>,public`) or with
    /// `ALTER ROLE <role> SET search_path = <schema>, public` — so every unqualified reference
    /// resolves here; Reliar does not verify this for you (ADR 0047). **Lowercase only**
    /// (`[a-z_][a-z0-9_$]*`, at most 63 bytes) — an uppercase name is rejected with
    /// [`MigrateError::InvalidSchema`] rather than silently folded, since PostgreSQL itself would
    /// fold it inconsistently across an unquoted reference (ADR 0040 §5).
    pub schema: &'a str,
}

impl Default for MigrateOptions<'_> {
    fn default() -> Self {
        Self { schema: "reliar" }
    }
}

impl<'a> MigrateOptions<'a> {
    /// Sets [`Self::schema`]. `#[non_exhaustive]` forbids struct-literal construction outside
    /// this crate, so this is the only way to migrate into a non-default schema.
    ///
    /// ```
    /// use reliar_store_postgres::MigrateOptions;
    /// let options = MigrateOptions::default().schema("orders");
    /// assert_eq!(options.schema, "orders");
    /// ```
    #[must_use]
    pub const fn schema(mut self, schema: &'a str) -> Self {
        self.schema = schema;

        self
    }
}

/// [`migrate`]'s failure. **Provider-owned**, not a re-export of `sqlx::migrate::MigrateError`:
/// a rejected schema identifier has no variant in `sqlx`'s own type to
/// report it as, since that check happens before any `sqlx::migrate` code runs at all.
///
/// ```no_run
/// # async fn run(pool: sqlx::PgPool) -> Result<(), Box<dyn std::error::Error>> {
/// use reliar_store_postgres::{MigrateOptions, migrate};
///
/// let options = MigrateOptions::default().schema("Not-Lowercase");
/// if let Err(err) = migrate(&pool, options).await {
///     eprintln!("migration failed: {err}");
/// }
/// # Ok(())
/// # }
/// ```
#[derive(Debug)]
#[non_exhaustive]
pub enum MigrateError {
    /// `options.schema` is not a valid PostgreSQL identifier (`[a-z_][a-z0-9_$]*`, at most 63
    /// bytes, **lowercase only** — PostgreSQL folds an unquoted identifier to lowercase, so an
    /// uppercase name would migrate into a schema the rest of the crate, and the host's own
    /// `search_path`, can never consistently resolve; ADR 0040 §5). Checked **before** the name
    /// reaches `dangerous_set_table_name`, which is string interpolation into DDL.
    InvalidSchema {
        /// The rejected schema name.
        schema: String,
    },

    /// Any failure from `sqlx::migrate::Migrator::run` or the dedicated connection's own setup
    /// (a connection failure, a checksum mismatch against an already-applied file, …).
    Sqlx {
        /// The underlying `sqlx` migration error.
        source: sqlx::migrate::MigrateError,
    },
}

impl fmt::Display for MigrateError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::InvalidSchema { schema } => write!(
                f,
                "{schema:?} is not a valid PostgreSQL identifier (expected \
                 [a-z_][a-z0-9_$]*, at most 63 bytes, lowercase only — PostgreSQL folds an \
                 unquoted identifier to lowercase, so an uppercase name would resolve \
                 inconsistently)"
            ),
            Self::Sqlx { source } => write!(f, "migration failed: {source}"),
        }
    }
}

impl std::error::Error for MigrateError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            Self::Sqlx { source } => Some(source),
            Self::InvalidSchema { .. } => None,
        }
    }
}

impl From<sqlx::migrate::MigrateError> for MigrateError {
    fn from(source: sqlx::migrate::MigrateError) -> Self {
        Self::Sqlx { source }
    }
}

impl From<sqlx::Error> for MigrateError {
    fn from(source: sqlx::Error) -> Self {
        Self::Sqlx {
            source: sqlx::migrate::MigrateError::Execute(source),
        }
    }
}

/// Reliar's own advisory-lock key for [`migrate`], derived from `schema` alone (ADR 0040
/// amendment A) — **never** sqlx's own `generate_lock_id` (private, and keyed on the database:
/// sharing it would serialize Reliar's migration behind the host's own migrator, dragging any
/// blocked statement of the host's into `0002`'s `CREATE INDEX CONCURRENTLY` wait set). Schema in
/// the key, not the database, because advisory locks are already per-database and a multi-tenant
/// host migrating several schemas should not run their index builds strictly in series
/// (`PROC_IN_SAFE_IC`, PostgreSQL 14+, this crate's floor is 18, lets two `CONCURRENTLY` builds on
/// different tables proceed without waiting on each other's snapshots).
///
/// FNV-1a 64, spelled out rather than reached for from `std::hash::DefaultHasher` (whose output
/// is explicitly not stable across processes or releases, so it cannot key a value two different
/// connections must agree on).
fn migration_lock_id(schema: &str) -> i64 {
    const OFFSET: u64 = 0xcbf2_9ce4_8422_2325;
    const PRIME: u64 = 0x0000_0100_0000_01b3;
    let mut hash = OFFSET;

    for byte in b"reliar.migrate:v1:".iter().chain(schema.as_bytes()) {
        hash ^= u64::from(*byte);

        hash = hash.wrapping_mul(PRIME);
    }

    i64::from_le_bytes(hash.to_le_bytes())
}

/// Serializes concurrent [`migrate`] callers on the same `schema` **without** sqlx's own
/// `Migrator::set_locking(true)` (ADR 0040 amendment A): that path takes the lock with a
/// *blocking* `SELECT pg_advisory_lock($1)`, and a blocked statement holds an open snapshot for
/// as long as it waits — which deadlocks against `0002`'s `CREATE INDEX CONCURRENTLY`, itself
/// waiting for every older snapshot to end, the moment a second caller's lock attempt overlaps
/// the first caller's index build. `pg_try_advisory_lock` returns at once whether or not it
/// acquired the lock, so between attempts this connection is genuinely idle: no open statement,
/// no snapshot, nothing for a concurrent index build to wait on.
///
/// **Session-level, not transaction-level** (`pg_try_advisory_lock`, not the `_xact_` variant): a
/// lock tied to a transaction would force one open across the whole run, defeating `0002`'s
/// `-- no-transaction` marker outright.
///
/// The wait is **unbounded**, deliberately: `CREATE INDEX CONCURRENTLY` on a large table can
/// legitimately take minutes, and a deadline short enough to matter would fail exactly the
/// deploy this exists to let through cleanly. A caller that wants a bound wraps the call to
/// [`migrate`] in `tokio::time::timeout` — dropping that future while this is polling releases
/// nothing, because nothing is held.
async fn acquire_migration_lock(conn: &mut PgConnection, lock_id: i64) -> Result<(), sqlx::Error> {
    let mut backoff = LOCK_POLL_FIRST_RETRY;

    loop {
        let acquired =
            sqlx::query_scalar!(r#"SELECT pg_try_advisory_lock($1) AS "acquired!""#, lock_id)
                .fetch_one(&mut *conn)
                .await?;

        if acquired {
            return Ok(());
        }

        if backoff == LOCK_POLL_FIRST_RETRY {
            tracing::info!(
                "another migrate() call holds Reliar's migration lock for this schema; waiting"
            );
        }

        tokio::time::sleep(backoff).await;

        backoff = (backoff * 2).min(LOCK_POLL_MAX_RETRY);
    }
}

/// Releases [`acquire_migration_lock`]'s lock. Best-effort: called on every path out of
/// [`migrate`] once the lock is held, but its own failure is never allowed to shadow the
/// migration run's result — ending the session (`conn.close()`, right after) releases the lock
/// regardless, and sqlx's own `run_direct` does not unlock on its error path either.
async fn release_migration_lock(conn: &mut PgConnection, lock_id: i64) {
    let result = sqlx::query_scalar!(r#"SELECT pg_advisory_unlock($1) AS "released!""#, lock_id)
        .fetch_one(&mut *conn)
        .await;

    if let Err(err) = result {
        tracing::warn!(error = %err, "failed to release Reliar's migration lock; ending the session releases it anyway");
    }
}

/// Applies Reliar's migrations. **Never invoked implicitly.** `pool` must reach a **PostgreSQL 18
/// or later** server — a hard requirement, with no older-version fallback. This is a stated
/// requirement, not a checked one: `migrate()` issues no version probe, and a server below the
/// floor fails later, at whichever migration file or query first needs a PostgreSQL 18 feature
/// (`uuidv7()`, in practice).
///
/// Creates `options.schema` if it does not exist, keeps bookkeeping in
/// `<schema>._migrations` — never `_sqlx_migrations` — and serializes concurrent callers with
/// **Reliar's own** advisory lock, acquired by polling (ADR 0040 amendment A; not
/// `sqlx::migrate`'s built-in blocking one), so every caller after the first observes `Ok(())`.
/// **Idempotent.**
/// Self-contained: does not depend on the caller's `search_path` (ADR 0018) — `create_schema`
/// plus the qualified bookkeeping table name make it work over a pool whose URL never set one.
///
/// ```no_run
/// # async fn run(pool: sqlx::PgPool) -> Result<(), Box<dyn std::error::Error>> {
/// use reliar_store_postgres::{MigrateOptions, migrate};
///
/// migrate(&pool, MigrateOptions::default()).await?;
/// # Ok(())
/// # }
/// ```
///
/// # The lock wait is unbounded, the connection must be a real session, and timing matters
///
/// A second concurrent caller can wait for the first for as long as that first run takes —
/// legitimately minutes for `CREATE INDEX CONCURRENTLY` on a large table — and this function
/// never times that wait out on its own; wrap the call in `tokio::time::timeout` if a bound is
/// needed. Dropping that future while it is still polling for the lock (before `migrator.run`
/// starts) is exactly as clean as it sounds — nothing is held between poll attempts, as noted
/// below. Dropping it **after** the lock is acquired, while `migrator.run` itself is executing
/// (e.g. mid-`CREATE INDEX CONCURRENTLY`), is different: the explicit unlock query never runs, so
/// the advisory lock is released only when the dropped connection's own teardown ends the
/// session, not by this function's normal path — and whatever DDL was in flight is left exactly
/// as any other interrupted `CONCURRENTLY` build would be (see the recovery step below).
/// `pool`'s connection URL **must not point at a transaction-mode pooler**: `migrate()`
/// needs one real session for the run's whole duration, both for `SET search_path` and for the
/// session-level advisory lock, and a pooler that hands out a different backend per statement
/// would silently break both (the `outbox_pgdog` test in this crate's suite migrates over a
/// direct connection for exactly this reason, before ever pooling). Finally, `CREATE INDEX
/// CONCURRENTLY` (in `0002_outbox_claimable_index.sql`) must wait for every transaction that was
/// already open when it started to finish, regardless of what table that transaction touches —
/// run `migrate()` when the database has no other long-running transaction in flight.
///
/// # Upgrading from 0.3.0
///
/// A host that only ever calls this function has nothing to do — `migrate()` applies
/// `0002`/`0003`/`0004` the same way it always applied `0001`. `0004_inbox.sql` (inbox contract
/// §3.2, ADR 0042) adds the `inbox` table used by [`crate::PostgresInboxStore`] — a brand-new,
/// empty table, so it runs in an ordinary transaction like `0001` and needs none of `0002`'s
/// `CREATE INDEX CONCURRENTLY` caveats below. A host that instead applies the published
/// `.sql` artifact through its own DBA pipeline (Flyway, Liquibase, sqitch, golang-migrate, a raw
/// `psql` invocation, …) **may not be** interchangeable with this function for `0002`: see
/// `docs/guides/postgres.md`'s "`migrate()` vs. the release SQL artifact" section for the
/// per-tool equivalent of "run this one file outside a transaction" that `0002`'s `CREATE INDEX
/// CONCURRENTLY` requires (`sqlx`'s own `-- no-transaction` marker means nothing to another
/// tool), and the same section's note on `0003`'s `SET LOCAL lock_timeout`, which needs an active
/// transaction to have any effect.
///
/// # Upgrading to 0.7.0 (`outbox` gains its own row identity)
///
/// `0005`–`0010` give the `outbox` row a database-assigned `id` (`pk_outbox`) separate from the
/// client-minted `message_id` it used to share one column with (ADR 0044). **Run `migrate()`
/// before starting 0.7.0 application code**: 0.7.0 reads/writes `message_id`, which does not exist
/// until `0005` applies. A 0.6.0 binary still running against the migrated schema keeps working for
/// every row it already leased or that predates the migration (`id == message_id` for those rows,
/// by construction — see `0006`'s backfill), but its own `enqueue` fails loudly on every new row,
/// since its `INSERT` no longer names every `NOT NULL` column — the caller's transaction rolls
/// back rather than writing a row nobody could later identify correctly. See
/// `docs/guides/postgres.md` for the full rolling-upgrade table and the recommended
/// stop-dispatchers-then-migrate procedure. `0006`'s backfill is the one step whose cost scales
/// with table size; its own doc comment carries the batched, restartable escape hatch for a
/// `statement_timeout` too short to let it complete in one statement.
///
/// # `0002_outbox_claimable_index.sql`, `0007`–`0009`, `0012`–`0013`, `0015` run outside a
/// transaction
///
/// Seven migrations issue `CREATE INDEX CONCURRENTLY` (ADR 0040 §2, ADR 0044 §4, ADR 0049
/// Amendment A, ADR 0050 §6), which PostgreSQL refuses inside a transaction block; sqlx's
/// `-- no-transaction` marker keeps each of them (and only them) out of one. `CONCURRENTLY`
/// cannot roll back on failure, so a connection drop or cancellation mid-build leaves an
/// **invalid** index rather than undoing itself:
///
/// ```text
/// ERROR: relation "ix_outbox_claimable" already exists
/// ```
///
/// (or `ix_outbox_id` / `ix_outbox_message_id` / `ix_outbox_dead_cursor` / `ix_outbox_claimable_id`
/// / `ix_outbox_ordering_key_id` / `ix_outbox_claimable_v2`) on the next `migrate()` call means
/// exactly that for the named index. Recover with, against the same schema:
///
/// ```sql
/// DROP INDEX CONCURRENTLY ix_outbox_claimable;   -- or ix_outbox_id / ix_outbox_message_id /
///                                                 -- ix_outbox_dead_cursor / ix_outbox_claimable_id /
///                                                 -- ix_outbox_ordering_key_id / ix_outbox_claimable_v2
/// ```
///
/// then re-run `migrate()` from the start — it is idempotent and will rebuild the index and
/// continue: `0003_drop_ix_outbox_pending.sql` refuses to drop `ix_outbox_pending` unless
/// `ix_outbox_claimable` exists and is valid, and `0010_outbox_primary_key_swap.sql` refuses to
/// promote `ix_outbox_id` to `pk_outbox` (or drop the two indexes `ix_outbox_dead_cursor`
/// supersedes) unless all three of `ix_outbox_id`, `ix_outbox_message_id` and
/// `ix_outbox_dead_cursor` exist and are valid.
///
/// `0014_drop_outbox_sequence.sql` guards the same way for `0012`/`0013`'s pair: it refuses to drop
/// the `sequence` column (and the indexes that name it) unless both `ix_outbox_claimable_id` and
/// `ix_outbox_ordering_key_id` exist and are valid, and names both in its error when they are not:
///
/// ```text
/// ix_outbox_claimable_id / ix_outbox_ordering_key_id are missing or invalid; rebuild them
/// concurrently, then re-run migrate()
/// ```
///
/// Recover the same way as above — `DROP INDEX CONCURRENTLY ix_outbox_claimable_id;` and/or
/// `ix_outbox_ordering_key_id`, whichever the message names, then re-run `migrate()`: `0012`/`0013`
/// rebuild the transient index and `0014` proceeds to drop `sequence` and promote the permanent
/// names.
///
/// `0016_drop_outbox_locked_until.sql` guards the same way, beside `0014`'s, for `0015`'s build: it
/// refuses to drop `locked_until`/`ck_outbox_lease` (and rename the transient index to its
/// permanent name) unless `ix_outbox_claimable_v2` exists and is valid, and names it in its error
/// when it is not:
///
/// ```text
/// ix_outbox_claimable_v2 is missing or invalid; rebuild it concurrently, then re-run migrate()
/// ```
///
/// Recover the same way as above — `DROP INDEX CONCURRENTLY ix_outbox_claimable_v2;` — then re-run
/// `migrate()`: `0015` rebuilds the transient index and `0016` proceeds to drop
/// `locked_until`/`ck_outbox_lease` and rename the transient index to its permanent name.
///
/// # Errors
///
/// Returns [`MigrateError::InvalidSchema`] when `options.schema` is not a valid PostgreSQL
/// identifier, or [`MigrateError::Sqlx`] for a connection failure, a checksum mismatch against an
/// already applied file, a server too old for a migration file's own SQL (`uuidv7()`, PostgreSQL
/// 18+), or any other failure `sqlx::migrate::Migrator::run` reports — including a `0003` run
/// against a missing/invalid `ix_outbox_claimable`, a `0014` run against a missing/invalid
/// `ix_outbox_claimable_id`/`ix_outbox_ordering_key_id`, or a `0016` run against a missing/invalid
/// `ix_outbox_claimable_v2` (see above).
pub async fn migrate(pool: &PgPool, options: MigrateOptions<'_>) -> Result<(), MigrateError> {
    // Validated once, before it is ever interpolated into `dangerous_set_table_name`/`SET
    // search_path` below, both of which build SQL text from this value rather than binding it
    // as data.
    if !is_valid_schema_name(options.schema) {
        return Err(MigrateError::InvalidSchema {
            schema: options.schema.to_owned(),
        });
    }

    // `Migrator` has no `Clone` impl, but every field is public (`migrate!()` relies on that to
    // construct the static in a const-promotable context), so a field-by-field copy is the
    // sanctioned way to get a mutable instance without touching the static (ADR 0018).
    let mut migrator = Migrator {
        migrations: MIGRATOR.migrations.clone(),
        ignore_missing: MIGRATOR.ignore_missing,
        locking: MIGRATOR.locking,
        no_tx: MIGRATOR.no_tx,
        table_name: MIGRATOR.table_name.clone(),
        create_schemas: MIGRATOR.create_schemas.clone(),
    };
    migrator.create_schema(options.schema.to_owned());
    migrator.dangerous_set_table_name(format!("{}._migrations", options.schema));
    // Reliar's own poll-based lock, not sqlx's blocking one — see `acquire_migration_lock`
    // (ADR 0040 amendment A). `set_locking(false)` (the default is `true`) turns off
    // `Migrator::run`'s built-in mutual exclusion so this is the only lock in play.
    migrator.set_locking(false);

    // `SET search_path` (unqualified migration SQL needs it, ADR 0018) is session-level, and so
    // is the migration lock below — both need a dedicated connection, never `pool.acquire()`,
    // since sqlx never resets a session-level GUC or an advisory lock when a pooled connection
    // is released. The URL behind `pool` must therefore not point at a transaction-mode pooler:
    // a session is required for both.
    let connect_options = pool.connect_options();
    let mut conn = PgConnection::connect_with(&connect_options).await?;

    conn.execute(sqlx::query(sqlx::AssertSqlSafe(format!(
        "SET search_path = \"{}\", public",
        options.schema.replace('"', "\"\"")
    ))))
    .await?;

    let lock_id = migration_lock_id(options.schema);
    acquire_migration_lock(&mut conn, lock_id).await?;
    let run_result = migrator.run(&mut conn).await;
    release_migration_lock(&mut conn, lock_id).await;
    run_result?;

    conn.close().await?;

    Ok(())
}

/// Validates a schema name against PostgreSQL's unquoted-identifier grammar, restricted to
/// **lowercase** (`[a-z_][a-z0-9_$]*`, at most 63 bytes — Postgres's own `NAMEDATALEN` limit)
/// **before** it is ever interpolated into `SET search_path`/`dangerous_set_table_name`, both of
/// which build SQL text from this value rather than binding it as data. The only caller left
/// after ADR 0047 (the stores themselves never validate a schema name — they never see one).
///
/// **Lowercase only, not merely case-insensitive (ADR 0040 §5).** PostgreSQL folds an *unquoted*
/// identifier to lowercase, so `schema = "Foo"` would migrate into a schema literally named
/// `"Foo"` (quoted) while every unqualified reference — the claim, `stats()`, the host's own
/// `search_path` — resolves the unquoted, lowercase-folded `foo` instead: a mismatch this crate
/// cannot detect from inside a single connection's `search_path`, since the host's own connection
/// string or `ALTER ROLE` also has to agree, and cannot be fixed here. Rejecting every uppercase
/// character removes the class of mismatch instead of chasing it through four call sites.
fn is_valid_schema_name(schema: &str) -> bool {
    let mut chars = schema.chars();
    let Some(first) = chars.next() else {
        return false;
    };

    if !(first.is_ascii_lowercase() || first == '_') {
        return false;
    }

    schema.len() <= 63
        && chars.all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '_' || c == '$')
}