huskies: merge 1194 bug db shadow-table tests flake when SHADOW_DB_PATH is not initialized

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<PipelineDb> = OnceLock::new();
 /// which test won the `PIPELINE_DB` init race.
 pub(crate) static SHADOW_DB_PATH: OnceLock<std::path::PathBuf> = 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() {
+    INIT.get_or_try_init(|| async {
-        return Ok(());
+        // Record the path before doing any real work so tests can always
-    }
+        // find the correct file — this closure is guaranteed to run at most
-    // Record the path before doing any real work so tests can always find the
+        // once, so there is no longer a race with the `PIPELINE_DB` set below.
-    // correct file even if two callers race — the OnceLock ensures only one
+        let _ = SHADOW_DB_PATH.set(db_path.to_path_buf());
-    // path wins, and whichever wins will also win 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
+        // Story 1087: before running the migration that splits `stage` into
-    // (`pipeline`, `status`), take a timestamped side-car copy of the live DB
+        // (`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
+        // 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
+        // not yet exist (fresh installs) or when the split-stage migration has
-    // already been applied (subsequent restarts).
+        // already been applied (subsequent restarts).
-    backup_pre_pipeline_status(db_path).await;
+        backup_pre_pipeline_status(db_path).await;
 
-    let options = SqliteConnectOptions::new()
+        let options = SqliteConnectOptions::new()
-        .filename(db_path)
+            .filename(db_path)
-        .create_if_missing(true);
+            .create_if_missing(true);
 
-    let pool = SqlitePool::connect_with(options).await?;
+        let pool = SqlitePool::connect_with(options).await?;
-    sqlx::migrate!("./migrations").run(&pool).await?;
+        sqlx::migrate!("./migrations").run(&pool).await?;
 
-    // Store pool in global static so other subsystems can reuse it.
+        // Store pool in global static so other subsystems can reuse it.
-    let _ = SHARED_POOL.set(pool.clone());
+        let _ = SHARED_POOL.set(pool.clone());
 
-    // Load existing content into the in-memory store.
+        // Load existing content into the in-memory store.
-    let rows: Vec<(String, Option<String>)> =
+        let rows: Vec<(String, Option<String>)> =
-        sqlx::query_as("SELECT id, content FROM pipeline_items WHERE content IS NOT NULL")
+            sqlx::query_as("SELECT id, content FROM pipeline_items WHERE content IS NOT NULL")
-            .fetch_all(&pool)
+                .fetch_all(&pool)
-            .await?;
+                .await?;
 
-    let mut content_map = HashMap::new();
+        let mut content_map = HashMap::new();
-    for (id, content) in rows {
+        for (id, content) in rows {
-        if let Some(c) = content {
+            if let Some(c) = content {
-            content_map.insert(id, c);
+                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::<PipelineWriteMsg>();
+        let (tx, mut rx) = mpsc::unbounded_channel::<PipelineWriteMsg>();
 
-    tokio::spawn(async move {
+        tokio::spawn(async move {
-        while let Some(msg) = rx.recv().await {
+            while let Some(msg) = rx.recv().await {
-            // The "deleted" sentinel means the caller wants the row gone.
+                // The "deleted" sentinel means the caller wants the row gone.
-            // Issue a real DELETE so the shadow table stays clean and
+                // Issue a real DELETE so the shadow table stays clean and
-            // sync_crdt_stages_from_db cannot resurrect a tombstoned item on
+                // sync_crdt_stages_from_db cannot resurrect a tombstoned item on
-            // the next restart.
+                // the next restart.
-            if msg.stage == "deleted" {
+                if msg.stage == "deleted" {
-                let result = sqlx::query("DELETE FROM pipeline_items WHERE id = ?1")
+                    let result = sqlx::query("DELETE FROM pipeline_items WHERE id = ?1")
-                    .bind(&msg.story_id)
+                        .bind(&msg.story_id)
-                    .execute(&pool)
+                        .execute(&pool)
-                    .await;
+                        .await;
-                if let Err(e) = result {
+                    if let Err(e) = result {
-                    slog!("[db] Shadow delete failed for '{}': {e}", msg.story_id);
+                        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 _ = PIPELINE_DB.set(PipelineDb { tx });
-            let result = sqlx::query(
+        Ok::<(), sqlx::Error>(())
-                "INSERT INTO pipeline_items \
+    })
-                     (id, name, stage, agent, retry_count, depends_on, content, created_at, updated_at) \
+    .await?;
-                 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 });
     Ok(())
 }