From d95120467ea53e9cb1f7bf33c587cbdb7aee2e97 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Fri, 10 Jul 2026 16:39:11 +0000 Subject: [PATCH] fix(sqlite): disable graph mmap across peer checkpoints --- src/db/connection.rs | 16 ++++-- src/migrate/consolidate/tests.rs | 85 ++++++++++++++++++++++++++++++++ 2 files changed, 97 insertions(+), 4 deletions(-) diff --git a/src/db/connection.rs b/src/db/connection.rs index b9db320e..f038ab29 100644 --- a/src/db/connection.rs +++ b/src/db/connection.rs @@ -45,6 +45,8 @@ pub(crate) fn platform_safe_mmap_size(mmap: u64) -> u64 { if cfg!(windows) { 0 } else { mmap } } +const GRAPH_STORE_MMAP_SIZE: u64 = 0; + /// Env var that, when set to `1`, switches every `TraceDecay` `SQLite` /// connection to `journal_mode=MEMORY` + `synchronous=OFF` on all platforms. /// @@ -344,8 +346,12 @@ impl Database { /// `cache_size` and `mmap_size` are scaled to the on-disk DB size so /// small projects don't pay the 320 MB baseline of a large project. async fn apply_pragmas(conn: &Connection, db_file_size: u64) -> Result<()> { - let (cache_kb, mmap) = adaptive_cache_sizes(db_file_size); - let mmap = platform_safe_mmap_size(mmap); + let (cache_kb, _) = adaptive_cache_sizes(db_file_size); + // Keep graph stores on ordinary file I/O. The daemon intentionally + // holds one connection open while watcher and hook processes open and + // close peers; mmap-backed peers can otherwise retain stale page views + // across WAL checkpoints, especially for legacy 4 KiB databases. + let mmap = GRAPH_STORE_MMAP_SIZE; let journal_mode = platform_safe_journal_mode(); let synchronous = platform_safe_synchronous_mode(); conn.execute_batch(&format!( @@ -367,8 +373,10 @@ impl Database { } async fn apply_read_only_pragmas(conn: &Connection, db_file_size: u64) -> Result<()> { - let (cache_kb, mmap) = adaptive_cache_sizes(db_file_size); - let mmap = platform_safe_mmap_size(mmap); + let (cache_kb, _) = adaptive_cache_sizes(db_file_size); + // Read-only graph handles may overlap a writer/checkpointer too, so + // they must use the same non-mmap access invariant. + let mmap = GRAPH_STORE_MMAP_SIZE; conn.execute_batch(&format!( "PRAGMA mmap_size = {mmap}; PRAGMA foreign_keys = ON; diff --git a/src/migrate/consolidate/tests.rs b/src/migrate/consolidate/tests.rs index 2e4f2b9f..a6e79d4c 100644 --- a/src/migrate/consolidate/tests.rs +++ b/src/migrate/consolidate/tests.rs @@ -419,6 +419,71 @@ async fn interrupted_apply_retries_without_duplicates_and_cuts_over_last() { assert_eq!(sqlite::count_rows(&sessions, "sessions").await.unwrap(), 2); } +#[tokio::test] +async fn mixed_page_destination_survives_overlapping_watcher_opens() { + let fixture = fixture().await; + let source = layout_for_id(&fixture.project, &fixture.profile, &fixture.source_id).unwrap(); + let target = layout_for_id(&fixture.project, &fixture.profile, &fixture.target_id).unwrap(); + rewrite_page_size(&source.graph_db_path, 8192).await; + rewrite_page_size(&target.graph_db_path, 4096).await; + assert_eq!(database_page_size(&source.graph_db_path).await, 8192); + assert_eq!(database_page_size(&target.graph_db_path).await, 4096); + + let options = fixture.options(); + let planned = plan(&options).await.unwrap(); + let applied = apply(&options, &planned.confirmation_token).await.unwrap(); + let destination = applied + .destination_data_root + .join(crate::config::DB_FILENAME); + assert_eq!(database_page_size(&destination).await, 4096); + + let open_options = TraceDecayOpenOptions { + profile_root: Some(fixture.profile.clone()), + global_db_path: Some(fixture.profile.join("global.db")), + }; + let cached = TraceDecay::open_with_options(&fixture.project, open_options.clone()) + .await + .unwrap(); + + for round in 0..2 { + fs::write( + fixture.project.join("lib.rs"), + format!("pub fn fixture() -> usize {{ {round} }}\n"), + ) + .unwrap(); + let watcher = TraceDecay::open_with_options(&fixture.project, open_options.clone()) + .await + .unwrap(); + watcher + .sync_if_stale_silent(&["lib.rs".to_string()]) + .await + .unwrap(); + drop(watcher); + + assert!(storage::has_sqlite_database_header(&destination).unwrap()); + let (verification, _) = Database::open_read_only(&destination).await.unwrap(); + let mut mmap_rows = verification + .conn() + .query("PRAGMA mmap_size", ()) + .await + .unwrap(); + assert_eq!( + mmap_rows + .next() + .await + .unwrap() + .unwrap() + .get::(0) + .unwrap(), + 0 + ); + drop(mmap_rows); + assert!(verification.quick_check().await.unwrap()); + verification.close(); + cached.get_all_files().await.unwrap(); + } +} + #[tokio::test] async fn destination_preparation_restarts_after_every_publish_boundary() { for stop in [ @@ -2370,6 +2435,26 @@ async fn execute_sql(path: &Path, sql: &str) { db.close(); } +async fn rewrite_page_size(path: &Path, page_size: i64) { + let (db, _) = Database::open(path).await.unwrap(); + db.checkpoint().await.unwrap(); + db.conn() + .execute_batch(&format!( + "PRAGMA journal_mode = DELETE; PRAGMA page_size = {page_size}; VACUUM;" + )) + .await + .unwrap(); + db.close(); +} + +async fn database_page_size(path: &Path) -> i64 { + let (db, _) = Database::open_read_only(path).await.unwrap(); + let mut rows = db.conn().query("PRAGMA page_size", ()).await.unwrap(); + let page_size = rows.next().await.unwrap().unwrap().get::(0).unwrap(); + db.close(); + page_size +} + async fn explain_query_plan(conn: &libsql::Connection, sql: &str) -> Vec { let mut rows = conn .query(&format!("EXPLAIN QUERY PLAN {sql}"), ())