Merge pull request #471 from ScriptedAlchemy/codex/graph-db-recovery

fix(storage): preflight dirty graph recovery read-only

Author: ScriptedAlchemy

src/tracedecay/lifecycle.rs

@@ -13,7 +13,9 @@ use crate::extraction::LanguageRegistry;
 use crate::global_db::{GraphScopeUpsert, StoreArtifactUpsert, StoreInstanceUpsert};
 use crate::storage::{self, StoreLayout};
 
-use super::locking::{clear_dirty_sentinel_at, has_dirty_sentinel_at};
+use super::locking::{
+    clear_dirty_sentinel_at, has_dirty_sentinel_at, try_acquire_graph_sync_locks,
+};
 use super::{TraceDecay, TraceDecayOpenOptions, current_timestamp};
 
 impl TraceDecay {
@@ -286,7 +288,7 @@ impl TraceDecay {
     /// Falls back to the nearest tracked ancestor DB with a warning only when
     /// the live branch cannot be auto-tracked, such as detached HEAD.
     /// If the previous operation was interrupted (dirty sentinel exists),
-    /// the database is integrity-checked and rebuilt if corrupted.
+    /// the database is integrity-checked before any writable open.
     pub async fn open(project_root: &Path) -> Result<Self> {
         Self::open_with_options(project_root, TraceDecayOpenOptions::default()).await
     }
@@ -337,6 +339,45 @@ impl TraceDecay {
             );
         }
 
+        // A dirty marker can also describe a sync that is still active in a
+        // peer process. Recovery must own both graph-local and legacy locks so
+        // it cannot race that writer or clear its sentinel. Preflight through
+        // the read-only connection before Database::open applies writable
+        // pragmas or migrations to a potentially damaged recovery set.
+        let _recovery_lock = if crashed {
+            Some(try_acquire_graph_sync_locks(
+                &active_graph_layout.sync_lock_path,
+                &store_layout.sync_lock_path,
+            )?)
+        } else {
+            None
+        };
+        if crashed {
+            let verification = match Database::open_read_only(&db_path).await {
+                Ok((db, _)) => db,
+                Err(error) => {
+                    print_corruption_warning(&db_path);
+                    return Err(recovery_required_error(&db_path, error));
+                }
+            };
+            let integrity = verification.quick_check().await;
+            verification.close();
+            match integrity {
+                Ok(true) => {}
+                Ok(false) => {
+                    print_corruption_warning(&db_path);
+                    return Err(recovery_required_error(
+                        &db_path,
+                        "read-only SQLite quick_check did not return ok",
+                    ));
+                }
+                Err(error) => {
+                    print_corruption_warning(&db_path);
+                    return Err(recovery_required_error(&db_path, error));
+                }
+            }
+        }
+
         // Ordinary opens never replace database files. A daemon or another MCP
         // process may still hold the current DB/WAL/SHM inodes, and deleting
         // them here would split readers and writers across different stores.

src/tracedecay/locking.rs

@@ -54,17 +54,10 @@ pub(super) struct ActiveSyncLockGuard {
 
 impl super::TraceDecay {
     pub(super) fn try_acquire_active_sync_lock(&self) -> Result<ActiveSyncLockGuard> {
-        let active = try_acquire_sync_lock_at(&self.active_graph_layout.sync_lock_path)?;
-        let legacy = if self.active_graph_layout.sync_lock_path == self.store_layout.sync_lock_path
-        {
-            None
-        } else {
-            Some(try_acquire_sync_lock_at(&self.store_layout.sync_lock_path)?)
-        };
-        Ok(ActiveSyncLockGuard {
-            _active: active,
-            _legacy: legacy,
-        })
+        try_acquire_graph_sync_locks(
+            &self.active_graph_layout.sync_lock_path,
+            &self.store_layout.sync_lock_path,
+        )
     }
 
     pub(super) fn write_active_dirty_sentinels(&self) {
@@ -82,6 +75,22 @@ impl super::TraceDecay {
     }
 }
 
+pub(super) fn try_acquire_graph_sync_locks(
+    active_path: &Path,
+    legacy_path: &Path,
+) -> Result<ActiveSyncLockGuard> {
+    let active = try_acquire_sync_lock_at(active_path)?;
+    let legacy = if active_path == legacy_path {
+        None
+    } else {
+        Some(try_acquire_sync_lock_at(legacy_path)?)
+    };
+    Ok(ActiveSyncLockGuard {
+        _active: active,
+        _legacy: legacy,
+    })
+}
+
 impl Drop for SyncLockGuard {
     fn drop(&mut self) {
         let _ = std::fs::remove_file(&self.path);

tests/storage_suite/corruption_test.rs

@@ -394,6 +394,140 @@ async fn open_preserves_corrupt_store_and_dirty_sentinel_for_offline_repair()
     Ok(())
 }
 
+#[tokio::test]
+async fn dirty_open_checks_integrity_before_writable_migration()
+-> std::result::Result<(), Box<dyn std::error::Error>> {
+    let dir = TempDir::new()?;
+    let project_root = dir.path().join("repo");
+    std::fs::create_dir_all(&project_root)?;
+    let open_options = TraceDecayOpenOptions {
+        profile_root: Some(dir.path().join("profile")),
+        global_db_path: Some(dir.path().join("global.db")),
+    };
+
+    let ts = TraceDecay::init_with_options(&project_root, open_options.clone()).await?;
+    let layout = ts.store_layout().clone();
+    ts.db()
+        .conn()
+        .execute_batch("PRAGMA user_version = 17")
+        .await?;
+    ts.checkpoint().await?;
+    ts.close();
+
+    let mut file = std::fs::OpenOptions::new()
+        .read(true)
+        .write(true)
+        .open(&layout.graph_db_path)?;
+    let offset = std::cmp::min(file.metadata()?.len() / 2, 8192);
+    file.seek(std::io::SeekFrom::Start(offset))?;
+    file.write_all(&[0xFF; 256])?;
+    file.sync_all()?;
+    drop(file);
+    std::fs::write(&layout.dirty_path, "pid=99999\nversion=test")?;
+
+    let before = std::fs::read(&layout.graph_db_path)?;
+    let result = TraceDecay::open_with_options(&project_root, open_options).await;
+    assert!(result.is_err(), "damaged dirty store must require recovery");
+    assert_eq!(
+        std::fs::read(&layout.graph_db_path)?,
+        before,
+        "integrity failure must be detected before writable migration"
+    );
+    assert!(layout.dirty_path.exists());
+    Ok(())
+}
+
+#[tokio::test]
+async fn dirty_open_does_not_race_an_active_sync_lock()
+-> std::result::Result<(), Box<dyn std::error::Error>> {
+    let dir = TempDir::new()?;
+    let project_root = dir.path().join("repo");
+    std::fs::create_dir_all(&project_root)?;
+    let open_options = TraceDecayOpenOptions {
+        profile_root: Some(dir.path().join("profile")),
+        global_db_path: Some(dir.path().join("global.db")),
+    };
+
+    let ts = TraceDecay::init_with_options(&project_root, open_options.clone()).await?;
+    let layout = ts.store_layout().clone();
+    ts.close();
+    let active_lock = layout.graph_db_path.with_file_name(format!(
+        "{}.sync.lock",
+        layout.graph_db_path.file_name().unwrap().to_string_lossy()
+    ));
+    std::fs::write(&active_lock, std::process::id().to_string())?;
+    std::fs::write(&layout.dirty_path, "pid=99999\nversion=test")?;
+    let before = std::fs::read(&layout.graph_db_path)?;
+
+    let error = match TraceDecay::open_with_options(&project_root, open_options).await {
+        Ok(_) => panic!("active writer lock must block recovery"),
+        Err(error) => error,
+    };
+    assert!(
+        error
+            .to_string()
+            .contains("another sync is already in progress")
+    );
+    assert_eq!(std::fs::read(&layout.graph_db_path)?, before);
+    assert!(layout.dirty_path.exists());
+    Ok(())
+}
+
+#[tokio::test]
+async fn dirty_open_recovers_committed_rows_before_clearing_sentinel()
+-> std::result::Result<(), Box<dyn std::error::Error>> {
+    let dir = TempDir::new()?;
+    let project_root = dir.path().join("repo");
+    std::fs::create_dir_all(&project_root)?;
+    let open_options = TraceDecayOpenOptions {
+        profile_root: Some(dir.path().join("profile")),
+        global_db_path: Some(dir.path().join("global.db")),
+    };
+
+    let ts = TraceDecay::init_with_options(&project_root, open_options.clone()).await?;
+    let layout = ts.store_layout().clone();
+    ts.db()
+        .conn()
+        .execute_batch("PRAGMA wal_autocheckpoint = 0")
+        .await?;
+    let mut journal_rows = ts.db().conn().query("PRAGMA journal_mode", ()).await?;
+    let journal_mode = journal_rows
+        .next()
+        .await?
+        .expect("journal mode row")
+        .get::<String>(0)?;
+    drop(journal_rows);
+    let node = sample_node("wal-recovery-node", "wal_recovery_node");
+    ts.db().insert_nodes(std::slice::from_ref(&node)).await?;
+    if journal_mode.eq_ignore_ascii_case("wal") {
+        let mut wal_path = layout.graph_db_path.as_os_str().to_os_string();
+        wal_path.push("-wal");
+        assert!(
+            std::fs::metadata(std::path::PathBuf::from(wal_path))?.len() > 0,
+            "disabled autocheckpoint must retain committed WAL frames"
+        );
+    } else {
+        assert!(
+            matches!(
+                journal_mode.to_ascii_lowercase().as_str(),
+                "delete" | "memory"
+            ),
+            "production recovery fixture must use a platform-safe non-WAL journal"
+        );
+    }
+    std::fs::write(&layout.dirty_path, "pid=99999\nversion=test")?;
+
+    let recovered = TraceDecay::open_with_options(&project_root, open_options).await?;
+    assert!(recovered.get_node(&node.id).await?.is_some());
+    assert!(
+        !layout.dirty_path.exists(),
+        "sentinel clears only after WAL-aware quick_check succeeds"
+    );
+    recovered.close();
+    ts.close();
+    Ok(())
+}
+
 #[tokio::test]
 async fn corrupt_db_detected_and_repaired_on_reopen() {
     let dir = TempDir::new().unwrap();