diff --git a/crates/tracedecay-global-db/src/observation/canonical_body_gc.rs b/crates/tracedecay-global-db/src/observation/canonical_body_gc.rs index 6c5af50fea3..a95b950ddab 100644 --- a/crates/tracedecay-global-db/src/observation/canonical_body_gc.rs +++ b/crates/tracedecay-global-db/src/observation/canonical_body_gc.rs @@ -6,12 +6,16 @@ //! bootstrap frontiers are complete. A candidate and its indexed references are //! checked under one guarded writer, so another owner cannot race collection. +use std::time::Instant; + use serde::{Deserialize, Serialize}; use tracedecay_domain::errors::{Result, TraceDecayError}; use tracedecay_runtime_core::db::Database; -use tracedecay_runtime_core::db::engine::{Executor, QueryExecutor, params}; +use tracedecay_runtime_core::db::engine::{Executor, QueryExecutor, WriteStatement, params}; use tracedecay_store::collect_body_refs; +#[cfg(test)] +mod bootstrap_write_tests; pub(crate) mod schema; #[cfg(test)] mod tests; @@ -25,6 +29,15 @@ pub const MAX_CANONICAL_BODY_GC_ROWS: usize = 48; pub struct CanonicalBodyGcReport { pub roots_scanned: u64, pub bodies_scanned: u64, + /// Wall-clock duration from writer admission through commit, not CPU time. + pub elapsed_ms: Option, + /// Candidates discharged in the committed page, even if their body stays live. + pub candidates_discharged: u64, + /// Phase-qualified SQL bootstrap cursor before the committed page. + /// The value is absent during candidate collection and idle ticks. + pub bootstrap_frontier_before: Option, + /// The corresponding persisted cursor, verified in the same writer. + pub bootstrap_frontier_after: Option, pub bodies_deleted: u64, /// Stored BLOB bytes deleted, including the effect of their existing /// compression. This is not database-file or filesystem space reclaimed. @@ -34,6 +47,16 @@ pub struct CanonicalBodyGcReport { pub has_more: bool, } +impl CanonicalBodyGcReport { + /// A bootstrap page can commit its remaining frontier after every owner in it was deleted. + pub fn has_committed_work(&self) -> bool { + self.roots_scanned != 0 + || self.bodies_scanned != 0 + || self.bootstrap_frontier_after.is_some() + } +} + +#[derive(Clone, Debug, PartialEq, Eq)] struct Progress { raw_through: i64, raw_until: i64, @@ -118,15 +141,25 @@ async fn save_progress(conn: &(impl Executor + ?Sized), progress: &Progress) -> if changed != 1 { return Err(integrity_error("canonical body ownership frontier changed")); } + // The UPDATE can report one affected row even if a database trigger + // rewrites the checkpoint. Verify the durable value before committing the + // references or candidate seed that depended on this bootstrap page. + if read_progress(conn).await? != *progress { + return Err(integrity_error( + "canonical body ownership bootstrap frontier was not persisted", + )); + } Ok(()) } #[tracing::instrument(name = "global_db.canonical_bodies.gc", level = "trace", skip_all)] pub(crate) async fn run(database: &Database, max_rows: usize) -> Result { + let started = Instant::now(); let transaction = database.begin_write_transaction(OPERATION).await?; match step(&transaction, max_rows).await { - Ok(report) => { + Ok(mut report) => { transaction.commit().await?; + report.elapsed_ms = Some(started.elapsed().as_millis()); Ok(report) } Err(error) => match transaction.rollback().await { @@ -144,14 +177,34 @@ async fn step(conn: &(impl Executor + ?Sized), max_rows: usize) -> Result Result { let mut rows = conn .query( - "SELECT store_id, CASE WHEN storage_kind = 'inline' AND content IS NULL - THEN content_hash END - FROM lcm_raw_messages WHERE store_id > ?1 AND store_id <= ?2 - ORDER BY store_id LIMIT ?3", + "SELECT store_id FROM lcm_raw_messages + WHERE store_id > ?1 AND store_id <= ?2 ORDER BY store_id LIMIT ?3", params![progress.raw_through, progress.raw_until, limit], ) .await .map_err(db_error)?; let mut page = Vec::new(); while let Some(row) = rows.next().await.map_err(db_error)? { - page.push(( - row.get::(0).map_err(db_error)?, - row.get::>(1).map_err(db_error)?, - )); + page.push(row.get::(0).map_err(db_error)?); } drop(rows); - for (owner, hash) in &page { - if let Some(hash) = hash { - conn.execute( - "INSERT OR IGNORE INTO canonical_body_gc_refs(owner_kind, owner_id, content_hash) - VALUES ('raw', ?1, ?2)", - params![*owner, hash.as_str()], + if let Some(&last) = page.last() { + // Both reads share the writer transaction. The last selected identity + // bounds this indexed insert to the same page, including sparse IDs, + // without one worker submission for every referenced owner. + conn.execute( + "INSERT OR IGNORE INTO canonical_body_gc_refs(owner_kind, owner_id, content_hash) + SELECT 'raw', store_id, content_hash FROM lcm_raw_messages + WHERE store_id > ?1 AND store_id <= ?2 + AND storage_kind = 'inline' AND content IS NULL", + params![progress.raw_through, last], + ) + .await + .map_err(db_error)?; + // A suppressed INSERT must not advance bootstrap past an owner whose + // canonical body would subsequently appear unreferenced. + let mut missing = conn + .query( + "SELECT 1 FROM lcm_raw_messages raw + WHERE raw.store_id > ?1 AND raw.store_id <= ?2 + AND raw.storage_kind = 'inline' AND raw.content IS NULL + AND NOT EXISTS ( + SELECT 1 FROM canonical_body_gc_refs r + WHERE r.owner_kind = 'raw' AND r.owner_id = raw.store_id + AND r.content_hash = raw.content_hash + ) LIMIT 1", + params![progress.raw_through, last], ) .await .map_err(db_error)?; + let incomplete = missing.next().await.map_err(db_error)?.is_some(); + drop(missing); + if incomplete { + return Err(integrity_error( + "canonical raw bootstrap reference was not indexed", + )); } - progress.raw_through = *owner; + progress.raw_through = last; } if page.len() < limit as usize { progress.raw_through = progress.raw_until; @@ -226,17 +300,60 @@ async fn bootstrap_observations( )); } drop(rows); + // Parsing preserves the canonical Rust reference contract, including + // value deduplication. Submit validated writes in bounded groups so an + // observation page cannot schedule one storage-worker hop per body hash. + // The caller's writer transaction rolls back every group on any failure. + const MAX_PENDING_OBSERVATION_REFS: usize = 128; + let mut pending = Vec::new(); + let mut expected = Vec::new(); for (owner, json) in &page { for hash in collect_body_refs(json).map_err(db_error)? { - conn.execute( - "INSERT OR IGNORE INTO canonical_body_gc_refs(owner_kind, owner_id, content_hash) - VALUES ('observation', ?1, ?2)", - params![*owner, hash], + expected.push((*owner, hash.clone())); + pending.push( + WriteStatement::new( + "INSERT OR IGNORE INTO canonical_body_gc_refs(owner_kind, owner_id, content_hash) + VALUES ('observation', ?1, ?2)", + params![*owner, hash], + ) + .map_err(db_error)?, + ); + if pending.len() == MAX_PENDING_OBSERVATION_REFS { + conn.execute_statements(std::mem::take(&mut pending)) + .await + .map_err(db_error)?; + } + } + progress.observations_through = *owner; + } + if !pending.is_empty() { + conn.execute_statements(pending).await.map_err(db_error)?; + } + if !expected.is_empty() { + // Verify the Rust-validated reference set, not a second parse of raw + // observation JSON. INSERT OR IGNORE and SQLite trigger RAISE(IGNORE) + // can both report success without creating a missing reference. + let expected_json = serde_json::to_string(&expected).map_err(db_error)?; + let mut missing = conn + .query( + "SELECT 1 FROM json_each(?1) pair + WHERE NOT EXISTS ( + SELECT 1 FROM canonical_body_gc_refs r + WHERE r.owner_kind = 'observation' + AND r.owner_id = json_extract(pair.value, '$[0]') + AND r.content_hash = json_extract(pair.value, '$[1]') + ) LIMIT 1", + params![expected_json], ) .await .map_err(db_error)?; + let incomplete = missing.next().await.map_err(db_error)?.is_some(); + drop(missing); + if incomplete { + return Err(integrity_error( + "canonical observation bootstrap reference was not indexed", + )); } - progress.observations_through = *owner; } if page.len() < limit as usize { progress.observations_through = progress.observations_until; @@ -266,14 +383,39 @@ async fn bootstrap_bodies( page.push(row.get::(0).map_err(db_error)?); } drop(rows); - for hash in &page { + if let Some(last) = page.last() { + // Seed the selected key range in one statement, not one asynchronous + // writer submission per body. No body payload is materialized here. conn.execute( - "INSERT OR IGNORE INTO canonical_body_gc_candidates(content_hash) VALUES (?1)", - params![hash.as_str()], + "INSERT OR IGNORE INTO canonical_body_gc_candidates(content_hash) + SELECT content_hash FROM session_canonical_bodies + WHERE content_hash > ?1 AND content_hash <= ?2", + params![progress.bodies_after.as_str(), last.as_str()], ) .await .map_err(db_error)?; - progress.bodies_after.clone_from(hash); + // A vetoed candidate insert otherwise leaves an orphaned body behind + // the permanently committed bootstrap cursor, with no future seed. + let mut missing = conn + .query( + "SELECT 1 FROM session_canonical_bodies b + WHERE b.content_hash > ?1 AND b.content_hash <= ?2 + AND NOT EXISTS ( + SELECT 1 FROM canonical_body_gc_candidates c + WHERE c.content_hash = b.content_hash + ) LIMIT 1", + params![progress.bodies_after.as_str(), last.as_str()], + ) + .await + .map_err(db_error)?; + let incomplete = missing.next().await.map_err(db_error)?.is_some(); + drop(missing); + if incomplete { + return Err(integrity_error( + "canonical body bootstrap candidate was not indexed", + )); + } + progress.bodies_after.clone_from(last); } if page.len() < limit as usize { progress.bodies_after.clone_from(&progress.bodies_until); @@ -321,15 +463,65 @@ async fn collect_candidates( report.stored_bytes_deleted = report.stored_bytes_deleted.saturating_add(bytes); } drop(deleted); + // An ignored or otherwise suppressed body DELETE must not silently + // discharge an unreferenced candidate. Once that candidate disappears, + // no owner-triggered re-enqueue is guaranteed, and the body can persist + // indefinitely without another collection attempt. + let mut uncollected = conn + .query( + "SELECT 1 FROM session_canonical_bodies b + WHERE b.content_hash IN (SELECT value FROM json_each(?1)) + AND NOT EXISTS (SELECT 1 FROM canonical_body_gc_refs r + WHERE r.content_hash = b.content_hash) + LIMIT 1", + params![hashes.as_str()], + ) + .await + .map_err(db_error)?; + if uncollected.next().await.map_err(db_error)?.is_some() { + return Err(integrity_error( + "canonical body collection left an unreferenced body behind", + )); + } + drop(uncollected); // Referenced candidates are discharged too. The final owner removal // records them again; idle ticks never revisit every retained body. - conn.execute( - "DELETE FROM canonical_body_gc_candidates - WHERE content_hash IN (SELECT value FROM json_each(?1))", - params![hashes], - ) - .await - .map_err(db_error)?; + // Deleting an unreferenced body cascades to its candidate row. The + // remaining explicit removals discharge referenced bodies that stay live. + let explicitly_discharged = conn + .execute( + "DELETE FROM canonical_body_gc_candidates + WHERE content_hash IN (SELECT value FROM json_each(?1))", + params![hashes.as_str()], + ) + .await + .map_err(db_error)?; + // A DELETE row count is not a durable discharge receipt: an AFTER DELETE + // trigger may have re-enqueued the very same candidates. Refuse the page + // under this writer instead of recommitting an identical 48-row batch. + let mut remaining = conn + .query( + "SELECT 1 FROM canonical_body_gc_candidates + WHERE content_hash IN (SELECT value FROM json_each(?1)) LIMIT 1", + params![hashes.as_str()], + ) + .await + .map_err(db_error)?; + if remaining.next().await.map_err(db_error)?.is_some() { + return Err(integrity_error( + "canonical body candidate discharge was not persisted", + )); + } + drop(remaining); + let discharged = explicitly_discharged + .checked_add(report.bodies_deleted) + .ok_or_else(|| integrity_error("canonical body candidate discharge overflow"))?; + if discharged != report.bodies_scanned { + return Err(integrity_error( + "canonical body candidate page was not fully discharged", + )); + } + report.candidates_discharged = discharged; Ok(()) } diff --git a/crates/tracedecay-global-db/src/observation/canonical_body_gc/bootstrap_write_tests.rs b/crates/tracedecay-global-db/src/observation/canonical_body_gc/bootstrap_write_tests.rs new file mode 100644 index 00000000000..5e71811ca92 --- /dev/null +++ b/crates/tracedecay-global-db/src/observation/canonical_body_gc/bootstrap_write_tests.rs @@ -0,0 +1,613 @@ +//! Measure real bootstrap write submissions without replacing the SQLite executor. + +use std::sync::atomic::{AtomicUsize, Ordering}; + +use tracedecay_runtime_core::db::engine::{ + Executor, IntoParams, QueryExecutor, Result as EngineResult, Rows, TestConnection, + TransactionBehavior, WriteStatement, params, +}; +use tracedecay_store::{CANONICAL_BODIES_TABLE_SQL, StoredCanonicalBody, UPSERT_CANONICAL_BODY_SQL}; + +use super::{CanonicalBodyGcReport, read_progress, schema, step}; + +struct CountWrites<'a, E: ?Sized> { + inner: &'a E, + submissions: AtomicUsize, +} + +impl QueryExecutor for CountWrites<'_, E> { + async fn query

(&self, sql: &str, params: P) -> EngineResult + where + P: IntoParams, + { + self.inner.query(sql, params).await + } +} + +impl Executor for CountWrites<'_, E> { + async fn execute

(&self, sql: &str, params: P) -> EngineResult + where + P: IntoParams, + { + self.submissions.fetch_add(1, Ordering::Relaxed); + self.inner.execute(sql, params).await + } + + async fn execute_statements(&self, statements: Vec) -> EngineResult> { + self.submissions.fetch_add(1, Ordering::Relaxed); + self.inner.execute_statements(statements).await + } + + async fn execute_batch(&self, sql: &str) -> EngineResult<()> { + self.submissions.fetch_add(1, Ordering::Relaxed); + self.inner.execute_batch(sql).await + } +} + +struct Fixture { + conn: TestConnection, + bodies: Vec, + _directory: tempfile::TempDir, +} + +impl Fixture { + async fn open() -> Self { + Self::with_observations(false).await + } + + async fn with_observations(include_observations: bool) -> Self { + let directory = tempfile::tempdir().unwrap(); + let conn = TestConnection::open(&directory.path().join("bootstrap-writes.db")); + conn.execute_batch(CANONICAL_BODIES_TABLE_SQL).await.unwrap(); + conn.execute_batch( + "CREATE TABLE lcm_raw_messages ( + store_id INTEGER PRIMARY KEY AUTOINCREMENT, + storage_kind TEXT NOT NULL, content TEXT, content_hash TEXT NOT NULL + ); + CREATE TABLE observations ( + sequence INTEGER PRIMARY KEY AUTOINCREMENT, + observation_json TEXT NOT NULL + );", + ) + .await + .unwrap(); + let transaction = conn + .transaction_with_behavior(TransactionBehavior::Immediate) + .await + .unwrap(); + let mut bodies = Vec::new(); + for ordinal in 1..=52_i64 { + let text = format!("retained bootstrap body {ordinal}"); + let body = StoredCanonicalBody::pack(text.as_bytes()).unwrap(); + transaction + .execute( + UPSERT_CANONICAL_BODY_SQL, + params![ + body.content_hash.as_str(), + body.encoding, + body.blob.as_slice(), + body.uncompressed_bytes + ], + ) + .await + .unwrap(); + // Holes in the key space must not turn the page into an ID count. + transaction + .execute( + "INSERT INTO lcm_raw_messages(store_id, storage_kind, content_hash) + VALUES (?1, 'inline', ?2)", + params![ordinal * 2, body.content_hash.as_str()], + ) + .await + .unwrap(); + if include_observations { + // Three distinct hashes from the third owner onward, so a + // 48-owner page crosses the 128-write submission boundary. + let current = body.content_hash.as_str(); + let previous = bodies + .last() + .map_or(current, |body: &StoredCanonicalBody| body.content_hash.as_str()); + let earlier = bodies + .iter() + .rev() + .nth(1) + .map_or(current, |body| body.content_hash.as_str()); + let json = serde_json::json!({ + "payload": {"a": null, "b": null, "c": null}, + "_tracedecay_canonical_body_refs": { + "/payload/a": current, + "/payload/b": previous, + "/payload/c": earlier + } + }) + .to_string(); + transaction + .execute( + "INSERT INTO observations(sequence, observation_json) VALUES (?1, ?2)", + params![ordinal * 2, json], + ) + .await + .unwrap(); + } + bodies.push(body); + } + schema::install(&transaction).await.unwrap(); + transaction.commit().await.unwrap(); + Self { + conn, + bodies, + _directory: directory, + } + } + + async fn tick(&self) -> (CanonicalBodyGcReport, usize) { + let transaction = self + .conn + .transaction_with_behavior(TransactionBehavior::Immediate) + .await + .unwrap(); + let (report, submissions) = { + let counted = CountWrites { + inner: &transaction, + submissions: AtomicUsize::new(0), + }; + let report = step(&counted, 48).await.unwrap(); + (report, counted.submissions.load(Ordering::Relaxed)) + }; + transaction.commit().await.unwrap(); + (report, submissions) + } + + async fn assert_bodies_preserved(&self) { + let count = scalar(&self.conn, "SELECT COUNT(*) FROM session_canonical_bodies").await; + assert_eq!(count, self.bodies.len() as i64); + for body in &self.bodies { + let mut rows = self + .conn + .query( + "SELECT body FROM session_canonical_bodies WHERE content_hash = ?1", + params![body.content_hash.as_str()], + ) + .await + .unwrap(); + let stored = rows.next().await.unwrap().unwrap(); + assert_eq!(stored.get::>(0).unwrap(), body.blob); + } + } +} + +async fn scalar(conn: &(impl QueryExecutor + ?Sized), sql: &str) -> i64 { + conn.query(sql, ()) + .await + .unwrap() + .next() + .await + .unwrap() + .unwrap() + .get(0) + .unwrap() +} + +async fn raw_owners(conn: &(impl QueryExecutor + ?Sized)) -> Vec { + let mut rows = conn + .query( + "SELECT owner_id FROM canonical_body_gc_refs WHERE owner_kind = 'raw' + ORDER BY owner_id", + (), + ) + .await + .unwrap(); + let mut owners = Vec::new(); + while let Some(row) = rows.next().await.unwrap() { + owners.push(row.get(0).unwrap()); + } + owners +} + +#[asupersync::test] +async fn bootstrap_pages_batch_writes_and_preserve_sparse_frontiers() { + let fixture = Fixture::open().await; + for (scanned, through, owners) in [(48_u64, 96_i64, 48_i64), (4, 104, 52)] { + let (report, submissions) = fixture.tick().await; + let progress = read_progress(&fixture.conn).await.unwrap(); + assert_eq!(report.roots_scanned, scanned); + assert_eq!(submissions, 2, "one seed submission plus the checkpoint"); + assert_eq!(progress.raw_through, through); + assert_eq!( + raw_owners(&fixture.conn).await, + (1..=owners).map(|ordinal| ordinal * 2).collect::>() + ); + assert_eq!(report.bodies_deleted, 0); + assert!(report.has_more); + } + + let mut hashes = fixture + .bodies + .iter() + .map(|body| body.content_hash.clone()) + .collect::>(); + hashes.sort(); + for (scanned, total) in [(48_u64, 48_usize), (4, 52)] { + let (report, submissions) = fixture.tick().await; + let progress = read_progress(&fixture.conn).await.unwrap(); + let candidates = scalar( + &fixture.conn, + "SELECT COUNT(*) FROM canonical_body_gc_candidates", + ) + .await; + assert_eq!(report.bodies_scanned, scanned); + assert_eq!(submissions, 2, "one candidate seed plus the checkpoint"); + assert_eq!(progress.bodies_after, hashes[total - 1]); + assert_eq!(candidates, total as i64); + assert_eq!(report.bodies_deleted, 0); + assert!(report.has_more); + } + + for scanned in [48_u64, 4] { + let (report, _) = fixture.tick().await; + assert_eq!(report.bodies_scanned, scanned); + assert_eq!(report.candidates_discharged, scanned); + assert_eq!(report.bodies_deleted, 0); + assert_eq!(report.has_more, scanned == 48); + } + let (idle, submissions) = fixture.tick().await; + assert!(!idle.has_committed_work()); + assert!(!idle.has_more); + assert_eq!(submissions, 0); + fixture.assert_bodies_preserved().await; +} + +#[asupersync::test] +async fn bootstrap_batch_refusal_rolls_back_and_recovers() { + for bodies_phase in [false, true] { + let fixture = Fixture::open().await; + if bodies_phase { + fixture.tick().await; + fixture.tick().await; + } + let before = read_progress(&fixture.conn).await.unwrap(); + let table = if bodies_phase { + "canonical_body_gc_candidates" + } else { + "canonical_body_gc_refs" + }; + // FAIL deliberately leaves the first 47 inserts in the transaction. + // The collector's rollback must discard them with the uncommitted page. + fixture + .conn + .execute_batch(&format!( + "CREATE TRIGGER test_bootstrap_refusal BEFORE INSERT ON {table} + WHEN (SELECT COUNT(*) FROM {table}) = 47 BEGIN + SELECT RAISE(FAIL, 'injected bootstrap batch refusal'); + END" + )) + .await + .unwrap(); + let transaction = fixture + .conn + .transaction_with_behavior(TransactionBehavior::Immediate) + .await + .unwrap(); + let error = step(&transaction, 48).await.unwrap_err(); + assert!(error.to_string().contains("injected bootstrap batch refusal")); + transaction.rollback().await.unwrap(); + assert_eq!(read_progress(&fixture.conn).await.unwrap(), before); + let count = scalar(&fixture.conn, &format!("SELECT COUNT(*) FROM {table}")).await; + assert_eq!(count, 0); + fixture.assert_bodies_preserved().await; + fixture + .conn + .execute_batch("DROP TRIGGER test_bootstrap_refusal") + .await + .unwrap(); + for scanned in [48_u64, 4] { + let (report, submissions) = fixture.tick().await; + let actual = if bodies_phase { + report.bodies_scanned + } else { + report.roots_scanned + }; + assert_eq!(submissions, 2); + assert_eq!(actual, scanned); + assert_eq!(report.bodies_deleted, 0); + } + let count = scalar(&fixture.conn, &format!("SELECT COUNT(*) FROM {table}")).await; + assert_eq!(count, 52); + fixture.assert_bodies_preserved().await; + } +} + +#[asupersync::test] +async fn bootstrap_batch_eligibility_does_not_change_the_page_boundary() { + let fixture = Fixture::open().await; + fixture + .conn + .execute_batch( + "UPDATE lcm_raw_messages SET storage_kind = 'external' WHERE store_id = 94; + UPDATE lcm_raw_messages SET content = 'owned inline text' WHERE store_id = 96;", + ) + .await + .unwrap(); + let (report, submissions) = fixture.tick().await; + let progress = read_progress(&fixture.conn).await.unwrap(); + assert_eq!(report.roots_scanned, 48); + assert_eq!(submissions, 2); + assert_eq!(progress.raw_through, 96); + assert_eq!( + raw_owners(&fixture.conn).await, + (1..=46_i64).map(|ordinal| ordinal * 2).collect::>() + ); + assert!(report.has_more); + let (tail, submissions) = fixture.tick().await; + let progress = read_progress(&fixture.conn).await.unwrap(); + assert_eq!(tail.roots_scanned, 4); + assert_eq!(submissions, 2); + assert_eq!(progress.raw_through, 104); + let expected = (1..=52_i64) + .filter(|ordinal| ![47, 48].contains(ordinal)) + .map(|ordinal| ordinal * 2) + .collect::>(); + assert_eq!(raw_owners(&fixture.conn).await, expected); + fixture.assert_bodies_preserved().await; +} + +async fn observation_owners(conn: &(impl QueryExecutor + ?Sized)) -> Vec { + let mut rows = conn + .query( + "SELECT DISTINCT owner_id FROM canonical_body_gc_refs + WHERE owner_kind = 'observation' ORDER BY owner_id", + (), + ) + .await + .unwrap(); + let mut owners = Vec::new(); + while let Some(row) = rows.next().await.unwrap() { + owners.push(row.get(0).unwrap()); + } + owners +} + +#[asupersync::test] +async fn observation_bootstrap_batches_across_a_bounded_write_submission() { + let fixture = Fixture::with_observations(true).await; + for _ in 0..2 { + fixture.tick().await; + } + for (scanned, frontier, owners, refs, submissions) in [ + (48_u64, 96_i64, 48_i64, 141_i64, 3_usize), + (4, 104, 52, 153, 2), + ] { + let (report, actual_submissions) = fixture.tick().await; + assert_eq!(report.roots_scanned, scanned); + assert_eq!( + report.bootstrap_frontier_after, + Some(format!("observations:{frontier}")) + ); + assert_eq!( + read_progress(&fixture.conn).await.unwrap().observations_through, + frontier + ); + assert_eq!(actual_submissions, submissions); + assert_eq!( + observation_owners(&fixture.conn).await, + (1..=owners).map(|ordinal| ordinal * 2).collect::>() + ); + assert_eq!( + scalar(&fixture.conn, "SELECT COUNT(*) FROM canonical_body_gc_refs + WHERE owner_kind = 'observation'") + .await, + refs, + "duplicate references within an observation must stay deduplicated" + ); + assert_eq!(report.bodies_deleted, 0); + assert!(report.has_more); + } + let mut finished = false; + for _ in 0..6 { + let (report, _) = fixture.tick().await; + assert_eq!(report.bodies_deleted, 0); + if !report.has_more { + finished = true; + break; + } + } + assert!(finished, "bounded bootstrap and candidate discharge must terminate"); + fixture.assert_bodies_preserved().await; + let (idle, submissions) = fixture.tick().await; + assert!(!idle.has_more); + assert!(!idle.has_committed_work()); + assert_eq!(submissions, 0); +} + +#[asupersync::test] +async fn observation_bootstrap_failure_after_a_full_batch_rolls_back_and_recovers() { + let fixture = Fixture::with_observations(true).await; + for _ in 0..2 { + fixture.tick().await; + } + let before = read_progress(&fixture.conn).await.unwrap(); + fixture + .conn + .execute_batch( + "CREATE TRIGGER refuse_observation_seed BEFORE INSERT ON canonical_body_gc_refs + WHEN NEW.owner_kind = 'observation' AND NEW.owner_id = 94 BEGIN + SELECT RAISE(FAIL, 'injected observation batch refusal'); + END", + ) + .await + .unwrap(); + let transaction = fixture + .conn + .transaction_with_behavior(TransactionBehavior::Immediate) + .await + .unwrap(); + let error = step(&transaction, 48).await.unwrap_err(); + assert!(error.to_string().contains("injected observation batch refusal")); + transaction.rollback().await.unwrap(); + assert_eq!(read_progress(&fixture.conn).await.unwrap(), before); + assert!(observation_owners(&fixture.conn).await.is_empty()); + fixture.assert_bodies_preserved().await; + + fixture + .conn + .execute_batch("DROP TRIGGER refuse_observation_seed") + .await + .unwrap(); + for (frontier, references, submissions) in [(96_i64, 141_i64, 3_usize), (104, 153, 2)] { + let (report, actual_submissions) = fixture.tick().await; + assert_eq!(report.bootstrap_frontier_after, Some(format!("observations:{frontier}"))); + assert_eq!(actual_submissions, submissions); + assert_eq!( + scalar(&fixture.conn, "SELECT COUNT(*) FROM canonical_body_gc_refs + WHERE owner_kind = 'observation'") + .await, + references + ); + } + fixture.assert_bodies_preserved().await; +} + +async fn assert_refused_bootstrap_rolls_back(fixture: &Fixture, diagnosis: &str) { + let before = read_progress(&fixture.conn).await.unwrap(); + let transaction = fixture + .conn + .transaction_with_behavior(TransactionBehavior::Immediate) + .await + .unwrap(); + let error = step(&transaction, 48).await.unwrap_err(); + assert!( + error.to_string().contains(diagnosis), + "unexpected bootstrap refusal: {error}" + ); + transaction.rollback().await.unwrap(); + assert_eq!(read_progress(&fixture.conn).await.unwrap(), before); + fixture.assert_bodies_preserved().await; +} + +#[asupersync::test] +async fn ignored_raw_reference_insert_refuses_checkpoint_then_recovers() { + let fixture = Fixture::open().await; + fixture + .conn + .execute_batch( + "CREATE TRIGGER ignore_raw_ref BEFORE INSERT ON canonical_body_gc_refs + WHEN NEW.owner_kind = 'raw' AND NEW.owner_id = 94 + BEGIN SELECT RAISE(IGNORE); END", + ) + .await + .unwrap(); + assert_refused_bootstrap_rolls_back(&fixture, "raw bootstrap reference was not indexed").await; + assert!(raw_owners(&fixture.conn).await.is_empty()); + fixture + .conn + .execute_batch("DROP TRIGGER ignore_raw_ref") + .await + .unwrap(); + for (scanned, frontier) in [(48_u64, 96_i64), (4, 104)] { + let (report, _) = fixture.tick().await; + assert_eq!(report.roots_scanned, scanned); + assert_eq!( + read_progress(&fixture.conn).await.unwrap().raw_through, + frontier + ); + } + assert_eq!(raw_owners(&fixture.conn).await.len(), 52); + fixture.assert_bodies_preserved().await; +} + +#[asupersync::test] +async fn ignored_observation_reference_insert_refuses_checkpoint_then_recovers() { + let fixture = Fixture::with_observations(true).await; + fixture.tick().await; + fixture.tick().await; + fixture + .conn + .execute_batch( + "CREATE TRIGGER ignore_observation_ref BEFORE INSERT ON canonical_body_gc_refs + WHEN NEW.owner_kind = 'observation' AND NEW.owner_id = 94 + BEGIN SELECT RAISE(IGNORE); END", + ) + .await + .unwrap(); + assert_refused_bootstrap_rolls_back( + &fixture, + "observation bootstrap reference was not indexed", + ) + .await; + assert!(observation_owners(&fixture.conn).await.is_empty()); + fixture + .conn + .execute_batch("DROP TRIGGER ignore_observation_ref") + .await + .unwrap(); + for (frontier, references) in [(96_i64, 141_i64), (104, 153)] { + let (report, _) = fixture.tick().await; + assert_eq!(report.bootstrap_frontier_after, Some(format!("observations:{frontier}"))); + assert_eq!(read_progress(&fixture.conn).await.unwrap().observations_through, frontier); + assert_eq!( + scalar( + &fixture.conn, + "SELECT COUNT(*) FROM canonical_body_gc_refs + WHERE owner_kind = 'observation'", + ) + .await, + references, + ); + } + fixture.assert_bodies_preserved().await; +} + +#[asupersync::test] +async fn ignored_candidate_insert_refuses_checkpoint_then_recovers() { + let fixture = Fixture::open().await; + fixture.tick().await; + fixture.tick().await; + let mut hashes = fixture + .bodies + .iter() + .map(|body| body.content_hash.as_str()) + .collect::>(); + hashes.sort_unstable(); + let skipped = hashes[47]; + fixture + .conn + .execute_batch(&format!( + "CREATE TRIGGER ignore_candidate_seed BEFORE INSERT ON canonical_body_gc_candidates + WHEN NEW.content_hash = '{skipped}' + BEGIN SELECT RAISE(IGNORE); END" + )) + .await + .unwrap(); + assert_refused_bootstrap_rolls_back(&fixture, "body bootstrap candidate was not indexed").await; + assert_eq!( + scalar( + &fixture.conn, + "SELECT COUNT(*) FROM canonical_body_gc_candidates", + ) + .await, + 0 + ); + fixture + .conn + .execute_batch("DROP TRIGGER ignore_candidate_seed") + .await + .unwrap(); + for (scanned, candidate_total) in [(48_u64, 48_i64), (4, 52)] { + let (report, _) = fixture.tick().await; + assert_eq!(report.bodies_scanned, scanned); + assert_eq!( + scalar(&fixture.conn, "SELECT COUNT(*) FROM canonical_body_gc_candidates").await, + candidate_total + ); + } + for scanned in [48_u64, 4] { + let (report, _) = fixture.tick().await; + assert_eq!(report.bodies_scanned, scanned); + assert_eq!(report.candidates_discharged, scanned); + assert_eq!(report.bodies_deleted, 0, "referenced bodies are preserved"); + } + let (idle, submissions) = fixture.tick().await; + assert!(!idle.has_more); + assert_eq!(submissions, 0); + fixture.assert_bodies_preserved().await; +} diff --git a/crates/tracedecay-global-db/src/observation/canonical_body_gc/tests.rs b/crates/tracedecay-global-db/src/observation/canonical_body_gc/tests.rs index 9da4f654323..ba18b93368b 100644 --- a/crates/tracedecay-global-db/src/observation/canonical_body_gc/tests.rs +++ b/crates/tracedecay-global-db/src/observation/canonical_body_gc/tests.rs @@ -92,6 +92,14 @@ async fn body_exists(conn: &(impl QueryExecutor + ?Sized), hash: &str) -> bool { .is_some() } +async fn candidate_count(conn: &(impl QueryExecutor + ?Sized)) -> i64 { + let mut rows = conn + .query("SELECT COUNT(*) FROM canonical_body_gc_candidates", ()) + .await + .unwrap(); + rows.next().await.unwrap().unwrap().get(0).unwrap() +} + fn observation(hash: &str) -> String { serde_json::json!({ "payload": {"text": null}, @@ -506,3 +514,393 @@ async fn nonpositive_legacy_owner_ids_cannot_be_skipped_by_bootstrap() { ); } } + +#[asupersync::test] +async fn ignored_body_delete_rolls_back_candidates_then_reclaims_distinct_pages() { + let store = Store::open().await; + store.install().await; + let mut hashes = Vec::new(); + for ordinal in 0..52 { + let body = + StoredCanonicalBody::pack(format!("unreferenced candidate {ordinal}").as_bytes()) + .unwrap(); + insert_body(store.conn(), &body).await; + hashes.push(body.content_hash); + } + assert_eq!(candidate_count(store.conn()).await, 52); + store + .conn() + .execute_batch( + "CREATE TRIGGER refuse_body_gc_delete + BEFORE DELETE ON session_canonical_bodies BEGIN + SELECT RAISE(IGNORE); + END", + ) + .await + .unwrap(); + + let transaction = store + .conn() + .transaction_with_behavior(TransactionBehavior::Immediate) + .await + .unwrap(); + let error = step(&transaction, 48).await.unwrap_err(); + assert!( + error + .to_string() + .contains("collection left an unreferenced body behind") + ); + transaction.rollback().await.unwrap(); + assert_eq!( + candidate_count(store.conn()).await, + 52, + "a vetoed DELETE cannot discard the only future collection opportunity" + ); + for hash in &hashes { + assert!(body_exists(store.conn(), hash).await); + } + + store + .conn() + .execute_batch("DROP TRIGGER refuse_body_gc_delete") + .await + .unwrap(); + let first = store.tick(48).await; + assert_eq!(first.bodies_scanned, 48); + assert_eq!(first.bodies_deleted, 48); + assert_eq!(first.candidates_discharged, 48); + assert!(first.has_more); + assert_eq!(candidate_count(store.conn()).await, 4); + let second = store.tick(48).await; + assert_eq!(second.bodies_scanned, 4); + assert_eq!(second.bodies_deleted, 4); + assert_eq!(second.candidates_discharged, 4); + assert!(!second.has_more); + assert_eq!(candidate_count(store.conn()).await, 0); + for hash in &hashes { + assert!(!body_exists(store.conn(), hash).await); + } + assert_eq!( + store.tick(48).await, + CanonicalBodyGcReport { + bootstrap_complete: true, + ..CanonicalBodyGcReport::default() + } + ); +} + +#[asupersync::test] +async fn zero_delete_pages_discharge_distinct_candidates_and_quiesce() { + let store = Store::open().await; + for ordinal in 0..52 { + let body = + StoredCanonicalBody::pack(format!("still referenced {ordinal}").as_bytes()).unwrap(); + insert_body(store.conn(), &body).await; + store + .conn() + .execute( + "INSERT INTO lcm_raw_messages(storage_kind, content, content_hash) + VALUES ('inline', NULL, ?1)", + params![body.content_hash.as_str()], + ) + .await + .unwrap(); + } + store.install().await; + + let mut discharged_pages = Vec::new(); + for _ in 0..16 { + let report = store.tick(48).await; + assert_eq!(report.bodies_deleted, 0, "referenced bodies stay live"); + if report.candidates_discharged > 0 { + assert_eq!(report.candidates_discharged, report.bodies_scanned); + discharged_pages.push(report.candidates_discharged); + } + if !report.has_more { + break; + } + } + assert_eq!(discharged_pages, vec![48, 4]); + assert!(!has_candidates(store.conn()).await.unwrap()); + assert_eq!( + store.tick(48).await, + CanonicalBodyGcReport { + bootstrap_complete: true, + ..CanonicalBodyGcReport::default() + } + ); + let mut rows = store + .conn() + .query("SELECT COUNT(*) FROM session_canonical_bodies", ()) + .await + .unwrap(); + let count: i64 = rows.next().await.unwrap().unwrap().get(0).unwrap(); + assert_eq!(count, 52); +} + +#[asupersync::test] +async fn bootstrap_reports_each_committed_sql_frontier_without_claiming_collection() { + let store = Store::open().await; + let body = StoredCanonicalBody::pack(b"referenced bootstrap progress").unwrap(); + insert_body(store.conn(), &body).await; + store + .conn() + .execute( + "INSERT INTO lcm_raw_messages(storage_kind, content, content_hash) + VALUES ('inline', NULL, ?1)", + params![body.content_hash.as_str()], + ) + .await + .unwrap(); + store + .conn() + .execute( + "INSERT INTO observations(observation_json) VALUES (?1)", + params![observation(&body.content_hash)], + ) + .await + .unwrap(); + store.install().await; + + let frontiers = [ + ("raw:0".to_owned(), "raw:1".to_owned()), + ("observations:0".to_owned(), "observations:1".to_owned()), + ( + "bodies:".to_owned(), + format!("bodies:{}", body.content_hash), + ), + ]; + for (phase, (before, after)) in frontiers.into_iter().enumerate() { + let report = store.tick(1).await; + assert_eq!( + report.bootstrap_frontier_before.as_deref(), + Some(before.as_str()) + ); + assert_eq!( + report.bootstrap_frontier_after.as_deref(), + Some(after.as_str()) + ); + let persisted = read_progress(store.conn()).await.unwrap(); + let stored_after = match phase { + 0 => format!("raw:{}", persisted.raw_through), + 1 => format!("observations:{}", persisted.observations_through), + _ => format!("bodies:{}", persisted.bodies_after), + }; + assert_eq!(stored_after, after, "reported cursor must be durable"); + assert_eq!(report.roots_scanned + report.bodies_scanned, 1); + assert_eq!(report.bodies_deleted, 0); + assert_eq!(report.candidates_discharged, 0); + assert!(report.has_more); + } + let discharged = store.tick(1).await; + assert_eq!(discharged.candidates_discharged, 1); + assert_eq!(discharged.bodies_deleted, 0); + assert!(discharged.bootstrap_frontier_before.is_none()); + assert!(discharged.bootstrap_frontier_after.is_none()); + assert!(!discharged.has_more); + assert!(body_exists(store.conn(), &body.content_hash).await); + assert_eq!( + store.tick(1).await, + CanonicalBodyGcReport { + bootstrap_complete: true, + ..CanonicalBodyGcReport::default() + } + ); +} + +#[asupersync::test] +async fn empty_bootstrap_page_reports_its_committed_frontier_once() { + let store = Store::open().await; + store.conn().execute( + "INSERT INTO lcm_raw_messages(storage_kind, content, content_hash) VALUES ('inline', 'text', 'fixture')", () + ).await.unwrap(); + store.install().await; + store + .conn() + .execute("DELETE FROM lcm_raw_messages", ()) + .await + .unwrap(); + + let page = store.tick(48).await; + assert_eq!((page.roots_scanned, page.bodies_scanned), (0, 0)); + assert_eq!(page.bootstrap_frontier_before.as_deref(), Some("raw:0")); + assert_eq!(page.bootstrap_frontier_after.as_deref(), Some("raw:1")); + assert_eq!(read_progress(store.conn()).await.unwrap().raw_through, 1); + assert!(page.has_committed_work()); + assert!(!store.tick(48).await.has_committed_work()); +} + +#[asupersync::test] +async fn rewritten_bootstrap_checkpoint_refuses_commit_and_recovers_across_48_row_pages() { + let store = Store::open().await; + let shared = StoredCanonicalBody::pack(b"shared canonical body").unwrap(); + insert_body(store.conn(), &shared).await; + for _ in 0..52 { + store + .conn() + .execute( + "INSERT INTO lcm_raw_messages(storage_kind, content, content_hash) + VALUES ('inline', NULL, ?1)", + params![shared.content_hash.as_str()], + ) + .await + .unwrap(); + } + store.install().await; + // Even with recursive triggers enabled, the inner write terminates at zero. + // SQLite reports one updated row for the outer statement, but a plain + // affected-row check would miss the nonadvancing durable frontier. + store + .conn() + .execute_batch( + "CREATE TRIGGER corrupt_gc_bootstrap AFTER UPDATE ON canonical_body_gc_progress + WHEN NEW.raw_through > 0 BEGIN + UPDATE canonical_body_gc_progress SET raw_through = 0 WHERE singleton = 1; + END", + ) + .await + .unwrap(); + let transaction = store + .conn() + .transaction_with_behavior(TransactionBehavior::Immediate) + .await + .unwrap(); + let error = step(&transaction, 48).await.unwrap_err(); + assert!(error.to_string().contains("frontier was not persisted")); + let mut staged = transaction + .query("SELECT COUNT(*) FROM canonical_body_gc_refs", ()) + .await + .unwrap(); + let staged_count: i64 = staged.next().await.unwrap().unwrap().get(0).unwrap(); + assert_eq!( + staged_count, 48, + "the page was enrolled before checkpoint refusal" + ); + drop(staged); + transaction.rollback().await.unwrap(); + assert_eq!(read_progress(store.conn()).await.unwrap().raw_through, 0); + let mut durable = store + .conn() + .query("SELECT COUNT(*) FROM canonical_body_gc_refs", ()) + .await + .unwrap(); + let durable_count: i64 = durable.next().await.unwrap().unwrap().get(0).unwrap(); + assert_eq!(durable_count, 0, "the refused page must roll back its refs"); + drop(durable); + assert!(body_exists(store.conn(), &shared.content_hash).await); + + store + .conn() + .execute_batch("DROP TRIGGER corrupt_gc_bootstrap") + .await + .unwrap(); + let first = store.tick(48).await; + assert_eq!(first.roots_scanned, 48); + assert_eq!(first.bootstrap_frontier_before.as_deref(), Some("raw:0")); + assert_eq!(first.bootstrap_frontier_after.as_deref(), Some("raw:48")); + assert!(first.has_more); + assert_eq!(read_progress(store.conn()).await.unwrap().raw_through, 48); + let second = store.tick(48).await; + assert_eq!(second.roots_scanned, 4); + assert_eq!(second.bootstrap_frontier_before.as_deref(), Some("raw:48")); + assert_eq!(second.bootstrap_frontier_after.as_deref(), Some("raw:52")); + assert_eq!(read_progress(store.conn()).await.unwrap().raw_through, 52); + let (deleted, bytes) = drain(&store, 48).await; + assert_eq!((deleted, bytes), (0, 0), "live bodies remain referenced"); + assert!(body_exists(store.conn(), &shared.content_hash).await); + assert!(!has_candidates(store.conn()).await.unwrap()); + let idle = store.tick(48).await; + assert!(idle.bootstrap_complete); + assert_eq!(idle.roots_scanned, 0); + assert_eq!(idle.bodies_scanned, 0); + assert!(!idle.has_more); +} + +#[asupersync::test] +async fn reinserted_candidate_page_refuses_false_discharge_and_recovers() { + let store = Store::open().await; + for ordinal in 0..52 { + let body = StoredCanonicalBody::pack(format!("retained {ordinal}").as_bytes()).unwrap(); + insert_body(store.conn(), &body).await; + store + .conn() + .execute( + "INSERT INTO lcm_raw_messages(storage_kind, content, content_hash) + VALUES ('inline', NULL, ?1)", + params![body.content_hash.as_str()], + ) + .await + .unwrap(); + } + store.install().await; + let mut bootstrap_finished = false; + for _ in 0..16 { + if store.tick(48).await.bootstrap_complete { + bootstrap_finished = true; + break; + } + } + assert!( + bootstrap_finished, + "candidate collection must become eligible" + ); + assert_eq!(candidate_count(store.conn()).await, 52); + + // The explicit DELETE returns 48 affected rows even when an AFTER DELETE + // trigger restores all of them. Only a same-transaction readback catches + // the false progress that otherwise repeats this exact 48-row page. + store + .conn() + .execute_batch( + "CREATE TRIGGER keep_candidates_after_delete + AFTER DELETE ON canonical_body_gc_candidates BEGIN + INSERT OR IGNORE INTO canonical_body_gc_candidates(content_hash) + VALUES (OLD.content_hash); + END", + ) + .await + .unwrap(); + let transaction = store + .conn() + .transaction_with_behavior(TransactionBehavior::Immediate) + .await + .unwrap(); + let error = step(&transaction, 48).await.unwrap_err(); + assert!( + error + .to_string() + .contains("candidate discharge was not persisted"), + "an apparent 48-row DELETE must not count as committed progress: {error}" + ); + assert_eq!( + candidate_count(&transaction).await, + 52, + "the trigger left every candidate in the still-open transaction" + ); + transaction.rollback().await.unwrap(); + assert_eq!( + candidate_count(store.conn()).await, + 52, + "the refused page must leave all candidate identities durable" + ); + store + .conn() + .execute_batch("DROP TRIGGER keep_candidates_after_delete") + .await + .unwrap(); + + let first = store.tick(48).await; + assert_eq!(first.bodies_scanned, 48); + assert_eq!(first.candidates_discharged, 48); + assert_eq!(first.bodies_deleted, 0); + assert!(first.has_more); + let second = store.tick(48).await; + assert_eq!(second.bodies_scanned, 4); + assert_eq!(second.candidates_discharged, 4); + assert_eq!(second.bodies_deleted, 0); + assert!(!second.has_more); + assert_eq!(candidate_count(store.conn()).await, 0); + let idle = store.tick(48).await; + assert_eq!(idle.bodies_scanned, 0); + assert!(!idle.has_more); +} diff --git a/crates/tracedecay-lcm/src/retention/replay_metadata.rs b/crates/tracedecay-lcm/src/retention/replay_metadata.rs index b7be38726d9..a948c9dfd1b 100644 --- a/crates/tracedecay-lcm/src/retention/replay_metadata.rs +++ b/crates/tracedecay-lcm/src/retention/replay_metadata.rs @@ -5,6 +5,9 @@ //! representations keep their own content. A durable rowid frontier and a byte //! budget keep this repair off the foreground ingest path. +use std::collections::BTreeSet; +use std::time::Instant; + use serde::{Deserialize, Serialize}; use serde_json::Value; use tracedecay_runtime_core::db::Database; @@ -12,6 +15,8 @@ use tracedecay_runtime_core::db::engine::{Executor, QueryExecutor, params}; use crate::{LcmError, LcmRawMessage, gc, payload, raw, schema, security, summary_convergence}; +#[cfg(all(test, target_os = "linux"))] +mod read_cost; #[cfg(test)] mod tests; @@ -19,19 +24,29 @@ const FRONTIER_KEY: &str = "replay_metadata_compaction_store_id_v1"; const OPERATION: &str = "compact duplicate LCM replay metadata"; pub const MAX_REPLAY_METADATA_COMPACTION_ROWS: usize = 48; // Bound both receipt verification and metadata sanitization while a writer is -// held. The raw metadata envelope itself is capped at 1 MiB by ingestion. +// held. Eligibility parsing has its own equal byte budget: ineligible rows +// still cost JSON work. Ingestion caps each metadata envelope at 1 MiB. const MAX_BATCH_BYTES: u64 = 4 * 1024 * 1024; #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] pub struct ReplayMetadataCompactionReport { pub rows_scanned: u64, + /// Wall-clock duration from initial frontier read through commit, not CPU time. + pub elapsed_ms: Option, + /// Transactional store-id frontier consumed by this page, absent on idle reads. + pub frontier_before: Option, + pub frontier_after: Option, pub rows_compacted: u64, - /// Eligible rows retained because their representation, sanitizer result, - /// or byte bound could not prove this narrow lossless transformation. + /// Rows retained because their representation, sanitizer result, or byte + /// bound could not prove this narrow lossless transformation, including + /// rows rejected by the eligibility check before body verification. pub rows_skipped: u64, /// Subset of skipped rows whose metadata plus canonical/stored body /// representations exceed this pass's 4 MiB read and verification budget. pub rows_skipped_oversized: u64, + /// Metadata bytes admitted to SQL eligibility checks, including ineligible + /// rows. Work deferred by the body verification budget is not parsed. + pub metadata_bytes_examined: u64, /// UTF-8 metadata bytes removed. This does not measure file shrinkage. pub metadata_bytes_removed: u64, pub has_more: bool, @@ -70,6 +85,7 @@ pub async fn compact_replay_metadata( database: &Database, max_rows: usize, ) -> Result { + let started = Instant::now(); let read = database.read_connection(); if !has_work(&read, frontier(&read).await?).await? { return Ok(ReplayMetadataCompactionReport::default()); @@ -79,8 +95,9 @@ pub async fn compact_replay_metadata( .await .map_err(db_error)?; match step(&transaction, max_rows).await { - Ok(report) => { + Ok(mut report) => { transaction.commit().await.map_err(db_error)?; + report.elapsed_ms = Some(started.elapsed().as_millis()); Ok(report) } Err(error) => match transaction.rollback().await { @@ -93,39 +110,48 @@ pub async fn compact_replay_metadata( struct Candidate { store_id: i64, eligible: bool, - read_bytes: Option, + oversized: bool, + body_size_known: bool, } -/// Read only sizes and identities until each candidate earns the byte budget. +/// Admit a prefix bounded by both metadata parsing and body-read estimates +/// before testing eligibility. Deferred rows must not be JSON-parsed again. async fn candidates( conn: &(impl QueryExecutor + ?Sized), after: i64, limit: usize, -) -> Result, LcmError> { +) -> Result<(Vec, u64), LcmError> { + // octet_length reads record metadata, unlike length(CAST(text AS BLOB)). + // Even a small scalar after a large body can load its overflow chain: + // defer storage_kind to eligibility and avoid unnecessary canonical joins. let mut rows = conn .query( - "SELECT r.store_id, - CASE WHEN r.storage_kind = 'inline' AND json_valid(r.metadata_json) - THEN json_type(r.metadata_json, '$.lcm_active_replay') = 'true' - AND json_type(r.metadata_json, '$.active_replay.content') = 'text' - ELSE 0 END, - LENGTH(CAST(r.metadata_json AS BLOB)), - COALESCE(LENGTH(CAST(r.content AS BLOB)), b.uncompressed_bytes), + "SELECT r.store_id, octet_length(r.metadata_json), + COALESCE(octet_length(r.content), + CASE WHEN LENGTH(b.body) <= ?3 THEN b.uncompressed_bytes END), COALESCE(LENGTH(b.body), 0) FROM lcm_raw_messages r - LEFT JOIN session_canonical_bodies b ON b.content_hash = r.content_hash + LEFT JOIN session_canonical_bodies b ON b.content_hash = CASE + WHEN r.content IS NULL AND octet_length(r.metadata_json) <= ?3 + THEN r.content_hash END WHERE r.store_id > ?1 ORDER BY r.store_id LIMIT ?2", params![ after, - limit.clamp(1, MAX_REPLAY_METADATA_COMPACTION_ROWS) as i64 + limit.clamp(1, MAX_REPLAY_METADATA_COMPACTION_ROWS) as i64, + MAX_BATCH_BYTES as i64 ], ) .await?; let mut result = Vec::new(); + let mut eligibility_ids = Vec::new(); + let mut remaining = MAX_BATCH_BYTES; + let mut read_remaining = MAX_BATCH_BYTES; while let Some(row) = rows.next().await? { - let metadata: Option = row.get(2)?; - let body: Option = row.get(3)?; - let stored: i64 = row.get(4)?; + let store_id: i64 = row.get(0)?; + let metadata: Option = row.get(1)?; + let body: Option = row.get(2)?; + let stored: i64 = row.get(3)?; + let metadata_bytes = u64::try_from(metadata.unwrap_or(0)).map_err(db_error)?; let read_bytes = metadata .zip(body) .and_then(|(metadata, body)| { @@ -136,62 +162,116 @@ async fn candidates( )) }) .and_then(|(metadata, body, stored)| metadata.checked_add(body)?.checked_add(stored)); + let oversized = metadata_bytes > MAX_BATCH_BYTES + || stored > MAX_BATCH_BYTES as i64 + || read_bytes.is_some_and(|bytes| bytes > MAX_BATCH_BYTES); + if !oversized { + if metadata_bytes > remaining { + break; + } + // Eligibility is not yet known. Charge each potential body read + // before parsing so deferred rows are not revisited for JSON work. + if let Some(bytes) = read_bytes { + if bytes > read_remaining { + break; + } + read_remaining -= bytes; + } + remaining -= metadata_bytes; + if metadata.is_some() { + eligibility_ids.push(store_id); + } + } result.push(Candidate { - store_id: row.get(0)?, - eligible: row.get::>(1)?.unwrap_or(0) != 0, - read_bytes, + store_id, + eligible: false, + oversized, + body_size_known: read_bytes.is_some(), }); } - Ok(result) + drop(rows); + // Resolve the admitted prefix in one query, not one query per raw row. + // Both reads share the writer transaction, so sizes and identities cannot + // change between admission and parsing. Oversized documents stay untouched. + if !eligibility_ids.is_empty() { + let ids = serde_json::to_string(&eligibility_ids).map_err(db_error)?; + let mut rows = conn + .query( + "SELECT store_id FROM lcm_raw_messages + WHERE store_id IN (SELECT value FROM json_each(?1)) + AND CASE WHEN storage_kind = 'inline' AND json_valid(metadata_json) + THEN json_type(metadata_json, '$.lcm_active_replay') = 'true' + AND json_type(metadata_json, '$.active_replay.content') = 'text' + ELSE 0 END", + params![ids], + ) + .await?; + let mut eligible = BTreeSet::new(); + while let Some(row) = rows.next().await? { + eligible.insert(row.get::(0)?); + } + for candidate in &mut result { + candidate.eligible = eligible.contains(&candidate.store_id); + } + } + Ok((result, MAX_BATCH_BYTES - remaining)) } +#[tracing::instrument( + name = "lcm.retention.replay_metadata.page", + level = "trace", + skip_all, + fields(metadata_bytes_examined = tracing::field::Empty), + err +)] async fn step( conn: &(impl Executor + ?Sized), max_rows: usize, ) -> Result { let mut after = frontier(conn).await?; - let candidates = candidates(conn, after, max_rows).await?; - let mut report = ReplayMetadataCompactionReport::default(); - let mut remaining = MAX_BATCH_BYTES; + let before = after; + let (candidates, metadata_bytes_examined) = candidates(conn, after, max_rows).await?; + tracing::Span::current().record("metadata_bytes_examined", metadata_bytes_examined); + let mut report = ReplayMetadataCompactionReport { + metadata_bytes_examined, + ..ReplayMetadataCompactionReport::default() + }; let mut guard_suspended = false; for candidate in candidates { - if candidate.eligible { - let bytes = candidate.read_bytes.ok_or_else(|| { - db_error(format!( + if candidate.oversized { + report.rows_skipped += 1; + report.rows_skipped_oversized += 1; + } else if candidate.eligible { + if !candidate.body_size_known { + return Err(db_error(format!( "raw row {} has no bounded canonical body", candidate.store_id - )) - })?; - if bytes > MAX_BATCH_BYTES { - report.rows_skipped += 1; - report.rows_skipped_oversized += 1; - } else if bytes > remaining { - break; - } else { - remaining -= bytes; - let message = raw::load_raw_message_by_store_id(conn, candidate.store_id).await?; - if let Some(metadata) = compacted_metadata(&message)? { - if !guard_suspended { - summary_convergence::suspend_raw_update_guard(conn).await?; - guard_suspended = true; - } - let changed = - apply_metadata(conn, candidate.store_id, &message, &metadata).await?; - if changed { - report.rows_compacted += 1; - report.metadata_bytes_removed += message - .metadata_json - .as_deref() - .map_or(0, str::len) - .saturating_sub(metadata.len()) - as u64; - } else { - report.rows_skipped += 1; - } + ))); + } + let message = raw::load_raw_message_by_store_id(conn, candidate.store_id).await?; + if let Some(metadata) = compacted_metadata(&message)? { + if !guard_suspended { + summary_convergence::suspend_raw_update_guard(conn).await?; + guard_suspended = true; + } + let changed = apply_metadata(conn, candidate.store_id, &message, &metadata).await?; + if changed { + report.rows_compacted += 1; + report.metadata_bytes_removed += message + .metadata_json + .as_deref() + .map_or(0, str::len) + .saturating_sub(metadata.len()) + as u64; } else { report.rows_skipped += 1; } + } else { + report.rows_skipped += 1; } + } else { + // Eligibility refusal retains the original representation too. + report.rows_skipped += 1; } report.rows_scanned += 1; after = candidate.store_id; @@ -199,7 +279,25 @@ async fn step( if guard_suspended { summary_convergence::restore_raw_update_guard(conn).await?; } - schema::set_gc_meta(conn, FRONTIER_KEY, &after.to_string()).await?; + if report.rows_scanned > 0 { + if after <= before { + return Err(db_error( + "replay metadata page did not advance its frontier", + )); + } + schema::set_gc_meta(conn, FRONTIER_KEY, &after.to_string()).await?; + // An advancing local key is not proof that its checkpoint was stored. + // Verify inside the same writer so a refused or rewritten checkpoint + // rolls back this page instead of committing a false progress report. + let persisted = frontier(conn).await?; + if persisted != after { + return Err(db_error(format!( + "replay metadata frontier was not persisted: expected {after}, found {persisted}" + ))); + } + report.frontier_before = Some(before); + report.frontier_after = Some(persisted); + } report.has_more = has_work(conn, after).await?; Ok(report) } diff --git a/crates/tracedecay-lcm/src/retention/replay_metadata/read_cost.rs b/crates/tracedecay-lcm/src/retention/replay_metadata/read_cost.rs new file mode 100644 index 00000000000..548348e2cd6 --- /dev/null +++ b/crates/tracedecay-lcm/src/retention/replay_metadata/read_cost.rs @@ -0,0 +1,197 @@ +//! Isolate process-wide read accounting from concurrently running unit tests. + +use std::process::Command; + +use tracedecay_privacy::sanitize_lcm_payload_text; +use tracedecay_runtime_core::db::engine::{ + Executor, QueryExecutor, TestConnection, TransactionBehavior, params, +}; + +use super::{MAX_BATCH_BYTES, ReplayMetadataCompactionReport, frontier, step}; +use crate::{raw, retrieval_content::projected_content_hash, schema, summary_convergence}; + +fn read_chars() -> u64 { + std::fs::read_to_string("/proc/self/io") + .unwrap() + .lines() + .find_map(|line| line.strip_prefix("rchar:")) + .unwrap() + .trim() + .parse() + .unwrap() +} + +#[asupersync::test] +async fn oversized_rows_skip_payload_io() { + const CHILD: &str = "TRACEDECAY_REPLAY_READ_COST_CHILD"; + const TEST: &str = "retention::replay_metadata::read_cost::oversized_rows_skip_payload_io"; + if std::env::var_os(CHILD).is_none() { + let output = Command::new(std::env::current_exe().unwrap()) + .args(["--exact", TEST, "--test-threads=1"]) + .env(CHILD, "1") + .output() + .unwrap(); + let stdout = String::from_utf8_lossy(&output.stdout); + assert!( + output.status.success(), + "isolated read-cost test failed:\n{stdout}\n{}", + String::from_utf8_lossy(&output.stderr) + ); + assert!(stdout.contains("1 passed; 0 failed"), "{stdout}"); + return; + } + + let directory = tempfile::tempdir().unwrap(); + let database_path = directory.path().join("replay-read-cost.db"); + let conn = TestConnection::open(&database_path); + conn.execute_batch( + "CREATE TABLE sessions ( + provider TEXT NOT NULL, session_id TEXT NOT NULL, + PRIMARY KEY(provider, session_id) + ); INSERT INTO sessions VALUES ('codex', 'session');", + ) + .await + .unwrap(); + schema::ensure_lcm_schema(&conn).await.unwrap(); + + let small = "verified replay after oversized rows ".repeat(80); + let sanitized = sanitize_lcm_payload_text(&small).unwrap(); + let metadata = serde_json::json!({ + "lcm_active_replay": true, + "active_replay": {"role": "assistant", "content": small}, + "ingest_protection": {"sanitization_receipt": sanitized.receipt()} + }); + let metadata = raw::resanitize_stored_provider_metadata_json(&metadata.to_string()).unwrap(); + let large = "x".repeat(2 * MAX_BATCH_BYTES as usize); + let padded = format!("{metadata}{}", " ".repeat(2 * MAX_BATCH_BYTES as usize)); + let large_hash = projected_content_hash(&large); + let small_hash = projected_content_hash(&small); + conn.execute( + "INSERT INTO session_canonical_bodies(content_hash, encoding, body, uncompressed_bytes) + VALUES (?1, 'identity', ?2, ?3)", + params![large_hash.as_str(), large.as_bytes(), large.len() as i64], + ) + .await + .unwrap(); + // Direct fixture writes represent oversized retained data, not new ingress. + // Each overflow location must be rejected before its bytes are loaded: + // inline body, metadata, then a separately stored canonical body. + let fixtures = [ + ( + 1_i64, + Some(large.as_str()), + large_hash.as_str(), + metadata.as_str(), + ), + ( + 2, + Some(small.as_str()), + small_hash.as_str(), + padded.as_str(), + ), + (3, None, large_hash.as_str(), metadata.as_str()), + ( + 4, + Some(small.as_str()), + small_hash.as_str(), + metadata.as_str(), + ), + ]; + for (id, content, hash, document) in fixtures { + conn.execute( + "INSERT INTO lcm_raw_messages(provider, message_id, session_id, role, ordinal, + content, content_hash, storage_kind, metadata_json) + VALUES ('codex', ?1, 'session', 'assistant', ?2, ?3, ?4, 'inline', ?5)", + params![format!("message-{id}"), id, content, hash, document], + ) + .await + .unwrap(); + } + let original = raw::load_raw_message_by_store_id(&conn, 4).await.unwrap(); + drop(conn); + + for id in 1..=3 { + // Fresh connections preserve the oversized-read measurement without + // issuing PRAGMAs forbidden by the exact-SQL test authority. + let conn = TestConnection::open(&database_path); + let transaction = conn + .transaction_with_behavior(TransactionBehavior::Immediate) + .await + .unwrap(); + let before = read_chars(); + let report = step(&transaction, 1).await.unwrap(); + transaction.commit().await.unwrap(); + let read = read_chars() - before; + assert!( + read < MAX_BATCH_BYTES, + "rejected row {id} read {read} bytes before admission; budget={MAX_BATCH_BYTES}" + ); + assert_eq!(report.rows_scanned, 1); + assert_eq!(report.rows_skipped_oversized, 1); + assert_eq!(report.rows_compacted, 0); + assert_eq!(report.metadata_bytes_examined, 0); + assert_eq!(report.frontier_before, Some(id - 1)); + assert_eq!(report.frontier_after, Some(id)); + assert_eq!(frontier(&conn).await.unwrap(), id); + assert!(report.has_more); + } + + let conn = TestConnection::open(&database_path); + let transaction = conn + .transaction_with_behavior(TransactionBehavior::Immediate) + .await + .unwrap(); + let report = step(&transaction, 48).await.unwrap(); + transaction.commit().await.unwrap(); + assert_eq!(report.rows_scanned, 1); + assert_eq!(report.rows_compacted, 1); + assert!(report.metadata_bytes_removed > 0); + assert_eq!(report.frontier_after, Some(4)); + assert!(!report.has_more); + let repaired = raw::load_raw_message_by_store_id(&conn, 4).await.unwrap(); + assert_eq!(repaired.content, original.content); + assert_eq!( + crate::replay_transactions::raw_replay_message(&repaired), + crate::replay_transactions::raw_replay_message(&original) + ); + summary_convergence::validate_raw_update_guard(&conn) + .await + .unwrap(); + + // Check complete retained bytes outside the measured interval. + for (id, content, _, document) in &fixtures[..3] { + let mut rows = conn + .query( + "SELECT content, metadata_json FROM lcm_raw_messages WHERE store_id = ?1", + params![*id], + ) + .await + .unwrap(); + let row = rows.next().await.unwrap().unwrap(); + assert_eq!(row.get::>(0).unwrap().as_deref(), *content); + assert_eq!(row.get::(1).unwrap(), *document); + } + let mut rows = conn + .query("SELECT body FROM session_canonical_bodies", ()) + .await + .unwrap(); + assert_eq!( + rows.next() + .await + .unwrap() + .unwrap() + .get::>(0) + .unwrap(), + large.as_bytes() + ); + drop(rows); + let transaction = conn + .transaction_with_behavior(TransactionBehavior::Immediate) + .await + .unwrap(); + assert_eq!( + step(&transaction, 48).await.unwrap(), + ReplayMetadataCompactionReport::default() + ); + transaction.commit().await.unwrap(); +} diff --git a/crates/tracedecay-lcm/src/retention/replay_metadata/tests.rs b/crates/tracedecay-lcm/src/retention/replay_metadata/tests.rs index 1bc50de1ce3..fae7245359a 100644 --- a/crates/tracedecay-lcm/src/retention/replay_metadata/tests.rs +++ b/crates/tracedecay-lcm/src/retention/replay_metadata/tests.rs @@ -250,6 +250,8 @@ async fn structured_different_media_and_external_representations_are_retained() let report = store.tick(48).await; assert_eq!(report.rows_scanned, 5); assert_eq!(report.rows_compacted, 0); + assert_eq!(report.rows_skipped, 5); + assert_eq!(report.rows_skipped_oversized, 0); assert_eq!(report.metadata_bytes_removed, 0); // All five provider representations still carry their original content. assert_eq!(scalar(&store.conn, "SELECT COUNT(*) FROM lcm_raw_messages WHERE json_type(metadata_json, '$.active_replay.content') IS NOT NULL").await, 5); @@ -348,14 +350,20 @@ async fn stale_metadata_cannot_be_overwritten_and_missing_guard_refuses_repair() async fn row_and_byte_budgets_advance_a_durable_frontier_and_finished_pages_are_noops() { let store = Store::open().await; let text = "bounded replay text ".repeat(35_000); + let mut metadata_bytes = Vec::new(); for ordinal in 1..=4 { - store + let id = store .insert(ordinal, &text, Value::String(text.clone()), true) .await; + let message = raw::load_raw_message_by_store_id(&store.conn, id) + .await + .unwrap(); + metadata_bytes.push(message.metadata_json.unwrap().len() as u64); } let first = store.tick(1).await; assert_eq!(first.rows_scanned, 1); assert_eq!(first.rows_compacted, 1); + assert_eq!(first.metadata_bytes_examined, metadata_bytes[0]); assert!(first.has_more); assert_eq!(frontier(&store.conn).await.unwrap(), 1); let second = store.tick(48).await; @@ -364,8 +372,339 @@ async fn row_and_byte_budgets_advance_a_durable_frontier_and_finished_pages_are_ "byte cap stops before the third large row" ); assert!(second.has_more); + assert_eq!( + second.metadata_bytes_examined, + metadata_bytes[1..1 + second.rows_scanned as usize] + .iter() + .sum::(), + "the body budget must not pre-parse replay metadata for deferred rows" + ); let last = store.tick(48).await; + assert_eq!( + last.metadata_bytes_examined, + metadata_bytes[1 + second.rows_scanned as usize..] + .iter() + .sum::() + ); assert!(!last.has_more); let idle = store.tick(48).await; assert_eq!(idle, ReplayMetadataCompactionReport::default()); } + +#[asupersync::test] +async fn zero_compaction_pages_advance_distinct_durable_frontiers() { + let store = Store::open().await; + for ordinal in 1..=52 { + store + .insert( + ordinal, + "canonical source", + Value::String("different replay".to_owned()), + false, + ) + .await; + } + let first = store.tick(48).await; + assert_eq!(first.rows_scanned, 48); + assert_eq!(first.rows_compacted, 0); + assert_eq!(first.rows_skipped, 48); + assert_eq!(first.rows_skipped_oversized, 0); + assert_eq!(first.frontier_before, Some(0)); + assert_eq!(first.frontier_after, Some(48)); + assert!(first.has_more); + + let second = store.tick(48).await; + assert_eq!(second.rows_scanned, 4); + assert_eq!(second.rows_compacted, 0); + assert_eq!(second.rows_skipped, 4); + assert_eq!(second.rows_skipped_oversized, 0); + assert_eq!(second.frontier_before, Some(48)); + assert_eq!(second.frontier_after, Some(52)); + assert!(!second.has_more); + assert_eq!(frontier(&store.conn).await.unwrap(), 52); + + assert_eq!( + store.tick(48).await, + ReplayMetadataCompactionReport::default() + ); +} + +#[asupersync::test] +async fn large_ineligible_metadata_is_bounded_before_eligibility_checks() { + let store = Store::open().await; + let structured = serde_json::json!([{ + "type": "text", + "text": "retained structured replay ".repeat(20_000) + }]); + let mut originals = Vec::new(); + for ordinal in 1..=9 { + let id = store + .insert(ordinal, "small verified body", structured.clone(), false) + .await; + originals.push(( + id, + raw::load_raw_message_by_store_id(&store.conn, id) + .await + .unwrap(), + )); + } + let total_metadata_bytes: u64 = originals + .iter() + .map(|(_, message)| message.metadata_json.as_ref().unwrap().len() as u64) + .sum(); + assert!(total_metadata_bytes > MAX_BATCH_BYTES); + assert!( + originals + .iter() + .all(|(_, message)| { message.metadata_json.as_ref().unwrap().len() < 1024 * 1024 }) + ); + + let mut after = 0; + let mut examined = 0; + let mut scanned = 0; + let mut ticks = 0; + for _ in 0..originals.len() { + let report = store.tick(MAX_REPLAY_METADATA_COMPACTION_ROWS).await; + ticks += 1; + assert!(report.rows_scanned > 0); + assert_eq!(report.rows_compacted, 0); + assert_eq!(report.rows_skipped, report.rows_scanned); + assert_eq!(report.rows_skipped_oversized, 0); + assert_eq!(report.metadata_bytes_removed, 0); + assert!(report.metadata_bytes_examined > 0); + assert!(report.metadata_bytes_examined <= MAX_BATCH_BYTES); + assert_eq!(report.frontier_before, Some(after)); + let next = report.frontier_after.unwrap(); + assert!(next > after); + assert_eq!(frontier(&store.conn).await.unwrap(), next); + after = next; + examined += report.metadata_bytes_examined; + scanned += report.rows_scanned; + if !report.has_more { + break; + } + } + assert!(ticks > 1, "ineligible JSON must obey the byte budget too"); + assert_eq!(scanned, originals.len() as u64); + assert_eq!(examined, total_metadata_bytes, "no document is revisited"); + assert_eq!(after, originals.last().unwrap().0); + assert_eq!( + store.tick(MAX_REPLAY_METADATA_COMPACTION_ROWS).await, + ReplayMetadataCompactionReport::default() + ); + for (id, original) in originals { + assert_eq!( + raw::load_raw_message_by_store_id(&store.conn, id) + .await + .unwrap(), + original, + "bounded scanning must preserve the body, replay and receipts" + ); + } +} + +#[asupersync::test] +async fn oversized_metadata_is_preserved_without_blocking_later_compaction() { + let store = Store::open().await; + let text = "verified duplicate content ".repeat(80); + let oversized_id = store + .insert(1, &text, Value::String(text.clone()), false) + .await; + let next_id = store + .insert(2, &text, Value::String(text.clone()), false) + .await; + // Bypass ingress only to exercise a pre-existing oversized stored document. + // The repair must not parse or rewrite it, nor pin the durable frontier. + let oversized = serde_json::json!({ + "lcm_active_replay": true, + "active_replay": {"content": text}, + "padding": "x".repeat(MAX_BATCH_BYTES as usize) + }) + .to_string(); + store + .conn + .execute( + "UPDATE lcm_raw_messages SET metadata_json = ?1 WHERE store_id = ?2", + params![oversized.as_str(), oversized_id], + ) + .await + .unwrap(); + let report = store.tick(MAX_REPLAY_METADATA_COMPACTION_ROWS).await; + assert_eq!(report.rows_scanned, 2); + assert_eq!(report.rows_skipped, 1); + assert_eq!(report.rows_skipped_oversized, 1); + assert_eq!(report.rows_compacted, 1); + assert!(report.metadata_bytes_examined > 0); + assert!(report.metadata_bytes_examined < MAX_BATCH_BYTES); + assert_eq!(report.frontier_after, Some(next_id)); + assert_eq!(frontier(&store.conn).await.unwrap(), next_id); + assert!(!report.has_more); + let mut rows = store + .conn + .query( + "SELECT metadata_json FROM lcm_raw_messages WHERE store_id = ?1", + params![oversized_id], + ) + .await + .unwrap(); + assert_eq!( + rows.next() + .await + .unwrap() + .unwrap() + .get::(0) + .unwrap(), + oversized + ); + drop(rows); + assert_eq!( + store.tick(MAX_REPLAY_METADATA_COMPACTION_ROWS).await, + ReplayMetadataCompactionReport::default() + ); + summary_convergence::validate_raw_update_guard(&store.conn) + .await + .unwrap(); +} + +async fn assert_checkpoint_refusal_is_atomic(store: &Store, id: i64, trigger_sql: &str) { + let checkpoint = schema::get_gc_meta(&store.conn, FRONTIER_KEY) + .await + .unwrap(); + let before = frontier(&store.conn).await.unwrap(); + let original = raw::load_raw_message_by_store_id(&store.conn, id) + .await + .unwrap(); + assert!(compacted_metadata(&original).unwrap().is_some()); + let revision = scalar( + &store.conn, + "SELECT raw_revision_generation FROM lcm_summary_convergence_queue", + ) + .await; + store.conn.execute_batch(trigger_sql).await.unwrap(); + let transaction = store + .conn + .transaction_with_behavior(TransactionBehavior::Immediate) + .await + .unwrap(); + let error = step(&transaction, 48).await.unwrap_err(); + assert!(error.to_string().contains("frontier was not persisted")); + let staged = raw::load_raw_message_by_store_id(&transaction, id) + .await + .unwrap(); + assert_ne!(staged.metadata_json, original.metadata_json); + assert!(metadata(&staged)["active_replay"].get("content").is_none()); + transaction.rollback().await.unwrap(); + + assert_eq!( + schema::get_gc_meta(&store.conn, FRONTIER_KEY) + .await + .unwrap(), + checkpoint, + "rollback preserves even an absent checkpoint" + ); + assert_eq!( + raw::load_raw_message_by_store_id(&store.conn, id) + .await + .unwrap(), + original, + "a refused checkpoint must restore the complete message and receipts" + ); + assert_eq!( + scalar( + &store.conn, + "SELECT raw_revision_generation FROM lcm_summary_convergence_queue" + ) + .await, + revision + ); + summary_convergence::validate_raw_update_guard(&store.conn) + .await + .unwrap(); + + store + .conn + .execute_batch("DROP TRIGGER refuse_replay_checkpoint") + .await + .unwrap(); + let recovered = store.tick(48).await; + assert_eq!(recovered.rows_scanned, 1); + assert_eq!(recovered.rows_compacted, 1); + assert!(recovered.metadata_bytes_removed > 0); + assert_eq!(recovered.frontier_before, Some(before)); + assert_eq!(recovered.frontier_after, Some(id)); + assert_eq!(frontier(&store.conn).await.unwrap(), id); + assert!(!recovered.has_more); + let repaired = raw::load_raw_message_by_store_id(&store.conn, id) + .await + .unwrap(); + assert_eq!(repaired.content, original.content); + assert_eq!(repaired.content_hash, original.content_hash); + assert_eq!( + crate::replay_transactions::raw_replay_message(&repaired), + crate::replay_transactions::raw_replay_message(&original) + ); + assert_eq!( + store.tick(48).await, + ReplayMetadataCompactionReport::default() + ); +} + +#[asupersync::test] +async fn ignored_initial_checkpoint_rolls_back_compaction() { + let store = Store::open().await; + let text = "verified replay checkpoint content ".repeat(80); + let id = store + .insert(1, &text, Value::String(text.clone()), false) + .await; + let trigger = format!( + "CREATE TRIGGER refuse_replay_checkpoint BEFORE INSERT ON lcm_gc_meta + WHEN NEW.key = '{FRONTIER_KEY}' BEGIN SELECT RAISE(IGNORE); END" + ); + assert_checkpoint_refusal_is_atomic(&store, id, &trigger).await; +} + +#[asupersync::test] +async fn rewritten_checkpoint_cannot_claim_durable_progress() { + let store = Store::open().await; + let text = "verified replay checkpoint content ".repeat(80); + let previous_id = store + .insert(1, &text, Value::String(text.clone()), false) + .await; + let first = store.tick(48).await; + assert_eq!(first.rows_compacted, 1); + assert_eq!(first.frontier_after, Some(previous_id)); + let id = store + .insert(2, &text, Value::String(text.clone()), false) + .await; + // A successful INSERT can still leave a different checkpoint behind. + // Counting affected rows alone cannot detect this failure. + let trigger = format!( + "CREATE TRIGGER refuse_replay_checkpoint AFTER INSERT ON lcm_gc_meta + WHEN NEW.key = '{FRONTIER_KEY}' BEGIN + UPDATE lcm_gc_meta SET value = '{previous_id}' WHERE key = NEW.key; + END" + ); + assert_checkpoint_refusal_is_atomic(&store, id, &trigger).await; +} + +#[asupersync::test] +async fn idle_page_does_not_initialize_a_checkpoint() { + let store = Store::open().await; + let trigger = format!( + "CREATE TRIGGER refuse_replay_checkpoint BEFORE INSERT ON lcm_gc_meta + WHEN NEW.key = '{FRONTIER_KEY}' BEGIN + SELECT RAISE(ABORT, 'idle page attempted a checkpoint write'); + END" + ); + store.conn.execute_batch(&trigger).await.unwrap(); + assert_eq!( + store.tick(48).await, + ReplayMetadataCompactionReport::default() + ); + assert_eq!( + schema::get_gc_meta(&store.conn, FRONTIER_KEY) + .await + .unwrap(), + None + ); +} diff --git a/crates/tracedecay/src/daemon/maintenance.rs b/crates/tracedecay/src/daemon/maintenance.rs index ea2070195f9..a7c1dce45d1 100644 --- a/crates/tracedecay/src/daemon/maintenance.rs +++ b/crates/tracedecay/src/daemon/maintenance.rs @@ -267,8 +267,10 @@ async fn run_registered_store_retention( Ok(_) => {} } log_payload_gc_outcome(&report.payload_gc); - let canonical_body_outcome = record_canonical_body_gc_outcome(&report.canonical_body_gc); - let replay_metadata_outcome = record_replay_metadata_outcome(&report.replay_metadata); + let canonical_body_outcome = + record_canonical_body_gc_outcome(database.db_path(), &report.canonical_body_gc); + let replay_metadata_outcome = + record_replay_metadata_outcome(database.db_path(), &report.replay_metadata); let outcome = if report.succeeded() { session_storage_outcome .combine(canonical_body_outcome) @@ -368,11 +370,18 @@ async fn run_registered_storage_continuation( ) .await; record_session_lcm_outcome(&session_lcm) - .combine(record_canonical_body_gc_outcome(&canonical_body_gc)) - .combine(record_replay_metadata_outcome(&replay_metadata)) + .combine(record_canonical_body_gc_outcome( + database.db_path(), + &canonical_body_gc, + )) + .combine(record_replay_metadata_outcome( + database.db_path(), + &replay_metadata, + )) } fn record_canonical_body_gc_outcome( + store: &Path, outcome: &Result< tracedecay_global_db::observation::CanonicalBodyGcReport, tracedecay_maintenance::retention::registered_store::RegisteredStoreRetentionErrorV1, @@ -380,12 +389,39 @@ fn record_canonical_body_gc_outcome( ) -> MaintenanceTickOutcome { match outcome { Ok(report) => { - if report.roots_scanned > 0 || report.bodies_scanned > 0 { + if report.has_committed_work() { log_daemon_event( "retention_canonical_bodies", &[ + ("store", store.display().to_string()), ("roots_scanned", report.roots_scanned.to_string()), ("bodies_scanned", report.bodies_scanned.to_string()), + ( + "bootstrap_frontier_before", + report + .bootstrap_frontier_before + .as_deref() + .unwrap_or("not_scanned") + .to_owned(), + ), + ( + "bootstrap_frontier_after", + report + .bootstrap_frontier_after + .as_deref() + .unwrap_or("not_scanned") + .to_owned(), + ), + ( + "elapsed_ms", + report + .elapsed_ms + .map_or_else(|| "unmeasured".to_owned(), |value| value.to_string()), + ), + ( + "candidates_discharged", + report.candidates_discharged.to_string(), + ), ("bodies_deleted", report.bodies_deleted.to_string()), ( "stored_bytes_deleted", @@ -407,6 +443,7 @@ fn record_canonical_body_gc_outcome( "retention_degraded", &[ ("pass", "canonical_body_gc".to_owned()), + ("store", store.display().to_string()), ("failure", error.diagnostic().to_owned()), ], ); @@ -416,6 +453,7 @@ fn record_canonical_body_gc_outcome( } fn record_replay_metadata_outcome( + store: &Path, outcome: &Result< tracedecay_lcm::retention::ReplayMetadataCompactionReport, tracedecay_maintenance::retention::registered_store::RegisteredStoreRetentionErrorV1, @@ -427,7 +465,32 @@ fn record_replay_metadata_outcome( log_daemon_event( "retention_replay_metadata", &[ + ("store", store.display().to_string()), ("rows_scanned", report.rows_scanned.to_string()), + ( + "elapsed_ms", + report + .elapsed_ms + .map_or_else(|| "unmeasured".to_owned(), |value| value.to_string()), + ), + ( + "metadata_bytes_examined", + report.metadata_bytes_examined.to_string(), + ), + ( + "frontier_before", + report.frontier_before.map_or_else( + || "not_scanned".to_owned(), + |value| value.to_string(), + ), + ), + ( + "frontier_after", + report.frontier_after.map_or_else( + || "not_scanned".to_owned(), + |value| value.to_string(), + ), + ), ("rows_compacted", report.rows_compacted.to_string()), ("rows_skipped", report.rows_skipped.to_string()), ( @@ -453,6 +516,7 @@ fn record_replay_metadata_outcome( "retention_degraded", &[ ("pass", "replay_metadata_compaction".to_owned()), + ("store", store.display().to_string()), ("failure", error.diagnostic().to_owned()), ], ); @@ -2019,4 +2083,82 @@ mod tests { .expect("native runtime"); assert_eq!(crate::test_support::native_runtime_hazards(), hazards); } + + /// Two mounted databases may commit identical 48-row pages in one tick. + /// Operator logs must not conflate their independent SQL frontiers. + #[test] + fn retention_progress_logs_distinguish_mounted_stores() { + const CHILD: &str = "TRACEDECAY_RETENTION_STORE_LOG_CHILD"; + const TEST: &str = + "daemon::maintenance::tests::retention_progress_logs_distinguish_mounted_stores"; + let stores = ["/project-a/sessions.db", "/project-b/sessions.db"]; + + if std::env::var_os(CHILD).is_some() { + for store in stores { + let canonical = tracedecay_global_db::observation::CanonicalBodyGcReport { + roots_scanned: 48, + bootstrap_frontier_before: Some("raw:0".to_owned()), + bootstrap_frontier_after: Some("raw:48".to_owned()), + has_more: true, + ..Default::default() + }; + let _ = super::record_canonical_body_gc_outcome(Path::new(store), &Ok(canonical)); + let replay = tracedecay_lcm::retention::ReplayMetadataCompactionReport { + rows_scanned: 48, + rows_skipped: 48, + frontier_before: Some(0), + frontier_after: Some(48), + has_more: true, + ..Default::default() + }; + let _ = super::record_replay_metadata_outcome(Path::new(store), &Ok(replay)); + } + return; + } + + // Capture the actual operator stderr lines in a child process. The + // normal daemon logger deliberately bypasses the test output capture. + let output = std::process::Command::new(std::env::current_exe().unwrap()) + .args(["--exact", TEST, "--nocapture", "--test-threads=1"]) + .env(CHILD, "1") + .output() + .expect("spawn isolated operator log test"); + let stdout = String::from_utf8_lossy(&output.stdout); + let stderr = String::from_utf8_lossy(&output.stderr); + assert!( + output.status.success() && stdout.contains("1 passed; 0 failed"), + "isolated test did not execute:\n{stdout}\n{stderr}" + ); + + let lines = stderr + .lines() + .filter(|line| { + line.contains("event=retention_canonical_bodies ") + || line.contains("event=retention_replay_metadata ") + }) + .collect::>(); + assert_eq!(lines.len(), 4, "expected two events per store: {stderr}"); + for store in stores { + for (event, frontier) in [ + ( + "retention_canonical_bodies", + "bootstrap_frontier_after=raw:48", + ), + ("retention_replay_metadata", "frontier_after=48"), + ] { + let count = lines + .iter() + .filter(|line| { + line.contains(&format!("event={event} ")) + && line.contains(&format!("store={store} ")) + && line.contains(frontier) + }) + .count(); + assert_eq!( + count, 1, + "missing distinct {event} page for {store}: {stderr}" + ); + } + } + } }