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. SHALL agree with
    /// [`crate::PostgresOutboxSettings::schema`] — [`crate::PostgresOutboxStore::connect`]'s
    /// startup verification fails otherwise, since `outbox` will not resolve where it expects.
    /// **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,
    },

    /// The connected server's `server_version_num` is below
    /// [`crate::MIN_SERVER_VERSION_NUM`] (PostgreSQL 18, ADR 0041) — **no
    /// older-version fallback**. Checked as the **first statement** on `migrate`'s dedicated
    /// connection, before `SET search_path`, schema creation, or any migration file runs — a
    /// refused `migrate()` leaves the database exactly as it found it. `migrate.rs` declares its
    /// own copy of this variant rather than re-exporting
    /// [`crate::PostgresOutboxError::UnsupportedServerVersion`] — the same shape as
    /// [`Self::InvalidSchema`] alongside [`crate::PostgresOutboxError::InvalidSchema`].
    UnsupportedServerVersion {
        /// [`crate::MIN_SERVER_VERSION_NUM`], restated on the value.
        required: u32,
        /// The `server_version_num` this connection reported.
        detected: u32,
    },
}

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}"),
            Self::UnsupportedServerVersion { required, detected } => write!(
                f,
                "PostgreSQL 18 or newer is required (server_version_num >= {required}); \
                 detected {detected} — there is no supported way to run Reliar below the floor"
            ),
        }
    }
}

impl std::error::Error for MigrateError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            Self::Sqlx { source } => Some(source),
            Self::InvalidSchema { .. } | Self::UnsupportedServerVersion { .. } => 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, checked here as the
/// first statement on this function's dedicated connection (ADR 0041):
/// a server below the floor returns [`MigrateError::UnsupportedServerVersion`] before the schema
/// is created or any migration file runs.
///
/// 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` run outside a transaction
///
/// Four migrations issue `CREATE INDEX CONCURRENTLY` (ADR 0040 §2, ADR 0044 §4), 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`) 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
/// ```
///
/// 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.
///
/// # Errors
///
/// Returns [`MigrateError::InvalidSchema`] when `options.schema` is not a valid PostgreSQL
/// identifier, [`MigrateError::UnsupportedServerVersion`] when `pool` reaches a server older
/// than [`crate::MIN_SERVER_VERSION_NUM`] (PostgreSQL 18), or [`MigrateError::Sqlx`] for a
/// connection failure, a checksum mismatch against an already applied file, or any other failure
/// `sqlx::migrate::Migrator::run` reports — including a `0003` run against a missing/invalid
/// `ix_outbox_claimable` (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 !crate::connection::schema::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?;

    // First statement on this connection (ADR 0041) — before `SET
    // search_path`, before `CREATE SCHEMA`, before any migration file runs, so a refused
    // `migrate()` leaves the database exactly as it found it.
    let detected = crate::connection::version::detected_server_version_num(&mut conn).await?;

    if detected < crate::MIN_SERVER_VERSION_NUM {
        return Err(MigrateError::UnsupportedServerVersion {
            required: crate::MIN_SERVER_VERSION_NUM,
            detected,
        });
    }

    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(())
}