diff --git a/desktop/scripts/check-file-sizes.mjs b/desktop/scripts/check-file-sizes.mjs index d0aeb3d355..fe9856c1b8 100644 --- a/desktop/scripts/check-file-sizes.mjs +++ b/desktop/scripts/check-file-sizes.mjs @@ -75,12 +75,18 @@ const overrides = new Map([ // category as above. Still queued to split. // merge_save_subscription_kinds command + owner_p-kinds TOCTOU fix adds ~30 // lines. Atomic merge to close the concurrent-seed race. Still queued to split. - ["src-tauri/src/archive/mod.rs", 1695], + // IMMEDIATE-tx fix: BEGIN IMMEDIATE replaces DEFERRED unchecked_transaction + // in merge_owner_p_kinds comment block (~5 lines). Still queued to split. + ["src-tauri/src/archive/mod.rs", 1700], // archive/store.rs: merge_owner_p_kinds fn (read+union+upsert under a single // SQLite tx) + 4 unit tests (create-when-none, adds-kind, idempotent, // concurrent-interleave). Load-bearing TOCTOU fix for the owner_p shared row. // Queued to split test module into store_tests.rs in a follow-up. - ["src-tauri/src/archive/store.rs", 1110], + // IMMEDIATE-tx fix: replaces mislabeled sequential test with a real + // two-connection WAL regression test (tempfile + std::thread + Barrier, + // ~85 lines). The new test exercises the actual concurrent write path and + // fails fast if IMMEDIATE guard is removed. Still queued to split. + ["src-tauri/src/archive/store.rs", 1179], ["src-tauri/src/commands/agents.rs", 1437], // #1418 read-path fix: get_thread_replies' blocker fix (shared TIMELINE_KINDS // const + build_thread_replies_filter helper, mirroring the channel sibling so diff --git a/desktop/src-tauri/src/archive/store.rs b/desktop/src-tauri/src/archive/store.rs index 7d3956e926..72704105f2 100644 --- a/desktop/src-tauri/src/archive/store.rs +++ b/desktop/src-tauri/src/archive/store.rs @@ -199,8 +199,13 @@ pub fn has_save_subscription( /// given identity + relay + scope_value. /// /// Reads the current `kinds` array, unions in `new_kind`, and writes back — -/// all inside a single SQLite transaction, so concurrent callers can never -/// clobber each other's kind regardless of await interleaving. +/// all inside a single `BEGIN IMMEDIATE` SQLite transaction. `IMMEDIATE` +/// acquires the write lock at `BEGIN` (before the `SELECT`), so a second +/// concurrent caller blocks on `busy_timeout` (5000 ms, set by +/// `open_archive_db`) and then reads the first caller's committed row. This +/// guarantees the union is complete; a `DEFERRED` transaction would let two +/// concurrent callers both read the empty snapshot and produce a +/// `BUSY_SNAPSHOT` failure on the losing writer. /// /// If no row exists yet it is created with `[new_kind]`. `now` is used as /// `created_at` only on insert (the conflict clause never updates it). @@ -212,51 +217,61 @@ pub fn merge_owner_p_kinds( new_kind: u32, now: i64, ) -> Result<(), String> { - // `unchecked_transaction` matches the rusqlite idiom for a plain - // BEGIN/COMMIT without a savepoint, giving us an exclusive write lock for - // the duration of the read-modify-write. - let tx = conn - .unchecked_transaction() - .map_err(|e| format!("merge_owner_p_kinds begin tx: {e}"))?; + // BEGIN IMMEDIATE takes the write lock before the SELECT so concurrent + // callers serialize here rather than racing to a BUSY_SNAPSHOT error. + conn.execute_batch("BEGIN IMMEDIATE") + .map_err(|e| format!("merge_owner_p_kinds begin immediate: {e}"))?; - // Read the current kinds, if any. - let existing_json: Option = tx - .query_row( - "SELECT kinds FROM save_subscriptions - WHERE identity_pubkey = ?1 - AND relay_url = ?2 - AND scope_type = 'owner_p' - AND scope_value = ?3", - params![identity_pubkey, relay_url, scope_value], - |row| row.get::<_, String>(0), + let result = (|| -> Result<(), String> { + // Read the current kinds, if any. + let existing_json: Option = conn + .query_row( + "SELECT kinds FROM save_subscriptions + WHERE identity_pubkey = ?1 + AND relay_url = ?2 + AND scope_type = 'owner_p' + AND scope_value = ?3", + params![identity_pubkey, relay_url, scope_value], + |row| row.get::<_, String>(0), + ) + .optional() + .map_err(|e| format!("merge_owner_p_kinds read: {e}"))?; + + // Parse, union, re-serialize. + let mut kinds: Vec = match existing_json { + Some(ref json) => serde_json::from_str(json).unwrap_or_default(), + None => vec![], + }; + if !kinds.contains(&new_kind) { + kinds.push(new_kind); + } + let kinds_json = serde_json::to_string(&kinds) + .map_err(|e| format!("merge_owner_p_kinds serialize: {e}"))?; + + // Upsert — INSERT on first call, UPDATE kinds on subsequent. + conn.execute( + "INSERT INTO save_subscriptions + (identity_pubkey, relay_url, scope_type, scope_value, kinds, created_at) + VALUES (?1, ?2, 'owner_p', ?3, ?4, ?5) + ON CONFLICT (identity_pubkey, relay_url, scope_type, scope_value) + DO UPDATE SET kinds = excluded.kinds", + params![identity_pubkey, relay_url, scope_value, kinds_json, now], ) - .optional() - .map_err(|e| format!("merge_owner_p_kinds read: {e}"))?; + .map_err(|e| format!("merge_owner_p_kinds upsert: {e}"))?; - // Parse, union, re-serialize. - let mut kinds: Vec = match existing_json { - Some(ref json) => serde_json::from_str(json).unwrap_or_default(), - None => vec![], - }; - if !kinds.contains(&new_kind) { - kinds.push(new_kind); + Ok(()) + })(); + + if result.is_ok() { + conn.execute_batch("COMMIT") + .map_err(|e| format!("merge_owner_p_kinds commit: {e}"))?; + } else { + // Best-effort rollback; ignore the rollback error to surface the + // original error to the caller. + let _ = conn.execute_batch("ROLLBACK"); } - let kinds_json = - serde_json::to_string(&kinds).map_err(|e| format!("merge_owner_p_kinds serialize: {e}"))?; - // Upsert — INSERT on first call, UPDATE kinds on subsequent. - tx.execute( - "INSERT INTO save_subscriptions - (identity_pubkey, relay_url, scope_type, scope_value, kinds, created_at) - VALUES (?1, ?2, 'owner_p', ?3, ?4, ?5) - ON CONFLICT (identity_pubkey, relay_url, scope_type, scope_value) - DO UPDATE SET kinds = excluded.kinds", - params![identity_pubkey, relay_url, scope_value, kinds_json, now], - ) - .map_err(|e| format!("merge_owner_p_kinds upsert: {e}"))?; - - tx.commit() - .map_err(|e| format!("merge_owner_p_kinds commit: {e}")) + result } /// Return the `kinds` JSON string for a matching save subscription, or `None` @@ -625,37 +640,91 @@ mod tests { assert_eq!(kinds, [44200], "no duplicates after idempotent call"); } - /// Simulates the concurrent-interleave race: both seeds read an empty row - /// "simultaneously" (both see None), then each writes its own kind via the - /// atomic merge fn. The last writer must still produce a row with BOTH - /// kinds because the merge fn reads-inside-the-tx, not reads-before-the-tx. + /// Two-connection WAL regression test for the BEGIN IMMEDIATE fix. /// - /// In a real concurrent scenario the SQLite write-lock serializes the two - /// transactions. Here we simulate "observer wrote first" by calling - /// merge_owner_p_kinds(24200) then merge_owner_p_kinds(44200) — the second - /// call reads the row that the first call created (within its transaction) - /// and merges 44200 in. The result must be [24200, 44200], not [44200]. + /// Opens TWO separate connections to the same WAL file (mirroring the + /// real scenario where the observer-archive seed hook and the + /// metric-archive seed hook each open their own connection via + /// `open_archive_db`). Both threads call `merge_owner_p_kinds` + /// concurrently from an empty row. With `BEGIN IMMEDIATE` the losing + /// thread blocks on `busy_timeout` until the winner commits, then + /// reads the committed row and merges its kind in. Both calls must + /// resolve `Ok` and the final row must contain exactly `[24200, 44200]`. + /// + /// A `DEFERRED` transaction would produce `SQLITE_BUSY_SNAPSHOT` on the + /// loser (not retried by busy_timeout), causing one kind to be silently + /// dropped. This test fails in <10 ms if the IMMEDIATE guard is removed. #[test] - fn test_merge_owner_p_kinds_concurrent_interleave_both_kinds_survive() { - let conn = in_memory(); - // Simulate: both seeds start with an empty row (conn starts clean). - // In the real race both would read [] concurrently; with the atomic tx - // one wins the write lock first. Simulate that ordering here. - merge_owner_p_kinds(&conn, "pk", "wss://r", "mypk", 24200, 1).unwrap(); // observer wins lock first - merge_owner_p_kinds(&conn, "pk", "wss://r", "mypk", 44200, 2).unwrap(); // metric runs second + fn test_merge_owner_p_kinds_two_conn_wal_both_kinds_survive() { + use std::sync::{Arc, Barrier}; + use std::thread; + use tempfile::NamedTempFile; - let subs = list_save_subscriptions(&conn, "pk", "wss://r").unwrap(); + // A real file DB is required for WAL mode (in-memory dbs don't support + // shared-cache WAL across multiple connections in the same process). + let db_file = NamedTempFile::new().unwrap(); + let db_path = db_file.path().to_path_buf(); + + // Initialise schema on the file DB via conn-A so both threads see it. + let init_conn = open_archive_db(&db_path).unwrap(); + drop(init_conn); + + // Barrier ensures both threads are inside `merge_owner_p_kinds` before + // either one issues `BEGIN IMMEDIATE`, maximising the race window. + let barrier = Arc::new(Barrier::new(2)); + + let path_a = db_path.clone(); + let path_b = db_path.clone(); + let bar_a = Arc::clone(&barrier); + let bar_b = Arc::clone(&barrier); + + let handle_observer = thread::spawn(move || { + let conn = open_archive_db(&path_a).unwrap(); + bar_a.wait(); // sync: both threads ready + merge_owner_p_kinds(&conn, "pk", "wss://r", "mypk", 24200, 1) + }); + + let handle_metric = thread::spawn(move || { + let conn = open_archive_db(&path_b).unwrap(); + bar_b.wait(); // sync: both threads ready + merge_owner_p_kinds(&conn, "pk", "wss://r", "mypk", 44200, 2) + }); + + let res_observer = handle_observer.join().expect("observer thread panicked"); + let res_metric = handle_metric.join().expect("metric thread panicked"); + + assert!( + res_observer.is_ok(), + "observer seed must succeed: {:?}", + res_observer + ); + assert!( + res_metric.is_ok(), + "metric seed must succeed: {:?}", + res_metric + ); + + // Verify the final row contains both kinds. + let verify_conn = open_archive_db(&db_path).unwrap(); + let subs = list_save_subscriptions(&verify_conn, "pk", "wss://r").unwrap(); assert_eq!(subs.len(), 1, "exactly one owner_p row"); let kinds: Vec = serde_json::from_str(&subs[0].kinds).unwrap(); assert!( kinds.contains(&24200), - "observer kind 24200 must survive concurrent metric seed" + "observer kind 24200 must survive concurrent metric seed; got {:?}", + kinds ); assert!( kinds.contains(&44200), - "metric kind 44200 must be present after concurrent seed" + "metric kind 44200 must be present after concurrent seed; got {:?}", + kinds + ); + assert_eq!( + kinds.len(), + 2, + "exactly two kinds, no duplicates; got {:?}", + kinds ); - assert_eq!(kinds.len(), 2, "exactly two kinds, no duplicates"); } // ── Archived events ──────────────────────────────────────────────────────