fix(sqlite): concurrent migrate retries

This commit is contained in:
2026-10-26 01:40:40 +01:00
parent 6ae3f54a36
commit 0908b7046a
3 changed files with 77 additions and 7 deletions

View File

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

View File

@@ -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;
}

View File

@@ -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.