Test the migration runner and separate drift from operational failures

.github/workflows/schema-deploy.yml

@@ -58,6 +58,15 @@ jobs:
         with:
           sync: false
 
+      # Resolve dependencies before interpreting the runner's exit status. A uv
+      # download or environment failure must fail even during a dry run.
+      - name: Prepare migration interpreter
+        run: |
+          set -Eeuo pipefail
+          uv venv "${RUNNER_TEMP}/migration-venv"
+          uv export --script scripts/migrate-schema.py --no-hashes > "${RUNNER_TEMP}/migration-requirements.txt"
+          uv pip install --python "${RUNNER_TEMP}/migration-venv/bin/python" -r "${RUNNER_TEMP}/migration-requirements.txt"
+
       - name: Configure AWS credentials via OIDC
         uses: aws-actions/configure-aws-credentials@v4
         with:
@@ -97,17 +106,14 @@ jobs:
           export PGSSLROOTCERT="${RUNNER_TEMP}/rds-global-bundle.pem"
           if [ "${DRY_RUN}" = "true" ]; then
             echo "dry_run: reporting status only, applying nothing"
-            # `status` exits 1 on drift (pending/orphaned migrations). For a dry_run that
-            # is the EXPECTED, informational result — not an infra error — so don't let it
-            # fail the step under `set -e`. (The `if` condition exempts the command from
-            # set -e.) The non-dry_run branch below intentionally lets a post-apply `status`
-            # non-zero fail the step: drift after an apply IS a real problem.
-            if uv run --no-project scripts/migrate-schema.py status; then
-              echo "dry_run: no drift — migration set matches the ledger."
-            else
-              echo "dry_run: drift detected (pending or orphaned migrations) — re-run without dry_run to apply."
-            fi
+            status=0
+            "${RUNNER_TEMP}/migration-venv/bin/python" scripts/migrate-schema.py status || status=$?
+            case "$status" in
+              0) echo "dry_run: migration set matches the ledger." ;;
+              1) echo "dry_run: pending or orphaned migrations detected." ;;
+              *) echo "::error::Migration status failed (exit $status)."; exit "$status" ;;
+            esac
           else
-            uv run --no-project scripts/migrate-schema.py apply
-            uv run --no-project scripts/migrate-schema.py status
+            "${RUNNER_TEMP}/migration-venv/bin/python" scripts/migrate-schema.py apply
+            "${RUNNER_TEMP}/migration-venv/bin/python" scripts/migrate-schema.py status
           fi

.github/workflows/web-ci.yml

@@ -76,5 +76,14 @@ jobs:
       - name: Require Docker for the integration suite
         run: docker info > /dev/null
 
+      - name: Install uv for migration tests
+        uses: spiraldb/actions/.github/actions/setup-uv@0.18.6
+        with:
+          sync: false
+
+      - name: Test migration runner
+        working-directory: .
+        run: uv run --no-project --with 'psycopg[binary]>=3.2' python -m unittest discover -s scripts/tests -v
+
       - name: Test
         run: pnpm test

migrations/README.md

@@ -126,3 +126,22 @@ role (another `CREATE ROLE`, `ALTER DEFAULT PRIVILEGES`, DDL on master-owned
 tables, or other superuser-only DDL) should carry
 `-- migrate-schema: requires-superuser` on a comment line so the same
 preflight guards it.
+
+## Runner checks
+
+`status` returns 0 for a matching ledger, 1 for confirmed pending or orphaned filenames, and 2
+for connection, permission, filesystem, or usage failures. `apply` returns 0 on success and 2
+on failure. The schema workflow resolves Python dependencies before interpreting these codes,
+so a dependency download failure cannot appear as an informational dry-run drift result.
+
+Run the CLI integration suite with Docker available:
+
+```bash
+uv run --no-project --with 'psycopg[binary]>=3.2' python -m unittest discover -s scripts/tests -v
+```
+
+The suite applies the real migrations to disposable PostgreSQL 16, checks a second apply,
+transaction rollback, ledger drift, role permissions, and a later migration as `migrator`.
+A non-superuser CREATEROLE bootstrap role exercises the role-grant path. The local `rds_iam`
+stand-in checks membership grants only. These tests do not verify RDS IAM authentication,
+AWS trust, or RDS-specific administrative privileges.

scripts/migrate-schema.py

@@ -28,10 +28,10 @@
 
 Exit codes
 ----------
-- `apply`  : 0 on success (zero or more migrations applied).
+- `apply`  : 0 on success, 2 on operational or usage failure.
 - `status` : 0 when the on-disk migration set matches the ledger;
              1 when there is drift (pending files OR applied-but-deleted
-             files). CI uses this for clean-tree gates.
+             files), 2 on operational or usage failure. CI must not treat 2 as drift.
 """
 
 import argparse
@@ -308,10 +308,7 @@ def main() -> int:
                 print(f"{count} migration(s) applied", file=sys.stderr)
                 return 0
             return status(conn, args.migrations)
-    except FileNotFoundError as e:
-        # Translate the typed exception from `discover` into a clean CLI
-        # error; callers that import this module get the original exception
-        # via `discover` instead of an opaque SystemExit.
+    except (psycopg.Error, OSError, ValueError) as e:
         print(str(e), file=sys.stderr)
         return 2
 

scripts/tests/test_migrate_schema.py

@@ -0,0 +1,138 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+"""Exercise the real CLI against disposable PostgreSQL 16 (Docker required)."""
+
+from pathlib import Path
+import shutil
+import subprocess
+import sys
+import tempfile
+import time
+import unittest
+import uuid
+
+import psycopg
+from psycopg.conninfo import make_conninfo
+
+ROOT = Path(__file__).resolve().parents[2]
+RUNNER = ROOT / "scripts/migrate-schema.py"
+
+
+class MigrationTests(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.container = subprocess.check_output(
+            ["docker", "run", "--rm", "-d", "-e", "POSTGRES_PASSWORD=test",
+             "-p", "127.0.0.1::5432", "postgres:16-alpine"], text=True
+        ).strip()
+        cls.addClassCleanup(subprocess.run, ["docker", "rm", "-f", cls.container],
+                            check=True, stdout=subprocess.DEVNULL)
+        port = subprocess.check_output(
+            ["docker", "port", cls.container, "5432"], text=True
+        ).strip().rsplit(":", 1)[1]
+        cls.dsn = make_conninfo(host="127.0.0.1", port=port, user="postgres",
+                               password="test", dbname="postgres", sslmode="disable",
+                               connect_timeout=2)
+        for _ in range(60):
+            try:
+                with psycopg.connect(cls.dsn):
+                    return
+            except psycopg.OperationalError:
+                time.sleep(0.5)
+        raise RuntimeError("Postgres did not start")
+
+    def setUp(self):
+        database = "test_" + uuid.uuid4().hex
+        with psycopg.connect(self.dsn, autocommit=True) as conn:
+            conn.execute(f'CREATE DATABASE "{database}"')
+        self.target = make_conninfo(self.dsn, dbname=database)
+        self.directory = tempfile.TemporaryDirectory()
+        self.addCleanup(self.directory.cleanup)
+        self.migrations = Path(self.directory.name)
+
+    def run_cli(self, command, expected, *, target=None, migrations=None):
+        result = subprocess.run(
+            [sys.executable, str(RUNNER), command, "--target", target or self.target,
+             "--migrations", str(migrations or self.migrations)],
+            capture_output=True, text=True,
+        )
+        self.assertEqual(result.returncode, expected, result.stdout + result.stderr)
+        return result
+
+    def test_status_distinguishes_drift_and_failure(self):
+        self.run_cli("status", 0)
+        (self.migrations / "001_first.sql").write_text("CREATE TABLE first (id int);")
+        self.run_cli("status", 1)
+        self.run_cli("apply", 0)
+        self.run_cli("status", 0)
+        (self.migrations / "001_first.sql").unlink()
+        self.run_cli("status", 1)
+        self.run_cli("status", 2, migrations=self.migrations / "missing")
+        self.run_cli("status", 2, target=make_conninfo(self.target, port=1))
+        self.run_cli("invalid-command", 2)
+        with psycopg.connect(self.target, autocommit=True) as conn:
+            conn.execute("CREATE ROLE denied_status LOGIN PASSWORD 'test'")
+        self.run_cli("status", 2, target=make_conninfo(self.target, user="denied_status"))
+
+    def test_failed_migration_rolls_back_only_its_ddl_and_ledger(self):
+        (self.migrations / "001_first.sql").write_text("CREATE TABLE first (id int);")
+        second = self.migrations / "002_second.sql"
+        second.write_text("CREATE TABLE second (id int); SELECT missing_column;")
+        self.run_cli("apply", 2)
+        with psycopg.connect(self.target) as conn:
+            self.assertEqual(conn.execute("SELECT filename FROM public._applied_migrations").fetchall(),
+                             [("001_first.sql",)])
+            self.assertIsNone(conn.execute("SELECT to_regclass('second')").fetchone()[0])
+            self.assertIsNotNone(conn.execute("SELECT to_regclass('first')").fetchone()[0])
+        second.write_text("CREATE TABLE second (id int);")
+        self.run_cli("apply", 0)
+        self.assertIn("0 migration(s) applied", self.run_cli("apply", 0).stderr)
+        self.run_cli("status", 0)
+
+    def test_real_bootstrap_and_steady_state_permissions(self):
+        migrations = sorted((ROOT / "migrations").glob("*.sql"))
+        bootstrap_end = max(i for i, path in enumerate(migrations)
+                            if "-- migrate-schema: requires-superuser" in path.read_text())
+        for migration in migrations[:bootstrap_end + 1]:
+            shutil.copy(migration, self.migrations)
+        # CREATEROLE models the non-superuser bootstrap path. The local rds_iam
+        # role exercises guarded grants only, not RDS IAM authentication.
+        with psycopg.connect(self.target, autocommit=True) as conn:
+            conn.execute("CREATE ROLE rds_iam")
+            conn.execute("CREATE ROLE bootstrap LOGIN CREATEROLE PASSWORD 'test'")
+            conn.execute("GRANT rds_iam TO bootstrap WITH ADMIN TRUE")
+            conn.execute("GRANT ALL ON SCHEMA public TO bootstrap WITH GRANT OPTION")
+        bootstrap = make_conninfo(self.target, user="bootstrap")
+        self.run_cli("apply", 0, target=bootstrap)
+        self.run_cli("status", 0, target=bootstrap)
+        with psycopg.connect(self.target, autocommit=True) as conn:
+            for role in ("migrator", "bench_ingest", "bench_read"):
+                conn.execute(f"ALTER ROLE {role} PASSWORD 'test'")
+        migrator = make_conninfo(self.target, user="migrator")
+        for migration in migrations[bootstrap_end + 1:]:
+            shutil.copy(migration, self.migrations)
+        self.run_cli("apply", 0, target=migrator)
+        self.run_cli("status", 0, target=migrator)
+        self.assertIn("0 migration(s) applied", self.run_cli("apply", 0, target=migrator).stderr)
+        (self.migrations / "999_future.sql").write_text("CREATE TABLE future (id int);")
+        self.run_cli("apply", 0, target=migrator)
+        for role, allowed in (("bench_read", "SELECT"), ("bench_ingest", "SELECT,INSERT,UPDATE")):
+            with psycopg.connect(make_conninfo(self.target, user=role)) as conn:
+                for privilege in allowed.split(","):
+                    self.assertTrue(conn.execute("SELECT has_table_privilege('future', %s)", (privilege,)).fetchone()[0])
+                self.assertFalse(conn.execute("SELECT has_table_privilege('future', 'DELETE')").fetchone()[0])
+                self.assertFalse(conn.execute("SELECT has_schema_privilege('public', 'CREATE')").fetchone()[0])
+                self.assertFalse(conn.execute("SELECT has_table_privilege('_applied_migrations', 'SELECT')").fetchone()[0])
+                if role == "bench_read":
+                    self.assertFalse(conn.execute("SELECT has_table_privilege('commits', 'INSERT')").fetchone()[0])
+                else:
+                    conn.execute("INSERT INTO future VALUES (1)")
+                    conn.execute("UPDATE future SET id = 2")
+        (self.migrations / "999_marked.sql").write_text(
+            "-- migrate-schema: requires-superuser\nCREATE TABLE forbidden (id int);"
+        )
+        result = self.run_cli("apply", 2, target=migrator)
+        self.assertIn("master-capable", result.stderr)
+        with psycopg.connect(self.target) as conn:
+            self.assertIsNone(conn.execute("SELECT to_regclass('forbidden')").fetchone()[0])

web/README.md

@@ -14,7 +14,7 @@ pnpm dev          # needs BENCH_DB_* pointing at a database (see below)
 pnpm format:check # prettier
 pnpm lint         # eslint
 pnpm build        # next build; deliberately works WITHOUT a database
-pnpm test         # vitest; the Postgres integration suite needs a Docker daemon
+pnpm test         # vitest; the Postgres integration suite needs Docker and uv
 ```
 
 `next build` never touches the database: every page and route is request-rendered

web/lib/test-harness.ts

@@ -6,18 +6,15 @@
  * is imported only by `*.test.ts` files; it never reaches a production bundle.
  *
  * It centralizes the pieces the suites previously copy-pasted (and that had
- * already begun to drift in name): the Docker probe, the migrations DDL, the
+ * already begun to drift in name): the Docker probe, the migration runner, the
  * container-boot + `BENCH_DB_*` env wiring, and the canonical three-commit
  * chart fixture mirroring `server/tests/common/mod.rs`.
  */
 
-import { execSync } from 'node:child_process';
-import { readdirSync, readFileSync } from 'node:fs';
-import { join } from 'node:path';
+import { execFileSync, execSync } from 'node:child_process';
 import { fileURLToPath } from 'node:url';
 import { PostgreSqlContainer, type StartedPostgreSqlContainer } from '@testcontainers/postgresql';
 import type { Pool } from 'pg';
-import { getPool } from './db';
 
 // Mirror the repo's Python `_docker_available()` precedent: the integration
 // tests need a Docker daemon, so they are skipped (not failed) when one is
@@ -31,44 +28,8 @@ export function dockerAvailable(): boolean {
   }
 }
 
-/** Absolute path of the repository's `migrations/` directory. */
-const MIGRATIONS_DIR = fileURLToPath(new URL('../../migrations', import.meta.url));
-
-/**
- * Every migration file in runner order (sorted filenames), so the suites
- * exercise the same DDL sequence `scripts/migrate-schema.py apply` runs and a
- * future schema migration is automatically covered by the web tests (the
- * web-deploy workflow gates on `migrations/**` for exactly this reason). The
- * full set is applicable here because migrations are substrate-portable by
- * policy (002/004 guard their `rds_iam` grants behind existence checks) and
- * the container connects as the superuser, which satisfies 004's
- * requires-superuser marker.
- */
-const MIGRATION_FILES: readonly string[] = readdirSync(MIGRATIONS_DIR, { withFileTypes: true })
-  .filter((entry) => entry.isFile() && entry.name.toLowerCase().endsWith('.sql'))
-  .map((entry) => entry.name)
-  .sort();
-
-/**
- * The migration-ledger DDL, kept in lockstep with `scripts/migrate-schema.py`
- * (`APPLIED_MIGRATIONS_DDL`): the runner creates the ledger BEFORE applying
- * any file, and `migrations/003` grants on it, so applying the files without
- * the ledger would fail.
- */
-const LEDGER_DDL = `CREATE TABLE IF NOT EXISTS public._applied_migrations (
-    filename TEXT PRIMARY KEY,
-    applied_at TIMESTAMPTZ NOT NULL DEFAULT now()
-)`;
-
-/**
- * Start a `postgres:16-alpine` testcontainer, point the connection lib at it
- * via the `BENCH_DB_*` env vars (`BENCH_DB_PASSWORD` set means the IAM token
- * path is bypassed), and apply the migration set unless `applySchema: false`.
- * Each file runs as one statement batch; the runner's `-- migrate-schema:`
- * directives are NOT interpreted here (none of the current migrations needs
- * that: requires-superuser is satisfied by the container superuser, and no
- * migration uses no-transaction). Callers own teardown: `await resetPool()`
- * then `await container.stop()` in `afterAll`.
+/** Start a disposable Postgres instance and apply the actual migration runner.
+ * Callers own teardown: resetPool(), then container.stop(). Requires uv on PATH.
  */
 export async function startBenchContainer(
   options: { applySchema?: boolean } = {},
@@ -81,10 +42,31 @@ export async function startBenchContainer(
   process.env.BENCH_DB_PASSWORD = container.getPassword();
   process.env.BENCH_DB_SSL = 'disable';
   if (options.applySchema !== false) {
-    const pool = getPool();
-    await pool.query(LEDGER_DDL);
-    for (const name of MIGRATION_FILES) {
-      await pool.query(readFileSync(join(MIGRATIONS_DIR, name), 'utf8'));
+    try {
+      execFileSync(
+        'uv',
+        [
+          'run',
+          '--no-project',
+          fileURLToPath(new URL('../../scripts/migrate-schema.py', import.meta.url)),
+          'apply',
+        ],
+        {
+          env: {
+            ...process.env,
+            PGHOST: container.getHost(),
+            PGPORT: String(container.getPort()),
+            PGDATABASE: container.getDatabase(),
+            PGUSER: container.getUsername(),
+            PGPASSWORD: container.getPassword(),
+            PGSSLMODE: 'disable',
+          },
+          stdio: 'pipe',
+        },
+      );
+    } catch (error) {
+      await container.stop();
+      throw error;
     }
   }
   return container;