From 8755788b5bd648c4a4ee7a42e171ff7d698328da Mon Sep 17 00:00:00 2001 From: franzer Date: Wed, 1 Jul 2026 12:40:39 -0600 Subject: [PATCH 1/2] feat: propagate child-record deletions via tombstones --- src/cli/commands/sync.rs | 476 +++++++++++++++++++++++++++- src/storage/db.rs | 661 +++++++++++++++++++++++++++++++++++++-- src/storage/models.rs | 29 ++ src/sync/store.rs | 71 ++++- 4 files changed, 1213 insertions(+), 24 deletions(-) diff --git a/src/cli/commands/sync.rs b/src/cli/commands/sync.rs index 5d02b14..c2382d4 100644 --- a/src/cli/commands/sync.rs +++ b/src/cli/commands/sync.rs @@ -25,18 +25,21 @@ use std::path::{Path, PathBuf}; use std::process::Command; use anyhow::{anyhow, bail, Context, Result}; -use chrono::{DateTime, Utc}; +use chrono::{DateTime, Duration, Utc}; use colored::Colorize; use serde::{Deserialize, Serialize}; use uuid::Uuid; use crate::cli::OutputFormat; use crate::config::Config; -use crate::storage::models::{Machine, Session}; +use crate::storage::models::{Machine, Session, Tombstone}; use crate::storage::Database; use crate::sync::gitref::{self, TreeEntry}; use crate::sync::keystore::{derive_store_key, generate_store_salt, store_id_from_salt, KeyStore}; -use crate::sync::store::{decrypt_session_record, encrypt_session_record, SessionRecord}; +use crate::sync::store::{ + decrypt_session_record, decrypt_tombstones, encrypt_session_record, encrypt_tombstones, + SessionRecord, +}; use crate::sync::SyncError; /// Full ref name for a repository's per-repo lore store. @@ -54,6 +57,19 @@ const GLOBAL_REMOTE: &str = "origin"; /// Minimum passphrase length for a newly created store. const MIN_PASSPHRASE_LEN: usize = 8; +/// Tree path of the encrypted tombstone set inside the store. +/// +/// Holds the JSON array of child-record deletions ({child_id, kind, session_id, +/// deleted_at}) so a deletion on one machine suppresses that record everywhere. +const TOMBSTONES_PATH: &str = "meta/tombstones"; + +/// Age past which a tombstone is garbage-collected during sync. +/// +/// A deletion older than this is assumed to have reached every machine, so +/// dropping its tombstone can no longer resurrect the deleted child. Bounds the +/// tombstone set so it does not grow without limit. +const TOMBSTONE_GC_DAYS: i64 = 90; + /// Maximum number of fetch/merge/build/push attempts before giving up. /// /// A concurrent local sync (compare-and-swap mismatch) or a remote that moved @@ -561,12 +577,27 @@ fn perform_sync_in_store( Vec::new() }; + // TOMBSTONES: union the remote deletion set into the local table BEFORE + // the merge so the merge suppresses re-adding any child that was deleted + // on another machine, then GC entries older than the retention window. + // A wrong key makes the remote tombstones undecryptable; that is treated + // as an empty set here (the merge below surfaces the wrong-key error). + let remote_tombstones = read_remote_tombstones(repo, &tracking_entries, key)?; + db.add_tombstones(&remote_tombstones)?; + db.prune_tombstones(Utc::now() - Duration::days(TOMBSTONE_GC_DAYS))?; + // MERGE remote -> local database (full records, newer-wins). A wrong key // surfaces here and aborts before anything is built or pushed. The store // selects which sync-tracking column an imported session is marked on. pulled_total += merge_remote_in_store(store, db, repo, &tracking_entries, key)?; merge_machines(db, repo, &tracking_entries)?; + // APPLY tombstones: remove any child that is present locally but has been + // deleted (on this or another machine). Suppression above stops a stale + // remote blob from re-adding it; this removes one already stored here. + let tombstones = db.list_tombstones()?; + db.apply_tombstones(&tombstones)?; + // BUILD the outgoing tree. Separate the TREE BASE (which entries the new // tree inherits) from the COMMIT PARENT (which commit it descends from): // @@ -629,6 +660,12 @@ fn perform_sync_in_store( add_meta_changes(db, repo, tree_base.as_deref(), salt, machine, &mut changes)?; + // Write the unioned, GC'd tombstone set back to the store, but only when + // it differs from what the base tree already holds so an unchanged set + // keeps its existing content-addressed blob (no churn from re-encrypting + // with a fresh nonce every sync). + add_tombstone_changes(db, repo, &remote_tombstones, key, &mut changes)?; + let tree = gitref::build_tree(repo, tree_base.as_deref(), &changes)?; let message = format!("lore: sync {} session(s)", sessions.len()); let commit = gitref::commit_tree(repo, &tree, commit_parent.as_deref(), &message)?; @@ -847,6 +884,69 @@ fn add_meta_changes( Ok(()) } +/// Reads and decrypts the remote store's tombstone set. +/// +/// Returns an empty set when the store has no `meta/tombstones` blob or the blob +/// cannot be decrypted (a wrong key, which the session merge reports separately; +/// treating it as empty here avoids a confusing early error). +fn read_remote_tombstones( + repo: &Path, + entries: &[TreeEntry], + key: &[u8], +) -> Result> { + match blob_at_path(repo, entries, TOMBSTONES_PATH)? { + Some(bytes) => match decrypt_tombstones(&bytes, key) { + Ok(tombstones) => Ok(tombstones), + Err(e) => { + tracing::debug!("Could not decrypt remote tombstones: {e}"); + Ok(Vec::new()) + } + }, + None => Ok(Vec::new()), + } +} + +/// Adds the encrypted tombstone blob to the outgoing tree when it changed. +/// +/// Compares the local tombstone set with what the base tree already holds +/// (`remote_tombstones`) by their `(child_id, kind)` keys. When the keys match, +/// nothing is written so the base's existing blob is preserved verbatim +/// (content-addressed dedup). Otherwise the full local set is re-encrypted and +/// written, which also shrinks the stored set after garbage collection. +fn add_tombstone_changes( + db: &Database, + repo: &Path, + remote_tombstones: &[Tombstone], + key: &[u8], + changes: &mut BTreeMap, +) -> Result<()> { + let local = db.list_tombstones()?; + if tombstone_keys_equal(&local, remote_tombstones) { + return Ok(()); + } + let blob = encrypt_tombstones(&local, key)?; + let sha = gitref::write_blob(repo, &blob)?; + changes.insert(TOMBSTONES_PATH.to_string(), sha); + Ok(()) +} + +/// Returns whether two tombstone sets hold the same `(child_id, kind)` keys. +/// +/// Only the identifying keys are compared (not `deleted_at` or `session_id`) so +/// two machines recording the same deletion at different times do not trigger a +/// perpetual rewrite of the store blob. +fn tombstone_keys_equal(a: &[Tombstone], b: &[Tombstone]) -> bool { + let keys_a: HashSet<(&str, &str)> = a + .iter() + .map(|t| (t.child_id.as_str(), t.kind.as_str())) + .collect(); + let keys_b: HashSet<(&str, &str)> = b + .iter() + .map(|t| (t.child_id.as_str(), t.kind.as_str())) + .collect(); + keys_a == keys_b +} + /// Assembles the complete reasoning record for a session from the database. fn assemble_record(db: &Database, session: &Session) -> Result { Ok(SessionRecord { @@ -2787,4 +2887,374 @@ mod tests { "global sync must NOT mark the per-repo track" ); } + + // ==================== tombstones ==================== + + /// Sets up a per-repo store on a fresh remote and returns (remote, url). + /// + /// Kept alive by the caller so the bare remote directory is not dropped. + fn setup_repo_with_store( + remote_url: &str, + passphrase: &str, + ) -> (TempDir, PathBuf, KeyStore, TempDir, MachineIdentity) { + let dir = tempfile::tempdir().unwrap(); + let repo = dir.path().to_path_buf(); + init_repo(&repo); + git(&repo, &["remote", "add", "origin", remote_url]); + let (keystore, kd) = test_keystore(); + let m = machine("machine-a", "Machine A"); + create_store(&repo, "origin", &keystore, &m, passphrase).unwrap(); + (dir, repo, keystore, kd, m) + } + + #[test] + fn test_link_deletion_propagates_via_tombstone() { + // Machine A deletes a link and syncs. Machine B (which has that link) + // syncs and the link is removed, and a further sync does not resurrect it. + let (_remote_dir, remote_url) = init_bare_remote(); + let passphrase = "shared team passphrase"; + + // Machine A: set up, seed a session with a link, sync. + let (_da, repo_a, keystore_a, _ka, ma) = setup_repo_with_store(&remote_url, passphrase); + let (mut db_a, _dba) = open_db(); + let session_id = seed_full_session(&mut db_a, "machine-a", &repo_dir(&repo_a)); + let (key_a, salt_a) = load_store_credentials(&repo_a, "origin", &keystore_a).unwrap(); + let sessions_a = scoped_unsynced(&db_a, &repo_a); + perform_sync( + &mut db_a, &repo_a, "origin", &key_a, &salt_a, &ma, sessions_a, + ) + .unwrap(); + + // Machine B: join, sync (pull the session and its link). + let dir_b = tempfile::tempdir().unwrap(); + let repo_b = dir_b.path(); + init_repo(repo_b); + git(repo_b, &["remote", "add", "origin", &remote_url]); + let (keystore_b, _kb) = test_keystore(); + let mb = machine("machine-b", "Machine B"); + gitref::fetch(repo_b, "origin", SESSIONS_REF).unwrap(); + let salt_b = read_store_salt(repo_b, "origin").unwrap().unwrap(); + join_store(repo_b, "origin", &keystore_b, &mb, &salt_b, passphrase).unwrap(); + let (mut db_b, _dbb) = open_db(); + let (key_b, salt_b2) = load_store_credentials(repo_b, "origin", &keystore_b).unwrap(); + let sessions_b = scoped_unsynced(&db_b, repo_b); + perform_sync( + &mut db_b, repo_b, "origin", &key_b, &salt_b2, &mb, sessions_b, + ) + .unwrap(); + assert_eq!( + db_b.get_links_by_session(&session_id).unwrap().len(), + 1, + "machine B must have pulled the link" + ); + + // Machine A: delete the link (records a tombstone) and sync. + assert!(db_a + .delete_link_by_session_and_commit(&session_id, "deadbeef") + .unwrap()); + let sessions_a = scoped_unsynced(&db_a, &repo_a); + perform_sync( + &mut db_a, &repo_a, "origin", &key_a, &salt_a, &ma, sessions_a, + ) + .unwrap(); + + // Machine B: sync. The tombstone removes the link locally. + let sessions_b = scoped_unsynced(&db_b, repo_b); + perform_sync( + &mut db_b, repo_b, "origin", &key_b, &salt_b2, &mb, sessions_b, + ) + .unwrap(); + assert!( + db_b.get_links_by_session(&session_id).unwrap().is_empty(), + "the deleted link must be removed on machine B" + ); + + // A subsequent sync must not resurrect it. + let sessions_b = scoped_unsynced(&db_b, repo_b); + perform_sync( + &mut db_b, repo_b, "origin", &key_b, &salt_b2, &mb, sessions_b, + ) + .unwrap(); + assert!( + db_b.get_links_by_session(&session_id).unwrap().is_empty(), + "the deleted link must stay removed after another sync" + ); + } + + #[test] + fn test_concurrent_add_survives_remote_deletion() { + // A deletes link X while B adds a different link Y to the same session. + // After both sync, X is gone everywhere but Y survives everywhere. Uses + // the global store so both machines can push the shared session (the + // per-repo store scopes pushes by working directory, which differs per + // temp repo, so it cannot model a two-way concurrent edit in a test). + let (_remote_dir, remote_url) = init_bare_remote(); + let passphrase = "shared global passphrase"; + + // A: set up the global store, seed a session with link X, sync. + let (_store_a, store_a) = init_global_store(&remote_url); + let (keystore_a, _ka) = test_keystore(); + let ma = machine("machine-a", "Machine A"); + create_store(&store_a, GLOBAL_REMOTE, &keystore_a, &ma, passphrase).unwrap(); + let (mut db_a, _dba) = open_db(); + let session_id = seed_full_session(&mut db_a, "machine-a", "/projects/repo-one"); + let (key_a, salt_a) = load_store_credentials(&store_a, GLOBAL_REMOTE, &keystore_a).unwrap(); + let run_a = |db: &mut Database| { + let sessions = db.get_unsynced_global_sessions().unwrap(); + perform_sync_in_store( + SyncStore::Global, + db, + &store_a, + GLOBAL_REMOTE, + &key_a, + &salt_a, + &ma, + sessions, + ) + .unwrap(); + }; + run_a(&mut db_a); + + // B: join and sync (has X). + let (_store_b, store_b) = init_global_store(&remote_url); + let (keystore_b, _kb) = test_keystore(); + let mb = machine("machine-b", "Machine B"); + gitref::fetch(&store_b, GLOBAL_REMOTE, SESSIONS_REF).unwrap(); + let salt_b = read_store_salt(&store_b, GLOBAL_REMOTE).unwrap().unwrap(); + join_store( + &store_b, + GLOBAL_REMOTE, + &keystore_b, + &mb, + &salt_b, + passphrase, + ) + .unwrap(); + let (mut db_b, _dbb) = open_db(); + let (key_b, salt_b2) = + load_store_credentials(&store_b, GLOBAL_REMOTE, &keystore_b).unwrap(); + let run_b = |db: &mut Database| { + let sessions = db.get_unsynced_global_sessions().unwrap(); + perform_sync_in_store( + SyncStore::Global, + db, + &store_b, + GLOBAL_REMOTE, + &key_b, + &salt_b2, + &mb, + sessions, + ) + .unwrap(); + }; + run_b(&mut db_b); + + // A: delete X, sync. + assert!(db_a + .delete_link_by_session_and_commit(&session_id, "deadbeef") + .unwrap()); + run_a(&mut db_a); + + // B: add a DIFFERENT link Y, then sync. + let y_id = Uuid::new_v4(); + db_b.insert_link(&SessionLink { + id: y_id, + session_id, + link_type: LinkType::Commit, + commit_sha: Some("feedface".to_string()), + branch: Some("main".to_string()), + remote: Some("origin".to_string()), + created_at: Utc::now(), + created_by: LinkCreator::User, + confidence: Some(0.9), + }) + .unwrap(); + run_b(&mut db_b); + + // B has only Y now (X removed by the tombstone). + let b_links = db_b.get_links_by_session(&session_id).unwrap(); + assert_eq!(b_links.len(), 1, "B keeps only its own new link"); + assert_eq!(b_links[0].id, y_id); + + // A: sync again to pull Y. + run_a(&mut db_a); + let a_links = db_a.get_links_by_session(&session_id).unwrap(); + assert_eq!(a_links.len(), 1, "A gets the concurrently added link"); + assert_eq!(a_links[0].id, y_id, "X stays deleted, Y propagates to A"); + } + + #[test] + fn test_tombstone_suppresses_stale_remote_blob() { + // Directly importing an older remote blob that still contains a + // tombstoned child must not re-add it. This is the resurrection path a + // tombstone must close even when the parent session blob is stale. + let (_remote_dir, remote_url) = init_bare_remote(); + let (_da, repo, keystore, _kd, _m) = + setup_repo_with_store(&remote_url, "passphrase abcdefgh"); + let (key, _salt) = load_store_credentials(&repo, "origin", &keystore).unwrap(); + + let (mut db, _dbd) = open_db(); + // A session with a link, then delete the link (records the tombstone). + let session_id = seed_full_session(&mut db, "machine-a", &repo_dir(&repo)); + let link = db.get_links_by_session(&session_id).unwrap()[0].clone(); + assert!(db + .delete_link_by_session_and_commit(&session_id, "deadbeef") + .unwrap()); + + // Build a stale remote blob that still holds the deleted link. + let session = db.get_session(&session_id).unwrap().unwrap(); + let record = SessionRecord { + session, + messages: vec![], + links: vec![link], + tags: vec![], + annotations: vec![], + summary: None, + }; + let blob = encrypt_session_record(&record, &key).unwrap(); + let sha = gitref::write_blob(&repo, &blob).unwrap(); + let entries = vec![TreeEntry { + mode: "100644".to_string(), + sha, + path: format!("sessions/{session_id}.enc"), + }]; + + // Merging the stale blob must not resurrect the tombstoned link. + merge_remote(&mut db, &repo, &entries, &key).unwrap(); + assert!( + db.get_links_by_session(&session_id).unwrap().is_empty(), + "a stale remote blob must not resurrect a tombstoned link" + ); + } + + #[test] + fn test_annotation_deletion_propagates_via_tombstone_global() { + // The same deletion-propagation guarantee holds for the global store and + // for annotations, exercising a second store path and child kind. + let (_remote_dir, remote_url) = init_bare_remote(); + let passphrase = "shared global passphrase"; + + // Machine A: set up the global store, seed a session, sync. + let (_store_a, store_a) = init_global_store(&remote_url); + let (keystore_a, _ka) = test_keystore(); + let ma = machine("machine-a", "Machine A"); + create_store(&store_a, GLOBAL_REMOTE, &keystore_a, &ma, passphrase).unwrap(); + let (mut db_a, _dba) = open_db(); + let session_id = seed_full_session(&mut db_a, "machine-a", "/projects/repo-one"); + let annotation_id = db_a.get_annotations(&session_id).unwrap()[0].id; + let (key_a, salt_a) = load_store_credentials(&store_a, GLOBAL_REMOTE, &keystore_a).unwrap(); + let run_a = |db: &mut Database| { + let sessions = db.get_unsynced_global_sessions().unwrap(); + perform_sync_in_store( + SyncStore::Global, + db, + &store_a, + GLOBAL_REMOTE, + &key_a, + &salt_a, + &ma, + sessions, + ) + .unwrap(); + }; + run_a(&mut db_a); + + // Machine B: join and sync (pull the annotation). + let (_store_b, store_b) = init_global_store(&remote_url); + let (keystore_b, _kb) = test_keystore(); + let mb = machine("machine-b", "Machine B"); + gitref::fetch(&store_b, GLOBAL_REMOTE, SESSIONS_REF).unwrap(); + let salt_b = read_store_salt(&store_b, GLOBAL_REMOTE).unwrap().unwrap(); + join_store( + &store_b, + GLOBAL_REMOTE, + &keystore_b, + &mb, + &salt_b, + passphrase, + ) + .unwrap(); + let (mut db_b, _dbb) = open_db(); + let (key_b, salt_b2) = + load_store_credentials(&store_b, GLOBAL_REMOTE, &keystore_b).unwrap(); + let run_b = |db: &mut Database| { + let sessions = db.get_unsynced_global_sessions().unwrap(); + perform_sync_in_store( + SyncStore::Global, + db, + &store_b, + GLOBAL_REMOTE, + &key_b, + &salt_b2, + &mb, + sessions, + ) + .unwrap(); + }; + run_b(&mut db_b); + assert_eq!( + db_b.get_annotations(&session_id).unwrap().len(), + 1, + "machine B must have pulled the annotation" + ); + + // Machine A: delete the annotation and sync. + assert!(db_a.delete_annotation(&annotation_id).unwrap()); + run_a(&mut db_a); + + // Machine B: sync. The tombstone removes the annotation. + run_b(&mut db_b); + assert!( + db_b.get_annotations(&session_id).unwrap().is_empty(), + "the deleted annotation must be removed on machine B via the global store" + ); + + // Not resurrected on a subsequent sync. + run_b(&mut db_b); + assert!( + db_b.get_annotations(&session_id).unwrap().is_empty(), + "the deleted annotation must stay removed" + ); + } + + #[test] + fn test_tombstone_blob_is_stable_across_unchanged_syncs() { + // Once the tombstone set is written, a sync that does not change it must + // reuse the existing content-addressed blob (no churn from re-encrypting + // with a fresh nonce every time). + let (_remote_dir, remote_url) = init_bare_remote(); + let (_da, repo, keystore, _kd, m) = + setup_repo_with_store(&remote_url, "passphrase abcdefgh"); + let (key, salt) = load_store_credentials(&repo, "origin", &keystore).unwrap(); + + let (mut db, _dbd) = open_db(); + let session_id = seed_full_session(&mut db, "machine-a", &repo_dir(&repo)); + let sessions = scoped_unsynced(&db, &repo); + perform_sync(&mut db, &repo, "origin", &key, &salt, &m, sessions).unwrap(); + + // Delete the link and sync so a tombstone blob is written. + assert!(db + .delete_link_by_session_and_commit(&session_id, "deadbeef") + .unwrap()); + let sessions = scoped_unsynced(&db, &repo); + perform_sync(&mut db, &repo, "origin", &key, &salt, &m, sessions).unwrap(); + let first = tombstone_blob_sha(&repo); + + // A no-op sync must not rewrite the tombstone blob. + let sessions = scoped_unsynced(&db, &repo); + perform_sync(&mut db, &repo, "origin", &key, &salt, &m, sessions).unwrap(); + let second = tombstone_blob_sha(&repo); + assert_eq!(first, second, "unchanged tombstone blob must be reused"); + } + + /// Returns the blob SHA of `meta/tombstones` in the local ref. + fn tombstone_blob_sha(repo: &Path) -> String { + let entries = gitref::read_tree(repo, SESSIONS_REF).unwrap(); + entries + .iter() + .find(|e| e.path == TOMBSTONES_PATH) + .expect("a tombstone blob should exist") + .sha + .clone() + } } diff --git a/src/storage/db.rs b/src/storage/db.rs index a6bb979..e398e51 100644 --- a/src/storage/db.rs +++ b/src/storage/db.rs @@ -13,9 +13,18 @@ use uuid::Uuid; use super::models::{ Annotation, Machine, Message, MessageContent, MessageRole, SearchResult, Session, SessionLink, - Summary, Tag, + Summary, Tag, Tombstone, }; +/// Tombstone kind for a deleted session-to-commit link. +const TOMBSTONE_KIND_LINK: &str = "link"; +/// Tombstone kind for a deleted tag. +const TOMBSTONE_KIND_TAG: &str = "tag"; +/// Tombstone kind for a deleted annotation. +const TOMBSTONE_KIND_ANNOTATION: &str = "annotation"; +/// Tombstone kind for a deleted summary. +const TOMBSTONE_KIND_SUMMARY: &str = "summary"; + /// Which sync-tracking column a merge or import marks on write. /// /// A session carries two independent sync tracks: the per-repo store @@ -316,6 +325,18 @@ impl Database { created_at TEXT NOT NULL ); + -- Tombstones record locally deleted child records so a deletion on + -- one machine propagates through the sync store instead of being + -- resurrected by the additive child merge. Keyed by (child_id, kind) + -- so a link and a tag can never collide on the same UUID. + CREATE TABLE IF NOT EXISTS tombstones ( + child_id TEXT NOT NULL, + kind TEXT NOT NULL CHECK(kind IN ('link','tag','annotation','summary')), + session_id TEXT, + deleted_at TEXT NOT NULL, + PRIMARY KEY (child_id, kind) + ); + -- Indexes for common queries CREATE INDEX IF NOT EXISTS idx_sessions_started_at ON sessions(started_at); CREATE INDEX IF NOT EXISTS idx_sessions_working_directory ON sessions(working_directory); @@ -325,6 +346,7 @@ impl Database { CREATE INDEX IF NOT EXISTS idx_annotations_session_id ON annotations(session_id); CREATE INDEX IF NOT EXISTS idx_tags_session_id ON tags(session_id); CREATE INDEX IF NOT EXISTS idx_tags_label ON tags(label); + CREATE INDEX IF NOT EXISTS idx_tombstones_deleted_at ON tombstones(deleted_at); "#, )?; @@ -923,6 +945,166 @@ impl Database { Ok(()) } + /// Records a tombstone for a locally deleted child record. + /// + /// Called only from the user-facing child delete methods (unlink, tag + /// remove, annotation delete, summary delete). The session-cascade delete + /// (`delete_session`, `delete_sessions_older_than`) deletes its children with + /// inline SQL and deliberately does NOT call this, so removing a whole + /// session never strips that session's children from the shared store. On a + /// re-delete the existing row's timestamp is refreshed so garbage collection + /// keys off the most recent deletion. + fn record_tombstone( + conn: &Connection, + child_id: &str, + kind: &str, + session_id: &Uuid, + deleted_at: DateTime, + ) -> Result<()> { + conn.execute( + r#" + INSERT INTO tombstones (child_id, kind, session_id, deleted_at) + VALUES (?1, ?2, ?3, ?4) + ON CONFLICT(child_id, kind) DO UPDATE SET + session_id = excluded.session_id, + deleted_at = excluded.deleted_at + "#, + params![ + child_id, + kind, + session_id.to_string(), + deleted_at.to_rfc3339() + ], + )?; + Ok(()) + } + + /// Returns the `id` values of a child table's rows for a session. + /// + /// Used by the bulk child delete methods to capture ids before deletion so + /// each removed row can be tombstoned. `table` is a fixed internal literal + /// (never user input), so interpolating it into the query is safe. + fn child_ids(&self, table: &str, session_id: &Uuid) -> Result> { + let sql = format!("SELECT id FROM {table} WHERE session_id = ?1"); + let mut stmt = self.conn.prepare(&sql)?; + let rows = stmt.query_map(params![session_id.to_string()], |row| row.get(0))?; + rows.collect::, _>>() + .context("Failed to read child ids") + } + + /// Returns whether a child record of the given kind is tombstoned. + /// + /// Used by the merge path to suppress re-adding a record that was deleted on + /// another machine. + fn is_tombstoned(conn: &Connection, child_id: &Uuid, kind: &str) -> Result { + let count: i64 = conn.query_row( + "SELECT COUNT(*) FROM tombstones WHERE child_id = ?1 AND kind = ?2", + params![child_id.to_string(), kind], + |row| row.get(0), + )?; + Ok(count > 0) + } + + /// Returns every locally recorded tombstone. + /// + /// Ordered by `(kind, child_id)` so the serialized store blob is stable + /// across repeated syncs on the same machine (content-addressed dedup). + pub fn list_tombstones(&self) -> Result> { + let mut stmt = self.conn.prepare( + "SELECT child_id, kind, session_id, deleted_at + FROM tombstones + ORDER BY kind ASC, child_id ASC", + )?; + let rows = stmt.query_map([], |row| { + let deleted_at: String = row.get(3)?; + Ok(Tombstone { + child_id: row.get(0)?, + kind: row.get(1)?, + session_id: row.get(2)?, + deleted_at: parse_datetime(&deleted_at)?, + }) + })?; + rows.collect::, _>>() + .context("Failed to list tombstones") + } + + /// Unions a set of remote tombstones into the local table. + /// + /// First-wins on `(child_id, kind)`: an existing local tombstone is left + /// untouched, so a machine's own deletion timestamp is preserved. Used by + /// the sync merge before importing remote records so the merge can suppress + /// any child that was deleted elsewhere. + pub fn add_tombstones(&self, tombstones: &[Tombstone]) -> Result<()> { + for t in tombstones { + self.conn.execute( + r#" + INSERT INTO tombstones (child_id, kind, session_id, deleted_at) + VALUES (?1, ?2, ?3, ?4) + ON CONFLICT(child_id, kind) DO NOTHING + "#, + params![t.child_id, t.kind, t.session_id, t.deleted_at.to_rfc3339()], + )?; + } + Ok(()) + } + + /// Deletes any locally-present child records that are tombstoned. + /// + /// Enforces, on this machine, a deletion that happened on another machine. + /// Deletes with plain SQL rather than the user-facing delete methods so it + /// does NOT record a fresh tombstone (the child is already tombstoned) and + /// cannot recurse. Each session whose child was actually removed is marked + /// unsynced so the cleaned record is re-exported on the next sync, replacing + /// the stale blob in the store. Suppression during merge already prevents + /// resurrection, so this re-export is a cleanup rather than a correctness + /// requirement. + pub fn apply_tombstones(&self, tombstones: &[Tombstone]) -> Result<()> { + let mut affected: HashSet = HashSet::new(); + for t in tombstones { + let deleted = match t.kind.as_str() { + TOMBSTONE_KIND_LINK => self.conn.execute( + "DELETE FROM session_links WHERE id = ?1", + params![t.child_id], + )?, + TOMBSTONE_KIND_TAG => self + .conn + .execute("DELETE FROM tags WHERE id = ?1", params![t.child_id])?, + TOMBSTONE_KIND_ANNOTATION => self + .conn + .execute("DELETE FROM annotations WHERE id = ?1", params![t.child_id])?, + TOMBSTONE_KIND_SUMMARY => self + .conn + .execute("DELETE FROM summaries WHERE id = ?1", params![t.child_id])?, + _ => 0, + }; + if deleted > 0 { + if let Some(sid) = &t.session_id { + if let Ok(uuid) = parse_uuid(sid) { + affected.insert(uuid); + } + } + } + } + for session_id in affected { + self.mark_session_unsynced(&session_id)?; + } + Ok(()) + } + + /// Prunes tombstones deleted before the given cutoff. + /// + /// Bounds the tombstone set so it does not grow without limit. A tombstone + /// older than the cutoff is assumed to have propagated to every machine, so + /// dropping it can no longer resurrect the deleted child. Returns the number + /// of tombstones removed. + pub fn prune_tombstones(&self, before: DateTime) -> Result { + let rows = self.conn.execute( + "DELETE FROM tombstones WHERE deleted_at < ?1", + params![before.to_rfc3339()], + )?; + Ok(rows) + } + /// Merges a full remote reasoning record into the database atomically. /// /// Used by git-ref sync when pulling a remote store. The session row, @@ -935,17 +1117,17 @@ impl Database { /// messages, or an equal message count with a later `ended_at`). When /// written, the session is marked synced with `synced_at`. /// - Links, tags, and annotations are additive and idempotent by id, so they - /// are always merged regardless of which session row is newer. This lets a - /// remote that added a child to an already-synced (equal or older) session - /// still deliver that child locally. - /// - The summary is applied through the newer-wins writer, so an older remote - /// summary never clobbers a newer local one. - /// - /// Known limitation: child merges are additive, so a link/tag/annotation - /// DELETED on another machine is not removed here (and can be resurrected on a - /// later export). Propagating deletions needs tombstones; tracked as a Phase - /// 29.8 follow-up. This is intentional for the first sync release: additive - /// merge never loses or corrupts data. + /// are merged regardless of which session row is newer, EXCEPT that a child + /// whose (child_id, kind) is tombstoned is suppressed. This lets a remote + /// that added a child to an already-synced session still deliver it, while + /// a child deleted on another machine is not resurrected. + /// - The summary is applied through the newer-wins writer (unless + /// tombstoned), so an older remote summary never clobbers a newer local one. + /// + /// Deletion propagation: the caller unions remote tombstones into the local + /// table before merging (so suppression here sees them) and applies them + /// afterward via [`Self::apply_tombstones`] to remove any child that was + /// already present locally. See `cli/commands/sync.rs`. /// /// Returns `true` when the session row and messages were written (the /// newer-wins branch ran), which callers use for the pulled count. @@ -1044,19 +1226,31 @@ impl Database { Self::write_session_with_messages(&tx, session, messages, Some(synced_at), track)?; } - // Child records are additive and idempotent by id: always merge them so a - // remote addition to an equal-or-older session is not lost. + // Child records are additive and idempotent by id: merge them so a + // remote addition to an equal-or-older session is not lost. A child that + // is tombstoned (deleted on this or another machine) is suppressed so a + // stale remote blob cannot resurrect it. Concurrent additions of other + // records are unaffected because suppression matches only the exact + // (child_id, kind) of a deleted record. for link in links { - Self::write_link(&tx, link, true)?; + if !Self::is_tombstoned(&tx, &link.id, TOMBSTONE_KIND_LINK)? { + Self::write_link(&tx, link, true)?; + } } for tag in tags { - Self::write_tag(&tx, tag, true)?; + if !Self::is_tombstoned(&tx, &tag.id, TOMBSTONE_KIND_TAG)? { + Self::write_tag(&tx, tag, true)?; + } } for annotation in annotations { - Self::write_annotation(&tx, annotation, true)?; + if !Self::is_tombstoned(&tx, &annotation.id, TOMBSTONE_KIND_ANNOTATION)? { + Self::write_annotation(&tx, annotation, true)?; + } } if let Some(summary) = summary { - Self::write_summary_newer(&tx, summary)?; + if !Self::is_tombstoned(&tx, &summary.id, TOMBSTONE_KIND_SUMMARY)? { + Self::write_summary_newer(&tx, summary)?; + } } tx.commit()?; @@ -1300,7 +1494,16 @@ impl Database { )?; if rows_affected > 0 { if let Some(sid) = session_id { - self.mark_session_unsynced(&parse_uuid(&sid)?)?; + let sid = parse_uuid(&sid)?; + // User-facing deletion: tombstone so the removal propagates. + Self::record_tombstone( + &self.conn, + &link_id.to_string(), + TOMBSTONE_KIND_LINK, + &sid, + Utc::now(), + )?; + self.mark_session_unsynced(&sid)?; } } Ok(rows_affected > 0) @@ -1310,11 +1513,18 @@ impl Database { /// /// Returns the number of links deleted. pub fn delete_links_by_session(&self, session_id: &Uuid) -> Result { + // Capture the link ids before deleting so each removal can be + // tombstoned and propagate to other machines. + let ids = self.child_ids("session_links", session_id)?; let rows_affected = self.conn.execute( "DELETE FROM session_links WHERE session_id = ?1", params![session_id.to_string()], )?; if rows_affected > 0 { + let now = Utc::now(); + for id in &ids { + Self::record_tombstone(&self.conn, id, TOMBSTONE_KIND_LINK, session_id, now)?; + } // Local edit: re-open the parent session for the next sync. self.mark_session_unsynced(session_id)?; } @@ -1331,11 +1541,26 @@ impl Database { commit_sha: &str, ) -> Result { let pattern = format!("{commit_sha}%"); + // Capture the matching link ids before deleting so each removal can be + // tombstoned and propagate to other machines. + let ids: Vec = { + let mut stmt = self.conn.prepare( + "SELECT id FROM session_links WHERE session_id = ?1 AND commit_sha LIKE ?2", + )?; + let rows = + stmt.query_map(params![session_id.to_string(), pattern], |row| row.get(0))?; + rows.collect::, _>>() + .context("Failed to read link ids")? + }; let rows_affected = self.conn.execute( "DELETE FROM session_links WHERE session_id = ?1 AND commit_sha LIKE ?2", params![session_id.to_string(), pattern], )?; if rows_affected > 0 { + let now = Utc::now(); + for id in &ids { + Self::record_tombstone(&self.conn, id, TOMBSTONE_KIND_LINK, session_id, now)?; + } // Local edit: re-open the parent session for the next sync. self.mark_session_unsynced(session_id)?; } @@ -2349,7 +2574,16 @@ impl Database { )?; if rows_affected > 0 { if let Some(sid) = session_id { - self.mark_session_unsynced(&parse_uuid(&sid)?)?; + let sid = parse_uuid(&sid)?; + // User-facing deletion: tombstone so the removal propagates. + Self::record_tombstone( + &self.conn, + &annotation_id.to_string(), + TOMBSTONE_KIND_ANNOTATION, + &sid, + Utc::now(), + )?; + self.mark_session_unsynced(&sid)?; } } Ok(rows_affected > 0) @@ -2360,11 +2594,18 @@ impl Database { /// Returns the number of annotations deleted. #[allow(dead_code)] pub fn delete_annotations_by_session(&self, session_id: &Uuid) -> Result { + // Capture the annotation ids before deleting so each removal can be + // tombstoned and propagate to other machines. + let ids = self.child_ids("annotations", session_id)?; let rows_affected = self.conn.execute( "DELETE FROM annotations WHERE session_id = ?1", params![session_id.to_string()], )?; if rows_affected > 0 { + let now = Utc::now(); + for id in &ids { + Self::record_tombstone(&self.conn, id, TOMBSTONE_KIND_ANNOTATION, session_id, now)?; + } // Local edit: re-open the parent session for the next sync. self.mark_session_unsynced(session_id)?; } @@ -2453,11 +2694,30 @@ impl Database { /// /// Returns `true` if a tag was deleted, `false` if not found. pub fn delete_tag(&self, session_id: &Uuid, label: &str) -> Result { + // Capture the tag id before deleting so the removal can be tombstoned + // and propagate to other machines. + let id: Option = self + .conn + .query_row( + "SELECT id FROM tags WHERE session_id = ?1 AND label = ?2", + params![session_id.to_string(), label], + |row| row.get(0), + ) + .optional()?; let rows_affected = self.conn.execute( "DELETE FROM tags WHERE session_id = ?1 AND label = ?2", params![session_id.to_string(), label], )?; if rows_affected > 0 { + if let Some(id) = id { + Self::record_tombstone( + &self.conn, + &id, + TOMBSTONE_KIND_TAG, + session_id, + Utc::now(), + )?; + } // Local edit: re-open the parent session for the next sync. self.mark_session_unsynced(session_id)?; } @@ -2469,11 +2729,18 @@ impl Database { /// Returns the number of tags deleted. #[allow(dead_code)] pub fn delete_tags_by_session(&self, session_id: &Uuid) -> Result { + // Capture the tag ids before deleting so each removal can be tombstoned + // and propagate to other machines. + let ids = self.child_ids("tags", session_id)?; let rows_affected = self.conn.execute( "DELETE FROM tags WHERE session_id = ?1", params![session_id.to_string()], )?; if rows_affected > 0 { + let now = Utc::now(); + for id in &ids { + Self::record_tombstone(&self.conn, id, TOMBSTONE_KIND_TAG, session_id, now)?; + } // Local edit: re-open the parent session for the next sync. self.mark_session_unsynced(session_id)?; } @@ -2631,11 +2898,30 @@ impl Database { /// Returns `true` if a summary was deleted, `false` if no summary existed. #[allow(dead_code)] pub fn delete_summary(&self, session_id: &Uuid) -> Result { + // Capture the summary id before deleting so the removal can be + // tombstoned and propagate to other machines. + let id: Option = self + .conn + .query_row( + "SELECT id FROM summaries WHERE session_id = ?1", + params![session_id.to_string()], + |row| row.get(0), + ) + .optional()?; let rows_affected = self.conn.execute( "DELETE FROM summaries WHERE session_id = ?1", params![session_id.to_string()], )?; if rows_affected > 0 { + if let Some(id) = id { + Self::record_tombstone( + &self.conn, + &id, + TOMBSTONE_KIND_SUMMARY, + session_id, + Utc::now(), + )?; + } // Local edit: re-open the parent session for the next sync. self.mark_session_unsynced(session_id)?; } @@ -7394,4 +7680,339 @@ mod tests { // Compare at second granularity to avoid RFC3339 sub-second rounding. assert_eq!(last.timestamp(), when.timestamp()); } + + // ==================== Tombstone Tests ==================== + + /// Seeds a session with one link, tag, annotation, and summary. + /// + /// Returns the ids so a test can delete a specific child and assert on the + /// resulting tombstone. + fn seed_session_with_children(db: &Database) -> (Uuid, Uuid, Uuid, Uuid, Uuid) { + let session = create_test_session("claude-code", "/project", Utc::now(), None); + db.insert_session(&session).expect("insert session"); + + let link = create_test_link(session.id, Some("deadbeef"), LinkType::Commit); + db.insert_link(&link).expect("insert link"); + + let tag = Tag { + id: Uuid::new_v4(), + session_id: session.id, + label: "feature".to_string(), + created_at: Utc::now(), + }; + db.insert_tag(&tag).expect("insert tag"); + + let annotation = Annotation { + id: Uuid::new_v4(), + session_id: session.id, + content: "note".to_string(), + created_at: Utc::now(), + }; + db.insert_annotation(&annotation) + .expect("insert annotation"); + + let summary = Summary { + id: Uuid::new_v4(), + session_id: session.id, + content: "summary".to_string(), + generated_at: Utc::now(), + }; + db.insert_summary(&summary).expect("insert summary"); + + (session.id, link.id, tag.id, annotation.id, summary.id) + } + + #[test] + fn test_delete_link_records_tombstone() { + let (db, _dir) = create_test_db(); + let (_sid, link_id, _tag, _ann, _sum) = seed_session_with_children(&db); + + assert!(db.delete_link(&link_id).unwrap()); + + let tombstones = db.list_tombstones().unwrap(); + assert_eq!(tombstones.len(), 1, "one tombstone recorded"); + assert_eq!(tombstones[0].child_id, link_id.to_string()); + assert_eq!(tombstones[0].kind, TOMBSTONE_KIND_LINK); + } + + #[test] + fn test_delete_tag_records_tombstone() { + let (db, _dir) = create_test_db(); + let (sid, _link, tag_id, _ann, _sum) = seed_session_with_children(&db); + + assert!(db.delete_tag(&sid, "feature").unwrap()); + + let tombstones = db.list_tombstones().unwrap(); + assert_eq!(tombstones.len(), 1); + assert_eq!(tombstones[0].child_id, tag_id.to_string()); + assert_eq!(tombstones[0].kind, TOMBSTONE_KIND_TAG); + } + + #[test] + fn test_delete_annotation_records_tombstone() { + let (db, _dir) = create_test_db(); + let (_sid, _link, _tag, ann_id, _sum) = seed_session_with_children(&db); + + assert!(db.delete_annotation(&ann_id).unwrap()); + + let tombstones = db.list_tombstones().unwrap(); + assert_eq!(tombstones.len(), 1); + assert_eq!(tombstones[0].child_id, ann_id.to_string()); + assert_eq!(tombstones[0].kind, TOMBSTONE_KIND_ANNOTATION); + } + + #[test] + fn test_delete_summary_records_tombstone() { + let (db, _dir) = create_test_db(); + let (sid, _link, _tag, _ann, sum_id) = seed_session_with_children(&db); + + assert!(db.delete_summary(&sid).unwrap()); + + let tombstones = db.list_tombstones().unwrap(); + assert_eq!(tombstones.len(), 1); + assert_eq!(tombstones[0].child_id, sum_id.to_string()); + assert_eq!(tombstones[0].kind, TOMBSTONE_KIND_SUMMARY); + } + + #[test] + fn test_bulk_delete_links_records_tombstone_per_link() { + let (db, _dir) = create_test_db(); + let session = create_test_session("claude-code", "/project", Utc::now(), None); + db.insert_session(&session).expect("insert session"); + let link_a = create_test_link(session.id, Some("aaaa1111"), LinkType::Commit); + let link_b = create_test_link(session.id, Some("bbbb2222"), LinkType::Commit); + db.insert_link(&link_a).unwrap(); + db.insert_link(&link_b).unwrap(); + + assert_eq!(db.delete_links_by_session(&session.id).unwrap(), 2); + + let tombstones = db.list_tombstones().unwrap(); + assert_eq!(tombstones.len(), 2, "one tombstone per removed link"); + let ids: HashSet = tombstones.iter().map(|t| t.child_id.clone()).collect(); + assert!(ids.contains(&link_a.id.to_string())); + assert!(ids.contains(&link_b.id.to_string())); + assert!(tombstones.iter().all(|t| t.kind == TOMBSTONE_KIND_LINK)); + } + + #[test] + fn test_session_cascade_delete_records_no_tombstones() { + // Deleting a whole session is a purely local operation: it must not + // tombstone the session's children, or a teammate sharing the store + // would lose that session's reasoning. + let (db, _dir) = create_test_db(); + let (sid, _link, _tag, _ann, _sum) = seed_session_with_children(&db); + + db.delete_session(&sid).expect("delete session"); + + assert!( + db.list_tombstones().unwrap().is_empty(), + "session-cascade delete must not create tombstones" + ); + } + + #[test] + fn test_delete_sessions_older_than_records_no_tombstones() { + // The bulk age-based session purge is also a cascade delete and must not + // tombstone children. + let (db, _dir) = create_test_db(); + let old = create_test_session( + "claude-code", + "/project", + Utc::now() - Duration::days(400), + None, + ); + db.insert_session(&old).expect("insert session"); + let link = create_test_link(old.id, Some("deadbeef"), LinkType::Commit); + db.insert_link(&link).unwrap(); + + let cutoff = Utc::now() - Duration::days(30); + assert_eq!(db.delete_sessions_older_than(cutoff).unwrap(), 1); + + assert!( + db.list_tombstones().unwrap().is_empty(), + "age-based session purge must not create tombstones" + ); + } + + #[test] + fn test_add_tombstones_unions_first_wins() { + let (db, _dir) = create_test_db(); + let child = Uuid::new_v4().to_string(); + let session = Uuid::new_v4().to_string(); + + let earlier = Utc::now() - Duration::hours(2); + let later = Utc::now(); + db.add_tombstones(&[Tombstone { + child_id: child.clone(), + kind: TOMBSTONE_KIND_LINK.to_string(), + session_id: Some(session.clone()), + deleted_at: earlier, + }]) + .unwrap(); + // A second union of the same (child_id, kind) must not overwrite. + db.add_tombstones(&[Tombstone { + child_id: child.clone(), + kind: TOMBSTONE_KIND_LINK.to_string(), + session_id: Some(session), + deleted_at: later, + }]) + .unwrap(); + + let tombstones = db.list_tombstones().unwrap(); + assert_eq!(tombstones.len(), 1, "union is keyed by (child_id, kind)"); + assert_eq!( + tombstones[0].deleted_at.timestamp(), + earlier.timestamp(), + "first-wins keeps the earliest recorded deletion" + ); + } + + #[test] + fn test_apply_tombstones_removes_local_child() { + let (db, _dir) = create_test_db(); + let (sid, link_id, _tag, _ann, _sum) = seed_session_with_children(&db); + + // A tombstone that arrived from another machine for a link we still hold. + db.apply_tombstones(&[Tombstone { + child_id: link_id.to_string(), + kind: TOMBSTONE_KIND_LINK.to_string(), + session_id: Some(sid.to_string()), + deleted_at: Utc::now(), + }]) + .unwrap(); + + assert!( + db.get_links_by_session(&sid).unwrap().is_empty(), + "apply_tombstones must remove the locally-present tombstoned link" + ); + // The cleaned session is re-opened for the next sync. + assert!( + db.get_unsynced_sessions() + .unwrap() + .iter() + .any(|s| s.id == sid), + "cleaning a child must re-open the parent session for re-export" + ); + } + + #[test] + fn test_apply_tombstones_does_not_record_new_tombstone() { + // Applying a tombstone must not create a fresh tombstone (it is already + // recorded), so the set does not grow on every sync. + let (db, _dir) = create_test_db(); + let (sid, link_id, _tag, _ann, _sum) = seed_session_with_children(&db); + // Clear the delete-path tombstones so we start from a known state. + db.prune_tombstones(Utc::now() + Duration::days(1)).unwrap(); + assert!(db.list_tombstones().unwrap().is_empty()); + + db.apply_tombstones(&[Tombstone { + child_id: link_id.to_string(), + kind: TOMBSTONE_KIND_LINK.to_string(), + session_id: Some(sid.to_string()), + deleted_at: Utc::now(), + }]) + .unwrap(); + + assert!( + db.list_tombstones().unwrap().is_empty(), + "apply_tombstones must not record new tombstones" + ); + } + + #[test] + fn test_merge_suppresses_tombstoned_child() { + // A remote record whose link is tombstoned locally must not re-add it. + let (mut db, _dir) = create_test_db(); + let session = create_test_session("claude-code", "/project", Utc::now(), None); + let link = create_test_link(session.id, Some("deadbeef"), LinkType::Commit); + + // Tombstone the link before merging a remote record that still holds it. + db.add_tombstones(&[Tombstone { + child_id: link.id.to_string(), + kind: TOMBSTONE_KIND_LINK.to_string(), + session_id: Some(session.id.to_string()), + deleted_at: Utc::now(), + }]) + .unwrap(); + + db.merge_remote_record( + &session, + &[], + std::slice::from_ref(&link), + &[], + &[], + None, + Utc::now(), + ) + .unwrap(); + + assert!( + db.get_links_by_session(&session.id).unwrap().is_empty(), + "merge must suppress a tombstoned link" + ); + } + + #[test] + fn test_merge_keeps_non_tombstoned_child() { + // A concurrent add (a different link id) must survive even when another + // link on the same session is tombstoned. + let (mut db, _dir) = create_test_db(); + let session = create_test_session("claude-code", "/project", Utc::now(), None); + let deleted = create_test_link(session.id, Some("deadbeef"), LinkType::Commit); + let added = create_test_link(session.id, Some("feedface"), LinkType::Commit); + + db.add_tombstones(&[Tombstone { + child_id: deleted.id.to_string(), + kind: TOMBSTONE_KIND_LINK.to_string(), + session_id: Some(session.id.to_string()), + deleted_at: Utc::now(), + }]) + .unwrap(); + + db.merge_remote_record( + &session, + &[], + &[deleted.clone(), added.clone()], + &[], + &[], + None, + Utc::now(), + ) + .unwrap(); + + let links = db.get_links_by_session(&session.id).unwrap(); + assert_eq!(links.len(), 1, "only the non-tombstoned link survives"); + assert_eq!(links[0].id, added.id); + } + + #[test] + fn test_prune_tombstones_removes_old_only() { + let (db, _dir) = create_test_db(); + let old_child = Uuid::new_v4().to_string(); + let fresh_child = Uuid::new_v4().to_string(); + db.add_tombstones(&[ + Tombstone { + child_id: old_child.clone(), + kind: TOMBSTONE_KIND_TAG.to_string(), + session_id: None, + deleted_at: Utc::now() - Duration::days(120), + }, + Tombstone { + child_id: fresh_child.clone(), + kind: TOMBSTONE_KIND_TAG.to_string(), + session_id: None, + deleted_at: Utc::now(), + }, + ]) + .unwrap(); + + let pruned = db + .prune_tombstones(Utc::now() - Duration::days(90)) + .unwrap(); + assert_eq!(pruned, 1, "only the old tombstone is pruned"); + + let remaining = db.list_tombstones().unwrap(); + assert_eq!(remaining.len(), 1); + assert_eq!(remaining[0].child_id, fresh_child); + } } diff --git a/src/storage/models.rs b/src/storage/models.rs index 10e28cf..db0f7ec 100644 --- a/src/storage/models.rs +++ b/src/storage/models.rs @@ -467,6 +467,35 @@ pub struct Machine { pub created_at: String, } +/// A record of a locally deleted child record (link, tag, annotation, summary). +/// +/// Child records merge additively across machines, so without a tombstone a +/// child DELETED on one machine would be resurrected the next time another +/// machine re-exported the parent session. A tombstone captures the deleted +/// child's id and kind so the sync merge can suppress re-adding that specific +/// record on every machine, while leaving concurrent additions of other records +/// untouched. +/// +/// Only user-facing deletions (unlink, tag remove, annotation delete, summary +/// delete) record tombstones. Deleting a whole session is a purely local +/// operation and must NOT tombstone its children, or a teammate who shares the +/// store would lose that session's reasoning. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct Tombstone { + /// The id of the deleted child record (its UUID rendered as a string). + pub child_id: String, + + /// The kind of child record: `link`, `tag`, `annotation`, or `summary`. + pub kind: String, + + /// The session the deleted child belonged to, when known. Used to re-export + /// the cleaned parent session after a tombstone is applied. + pub session_id: Option, + + /// When the child was deleted, used for garbage-collecting stale tombstones. + pub deleted_at: DateTime, +} + /// A tracked git repository. /// /// Repositories are discovered when sessions reference working directories diff --git a/src/sync/store.rs b/src/sync/store.rs index 1db81a4..3f21b25 100644 --- a/src/sync/store.rs +++ b/src/sync/store.rs @@ -19,7 +19,7 @@ use serde::{Deserialize, Serialize}; use super::encryption::{decrypt_data, encrypt_data}; use super::SyncError; -use crate::storage::models::{Annotation, Message, Session, SessionLink, Summary, Tag}; +use crate::storage::models::{Annotation, Message, Session, SessionLink, Summary, Tag, Tombstone}; /// The complete reasoning record for a single session. /// @@ -104,6 +104,32 @@ pub fn decrypt_session_record(blob: &[u8], key: &[u8]) -> Result gzip -> encrypt_data` pipeline as [`encrypt_session_record`]. +pub fn encrypt_tombstones(tombstones: &[Tombstone], key: &[u8]) -> Result, SyncError> { + let json = serde_json::to_vec(tombstones) + .map_err(|e| SyncError::Serialization(format!("Failed to serialize tombstones: {e}")))?; + + let compressed = gzip_compress(&json)?; + + encrypt_data(&compressed, key) +} + +/// Decrypts and deserializes the tombstone set from git-blob bytes. +/// +/// Inverse of [`encrypt_tombstones`]. +pub fn decrypt_tombstones(blob: &[u8], key: &[u8]) -> Result, SyncError> { + let compressed = decrypt_data(blob, key)?; + + let json = gzip_decompress(&compressed)?; + + serde_json::from_slice(&json) + .map_err(|e| SyncError::Serialization(format!("Failed to deserialize tombstones: {e}"))) +} + /// Compresses bytes with gzip at the default compression level. fn gzip_compress(data: &[u8]) -> Result, SyncError> { let mut encoder = GzEncoder::new(Vec::new(), Compression::default()); @@ -372,6 +398,49 @@ mod tests { assert_eq!(read_back, fixture, "binary blob must survive byte-for-byte"); } + #[test] + fn test_encrypt_decrypt_tombstones_roundtrip() { + let salt = generate_salt(); + let key = derive_key("tombstone passphrase", &salt).unwrap(); + + let tombstones = vec![ + Tombstone { + child_id: Uuid::new_v4().to_string(), + kind: "link".to_string(), + session_id: Some(Uuid::new_v4().to_string()), + deleted_at: Utc::now(), + }, + Tombstone { + child_id: Uuid::new_v4().to_string(), + kind: "summary".to_string(), + session_id: None, + deleted_at: Utc::now(), + }, + ]; + + let blob = encrypt_tombstones(&tombstones, &key).unwrap(); + let restored = decrypt_tombstones(&blob, &key).unwrap(); + + assert_eq!(restored, tombstones); + } + + #[test] + fn test_decrypt_tombstones_wrong_key_fails() { + let salt = generate_salt(); + let key = derive_key("right", &salt).unwrap(); + let wrong = derive_key("wrong", &salt).unwrap(); + + let tombstones = vec![Tombstone { + child_id: Uuid::new_v4().to_string(), + kind: "tag".to_string(), + session_id: None, + deleted_at: Utc::now(), + }]; + let blob = encrypt_tombstones(&tombstones, &key).unwrap(); + + assert!(decrypt_tombstones(&blob, &wrong).is_err()); + } + #[test] fn test_record_with_no_summary() { let salt = generate_salt(); From 5c52af71f1faf1b753fa043ded858e627fd6b85a Mon Sep 17 00:00:00 2001 From: franzer Date: Wed, 1 Jul 2026 13:24:44 -0600 Subject: [PATCH 2/2] fix: don't GC tombstones during sync (prevents stale-blob resurrection) --- src/cli/commands/sync.rs | 94 +++++++++++++++++++++++++++++++++------- src/storage/db.rs | 13 ++++-- 2 files changed, 88 insertions(+), 19 deletions(-) diff --git a/src/cli/commands/sync.rs b/src/cli/commands/sync.rs index c2382d4..c91f2d3 100644 --- a/src/cli/commands/sync.rs +++ b/src/cli/commands/sync.rs @@ -25,7 +25,7 @@ use std::path::{Path, PathBuf}; use std::process::Command; use anyhow::{anyhow, bail, Context, Result}; -use chrono::{DateTime, Duration, Utc}; +use chrono::{DateTime, Utc}; use colored::Colorize; use serde::{Deserialize, Serialize}; use uuid::Uuid; @@ -63,13 +63,6 @@ const MIN_PASSPHRASE_LEN: usize = 8; /// deleted_at}) so a deletion on one machine suppresses that record everywhere. const TOMBSTONES_PATH: &str = "meta/tombstones"; -/// Age past which a tombstone is garbage-collected during sync. -/// -/// A deletion older than this is assumed to have reached every machine, so -/// dropping its tombstone can no longer resurrect the deleted child. Bounds the -/// tombstone set so it does not grow without limit. -const TOMBSTONE_GC_DAYS: i64 = 90; - /// Maximum number of fetch/merge/build/push attempts before giving up. /// /// A concurrent local sync (compare-and-swap mismatch) or a remote that moved @@ -579,12 +572,14 @@ fn perform_sync_in_store( // TOMBSTONES: union the remote deletion set into the local table BEFORE // the merge so the merge suppresses re-adding any child that was deleted - // on another machine, then GC entries older than the retention window. - // A wrong key makes the remote tombstones undecryptable; that is treated - // as an empty set here (the merge below surfaces the wrong-key error). + // on another machine. Tombstones are never pruned during sync: they are + // tiny (a few ids plus a timestamp) and dropping one that is still needed + // for suppression would resurrect the deleted child, so the full set is + // kept indefinitely. A wrong key makes the remote tombstones + // undecryptable; that is treated as an empty set here (the merge below + // surfaces the wrong-key error). let remote_tombstones = read_remote_tombstones(repo, &tracking_entries, key)?; db.add_tombstones(&remote_tombstones)?; - db.prune_tombstones(Utc::now() - Duration::days(TOMBSTONE_GC_DAYS))?; // MERGE remote -> local database (full records, newer-wins). A wrong key // surfaces here and aborts before anything is built or pushed. The store @@ -660,8 +655,8 @@ fn perform_sync_in_store( add_meta_changes(db, repo, tree_base.as_deref(), salt, machine, &mut changes)?; - // Write the unioned, GC'd tombstone set back to the store, but only when - // it differs from what the base tree already holds so an unchanged set + // Write the unioned tombstone set back to the store, but only when it + // differs from what the base tree already holds so an unchanged set // keeps its existing content-addressed blob (no churn from re-encrypting // with a fresh nonce every sync). add_tombstone_changes(db, repo, &remote_tombstones, key, &mut changes)?; @@ -912,7 +907,7 @@ fn read_remote_tombstones( /// (`remote_tombstones`) by their `(child_id, kind)` keys. When the keys match, /// nothing is written so the base's existing blob is preserved verbatim /// (content-addressed dedup). Otherwise the full local set is re-encrypted and -/// written, which also shrinks the stored set after garbage collection. +/// written. fn add_tombstone_changes( db: &Database, repo: &Path, @@ -3127,6 +3122,75 @@ mod tests { ); } + #[test] + fn test_old_remote_tombstone_suppresses_stale_blob_through_sync() { + // The exact resurrection repro from the review: the remote store holds a + // tombstone older than the former 90-day GC window alongside a stale + // session blob that still contains the deleted child. A fresh machine + // must keep the child deleted after a full sync. The sync path no longer + // prunes tombstones, so the old tombstone survives the union and + // suppresses the stale blob during the merge instead of being dropped + // first (which would resurrect the child). + let (_remote_dir, remote_url) = init_bare_remote(); + let passphrase = "shared team passphrase"; + + // Machine A: set up, seed a session with a link, sync so the remote + // holds the session blob (still containing the link). + let (_da, repo_a, keystore_a, _ka, ma) = setup_repo_with_store(&remote_url, passphrase); + let (mut db_a, _dba) = open_db(); + let session_id = seed_full_session(&mut db_a, "machine-a", &repo_dir(&repo_a)); + let link = db_a.get_links_by_session(&session_id).unwrap()[0].clone(); + let (key_a, salt_a) = load_store_credentials(&repo_a, "origin", &keystore_a).unwrap(); + let sessions_a = scoped_unsynced(&db_a, &repo_a); + perform_sync( + &mut db_a, &repo_a, "origin", &key_a, &salt_a, &ma, sessions_a, + ) + .unwrap(); + + // Machine A: record a tombstone for the link backdated well past the + // former 90-day window, then sync again. The session blob is already + // synced so it is carried forward unchanged (still holding the link), + // making the remote store deliberately stale. Only the tombstone blob is + // written. + db_a.add_tombstones(&[Tombstone { + child_id: link.id.to_string(), + kind: "link".to_string(), + session_id: Some(session_id.to_string()), + deleted_at: Utc::now() - chrono::Duration::days(120), + }]) + .unwrap(); + let sessions_a = scoped_unsynced(&db_a, &repo_a); + perform_sync( + &mut db_a, &repo_a, "origin", &key_a, &salt_a, &ma, sessions_a, + ) + .unwrap(); + + // Machine B: fresh clone, join, sync. It fetches the stale session blob + // and the 120-day-old tombstone together. The old tombstone must + // suppress the stale blob so the link is never resurrected. + let dir_b = tempfile::tempdir().unwrap(); + let repo_b = dir_b.path(); + init_repo(repo_b); + git(repo_b, &["remote", "add", "origin", &remote_url]); + let (keystore_b, _kb) = test_keystore(); + let mb = machine("machine-b", "Machine B"); + gitref::fetch(repo_b, "origin", SESSIONS_REF).unwrap(); + let salt_b = read_store_salt(repo_b, "origin").unwrap().unwrap(); + join_store(repo_b, "origin", &keystore_b, &mb, &salt_b, passphrase).unwrap(); + let (mut db_b, _dbb) = open_db(); + let (key_b, salt_b2) = load_store_credentials(repo_b, "origin", &keystore_b).unwrap(); + let sessions_b = scoped_unsynced(&db_b, repo_b); + perform_sync( + &mut db_b, repo_b, "origin", &key_b, &salt_b2, &mb, sessions_b, + ) + .unwrap(); + + assert!( + db_b.get_links_by_session(&session_id).unwrap().is_empty(), + "an old remote tombstone must still suppress a stale session blob" + ); + } + #[test] fn test_annotation_deletion_propagates_via_tombstone_global() { // The same deletion-propagation guarantee holds for the global store and diff --git a/src/storage/db.rs b/src/storage/db.rs index e398e51..d38cd3d 100644 --- a/src/storage/db.rs +++ b/src/storage/db.rs @@ -1093,10 +1093,15 @@ impl Database { /// Prunes tombstones deleted before the given cutoff. /// - /// Bounds the tombstone set so it does not grow without limit. A tombstone - /// older than the cutoff is assumed to have propagated to every machine, so - /// dropping it can no longer resurrect the deleted child. Returns the number - /// of tombstones removed. + /// Returns the number of tombstones removed. This is NOT called during sync: + /// an age-based prune there could drop a tombstone still needed to suppress a + /// stale session blob, resurrecting the deleted child. Tombstones are tiny, so + /// the sync path keeps the full set indefinitely. This method is retained for + /// a possible future safe garbage collection that only prunes tombstones + /// proven to have propagated to every machine. + // Retained for that future safe GC and covered by a direct unit test; it has + // no non-test caller today now that the sync path never prunes. + #[allow(dead_code)] pub fn prune_tombstones(&self, before: DateTime) -> Result { let rows = self.conn.execute( "DELETE FROM tombstones WHERE deleted_at < ?1",