diff --git a/crates/adapters/sqlite/src/migrations.rs b/crates/adapters/sqlite/src/migrations.rs index 3df60715..afcae43b 100644 --- a/crates/adapters/sqlite/src/migrations.rs +++ b/crates/adapters/sqlite/src/migrations.rs @@ -1,10 +1,34 @@ +use std::time::Duration; + use domain::errors::DomainError; use sqlx::SqlitePool; +/// Server and worker both migrate at startup, so two processes can race on the +/// same database. The loser's migration fails (e.g. duplicate column) because +/// the winner already applied it; re-running re-reads `_sqlx_migrations` and +/// skips what is done. Only an error that persists across all retries is returned. +const RETRY_BACKOFFS: [Duration; 5] = [ + Duration::from_millis(100), + Duration::from_millis(250), + Duration::from_millis(500), + Duration::from_secs(1), + Duration::from_secs(1), +]; + pub(crate) async fn run(pool: &SqlitePool) -> Result<(), DomainError> { - sqlx::migrate!("./migrations") - .set_ignore_missing(true) - .run(pool) - .await - .map_err(|e| DomainError::InfrastructureError(e.to_string())) + let mut migrator = sqlx::migrate!("./migrations"); + migrator.set_ignore_missing(true); + let mut backoffs = RETRY_BACKOFFS.iter(); + loop { + match migrator.run(pool).await { + Ok(()) => return Ok(()), + Err(e) => match backoffs.next() { + Some(delay) => { + tracing::warn!("migration failed, retrying (concurrent migrator?): {e}"); + tokio::time::sleep(*delay).await; + } + None => return Err(DomainError::InfrastructureError(e.to_string())), + }, + } + } } diff --git a/crates/adapters/sqlite/tests/migration_ledger_test.rs b/crates/adapters/sqlite/tests/migration_ledger_test.rs index b226c58a..fbc1c75d 100644 --- a/crates/adapters/sqlite/tests/migration_ledger_test.rs +++ b/crates/adapters/sqlite/tests/migration_ledger_test.rs @@ -56,3 +56,47 @@ async fn a_database_migrated_by_the_old_layout_migrates_again_cleanly() { assert_eq!(ledger(&pool).await, LEDGER); } + +async fn wal_pool(path: &Path, create: bool) -> SqlitePool { + let opts = SqliteConnectOptions::new() + .filename(path) + .create_if_missing(create) + .journal_mode(sqlx::sqlite::SqliteJournalMode::Wal) + .busy_timeout(std::time::Duration::from_secs(5)); + SqlitePoolOptions::new() + .max_connections(2) + .connect_with(opts) + .await + .unwrap() +} + +async fn race_two_migrators(start: Option<&Path>) { + for _ in 0..10 { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("race.db"); + if let Some(src) = start { + std::fs::copy(src, &path).unwrap(); + } + let a = wal_pool(&path, start.is_none()).await; + let b = wal_pool(&path, false).await; + + let (ra, rb) = tokio::join!(sqlite::migrate(&a), sqlite::migrate(&b)); + + ra.unwrap(); + rb.unwrap(); + assert_eq!(ledger(&a).await, LEDGER); + } +} + +#[tokio::test] +async fn two_processes_migrating_the_same_fresh_database_both_succeed() { + race_two_migrators(None).await; +} + +#[tokio::test] +async fn two_processes_migrating_a_database_at_the_old_version_both_succeed() { + race_two_migrators(Some( + &Path::new(env!("CARGO_MANIFEST_DIR")).join(PRE_MERGE_DB), + )) + .await; +} diff --git a/docs/adr/0015-transactional-outbox.md b/docs/adr/0015-transactional-outbox.md index 50ae4d24..19b50a09 100644 --- a/docs/adr/0015-transactional-outbox.md +++ b/docs/adr/0015-transactional-outbox.md @@ -79,7 +79,9 @@ backends dead-letter after the same number of attempts. and updates `duplicate_window` (1 h) and the consumer's `max_deliver`/`ack_wait` in place, with a warning in the log. Both are mutable on a live stream/consumer, so no recreate is needed; immutable consumer fields (name, filter, deliver policy) are never touched. -- Starting server and worker at the same moment against a database that still needs migration `0101` - can make one fail with `duplicate column name`; start the server first, or restart the loser. +- Server and worker both migrate at startup (docker compose starts them together). `sqlite::migrate` + retries with a bounded backoff (100 ms to 1 s, 5 retries), so the loser of a migration race + re-reads `_sqlx_migrations`, skips what the winner applied and succeeds; only a persistent error + is returned. - Handler authors must keep new handlers idempotent; the ledger narrows duplicates but cannot remove the crash-after-effect case.