From b585453da7d3c634794e9dde92bd8e83e4350f37 Mon Sep 17 00:00:00 2001 From: Vasanthdev2004 Date: Mon, 1 Jun 2026 09:42:22 +0530 Subject: [PATCH 1/2] feat(db): versioned schema migrations with idempotent backfill MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace the inline &str SQL array that re-runs on every node startup with a versioned migration system. Each migration is recorded in a new schema_migrations table and applied at most once per node, inside a single transaction. For nodes upgrading from a pre-migration-versioning build, the canonical `repos` table is used as a signal to mark v1 as already applied without re-running its ~140 statements — so existing operators see zero behavior change on first restart after this commit. Closes the gap called out in docs/OSS-READINESS-AUDIT.md:78. - Adds Db::migration_status() returning applied (version, name, applied_at) - Adds 6 unit tests validating the static migration catalogue (versions strictly increasing, names distinct, bodies non-empty, v1 name locked to initial_schema) - cargo fmt + clippy -D warnings + cargo test --workspace all clean --- crates/gitlawb-node/src/db/mod.rs | 269 ++++++++++++++++++++++++++++-- 1 file changed, 259 insertions(+), 10 deletions(-) diff --git a/crates/gitlawb-node/src/db/mod.rs b/crates/gitlawb-node/src/db/mod.rs index 6690025e..6a07d186 100644 --- a/crates/gitlawb-node/src/db/mod.rs +++ b/crates/gitlawb-node/src/db/mod.rs @@ -1,7 +1,8 @@ -use anyhow::Result; +use anyhow::{Context, Result}; use chrono::{DateTime, Utc}; use serde::{Deserialize, Serialize}; use sqlx::{PgPool, Row}; +use tracing::info; use uuid::Uuid; // ── Public data types ───────────────────────────────────────────────────────── @@ -197,8 +198,172 @@ impl Db { Ok(db) } + /// Run all pending versioned migrations in order, inside a single + /// transaction per migration. Idempotent — migrations whose version is + /// already recorded in `schema_migrations` are skipped. + /// + /// For an existing installation upgrading from a pre-migration-versioning + /// build, we detect the already-applied v1 schema (the `repos` table is + /// canonical and present on every production node) and mark v1 as + /// applied without re-running its statements. This lets us ship versioned + /// migrations on a live network without forcing operators to drop tables. async fn migrate(&self) -> Result<()> { - let stmts = [ + // Bootstrap: ensure the `schema_migrations` table itself exists. + sqlx::query( + r#"CREATE TABLE IF NOT EXISTS schema_migrations ( + version BIGINT NOT NULL PRIMARY KEY, + name TEXT NOT NULL, + applied_at TEXT NOT NULL + )"#, + ) + .execute(&self.pool) + .await + .context("creating schema_migrations table")?; + + // Backfill: if the schema is already in place (legacy install) but no + // rows exist in `schema_migrations`, mark v1 as already applied. + // We use the `repos` table as the canonical signal — it has existed + // since the first public release and is present on every production + // node. If it's there but `schema_migrations` is empty, the schema + // was created by the old inline migration system. + let v1_already_applied: bool = sqlx::query( + "SELECT EXISTS(SELECT 1 FROM schema_migrations WHERE version = 1) AS applied", + ) + .fetch_one(&self.pool) + .await? + .get::("applied"); + + if !v1_already_applied { + let legacy_present: bool = sqlx::query( + "SELECT EXISTS( + SELECT 1 FROM information_schema.tables + WHERE table_schema = 'public' AND table_name = 'repos' + ) AS present", + ) + .fetch_one(&self.pool) + .await? + .get::("present"); + + if legacy_present { + sqlx::query( + "INSERT INTO schema_migrations (version, name, applied_at) + VALUES ($1, $2, $3)", + ) + .bind(1_i64) + .bind(MIGRATION_V1_NAME) + .bind(Utc::now().to_rfc3339()) + .execute(&self.pool) + .await + .context("backfilling schema_migrations for legacy v1 schema")?; + info!( + name = MIGRATION_V1_NAME, + "detected legacy schema, marked v1 as already applied" + ); + } + } + + // Apply any pending migrations in version order. + for m in MIGRATIONS { + let already: bool = sqlx::query( + "SELECT EXISTS(SELECT 1 FROM schema_migrations WHERE version = $1) AS applied", + ) + .bind(m.version) + .fetch_one(&self.pool) + .await? + .get::("applied"); + + if already { + continue; + } + + let started = std::time::Instant::now(); + info!( + version = m.version, + name = m.name, + statements = m.stmts.len(), + "applying migration" + ); + + // Run the migration body in a single transaction so a failure + // mid-way leaves the database in its prior state rather than + // partially mutated. + let mut tx = self.pool.begin().await?; + for stmt in m.stmts { + sqlx::query(stmt).execute(&mut *tx).await.with_context(|| { + format!( + "migration v{} ({}) failed on statement: {}", + m.version, m.name, stmt + ) + })?; + } + sqlx::query( + "INSERT INTO schema_migrations (version, name, applied_at) + VALUES ($1, $2, $3)", + ) + .bind(m.version) + .bind(m.name) + .bind(Utc::now().to_rfc3339()) + .execute(&mut *tx) + .await + .context("recording migration as applied")?; + tx.commit() + .await + .with_context(|| format!("committing migration v{}", m.version))?; + + info!( + version = m.version, + name = m.name, + elapsed_ms = started.elapsed().as_millis() as u64, + "migration applied" + ); + } + + Ok(()) + } + + /// Returns `(version, name, applied_at)` for every applied migration, + /// oldest first. Useful for ops/observability — surface via `gl status` + /// or `/api/v1/stats` in a follow-up. + #[allow(dead_code)] + pub async fn migration_status(&self) -> Result> { + let rows = sqlx::query( + "SELECT version, name, applied_at FROM schema_migrations ORDER BY version ASC", + ) + .fetch_all(&self.pool) + .await?; + Ok(rows + .into_iter() + .map(|r| { + ( + r.get::("version"), + r.get("name"), + r.get("applied_at"), + ) + }) + .collect()) + } +} + +// ── Migration catalogue ────────────────────────────────────────────────────── +// +// All schema statements are bundled into a single v1 migration so we can ship +// versioned migrations on a live network without breaking the existing +// install base. Future schema changes MUST be added as v2, v3, … — never +// appended to v1. Operators can read `schema_migrations` to confirm a node +// is at the expected version. + +const MIGRATION_V1_NAME: &str = "initial_schema"; + +struct Migration { + version: i64, + name: &'static str, + stmts: &'static [&'static str], +} + +const MIGRATIONS: &[Migration] = &[Migration { + version: 1, + name: MIGRATION_V1_NAME, + stmts: &[ r#"CREATE TABLE IF NOT EXISTS repos ( id TEXT NOT NULL PRIMARY KEY, name TEXT NOT NULL, @@ -453,14 +618,8 @@ impl Db { "CREATE INDEX IF NOT EXISTS idx_bounties_status ON bounties(status)", "CREATE INDEX IF NOT EXISTS idx_bounties_repo ON bounties(repo_owner, repo_name)", "CREATE INDEX IF NOT EXISTS idx_bounties_claimant ON bounties(claimant_did)", - ]; - - for stmt in &stmts { - sqlx::query(stmt).execute(&self.pool).await?; - } - Ok(()) - } -} + ], +}]; // ── Repos ───────────────────────────────────────────────────────────────────── @@ -2189,3 +2348,93 @@ impl Db { } } } + +// ── Tests ───────────────────────────────────────────────────────────────────── +// +// These tests don't require a live Postgres connection. They validate the +// static migration catalogue is well-formed so a future maintainer can't +// ship a regression like duplicate versions, negative versions, or empty +// migration bodies. The actual SQL execution is exercised by integration +// tests / first-run on a real node. + +#[cfg(test)] +mod migration_tests { + use super::{MIGRATIONS, MIGRATION_V1_NAME}; + + #[test] + fn migrations_are_non_empty() { + assert!( + !MIGRATIONS.is_empty(), + "MIGRATIONS must contain at least the initial v1 schema" + ); + } + + #[test] + fn migration_versions_are_strictly_increasing() { + let mut last = i64::MIN; + for m in MIGRATIONS { + assert!( + m.version > last, + "migration versions must be strictly increasing; \ + found {} after {}", + m.version, + last + ); + last = m.version; + } + } + + #[test] + fn migration_versions_start_at_one() { + // A version of 0 (or negative) would be a footgun: any future + // `WHERE version > current_max` style query would skip it. + assert_eq!( + MIGRATIONS.first().map(|m| m.version), + Some(1), + "the first migration must have version 1" + ); + } + + #[test] + fn migration_names_are_non_empty_and_distinct() { + let mut seen = std::collections::HashSet::new(); + for m in MIGRATIONS { + assert!( + !m.name.is_empty(), + "migration v{} has empty name", + m.version + ); + assert!( + !m.name.contains(char::is_whitespace), + "migration v{} name {:?} contains whitespace", + m.version, + m.name + ); + assert!( + seen.insert(m.name), + "duplicate migration name: {:?}", + m.name + ); + } + } + + #[test] + fn migration_bodies_are_non_empty() { + for m in MIGRATIONS { + assert!( + !m.stmts.is_empty(), + "migration v{} ({}) has no SQL statements", + m.version, + m.name + ); + } + } + + #[test] + fn v1_name_is_the_initial_schema() { + // This is what the legacy-install backfill writes to + // `schema_migrations` when an existing node upgrades. If you rename + // it, you must also update the backfill. + assert_eq!(MIGRATIONS[0].name, MIGRATION_V1_NAME); + } +} From d3b8c1e20390ba32eff63338206a6d8f51f043f5 Mon Sep 17 00:00:00 2001 From: Vasanthdev2004 Date: Mon, 1 Jun 2026 09:59:29 +0530 Subject: [PATCH 2/2] fix(db): make legacy migration backfill safe and lock-guarded Address two robustness gaps in the v1 migration runner on the live network: - Remove the "repos table present => v1 complete" backfill short-circuit. It assumed every existing node already had the full current schema; a node that was behind would be marked v1-applied while still missing newer tables or columns. Every v1 statement is idempotent (CREATE TABLE/INDEX IF NOT EXISTS, ADD COLUMN IF NOT EXISTS), so legacy installs now simply run v1 once: existing objects are no-ops, missing ones get created. - Guard the whole runner with a Postgres advisory lock so two instances against the same database (blue/green or rolling deploy) can't race to apply the same migration and trip the schema_migrations primary key. The lock is released explicitly, and automatically if the connection drops. Also document that per-migration transactions preclude CREATE INDEX CONCURRENTLY in future migrations. Co-Authored-By: Claude Opus 4.8 --- crates/gitlawb-node/src/db/mod.rs | 94 ++++++++++++++++--------------- 1 file changed, 48 insertions(+), 46 deletions(-) diff --git a/crates/gitlawb-node/src/db/mod.rs b/crates/gitlawb-node/src/db/mod.rs index 6a07d186..63dcfb21 100644 --- a/crates/gitlawb-node/src/db/mod.rs +++ b/crates/gitlawb-node/src/db/mod.rs @@ -202,11 +202,19 @@ impl Db { /// transaction per migration. Idempotent — migrations whose version is /// already recorded in `schema_migrations` are skipped. /// - /// For an existing installation upgrading from a pre-migration-versioning - /// build, we detect the already-applied v1 schema (the `repos` table is - /// canonical and present on every production node) and mark v1 as - /// applied without re-running its statements. This lets us ship versioned - /// migrations on a live network without forcing operators to drop tables. + /// Concurrency: the whole routine is guarded by a Postgres advisory lock so + /// two node instances pointed at the same database (e.g. during a + /// blue/green or rolling deploy) cannot race to apply the same migration + /// and trip the `schema_migrations` primary key. + /// + /// Legacy installs: v1 bundles the entire pre-versioning schema, and every + /// statement in it is idempotent (`CREATE TABLE IF NOT EXISTS`, + /// `CREATE INDEX IF NOT EXISTS`, `ADD COLUMN IF NOT EXISTS`). So an existing + /// node that predates this system just runs v1 once: existing objects are + /// no-ops, and any objects it was missing are created. We deliberately do + /// *not* short-circuit on the presence of a single canonical table — a node + /// that was behind on schema would then be marked complete while still + /// missing newer objects. async fn migrate(&self) -> Result<()> { // Bootstrap: ensure the `schema_migrations` table itself exists. sqlx::query( @@ -220,49 +228,35 @@ impl Db { .await .context("creating schema_migrations table")?; - // Backfill: if the schema is already in place (legacy install) but no - // rows exist in `schema_migrations`, mark v1 as already applied. - // We use the `repos` table as the canonical signal — it has existed - // since the first public release and is present on every production - // node. If it's there but `schema_migrations` is empty, the schema - // was created by the old inline migration system. - let v1_already_applied: bool = sqlx::query( - "SELECT EXISTS(SELECT 1 FROM schema_migrations WHERE version = 1) AS applied", - ) - .fetch_one(&self.pool) - .await? - .get::("applied"); - - if !v1_already_applied { - let legacy_present: bool = sqlx::query( - "SELECT EXISTS( - SELECT 1 FROM information_schema.tables - WHERE table_schema = 'public' AND table_name = 'repos' - ) AS present", - ) - .fetch_one(&self.pool) - .await? - .get::("present"); + // Serialize migrations across processes: hold a session-level advisory + // lock on a dedicated connection for the whole run. Another instance + // starting up blocks here until we finish. The lock is released when we + // explicitly unlock below, or automatically if the connection is + // dropped (e.g. on panic), so a crash can't wedge future restarts. + let mut lock_conn = self + .pool + .acquire() + .await + .context("acquiring connection for migration advisory lock")?; + sqlx::query("SELECT pg_advisory_lock($1)") + .bind(MIGRATION_ADVISORY_LOCK) + .execute(&mut *lock_conn) + .await + .context("acquiring migration advisory lock")?; - if legacy_present { - sqlx::query( - "INSERT INTO schema_migrations (version, name, applied_at) - VALUES ($1, $2, $3)", - ) - .bind(1_i64) - .bind(MIGRATION_V1_NAME) - .bind(Utc::now().to_rfc3339()) - .execute(&self.pool) - .await - .context("backfilling schema_migrations for legacy v1 schema")?; - info!( - name = MIGRATION_V1_NAME, - "detected legacy schema, marked v1 as already applied" - ); - } - } + let result = self.run_pending_migrations().await; + + let _ = sqlx::query("SELECT pg_advisory_unlock($1)") + .bind(MIGRATION_ADVISORY_LOCK) + .execute(&mut *lock_conn) + .await; - // Apply any pending migrations in version order. + result + } + + /// Apply every migration whose version isn't yet recorded, in order. + /// Must be called while holding the migration advisory lock. + async fn run_pending_migrations(&self) -> Result<()> { for m in MIGRATIONS { let already: bool = sqlx::query( "SELECT EXISTS(SELECT 1 FROM schema_migrations WHERE version = $1) AS applied", @@ -351,6 +345,14 @@ impl Db { // install base. Future schema changes MUST be added as v2, v3, … — never // appended to v1. Operators can read `schema_migrations` to confirm a node // is at the expected version. +// +// Each migration runs in a single transaction, so statements that Postgres +// forbids inside a transaction (notably `CREATE INDEX CONCURRENTLY`) cannot be +// used here. Build such indexes the ordinary, transaction-safe way, or stage +// them as a dedicated out-of-band operational step. + +// Arbitrary but stable key for the migration advisory lock ("gitlawb_" bytes). +const MIGRATION_ADVISORY_LOCK: i64 = 0x6769_746C_6177_625F; const MIGRATION_V1_NAME: &str = "initial_schema";