From 85438f647b40de8ba5a3bc3e302111d65c172d9a Mon Sep 17 00:00:00 2001 From: Duncan Date: Wed, 12 Aug 2026 23:58:44 -0400 Subject: [PATCH] test(archive): harden M4 shape validation and pin the init barrier MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Resolve the three review findings on the retention Phase-1 branch. M4 now validates the COMPLETE expected shape of its objects inside the BEGIN IMMEDIATE transaction — every column's name, type, nullability and primary-key position via PRAGMA table_info, and the scope-age index's key order via pragma_index_info — instead of counting object names. CREATE ... IF NOT EXISTS preserves a wrong-shaped object that already carries the expected name, so name presence alone could mark a wrong-shaped named table valid and hand Phase 2 an unusable table or missing access path. A wrong-shaped named index is dropped and rebuilt (an index carries no data); an incompatible named table is rejected and the whole transaction rolls back with no marker. The init-barrier and guard-lifetime contracts are now pinned directly against the ArchiveDb OnceCell/RwLock orchestration through a cfg(test) path/hook seam, rather than via raw SQLite contention: production-shaped with_conn callers race a held init and prove none opens a connection until initialization completes and exactly one initialization runs; a separate test proves a write-lock contender cannot enter until a with_conn closure returns and its connection drops. Adds the full policy lifecycle on one evolving state (active -> kind disabled -> subscription deleted -> orphan edited -> orphan deleted) and a deterministic concurrent merge/remove/set interleaving asserting the subscription stays valid, the explicit choice survives, and no policy row is deleted by a kinds mutation. Co-authored-by: Will Pfleger Signed-off-by: Will Pfleger --- desktop/src-tauri/src/archive/archive_db.rs | 97 ++++++- .../src-tauri/src/archive/archive_db_tests.rs | 246 ++++++++++++++++++ desktop/src-tauri/src/archive/retention.rs | 139 +++++++++- .../src-tauri/src/archive/retention_tests.rs | 222 ++++++++++++++++ .../src-tauri/src/archive/store_migrations.rs | 49 ++-- 5 files changed, 720 insertions(+), 33 deletions(-) create mode 100644 desktop/src-tauri/src/archive/archive_db_tests.rs diff --git a/desktop/src-tauri/src/archive/archive_db.rs b/desktop/src-tauri/src/archive/archive_db.rs index 3b121fb47..7c69591b0 100644 --- a/desktop/src-tauri/src/archive/archive_db.rs +++ b/desktop/src-tauri/src/archive/archive_db.rs @@ -37,6 +37,22 @@ use tokio::sync::{OnceCell, RwLock}; use super::store; use crate::managed_agents::nest_dir; +/// A hook run once on the blocking pool at the start of the single init task. +/// Test-only: lets a test count initializations and hold the winner long +/// enough to prove concurrent callers await it. Never set in production. +#[cfg(test)] +type InitHook = std::sync::Arc; + +/// Test-only overrides so the barrier and guard-lifetime contracts can be +/// exercised without a real nest: a fixed DB path in place of `nest_dir()` and +/// an optional init hook. `Default` leaves this `None`, so production always +/// resolves the path from the nest and runs no hook. +#[cfg(test)] +struct TestSeam { + path: PathBuf, + on_init: Option, +} + /// Gated owner of every production archive DB connection. Lives in /// [`crate::app_state::AppState`]; commands call [`ArchiveDb::with_conn`]. #[derive(Default)] @@ -49,26 +65,51 @@ pub struct ArchiveDb { /// Maintenance lock. Ordinary connections hold the read guard for their /// whole lifetime; the Phase-4 conversion holds the write guard. maintenance: RwLock<()>, + /// Test-only path/hook overrides; always `None` in production. + #[cfg(test)] + test_seam: Option, } impl ArchiveDb { - /// Resolve the archive DB path from the nest directory. Errors only when - /// the nest cannot be resolved (fatal for archive access, same as the - /// former `open_db`). - fn db_path() -> Result { + /// Resolve the archive DB path. Production resolves from the nest + /// directory; a test seam (when present) supplies a fixed path so the + /// barrier can be exercised without a real nest. Errors only when the nest + /// cannot be resolved (fatal for archive access, same as the former + /// `open_db`). + fn db_path(&self) -> Result { + #[cfg(test)] + if let Some(seam) = &self.test_seam { + return Ok(seam.path.clone()); + } let nest = nest_dir().ok_or("cannot resolve nest directory for archive")?; Ok(nest.join("archive").join("archive.db")) } + /// The init hook, if a test installed one; always `None` in production. + #[cfg(test)] + fn init_hook(&self) -> Option { + self.test_seam.as_ref().and_then(|s| s.on_init.clone()) + } + /// Complete the one-time init: open the DB once on the blocking pool, /// running `SCHEMA` + all migrations (incl. M4), then drop the connection. /// Concurrent callers await the same single execution. Idempotent and /// cheap after the first success (the cached `()` short-circuits). async fn ensure_initialized(&self) -> Result<(), String> { + let path = self.db_path()?; + #[cfg(test)] + let hook = self.init_hook(); self.init .get_or_try_init(|| async { - tokio::task::spawn_blocking(|| { - let path = Self::db_path()?; + tokio::task::spawn_blocking(move || { + // Test hook runs at the very start of the single init task, + // before the migration opens the DB — this is where a test + // holds the winner past the busy timeout to prove ordinary + // callers await it. No-op in production. + #[cfg(test)] + if let Some(hook) = hook { + hook(); + } // Opening runs every migration; the connection exists only // to complete them behind the barrier, so drop it here. let conn = store::open_archive_db(&path)?; @@ -104,9 +145,9 @@ impl ArchiveDb { F: FnOnce(&Connection) -> Result + Send + 'static, { self.ensure_initialized().await?; + let path = self.db_path()?; let _guard = self.maintenance.read().await; tokio::task::spawn_blocking(move || { - let path = Self::db_path()?; let conn = store::open_archive_db(&path)?; task(&conn) }) @@ -114,3 +155,45 @@ impl ArchiveDb { .map_err(|e| format!("archive db task failed: {e}"))? } } + +#[cfg(test)] +impl ArchiveDb { + /// Build an adapter bound to a fixed DB path (no nest required), so the + /// barrier and guard-lifetime contracts can be exercised in isolation. + fn with_test_path(path: PathBuf) -> Self { + Self { + init: OnceCell::new(), + maintenance: RwLock::new(()), + test_seam: Some(TestSeam { + path, + on_init: None, + }), + } + } + + /// Build an adapter bound to a fixed path whose single initialization runs + /// `hook` first — used to count initializations and to hold the init task + /// open across the concurrent-caller window. + fn with_test_hook(path: PathBuf, hook: InitHook) -> Self { + Self { + init: OnceCell::new(), + maintenance: RwLock::new(()), + test_seam: Some(TestSeam { + path, + on_init: Some(hook), + }), + } + } + + /// Whether the maintenance WRITE guard can be taken right now. A live + /// `with_conn` connection holds the read guard, so this returns `false` + /// while any ordinary connection is open and `true` once all have dropped — + /// exactly the signal the Phase-4 sole-connection VACUUM will gate on. + fn maintenance_write_available(&self) -> bool { + self.maintenance.try_write().is_ok() + } +} + +#[cfg(test)] +#[path = "archive_db_tests.rs"] +mod archive_db_tests; diff --git a/desktop/src-tauri/src/archive/archive_db_tests.rs b/desktop/src-tauri/src/archive/archive_db_tests.rs new file mode 100644 index 000000000..59f619e4d --- /dev/null +++ b/desktop/src-tauri/src/archive/archive_db_tests.rs @@ -0,0 +1,246 @@ +//! Behavior tests for the [`ArchiveDb`] init barrier and maintenance-lock +//! guard lifetime — the two contracts Phase 1 introduced and Thufir's pass-1 +//! review required to be pinned directly (not via raw SQLite contention). +//! +//! These race PRODUCTION-shaped `with_conn` callers through the real +//! `OnceCell`/`RwLock` orchestration, using the `#[cfg(test)]` path/hook seam +//! on `ArchiveDb` to make timing deterministic instead of relying on a large +//! on-disk fixture or wall-clock sleeps to approach the 5s busy timeout. + +use super::*; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::{Arc, Condvar, Mutex}; +use std::time::{Duration, Instant}; +use tempfile::TempDir; + +/// A one-shot latch that blocks a blocking-pool thread until the test releases +/// it. Deterministic stand-in for "the M4 winner holds init open past the busy +/// timeout": the init/closure thread parks here, the test observes the frozen +/// state, then releases. `Mutex` + `Condvar` are `Sync`, so an `Arc` +/// captured by the `Send + Sync` init hook type-checks. +struct Latch { + open: Mutex, + cv: Condvar, +} + +impl Latch { + fn new() -> Arc { + Arc::new(Self { + open: Mutex::new(false), + cv: Condvar::new(), + }) + } + + /// Block until [`release`](Self::release) is called (returns immediately if + /// already released). + fn wait(&self) { + let mut open = self.open.lock().unwrap(); + while !*open { + open = self.cv.wait(open).unwrap(); + } + } + + fn release(&self) { + *self.open.lock().unwrap() = true; + self.cv.notify_all(); + } +} + +/// Poll `cond` on the async runtime until it holds or `timeout` elapses. +/// Panics on timeout so a broken barrier surfaces as a failure, never a hang. +async fn await_until(what: &str, timeout: Duration, cond: impl Fn() -> bool) { + let deadline = Instant::now() + timeout; + while !cond() { + assert!(Instant::now() < deadline, "timed out waiting for {what}"); + tokio::time::sleep(Duration::from_millis(5)).await; + } +} + +/// An archive DB path inside a fresh temp dir. The dir is returned so the +/// caller keeps it alive for the whole test (dropping it deletes the file). +fn temp_db() -> (TempDir, std::path::PathBuf) { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("archive").join("archive.db"); + (dir, path) +} + +/// The barrier serializes first-open: exactly one initialization runs while +/// every concurrent `with_conn` caller awaits it, and no ordinary connection +/// opens until that initialization has completed. +/// +/// The init hook holds the single init task open on a latch. While it is held +/// we prove no `with_conn` caller has opened its connection (`open_count == 0`) +/// — impossible if callers bypassed the `OnceCell` and opened independently. +/// Releasing the latch lets init finish; all callers then complete, exactly +/// one initialization ran, and every open happened after init. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn test_first_open_barrier_serializes_init_and_defers_opens() { + let (_dir, path) = temp_db(); + let init_count = Arc::new(AtomicUsize::new(0)); + let open_count = Arc::new(AtomicUsize::new(0)); + let release = Latch::new(); + + let hook = { + let init_count = Arc::clone(&init_count); + let release = Arc::clone(&release); + Arc::new(move || { + // Runs once, at the head of the single init task. Record the + // initialization, then park so the test can inspect the frozen + // pre-open state. + init_count.fetch_add(1, Ordering::SeqCst); + release.wait(); + }) as InitHook + }; + let db = Arc::new(ArchiveDb::with_test_hook(path, hook)); + + // Trigger the single init and wait until it is provably in flight (hook + // ran) and parked on the latch. + let warm = { + let db = Arc::clone(&db); + tokio::spawn(async move { db.warm_init().await }) + }; + await_until("init to start", Duration::from_secs(10), || { + init_count.load(Ordering::SeqCst) == 1 + }) + .await; + + // Fan out production-shaped callers while init is held. Each records that + // its task was scheduled (`entered`) and that its closure actually opened a + // connection (`open_count`). + let entered = Arc::new(AtomicUsize::new(0)); + let callers: Vec<_> = (0..4) + .map(|_| { + let db = Arc::clone(&db); + let open_count = Arc::clone(&open_count); + let entered = Arc::clone(&entered); + tokio::spawn(async move { + entered.fetch_add(1, Ordering::SeqCst); + db.with_conn(move |conn| { + open_count.fetch_add(1, Ordering::SeqCst); + // Touch the migrated schema to prove a usable connection. + conn.query_row("SELECT COUNT(*) FROM retention_policies", [], |r| { + r.get::<_, i64>(0) + }) + .map_err(|e| e.to_string()) + }) + .await + }) + }) + .collect(); + + // All four caller tasks are scheduled and running before we judge the + // barrier: they have entered `with_conn` and can only be parked on the + // init `OnceCell`. Without the barrier they would instead open independent + // connections here and bump `open_count` while init is still held. + await_until("callers to be scheduled", Duration::from_secs(10), || { + entered.load(Ordering::SeqCst) == 4 + }) + .await; + tokio::time::sleep(Duration::from_millis(100)).await; + assert_eq!( + open_count.load(Ordering::SeqCst), + 0, + "no ordinary connection may open until initialization completes" + ); + + // Let init finish; every caller now completes against the migrated DB. + release.release(); + assert!(warm.await.unwrap().is_ok(), "warm init must succeed"); + for caller in callers { + assert!( + caller.await.unwrap().is_ok(), + "every with_conn must succeed" + ); + } + + assert_eq!( + init_count.load(Ordering::SeqCst), + 1, + "exactly one initialization ran behind the barrier" + ); + assert_eq!( + open_count.load(Ordering::SeqCst), + 4, + "all callers opened, and only after init" + ); +} + +/// The maintenance read guard lives for the FULL lifetime of a `with_conn` +/// connection: a write-lock contender cannot enter until the closure returns +/// and its connection has dropped. This is the invariant the Phase-4 +/// sole-connection VACUUM depends on. +/// +/// A `with_conn` closure parks on a latch while holding its connection; a +/// separate task contends for the maintenance write guard. While the closure +/// is parked the writer must be blocked. Releasing the closure — which returns +/// and drops the connection — lets the writer finally acquire the guard. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn test_with_conn_read_guard_blocks_writer_until_connection_drops() { + let (_dir, path) = temp_db(); + let db = Arc::new(ArchiveDb::with_test_path(path)); + db.warm_init().await.expect("init must succeed"); + + let in_closure = Arc::new(AtomicUsize::new(0)); + let hold = Latch::new(); + + // A live connection: the closure parks holding it (and thus the read + // guard) until released. + let work = { + let db = Arc::clone(&db); + let in_closure = Arc::clone(&in_closure); + let hold = Arc::clone(&hold); + tokio::spawn(async move { + db.with_conn(move |_conn| { + in_closure.fetch_add(1, Ordering::SeqCst); + hold.wait(); + Ok(()) + }) + .await + }) + }; + await_until( + "closure to hold the connection", + Duration::from_secs(10), + || in_closure.load(Ordering::SeqCst) == 1, + ) + .await; + + // A genuine write-lock contender. + let write_entered = Arc::new(AtomicUsize::new(0)); + let writer = { + let db = Arc::clone(&db); + let write_entered = Arc::clone(&write_entered); + tokio::spawn(async move { + let _w = db.maintenance.write().await; + write_entered.fetch_add(1, Ordering::SeqCst); + }) + }; + + // While the connection is held, the writer must not have entered, and the + // write guard must be unavailable. + tokio::time::sleep(Duration::from_millis(200)).await; + assert_eq!( + write_entered.load(Ordering::SeqCst), + 0, + "writer must block while a with_conn connection is live" + ); + assert!( + !db.maintenance_write_available(), + "write guard is unavailable while the read guard is held" + ); + + // Release the closure: it returns and its connection drops, releasing the + // read guard so the writer can proceed. + hold.release(); + assert!(work.await.unwrap().is_ok(), "held with_conn must succeed"); + writer.await.unwrap(); + assert_eq!( + write_entered.load(Ordering::SeqCst), + 1, + "writer enters once the connection has dropped" + ); + assert!( + db.maintenance_write_available(), + "write guard is free again after the connection drops" + ); +} diff --git a/desktop/src-tauri/src/archive/retention.rs b/desktop/src-tauri/src/archive/retention.rs index dfb688499..618bfbbb6 100644 --- a/desktop/src-tauri/src/archive/retention.rs +++ b/desktop/src-tauri/src/archive/retention.rs @@ -12,7 +12,7 @@ //! the existing `metric_store.rs` / `pipeline.rs` / `store_migrations.rs` //! precedent. -use rusqlite::{params, Connection}; +use rusqlite::{params, Connection, OptionalExtension}; // ── Constants ──────────────────────────────────────────────────────────────── @@ -63,6 +63,143 @@ CREATE INDEX IF NOT EXISTS idx_archived_event_scopes_age (identity_pubkey, relay_url, scope_type, scope_value, archived_at, id); "; +/// Name of the scope-age index, shared by the DDL, the shape check, and the +/// M4 drop-and-rebuild repair path. +pub(super) const SCOPE_AGE_INDEX_NAME: &str = "idx_archived_event_scopes_age"; + +// ── Expected schema shape (the single source of truth M4 validates against) ──── + +/// One column's expected shape: `(name, declared_type, not_null, pk_position)`. +/// `pk_position` is 0 when the column is not part of the primary key, else its +/// 1-based position in the PK — this is exactly what `PRAGMA table_info` reports +/// in its `pk` field, so PK column ORDER is validated, not just membership. +type ColumnShape = (&'static str, &'static str, bool, i64); + +/// Expected `retention_policies` columns in `cid` order. Mirrors +/// [`RETENTION_SCHEMA`]; a drift here or there is caught by the M4 shape check. +const RETENTION_POLICIES_SHAPE: &[ColumnShape] = &[ + ("identity_pubkey", "TEXT", true, 1), + ("relay_url", "TEXT", true, 2), + ("scope_type", "TEXT", true, 3), + ("scope_value", "TEXT", true, 4), + ("kind", "INTEGER", true, 5), + ("days", "INTEGER", false, 0), + ("updated_at", "INTEGER", true, 0), +]; + +/// Expected `archive_meta` columns in `cid` order. A `TEXT PRIMARY KEY` is +/// nullable in SQLite (only `INTEGER PRIMARY KEY` implies NOT NULL), so `key` +/// carries `not_null = false` with `pk_position = 1`. +const ARCHIVE_META_SHAPE: &[ColumnShape] = &[("key", "TEXT", false, 1), ("value", "TEXT", true, 0)]; + +/// Expected key columns of [`SCOPE_AGE_INDEX_NAME`] in seqno order — the +/// age-range access path Phase 2 depends on. Order is load-bearing: a covering +/// seek needs the scope keys before `archived_at`. +const SCOPE_AGE_INDEX_SHAPE: &[&str] = &[ + "identity_pubkey", + "relay_url", + "scope_type", + "scope_value", + "archived_at", + "id", +]; + +// ── Shape validation (used by migration M4) ──────────────────────────────────── + +/// Whether `table`'s live columns exactly match `expected` (name, declared +/// type, nullability, and PK position, all in `cid` order). `table` is always a +/// compile-time constant from this module, never user input, so interpolating +/// it into the table-valued `pragma_table_info` call carries no injection risk. +fn table_shape_matches( + conn: &Connection, + table: &str, + expected: &[ColumnShape], +) -> Result { + let mut stmt = conn + .prepare(&format!( + "SELECT name, type, \"notnull\", pk FROM pragma_table_info('{table}') ORDER BY cid" + )) + .map_err(|e| format!("shape check: prepare table_info({table}): {e}"))?; + let actual: Vec<(String, String, i64, i64)> = stmt + .query_map([], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?))) + .map_err(|e| format!("shape check: query table_info({table}): {e}"))? + .collect::, _>>() + .map_err(|e| format!("shape check: read table_info({table}): {e}"))?; + + Ok(actual.len() == expected.len() + && actual.iter().zip(expected).all( + |((name, ty, not_null, pk), (exp_name, exp_ty, exp_not_null, exp_pk))| { + name == exp_name + && ty.eq_ignore_ascii_case(exp_ty) + && (*not_null != 0) == *exp_not_null + && pk == exp_pk + }, + )) +} + +/// Whether the scope-age index exists on `archived_event_scopes` with exactly +/// the expected key columns in order. False when the index is absent, sits on +/// the wrong table, or indexes the wrong columns — all of which M4 repairs by +/// dropping and recreating it (an index carries no data, so a rebuild is safe). +pub(super) fn scope_age_index_is_correct(conn: &Connection) -> Result { + let table: Option = conn + .query_row( + "SELECT tbl_name FROM sqlite_master WHERE type = 'index' AND name = ?1", + params![SCOPE_AGE_INDEX_NAME], + |r| r.get(0), + ) + .optional() + .map_err(|e| format!("shape check: scope-age index table: {e}"))?; + if table.as_deref() != Some("archived_event_scopes") { + return Ok(false); + } + + let mut stmt = conn + .prepare(&format!( + "SELECT name FROM pragma_index_info('{SCOPE_AGE_INDEX_NAME}') ORDER BY seqno" + )) + .map_err(|e| format!("shape check: prepare index_info: {e}"))?; + let columns: Vec = stmt + .query_map([], |r| r.get::<_, String>(0)) + .map_err(|e| format!("shape check: query index_info: {e}"))? + .collect::, _>>() + .map_err(|e| format!("shape check: read index_info: {e}"))?; + + Ok(columns.len() == SCOPE_AGE_INDEX_SHAPE.len() + && columns + .iter() + .zip(SCOPE_AGE_INDEX_SHAPE) + .all(|(actual, expected)| actual == expected)) +} + +/// Validate the COMPLETE expected shape of the M4 objects: both tables' columns +/// (names, types, nullability, PK positions) and the scope-age index's key +/// order. `CREATE ... IF NOT EXISTS` silently preserves a wrong-shaped object +/// that already carries the expected name, so name presence alone cannot +/// certify the schema — a wrong-shaped named table would otherwise let M4 mark +/// itself applied and hand Phase 2 an unusable table or missing access path. +/// Called inside M4's `BEGIN IMMEDIATE`; an `Err` rolls the transaction back +/// with no marker written. +pub(super) fn validate_retention_schema_shape(conn: &Connection) -> Result<(), String> { + if !table_shape_matches(conn, "retention_policies", RETENTION_POLICIES_SHAPE)? { + return Err( + "migration M4: retention_policies has an unexpected column or primary-key shape" + .to_string(), + ); + } + if !table_shape_matches(conn, "archive_meta", ARCHIVE_META_SHAPE)? { + return Err( + "migration M4: archive_meta has an unexpected column or primary-key shape".to_string(), + ); + } + if !scope_age_index_is_correct(conn)? { + return Err(format!( + "migration M4: {SCOPE_AGE_INDEX_NAME} has an unexpected shape after rebuild" + )); + } + Ok(()) +} + // ── Types ───────────────────────────────────────────────────────────────────── /// A retention policy row with its live-subscription status, returned by diff --git a/desktop/src-tauri/src/archive/retention_tests.rs b/desktop/src-tauri/src/archive/retention_tests.rs index d05b2d727..723ee9ae2 100644 --- a/desktop/src-tauri/src/archive/retention_tests.rs +++ b/desktop/src-tauri/src/archive/retention_tests.rs @@ -397,6 +397,75 @@ fn test_m4_partial_schema_marker_absent_recovers_on_next_open() { assert_eq!(objects, 3, "partial schema repaired to the full shape"); } +#[test] +fn test_m4_wrong_shaped_named_table_rejected_no_marker() { + let db = NamedTempFile::new().unwrap(); + build_pre_m4_db(db.path()); + // Precreate an `archive_meta` table carrying the expected NAME but the + // wrong shape (its required `value` column is missing). `CREATE ... IF NOT + // EXISTS` preserves it, so only the explicit shape check can catch it. + { + let conn = Connection::open(db.path()).unwrap(); + conn.execute_batch("CREATE TABLE archive_meta (key TEXT PRIMARY KEY);") + .unwrap(); + } + // M4 must reject the incompatible named table and roll back with no marker, + // rather than certify a table Phase 2 could not use. + assert!( + store::open_archive_db(db.path()).is_err(), + "M4 must reject a wrong-shaped named archive_meta" + ); + let verify = Connection::open(db.path()).unwrap(); + assert_eq!( + m4_marker_count(&verify), + 0, + "no marker may certify the incompatible table" + ); + // The rollback left the schema untouched: the bad table is still one-column + // and retention_policies was never committed. + assert!( + !scope_age_index_is_correct(&verify).unwrap(), + "the index build rolled back with the rest of the transaction" + ); + let value_cols: i64 = verify + .query_row( + "SELECT COUNT(*) FROM pragma_table_info('archive_meta') WHERE name = 'value'", + [], + |r| r.get(0), + ) + .unwrap(); + assert_eq!( + value_cols, 0, + "the wrong-shaped table was not silently altered" + ); +} + +#[test] +fn test_m4_wrong_index_order_dropped_and_rebuilt() { + let db = NamedTempFile::new().unwrap(); + build_pre_m4_db(db.path()); + // Precreate the scope-age index with the expected NAME on the right table + // but the WRONG key order (`archived_at` first). A covering age-range seek + // needs the scope keys before `archived_at`, so this order is unusable. + { + let conn = Connection::open(db.path()).unwrap(); + conn.execute_batch(&format!( + "CREATE INDEX {SCOPE_AGE_INDEX_NAME} + ON archived_event_scopes + (archived_at, identity_pubkey, relay_url, scope_type, scope_value, id);" + )) + .unwrap(); + } + // M4 rebuilds a wrong-shaped index (safe — an index carries no data) rather + // than rejecting, so the open succeeds and the marker lands. + let conn = fresh(&db); + assert_eq!(m4_marker_count(&conn), 1, "M4 completes after the rebuild"); + assert!( + scope_age_index_is_correct(&conn).unwrap(), + "the index was rebuilt into the correct key order" + ); +} + // ── Concurrency ─────────────────────────────────────────────────────────────── #[test] @@ -481,3 +550,156 @@ fn test_concurrent_merge_seeds_both_policies() { assert_eq!(policy_days(&verify, OBSERVER_FRAME_KIND), Some(30)); assert_eq!(policy_days(&verify, 44200), None); } + +/// Deterministic interleaving of merge / remove / set on ONE `(scope, kind)` +/// across two connections to the same WAL file. Each mutator takes `BEGIN +/// IMMEDIATE`, so a barrier held before both begin forces them to serialize on +/// `busy_timeout` rather than race to a `BUSY_SNAPSHOT`. Whichever order the OS +/// picks, the invariants must hold: the subscription stays valid, the explicit +/// policy choice set by one thread is never silently reset by the other's +/// merge-seed (which is `ON CONFLICT DO NOTHING`), and no policy row is deleted +/// by a `kinds` mutation. Removing the `BEGIN IMMEDIATE` guard from the +/// mutators makes one contender fail with `BUSY_SNAPSHOT`, tripping this test. +#[test] +fn test_concurrent_merge_remove_set_keeps_state_valid_and_choice() { + use std::thread; + let db = NamedTempFile::new().unwrap(); + let path = db.path().to_path_buf(); + // Start from a subscription that already lists 24200 with an EXPLICIT + // 7-day policy — the choice both racing threads must not clobber. + { + let conn = fresh(&db); + store::merge_owner_p_kinds(&conn, ID, RELAY, ID, 24200, 100).unwrap(); + set_policy(&conn, ID, RELAY, OWNER, ID, 24200, Some(7), 100).unwrap(); + } + + let barrier = Arc::new(Barrier::new(3)); + // T1: enable a second kind (seeds its default policy). + // T2: disable 24200 (removes it from `kinds`, must NOT touch its policy). + // T3: re-set 24200's explicit policy to 5 days. + let t1 = { + let (p, b) = (path.clone(), Arc::clone(&barrier)); + thread::spawn(move || { + let conn = store::open_archive_db(&p).unwrap(); + b.wait(); + store::merge_owner_p_kinds(&conn, ID, RELAY, ID, 44200, 200) + }) + }; + let t2 = { + let (p, b) = (path.clone(), Arc::clone(&barrier)); + thread::spawn(move || { + let conn = store::open_archive_db(&p).unwrap(); + b.wait(); + store::remove_owner_p_kind(&conn, ID, RELAY, ID, 24200) + }) + }; + let t3 = { + let (p, b) = (path.clone(), Arc::clone(&barrier)); + thread::spawn(move || { + let conn = store::open_archive_db(&p).unwrap(); + b.wait(); + set_policy(&conn, ID, RELAY, OWNER, ID, 24200, Some(5), 300) + }) + }; + assert!(t1.join().unwrap().is_ok(), "concurrent merge must succeed"); + assert!(t2.join().unwrap().is_ok(), "concurrent remove must succeed"); + assert!(t3.join().unwrap().is_ok(), "concurrent set must succeed"); + + let verify = store::open_archive_db(&path).unwrap(); + // Subscription is valid JSON and now lists 44200 (added) but not 24200 + // (removed) — the two `kinds` mutations composed cleanly. + let subs = store::list_save_subscriptions(&verify, ID, RELAY).unwrap(); + assert_eq!(subs.len(), 1, "one owner_p row survives the interleaving"); + let sub_kinds: Vec = serde_json::from_str(&subs[0].kinds).unwrap(); + assert!(sub_kinds.contains(&44200), "the merged kind is present"); + assert!(!sub_kinds.contains(&24200), "the removed kind is gone"); + // Both policy rows still exist — no `kinds` mutation deleted one. + let policies = list_policies(&verify, ID, RELAY).unwrap(); + assert_eq!(policies.len(), 2, "both policy rows survive; none deleted"); + // T3's explicit 5-day choice for 24200 is the final value: T1's default + // seed for 44200 never touches 24200, and T2's remove leaves policies + // alone, so the observer policy reflects the explicit set, not a reset. + assert_eq!( + policy_days(&verify, 24200), + Some(5), + "explicit choice survives the concurrent merge/remove" + ); + assert_eq!( + policy_days(&verify, 44200), + None, + "44200 kept its Forever default" + ); +} + +/// One evolving state walked through the full policy lifecycle in a single +/// test: an active observer policy → the kind is disabled (policy orphaned but +/// preserved) → the whole subscription is deleted (still orphaned) → the orphan +/// is edited → the orphan is explicitly deleted. This proves the transitions +/// compose on the SAME rows, which the separate per-transition tests cannot. +#[test] +fn test_policy_lifecycle_active_orphaned_edited_deleted_on_one_state() { + let db = NamedTempFile::new().unwrap(); + let conn = fresh(&db); + + // 1. Active: a subscription lists 24200, its policy is the seeded default. + create_subscription_with_policies(&conn, ID, RELAY, OWNER, ID, &[24200], "[24200]", 100) + .unwrap(); + let active = list_policies(&conn, ID, RELAY).unwrap(); + assert_eq!(active.len(), 1); + assert!( + active[0].active, + "policy is active while the kind is listed" + ); + assert_eq!(active[0].days, Some(30), "seeded observer default"); + + // 2. Kind disabled: 24200 leaves `kinds`; the policy stays but goes orphaned. + store::remove_owner_p_kind(&conn, ID, RELAY, ID, 24200).unwrap(); + // Removing the last kind deleted the subscription row entirely. + assert!( + store::list_save_subscriptions(&conn, ID, RELAY) + .unwrap() + .is_empty(), + "the last kind off removes the subscription row" + ); + let disabled = list_policies(&conn, ID, RELAY).unwrap(); + assert_eq!(disabled.len(), 1, "the policy is preserved, not deleted"); + assert!( + !disabled[0].active, + "policy is orphaned once the kind is gone" + ); + assert_eq!( + disabled[0].days, + Some(30), + "it keeps expiring historical data" + ); + + // 3. Subscription deleted: an explicit delete on an already-absent row is a + // no-op; the orphaned policy is unaffected. + assert!( + !store::delete_save_subscription(&conn, ID, RELAY, OWNER, ID).unwrap(), + "no subscription row remains to delete" + ); + let still_orphaned = list_policies(&conn, ID, RELAY).unwrap(); + assert_eq!(still_orphaned.len(), 1); + assert!(!still_orphaned[0].active, "policy remains orphaned"); + + // 4. Edit the orphan: a user can still change retention for historical data. + set_policy(&conn, ID, RELAY, OWNER, ID, 24200, Some(90), 400).unwrap(); + let edited = list_policies(&conn, ID, RELAY).unwrap(); + assert_eq!(edited.len(), 1); + assert_eq!(edited[0].days, Some(90), "orphaned policy is editable"); + assert!( + !edited[0].active, + "editing does not resurrect the subscription" + ); + + // 5. Delete the orphan: the only path that removes a policy row. + assert!( + delete_policy(&conn, ID, RELAY, OWNER, ID, 24200).unwrap(), + "the orphaned policy is deleted" + ); + assert!( + list_policies(&conn, ID, RELAY).unwrap().is_empty(), + "no policy rows remain after the lifecycle completes" + ); +} diff --git a/desktop/src-tauri/src/archive/store_migrations.rs b/desktop/src-tauri/src/archive/store_migrations.rs index 3a4cedf36..e9b9dad15 100644 --- a/desktop/src-tauri/src/archive/store_migrations.rs +++ b/desktop/src-tauri/src/archive/store_migrations.rs @@ -379,22 +379,35 @@ fn migrate_add_retention_policies_locked(conn: &Connection) -> Result<(), String return Ok(()); } - // Idempotent DDL — repairs a partial state left by any interrupted run. + // Idempotent DDL — creates any missing objects, repairing a partial state + // left by an interrupted earlier run. `IF NOT EXISTS` preserves an object + // that already carries the expected name, so it cannot fix a wrong SHAPE — + // that is what the explicit validation below is for. conn.execute_batch(super::retention::RETENTION_SCHEMA) .map_err(|e| format!("migration M4: create retention schema: {e}"))?; - conn.execute_batch(super::retention::SCOPE_AGE_INDEX_DDL) - .map_err(|e| format!("migration M4: create scope-age index: {e}"))?; - // Schema-shape recheck: confirm all three objects (two tables + one index) - // actually materialized before seeding or writing the marker, so a marker - // never certifies a half-built schema. - if retention_schema_object_count(conn)? != 3 { - return Err( - "migration M4: retention schema incomplete after DDL (expected 2 tables + 1 index)" - .to_string(), - ); + // Repair a missing or wrong-shaped scope-age index by dropping and + // recreating it. An index carries no data, so a rebuild is always safe — + // unlike a table, whose wrong shape we must reject rather than drop. + if !super::retention::scope_age_index_is_correct(conn)? { + conn.execute_batch(&format!( + "DROP INDEX IF EXISTS {}", + super::retention::SCOPE_AGE_INDEX_NAME + )) + .map_err(|e| format!("migration M4: drop wrong-shaped scope-age index: {e}"))?; + conn.execute_batch(super::retention::SCOPE_AGE_INDEX_DDL) + .map_err(|e| format!("migration M4: create scope-age index: {e}"))?; } + // Validate the COMPLETE expected shape (both tables' columns, types, + // nullability, and PK positions, plus the index key order) before seeding + // or writing the marker. A wrong-shaped named TABLE cannot be auto-repaired + // without risking archived data, so it is rejected here and the whole + // transaction rolls back with no marker — the next open re-runs M4 once the + // schema is corrected. This guarantees the marker never certifies a + // half-built or mis-shaped schema that Phase 2 would inherit. + super::retention::validate_retention_schema_shape(conn)?; + let now = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .unwrap_or_default() @@ -425,20 +438,6 @@ fn retention_migration_applied(conn: &Connection) -> Result { Ok(count > 0) } -/// Count the M4 schema objects present: the `retention_policies` and -/// `archive_meta` tables plus the `idx_archived_event_scopes_age` index. A -/// fully-created schema returns 3. -fn retention_schema_object_count(conn: &Connection) -> Result { - conn.query_row( - "SELECT COUNT(*) FROM sqlite_master - WHERE (type = 'table' AND name IN ('retention_policies', 'archive_meta')) - OR (type = 'index' AND name = 'idx_archived_event_scopes_age')", - [], - |r| r.get(0), - ) - .map_err(|e| format!("migration M4: schema-shape check: {e}")) -} - /// Seed a default retention policy for every `(subscription, kind)` pair in /// `save_subscriptions`: kind 24200 → 30 days, every other kind → an explicit /// `NULL` (Forever) row. Runs on the caller's transaction so seeding is atomic