diff --git a/server/src/db/shadow_write.rs b/server/src/db/shadow_write.rs index 424ccf0a..b66f15fe 100644 --- a/server/src/db/shadow_write.rs +++ b/server/src/db/shadow_write.rs @@ -14,6 +14,7 @@ use std::collections::HashMap; use std::collections::HashSet; use std::path::Path; use std::sync::OnceLock; +use tokio::sync::OnceCell; use tokio::sync::mpsc; /// One migration row in the live database that is not in the compiled-in set. @@ -66,102 +67,117 @@ pub(crate) static PIPELINE_DB: OnceLock = OnceLock::new(); /// which test won the `PIPELINE_DB` init race. pub(crate) static SHADOW_DB_PATH: OnceLock = OnceLock::new(); +/// Guards the body of [`init`] so concurrent callers race safely. +/// +/// Before this existed, `PIPELINE_DB` and `SHADOW_DB_PATH` were two separate +/// `OnceLock`s set at different points in the function (one right at the +/// start, one only after the pool/migrations/write-task were fully set up). +/// Two callers racing past the old top-of-function check could each open +/// their own SQLite file, with one call's `SHADOW_DB_PATH` winning while a +/// *different* call's `PIPELINE_DB` (and its background writer) won — +/// leaving readers pointed at a file that the writer never touched +/// (story 1194). `get_or_try_init` ensures only the first caller runs the +/// body below; every other caller — even racing with a different `db_path` +/// — awaits and observes that same result. +static INIT: OnceCell<()> = OnceCell::const_new(); + /// Initialise the pipeline database. /// /// Opens (or creates) the SQLite file at `db_path`, runs embedded migrations, /// loads existing story content into the in-memory store, and spawns the /// background write task. Safe to call only once; subsequent calls are no-ops. pub async fn init(db_path: &Path) -> Result<(), sqlx::Error> { - if PIPELINE_DB.get().is_some() { - return Ok(()); - } - // Record the path before doing any real work so tests can always find the - // correct file even if two callers race — the OnceLock ensures only one - // path wins, and whichever wins will also win the PIPELINE_DB set below. - let _ = SHADOW_DB_PATH.set(db_path.to_path_buf()); + INIT.get_or_try_init(|| async { + // Record the path before doing any real work so tests can always + // find the correct file — this closure is guaranteed to run at most + // once, so there is no longer a race with the `PIPELINE_DB` set below. + let _ = SHADOW_DB_PATH.set(db_path.to_path_buf()); - // Story 1087: before running the migration that splits `stage` into - // (`pipeline`, `status`), take a timestamped side-car copy of the live DB - // so the pre-split state is recoverable. Skip the copy when the file does - // not yet exist (fresh installs) or when the split-stage migration has - // already been applied (subsequent restarts). - backup_pre_pipeline_status(db_path).await; + // Story 1087: before running the migration that splits `stage` into + // (`pipeline`, `status`), take a timestamped side-car copy of the live DB + // so the pre-split state is recoverable. Skip the copy when the file does + // not yet exist (fresh installs) or when the split-stage migration has + // already been applied (subsequent restarts). + backup_pre_pipeline_status(db_path).await; - let options = SqliteConnectOptions::new() - .filename(db_path) - .create_if_missing(true); + let options = SqliteConnectOptions::new() + .filename(db_path) + .create_if_missing(true); - let pool = SqlitePool::connect_with(options).await?; - sqlx::migrate!("./migrations").run(&pool).await?; + let pool = SqlitePool::connect_with(options).await?; + sqlx::migrate!("./migrations").run(&pool).await?; - // Store pool in global static so other subsystems can reuse it. - let _ = SHARED_POOL.set(pool.clone()); + // Store pool in global static so other subsystems can reuse it. + let _ = SHARED_POOL.set(pool.clone()); - // Load existing content into the in-memory store. - let rows: Vec<(String, Option)> = - sqlx::query_as("SELECT id, content FROM pipeline_items WHERE content IS NOT NULL") - .fetch_all(&pool) - .await?; + // Load existing content into the in-memory store. + let rows: Vec<(String, Option)> = + sqlx::query_as("SELECT id, content FROM pipeline_items WHERE content IS NOT NULL") + .fetch_all(&pool) + .await?; - let mut content_map = HashMap::new(); - for (id, content) in rows { - if let Some(c) = content { - content_map.insert(id, c); + let mut content_map = HashMap::new(); + for (id, content) in rows { + if let Some(c) = content { + content_map.insert(id, c); + } } - } - super::content_store::init_content_store(content_map); + super::content_store::init_content_store(content_map); - let (tx, mut rx) = mpsc::unbounded_channel::(); + let (tx, mut rx) = mpsc::unbounded_channel::(); - tokio::spawn(async move { - while let Some(msg) = rx.recv().await { - // The "deleted" sentinel means the caller wants the row gone. - // Issue a real DELETE so the shadow table stays clean and - // sync_crdt_stages_from_db cannot resurrect a tombstoned item on - // the next restart. - if msg.stage == "deleted" { - let result = sqlx::query("DELETE FROM pipeline_items WHERE id = ?1") - .bind(&msg.story_id) - .execute(&pool) - .await; - if let Err(e) = result { - slog!("[db] Shadow delete failed for '{}': {e}", msg.story_id); + tokio::spawn(async move { + while let Some(msg) = rx.recv().await { + // The "deleted" sentinel means the caller wants the row gone. + // Issue a real DELETE so the shadow table stays clean and + // sync_crdt_stages_from_db cannot resurrect a tombstoned item on + // the next restart. + if msg.stage == "deleted" { + let result = sqlx::query("DELETE FROM pipeline_items WHERE id = ?1") + .bind(&msg.story_id) + .execute(&pool) + .await; + if let Err(e) = result { + slog!("[db] Shadow delete failed for '{}': {e}", msg.story_id); + } + continue; + } + + let now = chrono::Utc::now().to_rfc3339(); + let result = sqlx::query( + "INSERT INTO pipeline_items \ + (id, name, stage, agent, retry_count, depends_on, content, created_at, updated_at) \ + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?8) \ + ON CONFLICT(id) DO UPDATE SET \ + name = excluded.name, \ + stage = excluded.stage, \ + agent = excluded.agent, \ + retry_count = excluded.retry_count, \ + depends_on = excluded.depends_on, \ + content = COALESCE(excluded.content, pipeline_items.content), \ + updated_at = excluded.updated_at", + ) + .bind(&msg.story_id) + .bind(&msg.name) + .bind(&msg.stage) + .bind(&msg.agent) + .bind(msg.retry_count) + .bind(&msg.depends_on) + .bind(&msg.content) + .bind(&now) + .execute(&pool) + .await; + + if let Err(e) = result { + slog!("[db] Shadow write failed for '{}': {e}", msg.story_id); } - continue; } + }); - let now = chrono::Utc::now().to_rfc3339(); - let result = sqlx::query( - "INSERT INTO pipeline_items \ - (id, name, stage, agent, retry_count, depends_on, content, created_at, updated_at) \ - VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?8) \ - ON CONFLICT(id) DO UPDATE SET \ - name = excluded.name, \ - stage = excluded.stage, \ - agent = excluded.agent, \ - retry_count = excluded.retry_count, \ - depends_on = excluded.depends_on, \ - content = COALESCE(excluded.content, pipeline_items.content), \ - updated_at = excluded.updated_at", - ) - .bind(&msg.story_id) - .bind(&msg.name) - .bind(&msg.stage) - .bind(&msg.agent) - .bind(msg.retry_count) - .bind(&msg.depends_on) - .bind(&msg.content) - .bind(&now) - .execute(&pool) - .await; - - if let Err(e) = result { - slog!("[db] Shadow write failed for '{}': {e}", msg.story_id); - } - } - }); - - let _ = PIPELINE_DB.set(PipelineDb { tx }); + let _ = PIPELINE_DB.set(PipelineDb { tx }); + Ok::<(), sqlx::Error>(()) + }) + .await?; Ok(()) }