From b97e0dec0aecde46fe453f232e6bfff6074c677e Mon Sep 17 00:00:00 2001 From: Sylvain Cau Date: Sun, 5 Jul 2026 14:40:19 -0700 Subject: [PATCH 1/5] fix(migrate): cap insert batches to the destination bind-parameter limit Bulk inserts batched 1000 rows regardless of table width, but a multi-row insert binds rows*columns parameters. A wide table like book_metadata (66 columns) overflowed PostgreSQL's 65535 and SQLite's 32766 per-statement parameter limits during copy/import, failing with "too many arguments for query: 66000". Cap each table's batch so rows*columns stays under the destination backend's limit. Adds a wide-table reproduction (1000 book_metadata rows into PostgreSQL) and unit tests for the cap calculation. --- crates/codex-migrate/src/engine.rs | 41 ++++++++++++++ tests/migrate/mod.rs | 85 ++++++++++++++++++++++++++++++ 2 files changed, 126 insertions(+) diff --git a/crates/codex-migrate/src/engine.rs b/crates/codex-migrate/src/engine.rs index 6b0b3d92..b2fde45f 100644 --- a/crates/codex-migrate/src/engine.rs +++ b/crates/codex-migrate/src/engine.rs @@ -104,6 +104,7 @@ where C: ConnectionTrait, R: BufRead, { + let batch_size = safe_batch_size::(conn.get_database_backend(), batch_size); let mut batch: Vec = Vec::with_capacity(batch_size); let mut count = 0u64; for line in reader.lines() { @@ -133,6 +134,7 @@ where S: ConnectionTrait + StreamTrait, D: ConnectionTrait, { + let batch_size = safe_batch_size::(dst.get_database_backend(), batch_size); let mut stream = open_source_stream::(src).await?; let mut batch: Vec = Vec::with_capacity(batch_size); let mut count = 0u64; @@ -168,6 +170,19 @@ where Ok(E::delete_many().exec(conn).await?.rows_affected) } +/// Cap the batch so a multi-row insert can't exceed the destination's bind +/// parameter limit. A batch binds `rows × columns` parameters; PostgreSQL caps +/// a statement at 65535 and SQLite at 32766, so a wide table (e.g. +/// `book_metadata`, ~66 columns) overflows a naive 1000-row batch. +fn safe_batch_size(backend: DatabaseBackend, requested: usize) -> usize { + let columns = ::iter().count().max(1); + let param_limit = match backend { + DatabaseBackend::Postgres | DatabaseBackend::MySql => 65535, + DatabaseBackend::Sqlite => 32766, + }; + requested.min((param_limit / columns).max(1)) +} + /// Insert one batch, using `exec_without_returning` to avoid a per-row /// RETURNING clause on bulk loads. No-op (and no DB round-trip) when empty. async fn insert_batch(conn: &C, batch: Vec) -> Result @@ -183,3 +198,29 @@ where E::insert_many(batch).exec_without_returning(conn).await?; Ok(n) } + +#[cfg(test)] +mod tests { + use super::safe_batch_size; + use sea_orm::DatabaseBackend; + + #[test] + fn caps_wide_table_under_postgres_param_limit() { + // book_metadata has 66 columns; a naive 1000-row batch would bind + // 66000 parameters, over PostgreSQL's 65535 limit. + let n = safe_batch_size::( + DatabaseBackend::Postgres, + 1000, + ); + assert!(n < 1000, "wide table should be capped, got {n}"); + assert!(n * 66 <= 65535, "batch {n} still exceeds the limit"); + } + + #[test] + fn keeps_requested_size_for_narrow_table() { + // genres has 4 columns; 1000 rows is well within the limit. + let n = + safe_batch_size::(DatabaseBackend::Postgres, 1000); + assert_eq!(n, 1000); + } +} diff --git a/tests/migrate/mod.rs b/tests/migrate/mod.rs index 16225cf5..11dad697 100644 --- a/tests/migrate/mod.rs +++ b/tests/migrate/mod.rs @@ -374,3 +374,88 @@ async fn export_import_across_all_engine_pairs() { drop_pg("codex_test_pair_a").await; drop_pg("codex_test_pair_b").await; } + +// --------------------------------------------------------------------------- +// Wide-table batch: a table with many columns must not exceed the destination's +// bind-parameter limit (PostgreSQL 65535 / SQLite 32766). +// --------------------------------------------------------------------------- + +use codex::db::entities::book_metadata; +use sea_orm::PaginatorTrait; + +async fn seed_wide_rows(db: &Database, n: usize) { + let library = db + .create_library("Comics", "/lib", ScanningStrategy::Default) + .await + .unwrap(); + let series = db.create_series(library.id, "Saga").await.unwrap(); + let conn = db.sea_orm_connection(); + + let mut books_am = Vec::with_capacity(n); + let mut meta_am = Vec::with_capacity(n); + for i in 0..n { + let book_id = Uuid::new_v4(); + books_am.push(books::ActiveModel { + id: Set(book_id), + series_id: Set(series.id), + library_id: Set(library.id), + path: Set(format!("/lib/{i}.cbz")), + file_name: Set(format!("{i}.cbz")), + file_size: Set(1), + file_hash: Set(format!("h{i}")), + partial_hash: Set(format!("p{i}")), + format: Set("cbz".to_string()), + page_count: Set(1), + deleted: Set(false), + analyzed: Set(true), + modified_at: Set(Utc::now()), + created_at: Set(Utc::now()), + updated_at: Set(Utc::now()), + ..Default::default() + }); + meta_am.push(book_metadata::ActiveModel { + id: Set(Uuid::new_v4()), + book_id: Set(book_id), + search_title: Set(format!("title {i}")), + created_at: Set(Utc::now()), + updated_at: Set(Utc::now()), + ..Default::default() + }); + } + // Seed in modest chunks so seeding itself stays under the SQLite limit. + for chunk in books_am.chunks(400) { + books::Entity::insert_many(chunk.to_vec()) + .exec(conn) + .await + .unwrap(); + } + for chunk in meta_am.chunks(400) { + book_metadata::Entity::insert_many(chunk.to_vec()) + .exec(conn) + .await + .unwrap(); + } +} + +#[tokio::test] +#[ignore] // requires a PostgreSQL test database +async fn copy_wide_table_exceeding_param_limit() { + let (src, _sd) = setup_test_db_wrapper().await; + let Some(pg) = fresh_pg("codex_test_wide").await else { + return; // no PostgreSQL available + }; + + // 1000 book_metadata rows × 66 columns = 66000 bind params in one naive + // batch — over PostgreSQL's 65535 limit. The batch cap must split it. + seed_wide_rows(&src, 1000).await; + + transfer(src.sea_orm_connection(), &pg) + .await + .expect("wide-table copy must not exceed the bind-parameter limit"); + + let n = book_metadata::Entity::find().count(&pg).await.unwrap(); + assert_eq!(n, 1000, "all wide rows copied"); + + drop(pg); + drop_pg("codex_test_wide").await; +} From fac13b355da02ccf822bb171e283e4b4663e9c02 Mon Sep 17 00:00:00 2001 From: Sylvain Cau Date: Sun, 5 Jul 2026 14:51:48 -0700 Subject: [PATCH 2/5] fix(migrate): disable foreign keys without superuser on PostgreSQL MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit copy/import disabled FK enforcement with `SET session_replication_role = replica`, which requires a superuser. On managed PostgreSQL the app role is only the database owner, so it failed with "permission denied to set parameter session_replication_role". On PostgreSQL, drop the FK constraints before the load and recreate them afterward — recreating an FK revalidates the loaded rows, so referential integrity is still checked, and it needs only table ownership. SQLite keeps its deferred-FK check. Runs inside the same transaction, so a failure rolls back cleanly. Adds a test that performs the copy as a non-superuser database owner. --- crates/codex-migrate/src/fk.rs | 138 ++++++++++++++++++++++++++------ crates/codex-migrate/src/lib.rs | 8 +- tests/migrate/mod.rs | 80 ++++++++++++++++++ 3 files changed, 201 insertions(+), 25 deletions(-) diff --git a/crates/codex-migrate/src/fk.rs b/crates/codex-migrate/src/fk.rs index cb5fb8cc..f7b4c510 100644 --- a/crates/codex-migrate/src/fk.rs +++ b/crates/codex-migrate/src/fk.rs @@ -1,35 +1,127 @@ //! Foreign-key enforcement control for bulk loads. //! //! A 1:1 load inserts tables in an arbitrary order, so per-row FK enforcement -//! would reject rows whose parents are inserted later. Rather than maintain a -//! drift-prone topological insert order across ~60 tables (some -//! self-referential), we disable enforcement for the duration of the load and -//! rely on the source data already being internally consistent. -//! -//! These calls are transaction-scoped and must run on the same connection that -//! performs the load — pass a [`sea_orm::DatabaseTransaction`]. +//! would reject rows whose parents are inserted later. We suppress enforcement +//! for the duration of the load, using a mechanism that needs only the +//! privileges the migration role already has (it just ran the migrations): //! //! - **SQLite:** `PRAGMA defer_foreign_keys = ON` defers all FK checks until //! COMMIT (and auto-resets there), so a complete, consistent dataset is still //! validated as a whole when the transaction commits. -//! - **PostgreSQL:** `SET LOCAL session_replication_role = replica` suppresses -//! FK (and other) triggers for the transaction. This requires the connection -//! role to be a superuser or the table owner. There is no commit-time -//! re-check, so integrity rests on source consistency plus the row-count -//! verification performed after the load. +//! - **PostgreSQL:** drop every FK constraint before the load and recreate it +//! after. Recreating an FK **revalidates** the loaded rows, so integrity is +//! still checked. This needs only table ownership — unlike +//! `session_replication_role`, which requires a superuser and is therefore +//! unavailable on most managed Postgres. +//! - **MySQL:** `SET FOREIGN_KEY_CHECKS = 0` for the session. +//! +//! All of this is transaction-scoped: pass the same +//! [`sea_orm::DatabaseTransaction`] to [`before_load`] and [`after_load`] that +//! performs the load, so a failure rolls the whole thing back. use anyhow::{Context, Result}; -use sea_orm::{ConnectionTrait, DatabaseBackend}; +use sea_orm::{ConnectionTrait, DatabaseBackend, Statement}; + +/// State captured by [`before_load`] that [`after_load`] needs to restore. +pub enum FkGuard { + /// SQLite: checks deferred; the commit revalidates them. + Deferred, + /// PostgreSQL: FK constraints were dropped and must be recreated. + DroppedConstraints(Vec), + /// MySQL: session FK checks were disabled. + SessionDisabled, +} + +/// A captured PostgreSQL foreign-key constraint, enough to recreate it verbatim. +pub struct PgForeignKey { + table: String, + name: String, + definition: String, +} -/// Disable foreign-key enforcement for the current transaction. -pub async fn disable(conn: &C) -> Result<()> { - let sql = match conn.get_database_backend() { - DatabaseBackend::Sqlite => "PRAGMA defer_foreign_keys = ON", - DatabaseBackend::Postgres => "SET LOCAL session_replication_role = replica", - DatabaseBackend::MySql => "SET FOREIGN_KEY_CHECKS = 0", - }; - conn.execute_unprepared(sql) +/// Suppress foreign-key enforcement for the current transaction, returning the +/// state needed to restore it in [`after_load`]. +pub async fn before_load(conn: &C) -> Result { + match conn.get_database_backend() { + DatabaseBackend::Sqlite => { + conn.execute_unprepared("PRAGMA defer_foreign_keys = ON") + .await + .context("failed to defer SQLite foreign keys")?; + Ok(FkGuard::Deferred) + } + DatabaseBackend::Postgres => { + let fks = capture_postgres_fks(conn).await?; + for fk in &fks { + conn.execute_unprepared(&format!( + "ALTER TABLE {} DROP CONSTRAINT \"{}\"", + fk.table, fk.name + )) + .await + .with_context(|| format!("failed to drop FK {} on {}", fk.name, fk.table))?; + } + Ok(FkGuard::DroppedConstraints(fks)) + } + DatabaseBackend::MySql => { + conn.execute_unprepared("SET FOREIGN_KEY_CHECKS = 0") + .await + .context("failed to disable MySQL foreign-key checks")?; + Ok(FkGuard::SessionDisabled) + } + } +} + +/// Restore foreign-key enforcement after the load. On PostgreSQL this recreates +/// (and thereby revalidates) every constraint captured in [`before_load`]. +pub async fn after_load(conn: &C, guard: FkGuard) -> Result<()> { + match guard { + // The SQLite commit revalidates deferred FKs; nothing to do here. + FkGuard::Deferred => Ok(()), + FkGuard::DroppedConstraints(fks) => { + for fk in &fks { + conn.execute_unprepared(&format!( + "ALTER TABLE {} ADD CONSTRAINT \"{}\" {}", + fk.table, fk.name, fk.definition + )) + .await + .with_context(|| { + format!( + "failed to recreate FK {} on {} (referential integrity violated?)", + fk.name, fk.table + ) + })?; + } + Ok(()) + } + FkGuard::SessionDisabled => { + conn.execute_unprepared("SET FOREIGN_KEY_CHECKS = 1") + .await + .context("failed to re-enable MySQL foreign-key checks")?; + Ok(()) + } + } +} + +/// Read every foreign-key constraint in the `public` schema with a definition +/// string suitable for recreating it verbatim. +async fn capture_postgres_fks(conn: &C) -> Result> { + let rows = conn + .query_all(Statement::from_string( + DatabaseBackend::Postgres, + "SELECT conrelid::regclass::text AS tbl, conname, pg_get_constraintdef(oid) AS def \ + FROM pg_constraint \ + WHERE contype = 'f' AND connamespace = 'public'::regnamespace" + .to_string(), + )) .await - .with_context(|| format!("failed to disable foreign-key enforcement ({sql})"))?; - Ok(()) + .context("failed to read PostgreSQL foreign-key constraints")?; + + let mut fks = Vec::with_capacity(rows.len()); + for row in rows { + fks.push(PgForeignKey { + table: row.try_get::("", "tbl")?, + name: row.try_get::("", "conname")?, + definition: row.try_get::("", "def")?, + }); + } + Ok(fks) } diff --git a/crates/codex-migrate/src/lib.rs b/crates/codex-migrate/src/lib.rs index 9a00ff60..aa9c00ec 100644 --- a/crates/codex-migrate/src/lib.rs +++ b/crates/codex-migrate/src/lib.rs @@ -59,7 +59,7 @@ pub async fn transfer( .await .context("failed to open destination transaction")?; - fk::disable(&txn).await?; + let guard = fk::before_load(&txn).await?; registry::truncate_all(&txn) .await @@ -69,6 +69,8 @@ pub async fn transfer( .await .context("failed while copying tables")?; + fk::after_load(&txn, guard).await?; + txn.commit() .await .context("failed to commit destination transaction (foreign-key check may have failed)")?; @@ -87,7 +89,7 @@ pub async fn load_from_dir(dst: &DatabaseConnection, db_dir: &Path) -> Result Result String { + std::env::var("POSTGRES_TEST_URL") + .unwrap_or_else(|_| "postgres://codex:codex@localhost:54321/codex_test".to_string()) +} + +#[tokio::test] +#[ignore] // requires a PostgreSQL test database with a superuser admin +async fn copy_works_as_non_superuser_database_owner() { + let base = pg_base_url(); + let Ok(admin) = Database::new(&database_config_from_url(&base).unwrap()).await else { + return; // no PostgreSQL available + }; + let ac = admin.sea_orm_connection(); + + // Clean slate, then a NOSUPERUSER role that owns a fresh database. + ac.execute_unprepared("DROP DATABASE IF EXISTS codex_test_nosuper WITH (FORCE)") + .await + .ok(); + ac.execute_unprepared("DROP ROLE IF EXISTS codex_nosuper") + .await + .ok(); + ac.execute_unprepared("CREATE ROLE codex_nosuper LOGIN PASSWORD 'nosuperpw' NOSUPERUSER") + .await + .unwrap(); + ac.execute_unprepared("CREATE DATABASE codex_test_nosuper OWNER codex_nosuper") + .await + .unwrap(); + + // As admin, hand the schema to the role so it can create tables (PG-version safe). + let mut admin_newdb = database_config_from_url(&base).unwrap(); + admin_newdb.postgres.as_mut().unwrap().database_name = "codex_test_nosuper".to_string(); + let admin2 = Database::new(&admin_newdb).await.unwrap(); + admin2 + .sea_orm_connection() + .execute_unprepared("ALTER SCHEMA public OWNER TO codex_nosuper") + .await + .ok(); + + // Connect AS the non-superuser owner and run migrations + transfer. + let mut cfg = database_config_from_url(&base).unwrap(); + { + let pg = cfg.postgres.as_mut().unwrap(); + pg.username = "codex_nosuper".to_string(); + pg.password = "nosuperpw".to_string(); + pg.database_name = "codex_test_nosuper".to_string(); + } + let target = Database::new(&cfg).await.unwrap(); + target.run_migrations().await.unwrap(); + + let (src, _sd) = setup_test_db_wrapper().await; + seed_min(src.sea_orm_connection()).await; + + // The crux: this must succeed without superuser. + transfer(src.sea_orm_connection(), target.sea_orm_connection()) + .await + .expect("copy must work as a non-superuser database owner"); + + let n = users::Entity::find() + .count(target.sea_orm_connection()) + .await + .unwrap(); + assert_eq!(n, 1); + + // Cleanup. + drop(target); + drop(admin2); + ac.execute_unprepared("DROP DATABASE IF EXISTS codex_test_nosuper WITH (FORCE)") + .await + .ok(); + ac.execute_unprepared("DROP ROLE IF EXISTS codex_nosuper") + .await + .ok(); +} From f1cdb900555466d7e0f75f24153784ec90cede22 Mon Sep 17 00:00:00 2001 From: Sylvain Cau Date: Sun, 5 Jul 2026 15:07:13 -0700 Subject: [PATCH 3/5] feat(migrate): add --progress and default row-count verification Add a --progress flag to export/import/copy that logs per-table start and finish plus a periodic within-table row count. It's emitted as log lines (not a TTY bar) so it reads the same in an interactive terminal and in captured Kubernetes/CI logs. Verify row counts by default after import and copy: independently re-count the source (copy) or read the export manifest's recorded counts (import), count the target, compare per table, and fail the command on any mismatch. A --no-verify flag skips it. --- crates/codex-migrate/src/archive.rs | 7 ++- crates/codex-migrate/src/engine.rs | 36 +++++++++++-- crates/codex-migrate/src/lib.rs | 13 +++-- crates/codex-migrate/src/progress.rs | 52 +++++++++++++++++++ crates/codex-migrate/src/registry.rs | 30 ++++++++--- .../codex-migrate/tests/archive_roundtrip.rs | 2 + .../codex-migrate/tests/transfer_roundtrip.rs | 50 ++++++++++++------ src/commands/copy.rs | 22 ++++++-- src/commands/export.rs | 12 +++-- src/commands/import.rs | 42 ++++++++++++--- src/commands/mod.rs | 27 ++++++++++ src/main.rs | 40 +++++++++++++- tests/migrate/mod.rs | 34 ++++++++---- 13 files changed, 310 insertions(+), 57 deletions(-) create mode 100644 crates/codex-migrate/src/progress.rs diff --git a/crates/codex-migrate/src/archive.rs b/crates/codex-migrate/src/archive.rs index 41cafc88..7a3e8cda 100644 --- a/crates/codex-migrate/src/archive.rs +++ b/crates/codex-migrate/src/archive.rs @@ -20,6 +20,7 @@ use crate::manifest::{ ARCHIVE_FORMAT_VERSION, ArtifactEntry, ArtifactGroup, Manifest, TableCount, backend_name, schema_version, }; +use crate::progress::Progress; use crate::reroot::{self, RerootStats}; use crate::{TransferReport, registry}; @@ -54,12 +55,13 @@ pub async fn export_archive( conn: &DatabaseConnection, out_path: &Path, artifacts: &[ArtifactSource], + progress: Progress, ) -> Result { let staging = tempfile::tempdir().context("failed to create export staging dir")?; let db_dir = staging.path().join("db"); std::fs::create_dir_all(&db_dir)?; - let counts = registry::dump_all_to_dir(conn, &db_dir) + let counts = registry::dump_all_to_dir(conn, &db_dir, progress) .await .context("failed to dump database tables")?; let total_rows = counts.iter().map(|c| c.rows).sum(); @@ -111,11 +113,12 @@ pub async fn import_archive( conn: &DatabaseConnection, in_path: &Path, targets: &[ArtifactTarget], + progress: Progress, ) -> Result { let staging = tempfile::tempdir().context("failed to create import staging dir")?; let manifest = extract(in_path, staging.path())?; - let report = crate::load_from_dir(conn, &staging.path().join("db")) + let report = crate::load_from_dir(conn, &staging.path().join("db"), progress) .await .context("failed to load database from archive")?; diff --git a/crates/codex-migrate/src/engine.rs b/crates/codex-migrate/src/engine.rs index b2fde45f..8957763a 100644 --- a/crates/codex-migrate/src/engine.rs +++ b/crates/codex-migrate/src/engine.rs @@ -19,6 +19,8 @@ use sea_orm::{ use serde::Serialize; use serde::de::DeserializeOwned; +use crate::progress::{Progress, ROW_REPORT_INTERVAL}; + /// Default insert batch size. Bounds memory and round-trips while staying well /// under parameter limits for the widest tables. pub const DEFAULT_BATCH_SIZE: usize = 1000; @@ -78,35 +80,48 @@ fn iden_string(iden: &I) -> String { /// Stream every row of `E` and write it as one NDJSON line to `out`. /// Returns the number of rows written. -pub async fn dump_table(conn: &C, out: &mut W) -> Result +pub async fn dump_table(conn: &C, out: &mut W, progress: Progress) -> Result where E: EntityTrait, E::Model: Serialize, C: ConnectionTrait + StreamTrait, W: Write, { + let table = iden_string(&E::default()); let mut stream = open_source_stream::(conn).await?; let mut count = 0u64; + let mut next_report = ROW_REPORT_INTERVAL; while let Some(model) = stream.try_next().await? { serde_json::to_writer(&mut *out, &model)?; out.write_all(b"\n")?; count += 1; + if count >= next_report { + progress.table_rows(&table, count); + next_report += ROW_REPORT_INTERVAL; + } } Ok(count) } /// Read NDJSON lines from `reader`, deserialize each into `E::Model`, and /// insert them into `conn` in batches of `batch_size`. Returns rows inserted. -pub async fn load_table(conn: &C, reader: R, batch_size: usize) -> Result +pub async fn load_table( + conn: &C, + reader: R, + batch_size: usize, + progress: Progress, +) -> Result where E: EntityTrait, E::Model: DeserializeOwned + IntoActiveModel, C: ConnectionTrait, R: BufRead, { + let table = iden_string(&E::default()); let batch_size = safe_batch_size::(conn.get_database_backend(), batch_size); let mut batch: Vec = Vec::with_capacity(batch_size); let mut count = 0u64; + let mut next_report = ROW_REPORT_INTERVAL; for line in reader.lines() { let line = line?; if line.trim().is_empty() { @@ -116,6 +131,10 @@ where batch.push(model.into_active_model()); if batch.len() >= batch_size { count += insert_batch::(conn, std::mem::take(&mut batch)).await?; + if count >= next_report { + progress.table_rows(&table, count); + next_report += ROW_REPORT_INTERVAL; + } } } if !batch.is_empty() { @@ -127,21 +146,32 @@ where /// Stream every row of `E` directly from `src` into `dst` in batches, without /// an intermediate serialized form. This is the path used by the direct /// database-to-database `copy`. -pub async fn copy_table(src: &S, dst: &D, batch_size: usize) -> Result +pub async fn copy_table( + src: &S, + dst: &D, + batch_size: usize, + progress: Progress, +) -> Result where E: EntityTrait, E::Model: IntoActiveModel, S: ConnectionTrait + StreamTrait, D: ConnectionTrait, { + let table = iden_string(&E::default()); let batch_size = safe_batch_size::(dst.get_database_backend(), batch_size); let mut stream = open_source_stream::(src).await?; let mut batch: Vec = Vec::with_capacity(batch_size); let mut count = 0u64; + let mut next_report = ROW_REPORT_INTERVAL; while let Some(model) = stream.try_next().await? { batch.push(model.into_active_model()); if batch.len() >= batch_size { count += insert_batch::(dst, std::mem::take(&mut batch)).await?; + if count >= next_report { + progress.table_rows(&table, count); + next_report += ROW_REPORT_INTERVAL; + } } } if !batch.is_empty() { diff --git a/crates/codex-migrate/src/lib.rs b/crates/codex-migrate/src/lib.rs index aa9c00ec..d4a5f3a4 100644 --- a/crates/codex-migrate/src/lib.rs +++ b/crates/codex-migrate/src/lib.rs @@ -17,11 +17,13 @@ pub mod engine; pub mod fk; pub mod guard; pub mod manifest; +pub mod progress; pub mod registry; pub mod reroot; pub mod verify; pub use conn::database_config_from_url; +pub use progress::Progress; use std::path::Path; @@ -53,6 +55,7 @@ pub struct TransferReport { pub async fn transfer( src: &DatabaseConnection, dst: &DatabaseConnection, + progress: Progress, ) -> Result { let txn = dst .begin() @@ -65,7 +68,7 @@ pub async fn transfer( .await .context("failed to clear destination before load")?; - let tables = registry::copy_all(src, &txn, engine::DEFAULT_BATCH_SIZE) + let tables = registry::copy_all(src, &txn, engine::DEFAULT_BATCH_SIZE, progress) .await .context("failed while copying tables")?; @@ -83,7 +86,11 @@ pub async fn transfer( /// 1:1 mirror. Same transaction semantics as [`transfer`] (disable FK → /// truncate → load → commit); this is the archive-backed load used by /// `import`. -pub async fn load_from_dir(dst: &DatabaseConnection, db_dir: &Path) -> Result { +pub async fn load_from_dir( + dst: &DatabaseConnection, + db_dir: &Path, + progress: Progress, +) -> Result { let txn = dst .begin() .await @@ -95,7 +102,7 @@ pub async fn load_from_dir(dst: &DatabaseConnection, db_dir: &Path) -> Result Self { + if enabled { + Progress::Log + } else { + Progress::Silent + } + } + + fn on(self) -> bool { + matches!(self, Progress::Log) + } + + pub(crate) fn table_start(self, table: &str) { + if self.on() { + info!(" → {table}"); + } + } + + pub(crate) fn table_rows(self, table: &str, rows: u64) { + if self.on() { + info!(" {table}: {rows} rows…"); + } + } + + pub(crate) fn table_done(self, table: &str, rows: u64) { + if self.on() { + info!(" ✓ {table}: {rows} rows"); + } + } +} diff --git a/crates/codex-migrate/src/registry.rs b/crates/codex-migrate/src/registry.rs index e0f68482..9455cd89 100644 --- a/crates/codex-migrate/src/registry.rs +++ b/crates/codex-migrate/src/registry.rs @@ -12,6 +12,7 @@ use anyhow::Result; use sea_orm::{ConnectionTrait, EntityName, StreamTrait}; use crate::engine; +use crate::progress::Progress; /// Row count for a single table, produced by the collective operations. #[derive(Debug, Clone, PartialEq, Eq)] @@ -141,7 +142,12 @@ where } /// Stream every table directly from `src` into `dst` (the direct copy path). -pub async fn copy_all(src: &S, dst: &D, batch_size: usize) -> Result> +pub async fn copy_all( + src: &S, + dst: &D, + batch_size: usize, + progress: Progress, +) -> Result> where S: ConnectionTrait + StreamTrait, D: ConnectionTrait, @@ -151,7 +157,9 @@ where ($ent:ident) => {{ type E = codex_db::entities::$ent::Entity; let table = ::default().table_name().to_string(); - let n = engine::copy_table::(src, dst, batch_size).await?; + progress.table_start(&table); + let n = engine::copy_table::(src, dst, batch_size, progress).await?; + progress.table_done(&table, n); rows.push(TableRows { table, rows: n }); }}; } @@ -161,7 +169,7 @@ where /// Dump every table to `dir` as one `.ndjson` file each (the archive /// payload). `dir` must already exist. -pub async fn dump_all_to_dir(conn: &C, dir: &Path) -> Result> +pub async fn dump_all_to_dir(conn: &C, dir: &Path, progress: Progress) -> Result> where C: ConnectionTrait + StreamTrait, { @@ -171,10 +179,12 @@ where ($ent:ident) => {{ type E = codex_db::entities::$ent::Entity; let table = ::default().table_name().to_string(); + progress.table_start(&table); let path = dir.join(format!("{table}.ndjson")); let mut writer = std::io::BufWriter::new(std::fs::File::create(&path)?); - let n = engine::dump_table::(conn, &mut writer).await?; + let n = engine::dump_table::(conn, &mut writer, progress).await?; writer.flush()?; + progress.table_done(&table, n); rows.push(TableRows { table, rows: n }); }}; } @@ -184,7 +194,12 @@ where /// Load every table from a directory of `
.ndjson` files. Missing files /// are treated as empty tables. -pub async fn load_all_from_dir(conn: &C, dir: &Path, batch_size: usize) -> Result> +pub async fn load_all_from_dir( + conn: &C, + dir: &Path, + batch_size: usize, + progress: Progress, +) -> Result> where C: ConnectionTrait, { @@ -195,8 +210,11 @@ where let table = ::default().table_name().to_string(); let path = dir.join(format!("{table}.ndjson")); let n = if path.exists() { + progress.table_start(&table); let reader = std::io::BufReader::new(std::fs::File::open(&path)?); - engine::load_table::(conn, reader, batch_size).await? + let n = engine::load_table::(conn, reader, batch_size, progress).await?; + progress.table_done(&table, n); + n } else { 0 }; diff --git a/crates/codex-migrate/tests/archive_roundtrip.rs b/crates/codex-migrate/tests/archive_roundtrip.rs index dce30649..04792239 100644 --- a/crates/codex-migrate/tests/archive_roundtrip.rs +++ b/crates/codex-migrate/tests/archive_roundtrip.rs @@ -140,6 +140,7 @@ async fn archive_roundtrip_mirrors_data_and_reroots_paths() { source_dir: src_uploads.clone(), }, ], + codex_migrate::Progress::Silent, ) .await .expect("export should succeed"); @@ -165,6 +166,7 @@ async fn archive_roundtrip_mirrors_data_and_reroots_paths() { target_dir: dst_uploads.clone(), }, ], + codex_migrate::Progress::Silent, ) .await .expect("import should succeed"); diff --git a/crates/codex-migrate/tests/transfer_roundtrip.rs b/crates/codex-migrate/tests/transfer_roundtrip.rs index d479308f..4dafda0c 100644 --- a/crates/codex-migrate/tests/transfer_roundtrip.rs +++ b/crates/codex-migrate/tests/transfer_roundtrip.rs @@ -52,9 +52,13 @@ async fn transfer_mirrors_source_into_empty_target() { let (library_id, series_id, genre_id) = seed_source(&src).await; - let report = codex_migrate::transfer(src.sea_orm_connection(), dst.sea_orm_connection()) - .await - .expect("transfer should succeed"); + let report = codex_migrate::transfer( + src.sea_orm_connection(), + dst.sea_orm_connection(), + codex_migrate::Progress::Silent, + ) + .await + .expect("transfer should succeed"); // Every non-empty table copied at least what we seeded. assert!( @@ -108,9 +112,13 @@ async fn transfer_preserves_library_json_config_exactly() { let (dst, _dst_dir) = create_test_db().await; let (library_id, _, _) = seed_source(&src).await; - codex_migrate::transfer(src.sea_orm_connection(), dst.sea_orm_connection()) - .await - .unwrap(); + codex_migrate::transfer( + src.sea_orm_connection(), + dst.sea_orm_connection(), + codex_migrate::Progress::Silent, + ) + .await + .unwrap(); // Full-model equality confirms JSON config columns round-trip verbatim. let src_lib = codex_db::entities::libraries::Entity::find_by_id(library_id) @@ -136,9 +144,13 @@ async fn transfer_overwrites_existing_target_data() { let (dst_only_lib, _, _) = seed_source(&dst).await; // First transfer replaces the target's content with the source's. - codex_migrate::transfer(src.sea_orm_connection(), dst.sea_orm_connection()) - .await - .unwrap(); + codex_migrate::transfer( + src.sea_orm_connection(), + dst.sea_orm_connection(), + codex_migrate::Progress::Silent, + ) + .await + .unwrap(); // The target's original library is gone; the target now mirrors the source. let stale = codex_db::entities::libraries::Entity::find_by_id(dst_only_lib) @@ -152,9 +164,13 @@ async fn transfer_overwrites_existing_target_data() { assert!(verify::compare(&src_counts, &after_first).is_empty()); // A second transfer is idempotent (truncate → reload leaves the mirror). - codex_migrate::transfer(src.sea_orm_connection(), dst.sea_orm_connection()) - .await - .unwrap(); + codex_migrate::transfer( + src.sea_orm_connection(), + dst.sea_orm_connection(), + codex_migrate::Progress::Silent, + ) + .await + .unwrap(); let after_second = registry::count_all(dst.sea_orm_connection()).await.unwrap(); assert!(verify::compare(&src_counts, &after_second).is_empty()); } @@ -229,9 +245,13 @@ async fn transfer_reads_text_stored_uuids_from_sqlite() { .unwrap(); assert_eq!(row.try_get::("", "t").unwrap(), "text"); - codex_migrate::transfer(sconn, dst.sea_orm_connection()) - .await - .expect("transfer must read text-stored UUIDs"); + codex_migrate::transfer( + sconn, + dst.sea_orm_connection(), + codex_migrate::Progress::Silent, + ) + .await + .expect("transfer must read text-stored UUIDs"); let uuid = Uuid::parse_str(id).unwrap(); let genre = genres::Entity::find_by_id(uuid) diff --git a/src/commands/copy.rs b/src/commands/copy.rs index 85a3145c..11beb460 100644 --- a/src/commands/copy.rs +++ b/src/commands/copy.rs @@ -14,6 +14,7 @@ use tracing::{info, warn}; /// At least one side must be non-local. /// /// `copy` moves rows only; on-disk files are not transferred. +#[allow(clippy::too_many_arguments)] pub async fn copy_command( config_path: PathBuf, from: Option, @@ -21,6 +22,8 @@ pub async fn copy_command( from_config: Option, to_config: Option, replace: bool, + progress: bool, + no_verify: bool, ) -> Result<()> { // Local config: used for tracing and as the fallback for an omitted side. let (local_config, _created) = load_config(config_path.clone())?; @@ -81,9 +84,18 @@ pub async fn copy_command( } info!("Copying database rows from source to target..."); - let report = transfer(source.sea_orm_connection(), target.sea_orm_connection()) - .await - .context("Copy failed")?; + let report = transfer( + source.sea_orm_connection(), + target.sea_orm_connection(), + codex_migrate::Progress::from_flag(progress), + ) + .await + .context("Copy failed")?; + + if !no_verify { + let source_counts = codex_migrate::registry::count_all(source.sea_orm_connection()).await?; + super::verify_row_counts(&source_counts, target.sea_orm_connection()).await?; + } info!("========================================"); info!( @@ -190,6 +202,8 @@ files: None, None, false, + false, + false, ) .await .expect("copy should succeed"); @@ -205,7 +219,7 @@ files: async fn copy_requires_at_least_one_explicit_side() { let dir = TempDir::new().unwrap(); let cfg = write_config(dir.path(), "only"); - let err = copy_command(cfg, None, None, None, None, false) + let err = copy_command(cfg, None, None, None, None, false, false, false) .await .expect_err("copy with no explicit side must error"); assert!(err.to_string().contains("at least one explicit side")); diff --git a/src/commands/export.rs b/src/commands/export.rs index f9502f68..a9dd17fc 100644 --- a/src/commands/export.rs +++ b/src/commands/export.rs @@ -18,6 +18,7 @@ pub async fn export_command( no_thumbnails: bool, no_uploads: bool, no_plugins: bool, + progress: bool, ) -> Result<()> { let (config, _created) = load_config(config_path.clone())?; let _tracing = init_tracing(&config)?; @@ -56,9 +57,14 @@ pub async fn export_command( } info!("Exporting database to {}", output.display()); - let manifest = export_archive(db.sea_orm_connection(), &output, &artifacts) - .await - .context("Export failed")?; + let manifest = export_archive( + db.sea_orm_connection(), + &output, + &artifacts, + codex_migrate::Progress::from_flag(progress), + ) + .await + .context("Export failed")?; info!("========================================"); info!( diff --git a/src/commands/import.rs b/src/commands/import.rs index d8b91300..17cda6f5 100644 --- a/src/commands/import.rs +++ b/src/commands/import.rs @@ -12,7 +12,13 @@ use tracing::{info, warn}; /// migrations on the target, verifies the archive matches this schema, refuses /// a target that already holds user data (unless `--replace`), then loads the /// data, unpacks artifacts, and re-roots stored file paths. -pub async fn import_command(config_path: PathBuf, input: PathBuf, replace: bool) -> Result<()> { +pub async fn import_command( + config_path: PathBuf, + input: PathBuf, + replace: bool, + progress: bool, + no_verify: bool, +) -> Result<()> { let (config, _created) = load_config(config_path.clone())?; let _tracing = init_tracing(&config)?; info!("Loading configuration from {:?}", config_path); @@ -51,9 +57,28 @@ pub async fn import_command(config_path: PathBuf, input: PathBuf, replace: bool) } info!("Importing {} ...", input.display()); - let outcome = import_archive(conn, &input, &artifact_targets(&config)) - .await - .context("Import failed")?; + let outcome = import_archive( + conn, + &input, + &artifact_targets(&config), + codex_migrate::Progress::from_flag(progress), + ) + .await + .context("Import failed")?; + + if !no_verify { + // The manifest records the source's per-table counts at export time. + let source_counts: Vec = outcome + .manifest + .tables + .iter() + .map(|t| codex_migrate::TableRows { + table: t.table.clone(), + rows: t.rows, + }) + .collect(); + super::verify_row_counts(&source_counts, conn).await?; + } info!("========================================"); info!( @@ -167,19 +192,20 @@ files: false, false, false, + false, ) .await .expect("export should succeed"); assert!(archive.exists(), "archive written"); // Import into a fresh target. - import_command(tgt_cfg.clone(), archive.clone(), false) + import_command(tgt_cfg.clone(), archive.clone(), false, false, false) .await .expect("import into fresh target should succeed"); assert_eq!(library_names(&tgt_cfg).await, vec!["Comics".to_string()]); // A second import without --replace is refused (target now has data). - let err = import_command(tgt_cfg.clone(), archive.clone(), false) + let err = import_command(tgt_cfg.clone(), archive.clone(), false, false, false) .await .expect_err("import into non-fresh target should be refused"); assert!( @@ -188,7 +214,7 @@ files: ); // With --replace it succeeds and still mirrors the source. - import_command(tgt_cfg.clone(), archive.clone(), true) + import_command(tgt_cfg.clone(), archive.clone(), true, false, false) .await .expect("import --replace should succeed"); assert_eq!(library_names(&tgt_cfg).await, vec!["Comics".to_string()]); @@ -198,7 +224,7 @@ files: async fn import_rejects_missing_archive() { let dir = TempDir::new().unwrap(); let cfg = write_config(dir.path(), "tgt"); - let err = import_command(cfg, dir.path().join("nope.tar.gz"), false) + let err = import_command(cfg, dir.path().join("nope.tar.gz"), false, false, false) .await .expect_err("missing archive should error"); assert!(err.to_string().contains("archive not found")); diff --git a/src/commands/mod.rs b/src/commands/mod.rs index 8debb085..5ac8cbb3 100644 --- a/src/commands/mod.rs +++ b/src/commands/mod.rs @@ -13,6 +13,33 @@ pub mod worker; pub use copy::copy_command; pub use export::export_command; pub use import::import_command; + +use anyhow::{Result, bail}; +use codex_migrate::{TableRows, registry, verify}; +use sea_orm::DatabaseConnection; +use tracing::{info, warn}; + +/// Compare per-table source counts against a live target and fail on any +/// mismatch. Shared by `import` and `copy` for their post-load verification. +pub(crate) async fn verify_row_counts( + source_counts: &[TableRows], + target: &DatabaseConnection, +) -> Result<()> { + let target_counts = registry::count_all(target).await?; + let mismatches = verify::compare(source_counts, &target_counts); + if mismatches.is_empty() { + info!("✓ verification: {} tables match", source_counts.len()); + Ok(()) + } else { + for m in &mismatches { + warn!(" ✗ {}: source={} target={}", m.table, m.source, m.target); + } + bail!( + "row-count verification failed: {} table(s) differ", + mismatches.len() + ); + } +} pub use migrate::migrate_command; pub use openapi::{OpenApiFormat, openapi_command}; pub use scan::scan_command; diff --git a/src/main.rs b/src/main.rs index 53deb0bd..5de0c696 100644 --- a/src/main.rs +++ b/src/main.rs @@ -136,6 +136,10 @@ enum Commands { /// Skip bundling plugin data #[arg(long)] no_plugins: bool, + + /// Log per-table progress while exporting + #[arg(long)] + progress: bool, }, /// Import a .tar.gz archive produced by `export` into this instance @@ -151,6 +155,14 @@ enum Commands { /// Replace existing data in the target instead of refusing a non-empty target #[arg(long)] replace: bool, + + /// Log per-table progress while importing + #[arg(long)] + progress: bool, + + /// Skip the post-import row-count verification + #[arg(long)] + no_verify: bool, }, /// Copy database rows directly from one instance to another @@ -178,6 +190,14 @@ enum Commands { /// Replace existing data in the target instead of refusing a non-empty target #[arg(long)] replace: bool, + + /// Log per-table progress while copying + #[arg(long)] + progress: bool, + + /// Skip the post-copy row-count verification + #[arg(long)] + no_verify: bool, }, } @@ -230,6 +250,7 @@ async fn main() -> anyhow::Result<()> { no_thumbnails, no_uploads, no_plugins, + progress, } => { export_command( config, @@ -239,6 +260,7 @@ async fn main() -> anyhow::Result<()> { no_thumbnails, no_uploads, no_plugins, + progress, ) .await?; } @@ -246,8 +268,10 @@ async fn main() -> anyhow::Result<()> { config, input, replace, + progress, + no_verify, } => { - import_command(config, input, replace).await?; + import_command(config, input, replace, progress, no_verify).await?; } Commands::Copy { config, @@ -256,8 +280,20 @@ async fn main() -> anyhow::Result<()> { from_config, to_config, replace, + progress, + no_verify, } => { - copy_command(config, from, to, from_config, to_config, replace).await?; + copy_command( + config, + from, + to, + from_config, + to_config, + replace, + progress, + no_verify, + ) + .await?; } } diff --git a/tests/migrate/mod.rs b/tests/migrate/mod.rs index 30e3a677..6dba1313 100644 --- a/tests/migrate/mod.rs +++ b/tests/migrate/mod.rs @@ -148,9 +148,13 @@ async fn sqlite_to_postgres_roundtrip_mirrors_all_data() { }; let ids = seed_source(&src).await; - transfer(src.sea_orm_connection(), &pg) - .await - .expect("SQLite -> PostgreSQL transfer should succeed"); + transfer( + src.sea_orm_connection(), + &pg, + codex::migrate::Progress::Silent, + ) + .await + .expect("SQLite -> PostgreSQL transfer should succeed"); // Row-count parity across every table. let src_counts = registry::count_all(src.sea_orm_connection()).await.unwrap(); @@ -306,10 +310,10 @@ async fn export_import_pair(src: &DatabaseConnection, tgt: &DatabaseConnection, let dir = tempfile::tempdir().unwrap(); let archive = dir.path().join("export.tar.gz"); - export_archive(src, &archive, &[]) + export_archive(src, &archive, &[], codex::migrate::Progress::Silent) .await .unwrap_or_else(|e| panic!("{label}: export failed: {e:#}")); - import_archive(tgt, &archive, &[]) + import_archive(tgt, &archive, &[], codex::migrate::Progress::Silent) .await .unwrap_or_else(|e| panic!("{label}: import failed: {e:#}")); @@ -449,9 +453,13 @@ async fn copy_wide_table_exceeding_param_limit() { // batch — over PostgreSQL's 65535 limit. The batch cap must split it. seed_wide_rows(&src, 1000).await; - transfer(src.sea_orm_connection(), &pg) - .await - .expect("wide-table copy must not exceed the bind-parameter limit"); + transfer( + src.sea_orm_connection(), + &pg, + codex::migrate::Progress::Silent, + ) + .await + .expect("wide-table copy must not exceed the bind-parameter limit"); let n = book_metadata::Entity::find().count(&pg).await.unwrap(); assert_eq!(n, 1000, "all wide rows copied"); @@ -519,9 +527,13 @@ async fn copy_works_as_non_superuser_database_owner() { seed_min(src.sea_orm_connection()).await; // The crux: this must succeed without superuser. - transfer(src.sea_orm_connection(), target.sea_orm_connection()) - .await - .expect("copy must work as a non-superuser database owner"); + transfer( + src.sea_orm_connection(), + target.sea_orm_connection(), + codex::migrate::Progress::Silent, + ) + .await + .expect("copy must work as a non-superuser database owner"); let n = users::Entity::find() .count(target.sea_orm_connection()) From c44ce6f0c5509b85154901d9afe92305fb75bfef Mon Sep 17 00:00:00 2001 From: Sylvain Cau Date: Sun, 5 Jul 2026 15:21:58 -0700 Subject: [PATCH 4/5] feat(migrate): add --full-verification report for import and copy Add an opt-in --full-verification that compares every record's canonical content on both sides and prints a report of any differences. It normalizes representation differences that are semantically irrelevant across engines: integer-valued floats (1.0 == 1), JSON object key order (jsonb reorders keys), and timestamp precision (PostgreSQL truncates to microseconds). Each table is reduced to an order-independent digest computed identically from a database connection or the archive's NDJSON, so the check streams and stays O(1) memory. The report is informational and does not fail the command; the default row-count check remains the hard safety gate. Adds canonicalization unit tests and a cross-engine test that confirms matching content and detects a tampered row. --- crates/codex-migrate/src/archive.rs | 22 +- crates/codex-migrate/src/engine.rs | 38 ++++ crates/codex-migrate/src/full_verify.rs | 189 ++++++++++++++++++ crates/codex-migrate/src/lib.rs | 1 + crates/codex-migrate/src/registry.rs | 43 ++++ .../codex-migrate/tests/archive_roundtrip.rs | 1 + src/commands/copy.rs | 14 +- src/commands/import.rs | 25 ++- src/commands/mod.rs | 34 ++++ src/main.rs | 21 +- tests/migrate/mod.rs | 60 +++++- 11 files changed, 438 insertions(+), 10 deletions(-) create mode 100644 crates/codex-migrate/src/full_verify.rs diff --git a/crates/codex-migrate/src/archive.rs b/crates/codex-migrate/src/archive.rs index 7a3e8cda..5ab3bdc3 100644 --- a/crates/codex-migrate/src/archive.rs +++ b/crates/codex-migrate/src/archive.rs @@ -46,6 +46,9 @@ pub struct ImportOutcome { pub manifest: Manifest, pub report: TransferReport, pub reroot: RerootStats, + /// Per-table canonical-content mismatches, when full verification was + /// requested; empty otherwise. + pub full_verify: Vec, } /// Export `conn` and the given artifact trees into a `.tar.gz` at `out_path`. @@ -114,11 +117,13 @@ pub async fn import_archive( in_path: &Path, targets: &[ArtifactTarget], progress: Progress, + full_verify: bool, ) -> Result { let staging = tempfile::tempdir().context("failed to create import staging dir")?; let manifest = extract(in_path, staging.path())?; + let db_dir = staging.path().join("db"); - let report = crate::load_from_dir(conn, &staging.path().join("db"), progress) + let report = crate::load_from_dir(conn, &db_dir, progress) .await .context("failed to load database from archive")?; @@ -163,10 +168,25 @@ pub async fn import_archive( .await .context("failed to re-root artifact paths")?; + // Optional deep check: compare the canonical content of every row in the + // archive against the loaded target. + let full_verify_mismatches = if full_verify { + let source = registry::digest_all_from_ndjson_dir(&db_dir) + .await + .context("failed to digest archive contents")?; + let target = registry::digest_all_from_conn(conn) + .await + .context("failed to digest imported data")?; + crate::full_verify::compare_digests(&source, &target) + } else { + Vec::new() + }; + Ok(ImportOutcome { manifest, report, reroot, + full_verify: full_verify_mismatches, }) } diff --git a/crates/codex-migrate/src/engine.rs b/crates/codex-migrate/src/engine.rs index 8957763a..fd73b3f5 100644 --- a/crates/codex-migrate/src/engine.rs +++ b/crates/codex-migrate/src/engine.rs @@ -19,6 +19,7 @@ use sea_orm::{ use serde::Serialize; use serde::de::DeserializeOwned; +use crate::full_verify::{DigestAccumulator, TableDigest}; use crate::progress::{Progress, ROW_REPORT_INTERVAL}; /// Default insert batch size. Bounds memory and round-trips while staying well @@ -190,6 +191,43 @@ where Ok(E::find().count(conn).await?) } +/// Compute the canonical content digest of `E` read from a connection. +pub async fn digest_table_from_conn(conn: &C) -> Result +where + E: EntityTrait, + E::Model: Serialize, + C: ConnectionTrait + StreamTrait, +{ + let table = iden_string(&E::default()); + let mut stream = open_source_stream::(conn).await?; + let mut acc = DigestAccumulator::default(); + while let Some(model) = stream.try_next().await? { + acc.add(&model); + } + Ok(acc.finish(table)) +} + +/// Compute the canonical content digest of `E` read from an NDJSON reader (the +/// archive payload). +pub fn digest_table_from_ndjson(reader: R) -> Result +where + E: EntityTrait, + E::Model: DeserializeOwned + Serialize, + R: BufRead, +{ + let table = iden_string(&E::default()); + let mut acc = DigestAccumulator::default(); + for line in reader.lines() { + let line = line?; + if line.trim().is_empty() { + continue; + } + let model: E::Model = serde_json::from_str(&line)?; + acc.add(&model); + } + Ok(acc.finish(table)) +} + /// Delete every row of `E`. Used by `--replace` before a load. Callers must /// have foreign-key enforcement disabled (see [`crate::fk`]). pub async fn truncate_table(conn: &C) -> Result diff --git a/crates/codex-migrate/src/full_verify.rs b/crates/codex-migrate/src/full_verify.rs new file mode 100644 index 00000000..5f23abb9 --- /dev/null +++ b/crates/codex-migrate/src/full_verify.rs @@ -0,0 +1,189 @@ +//! Deep, per-record verification that compares the *canonical* value of every +//! row on both sides, so representation differences that are semantically +//! irrelevant do not read as mismatches: +//! +//! - numbers: `1.0` and `1` are equal (integer-valued floats normalize to ints) +//! - JSON objects: key order is normalized (PostgreSQL `jsonb` reorders keys) +//! - timestamps: normalized to microsecond precision (PostgreSQL truncates) +//! +//! Each table is reduced to an order-independent 128-bit digest (a wrapping sum +//! of per-row hashes) plus a row count, computed identically from a database +//! connection or from an archive's NDJSON. Matching digests mean the two sides +//! hold the same canonical data; the check streams rows, so it is O(1) memory. + +use std::collections::hash_map::DefaultHasher; +use std::hash::{Hash, Hasher}; + +use chrono::{DateTime, Utc}; +use serde::Serialize; +use serde_json::{Number, Value}; + +/// Per-table digest: an order-independent content checksum and a row count. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct TableDigest { + pub table: String, + pub checksum: u128, + pub rows: u64, +} + +/// A table whose source and target content or row count disagree. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct FullMismatch { + pub table: String, + pub source_rows: u64, + pub target_rows: u64, + /// True when the row counts match but the canonical content differs. + pub content_differs: bool, +} + +/// Accumulates a table's digest one row at a time. +#[derive(Default)] +pub(crate) struct DigestAccumulator { + checksum: u128, + rows: u64, +} + +impl DigestAccumulator { + /// Fold one row (any `Serialize` model) into the digest. + pub(crate) fn add(&mut self, model: &M) { + self.checksum = self + .checksum + .wrapping_add(u128::from(canonical_row_hash(model))); + self.rows += 1; + } + + pub(crate) fn finish(self, table: String) -> TableDigest { + TableDigest { + table, + checksum: self.checksum, + rows: self.rows, + } + } +} + +/// Hash of a row's canonical representation. +fn canonical_row_hash(model: &M) -> u64 { + let value = serde_json::to_value(model).unwrap_or(Value::Null); + let canonical = serde_json::to_string(&canonicalize(value)).unwrap_or_default(); + let mut hasher = DefaultHasher::new(); + canonical.hash(&mut hasher); + hasher.finish() +} + +/// Recursively rewrite a JSON value into a canonical form. +fn canonicalize(value: Value) -> Value { + match value { + Value::Number(n) => canonical_number(n), + Value::String(s) => Value::String(canonical_string(&s)), + Value::Array(items) => Value::Array(items.into_iter().map(canonicalize).collect()), + Value::Object(map) => { + // Sort keys so jsonb's reordering doesn't matter, and canonicalize + // each value. + let mut entries: Vec<(String, Value)> = map.into_iter().collect(); + entries.sort_by(|a, b| a.0.cmp(&b.0)); + let mut out = serde_json::Map::with_capacity(entries.len()); + for (k, v) in entries { + out.insert(k, canonicalize(v)); + } + Value::Object(out) + } + other => other, + } +} + +/// Normalize integer-valued floats to integers (`1.0` → `1`), so a value stored +/// as text JSON and re-read from `jsonb` compares equal. +fn canonical_number(n: Number) -> Value { + if let Some(f) = n.as_f64() + && f.is_finite() + && f.fract() == 0.0 + && f.abs() < 9.0e15 + { + return if f >= 0.0 { + Value::Number(Number::from(f as u64)) + } else { + Value::Number(Number::from(f as i64)) + }; + } + Value::Number(n) +} + +/// Truncate RFC 3339 timestamps to microsecond precision so a value that +/// PostgreSQL truncated matches the SQLite original. Non-timestamp strings pass +/// through unchanged (and identical strings on both sides stay identical). +fn canonical_string(s: &str) -> String { + if let Ok(dt) = DateTime::parse_from_rfc3339(s) { + return dt + .with_timezone(&Utc) + .format("%Y-%m-%dT%H:%M:%S%.6fZ") + .to_string(); + } + s.to_string() +} + +/// Compare two sets of per-table digests, returning every table whose row count +/// or canonical content differs. +pub fn compare_digests(source: &[TableDigest], target: &[TableDigest]) -> Vec { + let mut mismatches = Vec::new(); + for s in source { + let t = target.iter().find(|t| t.table == s.table); + let (target_rows, target_checksum) = match t { + Some(t) => (t.rows, t.checksum), + None => (0, 0), + }; + if s.rows != target_rows { + mismatches.push(FullMismatch { + table: s.table.clone(), + source_rows: s.rows, + target_rows, + content_differs: false, + }); + } else if s.checksum != target_checksum { + mismatches.push(FullMismatch { + table: s.table.clone(), + source_rows: s.rows, + target_rows, + content_differs: true, + }); + } + } + mismatches +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + #[test] + fn integer_valued_floats_normalize_to_ints() { + assert_eq!(canonicalize(json!(1.0)), json!(1)); + assert_eq!( + canonicalize(json!({"n": 1.0})), + canonicalize(json!({"n": 1})) + ); + } + + #[test] + fn object_key_order_is_normalized() { + let a = canonicalize(json!({"b": 1, "a": 2})); + let b = canonicalize(json!({"a": 2, "b": 1})); + assert_eq!(a, b); + } + + #[test] + fn timestamps_truncate_to_microseconds() { + // Same instant, nanosecond vs microsecond precision. + let nanos = canonical_string("2026-07-05T12:00:00.123456789Z"); + let micros = canonical_string("2026-07-05T12:00:00.123456Z"); + assert_eq!(nanos, micros); + } + + #[test] + fn genuinely_different_rows_hash_differently() { + assert_ne!( + canonical_row_hash(&json!({"a": 1})), + canonical_row_hash(&json!({"a": 2})) + ); + } +} diff --git a/crates/codex-migrate/src/lib.rs b/crates/codex-migrate/src/lib.rs index d4a5f3a4..49607e94 100644 --- a/crates/codex-migrate/src/lib.rs +++ b/crates/codex-migrate/src/lib.rs @@ -15,6 +15,7 @@ pub mod archive; pub mod conn; pub mod engine; pub mod fk; +pub mod full_verify; pub mod guard; pub mod manifest; pub mod progress; diff --git a/crates/codex-migrate/src/registry.rs b/crates/codex-migrate/src/registry.rs index 9455cd89..3e2a1b75 100644 --- a/crates/codex-migrate/src/registry.rs +++ b/crates/codex-migrate/src/registry.rs @@ -12,6 +12,7 @@ use anyhow::Result; use sea_orm::{ConnectionTrait, EntityName, StreamTrait}; use crate::engine; +use crate::full_verify::TableDigest; use crate::progress::Progress; /// Row count for a single table, produced by the collective operations. @@ -224,3 +225,45 @@ where for_each_entity!(load_one); Ok(rows) } + +/// Canonical content digest of every table read from a connection. +pub async fn digest_all_from_conn(conn: &C) -> Result> +where + C: ConnectionTrait + StreamTrait, +{ + let mut digests = Vec::new(); + macro_rules! digest_one { + ($ent:ident) => {{ + type E = codex_db::entities::$ent::Entity; + digests.push(engine::digest_table_from_conn::(conn).await?); + }}; + } + for_each_entity!(digest_one); + Ok(digests) +} + +/// Canonical content digest of every table read from a directory of +/// `
.ndjson` files. Missing files are treated as empty tables. +pub async fn digest_all_from_ndjson_dir(dir: &Path) -> Result> { + let mut digests = Vec::new(); + macro_rules! digest_one { + ($ent:ident) => {{ + type E = codex_db::entities::$ent::Entity; + let table = ::default().table_name().to_string(); + let path = dir.join(format!("{table}.ndjson")); + let digest = if path.exists() { + let reader = std::io::BufReader::new(std::fs::File::open(&path)?); + engine::digest_table_from_ndjson::(reader)? + } else { + TableDigest { + table, + checksum: 0, + rows: 0, + } + }; + digests.push(digest); + }}; + } + for_each_entity!(digest_one); + Ok(digests) +} diff --git a/crates/codex-migrate/tests/archive_roundtrip.rs b/crates/codex-migrate/tests/archive_roundtrip.rs index 04792239..09c159c2 100644 --- a/crates/codex-migrate/tests/archive_roundtrip.rs +++ b/crates/codex-migrate/tests/archive_roundtrip.rs @@ -167,6 +167,7 @@ async fn archive_roundtrip_mirrors_data_and_reroots_paths() { }, ], codex_migrate::Progress::Silent, + false, ) .await .expect("import should succeed"); diff --git a/src/commands/copy.rs b/src/commands/copy.rs index 11beb460..965b6afa 100644 --- a/src/commands/copy.rs +++ b/src/commands/copy.rs @@ -24,6 +24,7 @@ pub async fn copy_command( replace: bool, progress: bool, no_verify: bool, + full_verification: bool, ) -> Result<()> { // Local config: used for tracing and as the fallback for an omitted side. let (local_config, _created) = load_config(config_path.clone())?; @@ -97,6 +98,16 @@ pub async fn copy_command( super::verify_row_counts(&source_counts, target.sea_orm_connection()).await?; } + if full_verification { + info!("Running full per-record verification..."); + let src_digests = + codex_migrate::registry::digest_all_from_conn(source.sea_orm_connection()).await?; + let dst_digests = + codex_migrate::registry::digest_all_from_conn(target.sea_orm_connection()).await?; + let mismatches = codex_migrate::full_verify::compare_digests(&src_digests, &dst_digests); + super::report_full_verify(&mismatches, src_digests.len()); + } + info!("========================================"); info!( "✓ Copy complete: {} rows across {} tables", @@ -204,6 +215,7 @@ files: false, false, false, + false, ) .await .expect("copy should succeed"); @@ -219,7 +231,7 @@ files: async fn copy_requires_at_least_one_explicit_side() { let dir = TempDir::new().unwrap(); let cfg = write_config(dir.path(), "only"); - let err = copy_command(cfg, None, None, None, None, false, false, false) + let err = copy_command(cfg, None, None, None, None, false, false, false, false) .await .expect_err("copy with no explicit side must error"); assert!(err.to_string().contains("at least one explicit side")); diff --git a/src/commands/import.rs b/src/commands/import.rs index 17cda6f5..a267542c 100644 --- a/src/commands/import.rs +++ b/src/commands/import.rs @@ -18,6 +18,7 @@ pub async fn import_command( replace: bool, progress: bool, no_verify: bool, + full_verification: bool, ) -> Result<()> { let (config, _created) = load_config(config_path.clone())?; let _tracing = init_tracing(&config)?; @@ -62,10 +63,15 @@ pub async fn import_command( &input, &artifact_targets(&config), codex_migrate::Progress::from_flag(progress), + full_verification, ) .await .context("Import failed")?; + if full_verification { + super::report_full_verify(&outcome.full_verify, codex_migrate::table_names().len()); + } + if !no_verify { // The manifest records the source's per-table counts at export time. let source_counts: Vec = outcome @@ -199,13 +205,13 @@ files: assert!(archive.exists(), "archive written"); // Import into a fresh target. - import_command(tgt_cfg.clone(), archive.clone(), false, false, false) + import_command(tgt_cfg.clone(), archive.clone(), false, false, false, false) .await .expect("import into fresh target should succeed"); assert_eq!(library_names(&tgt_cfg).await, vec!["Comics".to_string()]); // A second import without --replace is refused (target now has data). - let err = import_command(tgt_cfg.clone(), archive.clone(), false, false, false) + let err = import_command(tgt_cfg.clone(), archive.clone(), false, false, false, false) .await .expect_err("import into non-fresh target should be refused"); assert!( @@ -214,7 +220,7 @@ files: ); // With --replace it succeeds and still mirrors the source. - import_command(tgt_cfg.clone(), archive.clone(), true, false, false) + import_command(tgt_cfg.clone(), archive.clone(), true, false, false, false) .await .expect("import --replace should succeed"); assert_eq!(library_names(&tgt_cfg).await, vec!["Comics".to_string()]); @@ -224,9 +230,16 @@ files: async fn import_rejects_missing_archive() { let dir = TempDir::new().unwrap(); let cfg = write_config(dir.path(), "tgt"); - let err = import_command(cfg, dir.path().join("nope.tar.gz"), false, false, false) - .await - .expect_err("missing archive should error"); + let err = import_command( + cfg, + dir.path().join("nope.tar.gz"), + false, + false, + false, + false, + ) + .await + .expect_err("missing archive should error"); assert!(err.to_string().contains("archive not found")); } } diff --git a/src/commands/mod.rs b/src/commands/mod.rs index 5ac8cbb3..236e90ef 100644 --- a/src/commands/mod.rs +++ b/src/commands/mod.rs @@ -15,6 +15,7 @@ pub use export::export_command; pub use import::import_command; use anyhow::{Result, bail}; +use codex_migrate::full_verify::FullMismatch; use codex_migrate::{TableRows, registry, verify}; use sea_orm::DatabaseConnection; use tracing::{info, warn}; @@ -40,6 +41,39 @@ pub(crate) async fn verify_row_counts( ); } } + +/// Print the outcome of a full (per-record) verification as a report. +/// +/// This is an informational deep check: it reports which tables differ but does +/// **not** fail the command (the default row-count check is the hard safety +/// gate). `tables_checked` is the total number of tables compared. +pub(crate) fn report_full_verify(mismatches: &[FullMismatch], tables_checked: usize) { + info!("---------- full verification ----------"); + if mismatches.is_empty() { + info!("✓ every row matches across all {tables_checked} tables (canonical comparison)"); + } else { + warn!( + "⚠ {} of {} table(s) differ:", + mismatches.len(), + tables_checked + ); + for m in mismatches { + if m.content_differs { + warn!( + " ✗ {}: {} rows, content differs (values not identical after canonicalization)", + m.table, m.source_rows + ); + } else { + warn!( + " ✗ {}: row count differs — source={} target={}", + m.table, m.source_rows, m.target_rows + ); + } + } + warn!("(full verification is informational; the command did not fail on these)"); + } + info!("---------------------------------------"); +} pub use migrate::migrate_command; pub use openapi::{OpenApiFormat, openapi_command}; pub use scan::scan_command; diff --git a/src/main.rs b/src/main.rs index 5de0c696..6119cc49 100644 --- a/src/main.rs +++ b/src/main.rs @@ -163,6 +163,10 @@ enum Commands { /// Skip the post-import row-count verification #[arg(long)] no_verify: bool, + + /// Also compare every record's canonical content between archive and target (slower) + #[arg(long)] + full_verification: bool, }, /// Copy database rows directly from one instance to another @@ -198,6 +202,10 @@ enum Commands { /// Skip the post-copy row-count verification #[arg(long)] no_verify: bool, + + /// Also compare every record's canonical content between source and target (slower) + #[arg(long)] + full_verification: bool, }, } @@ -270,8 +278,17 @@ async fn main() -> anyhow::Result<()> { replace, progress, no_verify, + full_verification, } => { - import_command(config, input, replace, progress, no_verify).await?; + import_command( + config, + input, + replace, + progress, + no_verify, + full_verification, + ) + .await?; } Commands::Copy { config, @@ -282,6 +299,7 @@ async fn main() -> anyhow::Result<()> { replace, progress, no_verify, + full_verification, } => { copy_command( config, @@ -292,6 +310,7 @@ async fn main() -> anyhow::Result<()> { replace, progress, no_verify, + full_verification, ) .await?; } diff --git a/tests/migrate/mod.rs b/tests/migrate/mod.rs index 6dba1313..4670f7f1 100644 --- a/tests/migrate/mod.rs +++ b/tests/migrate/mod.rs @@ -313,7 +313,7 @@ async fn export_import_pair(src: &DatabaseConnection, tgt: &DatabaseConnection, export_archive(src, &archive, &[], codex::migrate::Progress::Silent) .await .unwrap_or_else(|e| panic!("{label}: export failed: {e:#}")); - import_archive(tgt, &archive, &[], codex::migrate::Progress::Silent) + import_archive(tgt, &archive, &[], codex::migrate::Progress::Silent, false) .await .unwrap_or_else(|e| panic!("{label}: import failed: {e:#}")); @@ -551,3 +551,61 @@ async fn copy_works_as_non_superuser_database_owner() { .await .ok(); } + +// --------------------------------------------------------------------------- +// Full (per-record) verification: canonical content matches cross-engine, and +// a tampered row is detected. +// --------------------------------------------------------------------------- + +#[tokio::test] +#[ignore] // requires a PostgreSQL test database +async fn full_verification_matches_then_detects_tampering() { + use codex::migrate::full_verify::compare_digests; + + let (src, _sd) = setup_test_db_wrapper().await; + let Some(pg) = fresh_pg("codex_test_fullverify").await else { + return; + }; + // Rich fixture: JSON, a float, ints, timestamps, UUID FKs. + seed_source(&src).await; + + transfer( + src.sea_orm_connection(), + &pg, + codex::migrate::Progress::Silent, + ) + .await + .unwrap(); + + // Positive: canonical content matches across engines despite jsonb key + // reordering, number normalization, and timestamp precision differences. + let src_digests = registry::digest_all_from_conn(src.sea_orm_connection()) + .await + .unwrap(); + let pg_digests = registry::digest_all_from_conn(&pg).await.unwrap(); + let clean = compare_digests(&src_digests, &pg_digests); + assert!( + clean.is_empty(), + "content should match cross-engine: {clean:?}" + ); + + // Negative: mutate one row on the target; only that table should differ. + pg.execute_unprepared("UPDATE users SET username = 'tampered'") + .await + .unwrap(); + let pg_digests2 = registry::digest_all_from_conn(&pg).await.unwrap(); + let mismatches = compare_digests(&src_digests, &pg_digests2); + assert!( + mismatches + .iter() + .any(|m| m.table == "users" && m.content_differs), + "tampered users row should be detected: {mismatches:?}" + ); + assert!( + mismatches.iter().all(|m| m.table == "users"), + "only users should differ: {mismatches:?}" + ); + + drop(pg); + drop_pg("codex_test_fullverify").await; +} From 84cf1deba9060b05e98830fd1297824f4f3c44e4 Mon Sep 17 00:00:00 2001 From: Sylvain Cau Date: Sun, 5 Jul 2026 15:25:58 -0700 Subject: [PATCH 5/5] docs(migrate): document progress, verification, and owner-only FK handling - document --progress, --no-verify, and the --full-verification report on the export/import/copy page, and note row-count verification runs by default - correct the PostgreSQL privileges note: the load now drops and recreates FK constraints (owner-only), so no superuser is required, including on managed PostgreSQL - update the contributor page for the FK approach, batch-size capping, and the two verification levels --- docs/dev/contributing/data-export-import.md | 29 +++++++++++++--- .../backup-migration/export-import-copy.md | 33 ++++++++++++++++--- .../docs/backup-migration/migrate-postgres.md | 8 ++--- 3 files changed, 57 insertions(+), 13 deletions(-) diff --git a/docs/dev/contributing/data-export-import.md b/docs/dev/contributing/data-export-import.md index 9605568c..6a30d779 100644 --- a/docs/dev/contributing/data-export-import.md +++ b/docs/dev/contributing/data-export-import.md @@ -33,11 +33,12 @@ result. from that single list. A drift-guard test fails if a migration adds a table that isn't registered. - **Foreign keys** — a 1:1 load inserts tables in arbitrary order, so FK - enforcement is disabled for the duration of the load (SQLite - `defer_foreign_keys`; PostgreSQL `session_replication_role = replica`, which - needs owner/superuser) and the whole load runs in one destination - transaction. On SQLite the commit re-validates FKs; on PostgreSQL integrity - rests on source consistency plus a post-load row-count check. + enforcement is suppressed for the load. SQLite uses `defer_foreign_keys` (the + commit re-validates). PostgreSQL **drops the FK constraints and recreates + them** after the load — recreating revalidates the rows, and it needs only + table ownership (not a superuser, unlike `session_replication_role`), so it + works on managed PostgreSQL. The whole load runs in one destination + transaction, so a failure rolls back cleanly. - **Truncate before load** — migrations seed rows (e.g. `settings`), so a freshly-migrated target is *not* empty. A faithful mirror therefore truncates every table before loading. The CLI's fresh-target guard (refuse a target @@ -49,6 +50,24 @@ result. a query that coerces UUID columns back to blobs (`text → unhex(replace(col,'-',''))`, blobs pass through), so mixed storage decodes correctly and is canonicalized on the way out. +- **Batch sizing** — a multi-row insert binds `rows × columns` parameters, and + PostgreSQL caps a statement at 65535 (SQLite at 32766). The batch size is + capped per table so a wide table (e.g. `book_metadata`, ~66 columns) can't + overflow the destination's limit. + +## Verification + +Two levels, both surfaced by the CLI: + +- **Row-count parity** (`registry::count_all` + `verify::compare`) runs by + default after `import`/`copy` and fails the command on any mismatch. +- **Full verification** (`full_verify`) is opt-in and reports (does not fail): + each table is reduced to an order-independent digest of every row's + *canonical* value — integer-valued floats normalize (`1.0` == `1`), JSON keys + are sorted (jsonb reorders), and timestamps truncate to microseconds. Digests + are computed identically from a connection or the archive's NDJSON, so it + streams and stays O(1) memory. Genuine content differences (or a tampered + target) change the digest; representation differences do not. ## Archive format diff --git a/docs/docs/backup-migration/export-import-copy.md b/docs/docs/backup-migration/export-import-copy.md index 5427aeea..6f1d120d 100644 --- a/docs/docs/backup-migration/export-import-copy.md +++ b/docs/docs/backup-migration/export-import-copy.md @@ -44,6 +44,7 @@ covers**, and **plugin data**. Flags: | `--no-thumbnails` | Skip generated thumbnails | | `--no-uploads` | Skip uploaded/extracted covers | | `--no-plugins` | Skip plugin data | +| `--progress` | Log per-table progress while exporting | The archive contains a `manifest.json` (format and schema version, per-table row counts, bundled artifact groups), one `db/
.ndjson` per table, and the @@ -105,6 +106,28 @@ To avoid leaking a password via the process list, prefer the env vars or `--from-config` / `--to-config` over a `postgres://user:pass@…` URL on the command line. +## Progress & verification + +`import` and `copy` accept: + +| Flag | Effect | +|------|--------| +| `--progress` | Log per-table progress (and a periodic row count for large tables) — reads the same in a terminal or in captured logs | +| `--no-verify` | Skip the row-count verification | +| `--full-verification` | Additionally compare every record's content (see below) | + +**Row-count verification runs by default.** After the load, the source and +target are re-counted per table and compared; a mismatch **fails the command**. +This is your confirmation that no rows were dropped. + +**`--full-verification`** is an opt-in deeper check: it compares every record's +*canonical* content on both sides and prints a **report** (it does not fail the +command). "Canonical" means representation differences that don't change meaning +are ignored — `1.0` equals `1`, JSON object key order is normalized (PostgreSQL +`jsonb` reorders keys), and timestamps are compared at microsecond precision +(PostgreSQL truncates). It streams rows, so it stays memory-cheap even on large +tables, but it does re-read everything, so it adds time. + ## Setting up the target You never create tables — `import` and `copy` run the migrations themselves. @@ -121,7 +144,7 @@ What you need to prepare depends on the engine: ```sql CREATE DATABASE codex; CREATE USER codex WITH PASSWORD '...'; - ALTER DATABASE codex OWNER TO codex; -- owner/superuser: see the privilege note below + ALTER DATABASE codex OWNER TO codex; -- the role must own the database (see the note below) ``` Then point the config/env at it and import — the schema and data are created @@ -163,7 +186,9 @@ encryption key** as the source, or those values cannot be decrypted afterwards. ::: :::note PostgreSQL privileges -`import` and `copy` temporarily disable foreign-key enforcement during the bulk -load, which requires the target connection to be a **superuser or the database -owner**. This is normally the case when you provision the database yourself. +`import` and `copy` suppress foreign-key enforcement during the bulk load by +dropping and recreating the FK constraints, which requires only that the target +role **owns the tables** — no superuser needed. This works on managed +PostgreSQL, where you typically get a database owner but not a superuser. (The +role owns the tables because it ran the migrations that created them.) ::: diff --git a/docs/docs/backup-migration/migrate-postgres.md b/docs/docs/backup-migration/migrate-postgres.md index 197e9b5d..d1993c6f 100644 --- a/docs/docs/backup-migration/migrate-postgres.md +++ b/docs/docs/backup-migration/migrate-postgres.md @@ -65,10 +65,10 @@ cannot be decrypted after the move. ::: :::note PostgreSQL privileges -`import` temporarily disables foreign-key enforcement during the bulk load -(`SET session_replication_role = replica`), which requires the target -connection to be a **superuser or the database owner**. This is normally the -case when you provision the database yourself. +`import` suppresses foreign-key enforcement during the bulk load by dropping and +recreating the FK constraints, which requires only that the target role **owns +the tables** (it does, because it ran the migrations). No superuser is needed, +so this works on managed PostgreSQL. ::: ## Library files