From 8db58ec7ddbfbe29d295404cb37c84c4b2b458d0 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sat, 7 Nov 2026 16:58:05 -0800 Subject: [PATCH 01/27] fix(retention): verify and expose transactional GC progress Report candidate discharge and replay SQL frontiers for each scanned page. Reject incomplete candidate discharge or nonadvancing replay pages. Add >48-row behavioral regressions for zero-reclamation backlogs. Refs #4787 --- .../src/observation/canonical_body_gc.rs | 28 ++++++++--- .../observation/canonical_body_gc/tests.rs | 49 +++++++++++++++++++ .../src/retention/replay_metadata.rs | 13 +++++ .../src/retention/replay_metadata/tests.rs | 34 +++++++++++++ crates/tracedecay/src/daemon/maintenance.rs | 18 +++++++ 5 files changed, 135 insertions(+), 7 deletions(-) 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..7b5c0b056f4 100644 --- a/crates/tracedecay-global-db/src/observation/canonical_body_gc.rs +++ b/crates/tracedecay-global-db/src/observation/canonical_body_gc.rs @@ -25,6 +25,8 @@ pub const MAX_CANONICAL_BODY_GC_ROWS: usize = 48; pub struct CanonicalBodyGcReport { pub roots_scanned: u64, pub bodies_scanned: u64, + /// Candidates discharged in the committed page, even if their body stays live. + pub candidates_discharged: u64, pub bodies_deleted: u64, /// Stored BLOB bytes deleted, including the effect of their existing /// compression. This is not database-file or filesystem space reclaimed. @@ -323,13 +325,25 @@ async fn collect_candidates( drop(deleted); // 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], + ) + .await + .map_err(db_error)?; + 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/tests.rs b/crates/tracedecay-global-db/src/observation/canonical_body_gc/tests.rs index 9da4f654323..9b2accfd09e 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 @@ -506,3 +506,52 @@ async fn nonpositive_legacy_owner_ids_cannot_be_skipped_by_bootstrap() { ); } } + +#[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); +} diff --git a/crates/tracedecay-lcm/src/retention/replay_metadata.rs b/crates/tracedecay-lcm/src/retention/replay_metadata.rs index b7be38726d9..72910e4798a 100644 --- a/crates/tracedecay-lcm/src/retention/replay_metadata.rs +++ b/crates/tracedecay-lcm/src/retention/replay_metadata.rs @@ -25,6 +25,9 @@ const MAX_BATCH_BYTES: u64 = 4 * 1024 * 1024; #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] pub struct ReplayMetadataCompactionReport { pub rows_scanned: u64, + /// 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. @@ -150,6 +153,7 @@ async fn step( max_rows: usize, ) -> Result { let mut after = frontier(conn).await?; + let before = after; let candidates = candidates(conn, after, max_rows).await?; let mut report = ReplayMetadataCompactionReport::default(); let mut remaining = MAX_BATCH_BYTES; @@ -199,6 +203,15 @@ async fn step( if guard_suspended { summary_convergence::restore_raw_update_guard(conn).await?; } + if report.rows_scanned > 0 { + if after <= before { + return Err(db_error( + "replay metadata page did not advance its frontier", + )); + } + report.frontier_before = Some(before); + report.frontier_after = Some(after); + } schema::set_gc_meta(conn, FRONTIER_KEY, &after.to_string()).await?; report.has_more = has_work(conn, after).await?; Ok(report) diff --git a/crates/tracedecay-lcm/src/retention/replay_metadata/tests.rs b/crates/tracedecay-lcm/src/retention/replay_metadata/tests.rs index 1bc50de1ce3..38a405796cb 100644 --- a/crates/tracedecay-lcm/src/retention/replay_metadata/tests.rs +++ b/crates/tracedecay-lcm/src/retention/replay_metadata/tests.rs @@ -369,3 +369,37 @@ async fn row_and_byte_budgets_advance_a_durable_frontier_and_finished_pages_are_ 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.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.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() + ); +} diff --git a/crates/tracedecay/src/daemon/maintenance.rs b/crates/tracedecay/src/daemon/maintenance.rs index ea2070195f9..9e293532be4 100644 --- a/crates/tracedecay/src/daemon/maintenance.rs +++ b/crates/tracedecay/src/daemon/maintenance.rs @@ -386,6 +386,10 @@ fn record_canonical_body_gc_outcome( &[ ("roots_scanned", report.roots_scanned.to_string()), ("bodies_scanned", report.bodies_scanned.to_string()), + ( + "candidates_discharged", + report.candidates_discharged.to_string(), + ), ("bodies_deleted", report.bodies_deleted.to_string()), ( "stored_bytes_deleted", @@ -428,6 +432,20 @@ fn record_replay_metadata_outcome( "retention_replay_metadata", &[ ("rows_scanned", report.rows_scanned.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()), ( From 001b1abe47f636b05ed95c964e3cbd6f6ab5cdee Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sat, 7 Nov 2026 17:36:01 -0800 Subject: [PATCH 02/27] fix(retention): bound replay eligibility parsing before scanning Admit a size-bounded metadata prefix before running SQL JSON eligibility checks, including rows that cannot be compacted. Resolve admitted IDs in one query under the existing writer transaction. Preserve oversized metadata while advancing past it without parsing or blocking later rows. Expose examined metadata bytes through the page report and tracing span. Add database-backed regressions for large ineligible backlogs and an oversized stored document followed by valid compactable work. Refs #4787 --- .../src/retention/replay_metadata.rs | 95 ++++++++++--- .../src/retention/replay_metadata/tests.rs | 129 ++++++++++++++++++ 2 files changed, 208 insertions(+), 16 deletions(-) diff --git a/crates/tracedecay-lcm/src/retention/replay_metadata.rs b/crates/tracedecay-lcm/src/retention/replay_metadata.rs index 72910e4798a..da43f273f78 100644 --- a/crates/tracedecay-lcm/src/retention/replay_metadata.rs +++ b/crates/tracedecay-lcm/src/retention/replay_metadata.rs @@ -5,6 +5,8 @@ //! 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 serde::{Deserialize, Serialize}; use serde_json::Value; use tracedecay_runtime_core::db::Database; @@ -19,7 +21,8 @@ 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)] @@ -29,12 +32,15 @@ pub struct ReplayMetadataCompactionReport { 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. 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 parsing, including ineligible + /// rows and any prefetched rows deferred by the separate body byte budget. + 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, @@ -96,22 +102,20 @@ pub async fn compact_replay_metadata( struct Candidate { store_id: i64, eligible: bool, + metadata_oversized: bool, read_bytes: Option, } -/// Read only sizes and identities until each candidate earns the byte budget. +/// Admit a size-bounded prefix before parsing any metadata. Eligibility must +/// not parse all 48 documents before the body's separate byte budget is applied. async fn candidates( conn: &(impl QueryExecutor + ?Sized), after: i64, limit: usize, -) -> Result, LcmError> { +) -> Result<(Vec, u64), LcmError> { 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, + "SELECT r.store_id, r.storage_kind = 'inline', LENGTH(CAST(r.metadata_json AS BLOB)), COALESCE(LENGTH(CAST(r.content AS BLOB)), b.uncompressed_bytes), COALESCE(LENGTH(b.body), 0) @@ -125,10 +129,29 @@ async fn candidates( ) .await?; let mut result = Vec::new(); + let mut eligibility_ids = Vec::new(); + let mut remaining = MAX_BATCH_BYTES; while let Some(row) = rows.next().await? { + let store_id: i64 = row.get(0)?; + let inline = row.get::(1)? != 0; let metadata: Option = row.get(2)?; let body: Option = row.get(3)?; let stored: i64 = row.get(4)?; + let metadata_bytes = if inline { + u64::try_from(metadata.unwrap_or(0)).map_err(db_error)? + } else { + 0 + }; + let metadata_oversized = metadata_bytes > MAX_BATCH_BYTES; + if !metadata_oversized { + if metadata_bytes > remaining { + break; + } + remaining -= metadata_bytes; + if inline && metadata.is_some() { + eligibility_ids.push(store_id); + } + } let read_bytes = metadata .zip(body) .and_then(|(metadata, body)| { @@ -140,26 +163,66 @@ async fn candidates( }) .and_then(|(metadata, body, stored)| metadata.checked_add(body)?.checked_add(stored)); result.push(Candidate { - store_id: row.get(0)?, - eligible: row.get::>(1)?.unwrap_or(0) != 0, + store_id, + eligible: false, + metadata_oversized, read_bytes, }); } - 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 before = after; - let candidates = candidates(conn, after, max_rows).await?; - let mut report = ReplayMetadataCompactionReport::default(); + 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 remaining = MAX_BATCH_BYTES; let mut guard_suspended = false; for candidate in candidates { - if candidate.eligible { + if candidate.metadata_oversized { + report.rows_skipped += 1; + report.rows_skipped_oversized += 1; + } else if candidate.eligible { let bytes = candidate.read_bytes.ok_or_else(|| { db_error(format!( "raw row {} has no bounded canonical body", diff --git a/crates/tracedecay-lcm/src/retention/replay_metadata/tests.rs b/crates/tracedecay-lcm/src/retention/replay_metadata/tests.rs index 38a405796cb..a9b39a4c727 100644 --- a/crates/tracedecay-lcm/src/retention/replay_metadata/tests.rs +++ b/crates/tracedecay-lcm/src/retention/replay_metadata/tests.rs @@ -403,3 +403,132 @@ async fn zero_compaction_pages_advance_distinct_durable_frontiers() { 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, 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(); +} From 2a42006eef255c577137620543f9fa25cdbe96c0 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sat, 7 Nov 2026 18:18:36 -0800 Subject: [PATCH 03/27] style(retention): satisfy Bazel rustfmt for replay regressions Apply only the two exact Rustfmt changes identified by the blocked repository formatting gate on the last PR head. Refs #4787 --- .../src/retention/replay_metadata/tests.rs | 15 +++++++++++---- 1 file changed, 11 insertions(+), 4 deletions(-) diff --git a/crates/tracedecay-lcm/src/retention/replay_metadata/tests.rs b/crates/tracedecay-lcm/src/retention/replay_metadata/tests.rs index a9b39a4c727..204b56de215 100644 --- a/crates/tracedecay-lcm/src/retention/replay_metadata/tests.rs +++ b/crates/tracedecay-lcm/src/retention/replay_metadata/tests.rs @@ -428,9 +428,11 @@ async fn large_ineligible_metadata_is_bounded_before_eligibility_checks() { .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 - })); + assert!( + originals + .iter() + .all(|(_, message)| { message.metadata_json.as_ref().unwrap().len() < 1024 * 1024 }) + ); let mut after = 0; let mut examined = 0; @@ -520,7 +522,12 @@ async fn oversized_metadata_is_preserved_without_blocking_later_compaction() { .await .unwrap(); assert_eq!( - rows.next().await.unwrap().unwrap().get::(0).unwrap(), + rows.next() + .await + .unwrap() + .unwrap() + .get::(0) + .unwrap(), oversized ); drop(rows); From 91ffd680d7ec128dbf5cd14d9bcc721b4e457a21 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sat, 7 Nov 2026 19:20:15 -0800 Subject: [PATCH 04/27] fix(retention): verify stored replay checkpoints before commit Read back the replay frontier in the page's writer transaction before reporting progress. Refuse ignored or rewritten checkpoint writes so the existing transaction owner rolls back metadata changes rather than committing a page that will be visited again. Idle pages do not write. Add real database regressions for an ignored initial checkpoint, a rewritten existing checkpoint, rollback and successful recovery, and write-free idle pages. Preserve existing parsing and body budgets. Refs #4787 --- .../src/retention/replay_metadata.rs | 13 +- .../src/retention/replay_metadata/tests.rs | 143 ++++++++++++++++++ 2 files changed, 154 insertions(+), 2 deletions(-) diff --git a/crates/tracedecay-lcm/src/retention/replay_metadata.rs b/crates/tracedecay-lcm/src/retention/replay_metadata.rs index da43f273f78..6beaefecd52 100644 --- a/crates/tracedecay-lcm/src/retention/replay_metadata.rs +++ b/crates/tracedecay-lcm/src/retention/replay_metadata.rs @@ -272,10 +272,19 @@ async fn step( "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(after); + report.frontier_after = Some(persisted); } - schema::set_gc_meta(conn, FRONTIER_KEY, &after.to_string()).await?; report.has_more = has_work(conn, after).await?; Ok(report) } diff --git a/crates/tracedecay-lcm/src/retention/replay_metadata/tests.rs b/crates/tracedecay-lcm/src/retention/replay_metadata/tests.rs index 204b56de215..23a0b3962af 100644 --- a/crates/tracedecay-lcm/src/retention/replay_metadata/tests.rs +++ b/crates/tracedecay-lcm/src/retention/replay_metadata/tests.rs @@ -539,3 +539,146 @@ async fn oversized_metadata_is_preserved_without_blocking_later_compaction() { .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 + ); +} From ad3729b860d9331d5913c41edc1a479bf3a18f08 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sat, 7 Nov 2026 19:44:16 -0800 Subject: [PATCH 05/27] fix(retention): verify canonical GC bootstrap checkpoints Refuse an unchanged bootstrap page or a checkpoint that did not persist, even when its SQL UPDATE reports one affected row. Roll back enrolled refs and seed writes through the existing writer transaction on refusal. Add a database-backed 52-owner regression proving repeated 48-row pages fail closed when a trigger rewrites their frontier, then resume through 48+4 pages and quiesce without collecting the live shared body. Refs #4787 --- .../src/observation/canonical_body_gc.rs | 15 ++++ .../observation/canonical_body_gc/tests.rs | 79 +++++++++++++++++++ 2 files changed, 94 insertions(+) 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 7b5c0b056f4..819fa15afcf 100644 --- a/crates/tracedecay-global-db/src/observation/canonical_body_gc.rs +++ b/crates/tracedecay-global-db/src/observation/canonical_body_gc.rs @@ -36,6 +36,7 @@ pub struct CanonicalBodyGcReport { pub has_more: bool, } +#[derive(Clone, Debug, PartialEq, Eq)] struct Progress { raw_through: i64, raw_until: i64, @@ -120,6 +121,14 @@ 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(()) } @@ -146,6 +155,7 @@ async fn step(conn: &(impl Executor + ?Sized), max_rows: usize) -> Result Result 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!(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!(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); +} From 21032e3de2e90d2860d8d73b8e10e3c8faf6ffd7 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sat, 7 Nov 2026 20:24:55 -0800 Subject: [PATCH 06/27] style(retention): satisfy Bazel rustfmt for GC regression Apply the exact formatting required by the failing Bazel rustfmt gate in the new canonical-body frontier rollback regression. Refs #4787 --- .../src/observation/canonical_body_gc/tests.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) 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 9c405001fa8..a81eb82aeef 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 @@ -598,7 +598,10 @@ async fn rewritten_bootstrap_checkpoint_refuses_commit_and_recovers_across_48_ro .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"); + 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); From 8e044d468ac8cf755752792ca9f7ef7212f77578 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sat, 7 Nov 2026 21:30:02 -0800 Subject: [PATCH 07/27] perf(retention): avoid repeated replay eligibility parsing Bound replay eligibility admission by the conservative body verification budget before parsing metadata. Deferred rows no longer incur repeated JSON validation on successive ticks. Assert exactly-once metadata admission across database-backed multi-page body-budget regressions. Refs #4787 --- .../src/retention/replay_metadata.rs | 35 ++++++++++++------- .../src/retention/replay_metadata/tests.rs | 21 ++++++++++- 2 files changed, 43 insertions(+), 13 deletions(-) diff --git a/crates/tracedecay-lcm/src/retention/replay_metadata.rs b/crates/tracedecay-lcm/src/retention/replay_metadata.rs index 6beaefecd52..6e435e32ce5 100644 --- a/crates/tracedecay-lcm/src/retention/replay_metadata.rs +++ b/crates/tracedecay-lcm/src/retention/replay_metadata.rs @@ -39,7 +39,7 @@ pub struct ReplayMetadataCompactionReport { /// representations exceed this pass's 4 MiB read and verification budget. pub rows_skipped_oversized: u64, /// Metadata bytes admitted to SQL eligibility parsing, including ineligible - /// rows and any prefetched rows deferred by the separate body byte budget. + /// 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, @@ -106,8 +106,8 @@ struct Candidate { read_bytes: Option, } -/// Admit a size-bounded prefix before parsing any metadata. Eligibility must -/// not parse all 48 documents before the body's separate byte budget is applied. +/// 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, @@ -131,6 +131,7 @@ async fn candidates( 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 store_id: i64 = row.get(0)?; let inline = row.get::(1)? != 0; @@ -143,15 +144,6 @@ async fn candidates( 0 }; let metadata_oversized = metadata_bytes > MAX_BATCH_BYTES; - if !metadata_oversized { - if metadata_bytes > remaining { - break; - } - remaining -= metadata_bytes; - if inline && metadata.is_some() { - eligibility_ids.push(store_id); - } - } let read_bytes = metadata .zip(body) .and_then(|(metadata, body)| { @@ -162,6 +154,25 @@ async fn candidates( )) }) .and_then(|(metadata, body, stored)| metadata.checked_add(body)?.checked_add(stored)); + if !metadata_oversized { + if metadata_bytes > remaining { + break; + } + // Eligibility is not yet known. Charge every inline row against + // the verification budget so deferred rows are not JSON-parsed + // again on the next tick. Individually oversized rows still + // advance without loading their bodies. + if inline && let Some(bytes) = read_bytes && bytes <= MAX_BATCH_BYTES { + if bytes > read_remaining { + break; + } + read_remaining -= bytes; + } + remaining -= metadata_bytes; + if inline && metadata.is_some() { + eligibility_ids.push(store_id); + } + } result.push(Candidate { store_id, eligible: false, diff --git a/crates/tracedecay-lcm/src/retention/replay_metadata/tests.rs b/crates/tracedecay-lcm/src/retention/replay_metadata/tests.rs index 23a0b3962af..5c794d2ed99 100644 --- a/crates/tracedecay-lcm/src/retention/replay_metadata/tests.rs +++ b/crates/tracedecay-lcm/src/retention/replay_metadata/tests.rs @@ -348,14 +348,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,7 +370,20 @@ 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()); From a1b2bcf0572bd0b654ae10983329469b776c0238 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sat, 7 Nov 2026 22:08:31 -0800 Subject: [PATCH 08/27] feat(retention): record committed page latency and replay parsing work Measure each canonical-body GC and replay-metadata pass from admission through successful writer commit. Include elapsed milliseconds and admitted metadata bytes in the existing daemon progress events so an affected host can distinguish stalled page retries from costly but advancing backlogs. Preserve the transactional progress guards. Refs #4787 --- .../src/observation/canonical_body_gc.rs | 8 +++++++- .../src/retention/replay_metadata.rs | 7 ++++++- crates/tracedecay/src/daemon/maintenance.rs | 18 ++++++++++++++++++ 3 files changed, 31 insertions(+), 2 deletions(-) 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 819fa15afcf..34dbdf30ad2 100644 --- a/crates/tracedecay-global-db/src/observation/canonical_body_gc.rs +++ b/crates/tracedecay-global-db/src/observation/canonical_body_gc.rs @@ -6,6 +6,8 @@ //! 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; @@ -25,6 +27,8 @@ 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, pub bodies_deleted: u64, @@ -134,10 +138,12 @@ async fn save_progress(conn: &(impl Executor + ?Sized), progress: &Progress) -> #[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 { diff --git a/crates/tracedecay-lcm/src/retention/replay_metadata.rs b/crates/tracedecay-lcm/src/retention/replay_metadata.rs index 6e435e32ce5..48b7128b81c 100644 --- a/crates/tracedecay-lcm/src/retention/replay_metadata.rs +++ b/crates/tracedecay-lcm/src/retention/replay_metadata.rs @@ -6,6 +6,7 @@ //! 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; @@ -28,6 +29,8 @@ 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, @@ -79,6 +82,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()); @@ -88,8 +92,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 { diff --git a/crates/tracedecay/src/daemon/maintenance.rs b/crates/tracedecay/src/daemon/maintenance.rs index 9e293532be4..967bf06dbba 100644 --- a/crates/tracedecay/src/daemon/maintenance.rs +++ b/crates/tracedecay/src/daemon/maintenance.rs @@ -386,6 +386,13 @@ fn record_canonical_body_gc_outcome( &[ ("roots_scanned", report.roots_scanned.to_string()), ("bodies_scanned", report.bodies_scanned.to_string()), + ( + "elapsed_ms", + report.elapsed_ms.map_or_else( + || "unmeasured".to_owned(), + |value| value.to_string(), + ), + ), ( "candidates_discharged", report.candidates_discharged.to_string(), @@ -432,6 +439,17 @@ fn record_replay_metadata_outcome( "retention_replay_metadata", &[ ("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( From 3924b36324044b9770ebb81fc201b40ae0e55a10 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sat, 7 Nov 2026 23:22:21 -0800 Subject: [PATCH 09/27] style(retention): fix Bazel rustfmt for per-tick measurements Apply the exact three rustfmt transformations in the latest failed repository gate without changing retention behavior. Refs #4787 --- .../src/retention/replay_metadata.rs | 5 ++++- crates/tracedecay/src/daemon/maintenance.rs | 14 ++++++-------- 2 files changed, 10 insertions(+), 9 deletions(-) diff --git a/crates/tracedecay-lcm/src/retention/replay_metadata.rs b/crates/tracedecay-lcm/src/retention/replay_metadata.rs index 48b7128b81c..50d9e115c24 100644 --- a/crates/tracedecay-lcm/src/retention/replay_metadata.rs +++ b/crates/tracedecay-lcm/src/retention/replay_metadata.rs @@ -167,7 +167,10 @@ async fn candidates( // the verification budget so deferred rows are not JSON-parsed // again on the next tick. Individually oversized rows still // advance without loading their bodies. - if inline && let Some(bytes) = read_bytes && bytes <= MAX_BATCH_BYTES { + if inline + && let Some(bytes) = read_bytes + && bytes <= MAX_BATCH_BYTES + { if bytes > read_remaining { break; } diff --git a/crates/tracedecay/src/daemon/maintenance.rs b/crates/tracedecay/src/daemon/maintenance.rs index 967bf06dbba..8916ab52886 100644 --- a/crates/tracedecay/src/daemon/maintenance.rs +++ b/crates/tracedecay/src/daemon/maintenance.rs @@ -388,10 +388,9 @@ fn record_canonical_body_gc_outcome( ("bodies_scanned", report.bodies_scanned.to_string()), ( "elapsed_ms", - report.elapsed_ms.map_or_else( - || "unmeasured".to_owned(), - |value| value.to_string(), - ), + report + .elapsed_ms + .map_or_else(|| "unmeasured".to_owned(), |value| value.to_string()), ), ( "candidates_discharged", @@ -441,10 +440,9 @@ fn record_replay_metadata_outcome( ("rows_scanned", report.rows_scanned.to_string()), ( "elapsed_ms", - report.elapsed_ms.map_or_else( - || "unmeasured".to_owned(), - |value| value.to_string(), - ), + report + .elapsed_ms + .map_or_else(|| "unmeasured".to_owned(), |value| value.to_string()), ), ( "metadata_bytes_examined", From a945a8b85b4a1f3f146872ff226d3319014b9b21 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sun, 8 Nov 2026 02:07:28 -0800 Subject: [PATCH 10/27] fix(retention): reject oversized payloads before sizing reads Use SQLite record byte lengths without materializing raw text. Defer storage-kind reads and unnecessary canonical lookups until admission, and do not read canonical size scalars beyond an oversized BLOB. Skip eligibility parsing when the stored representations already exceed the page budget, while preserving the rows and advancing the frontier. Add an isolated Linux database regression measuring actual read syscall bytes for oversized inline, metadata, and canonical payloads. Verify subsequent valid compaction, complete byte preservation, and quiescence. Refs #4787 --- .../src/retention/replay_metadata.rs | 58 +++--- .../retention/replay_metadata/read_cost.rs | 195 ++++++++++++++++++ 2 files changed, 224 insertions(+), 29 deletions(-) create mode 100644 crates/tracedecay-lcm/src/retention/replay_metadata/read_cost.rs diff --git a/crates/tracedecay-lcm/src/retention/replay_metadata.rs b/crates/tracedecay-lcm/src/retention/replay_metadata.rs index 50d9e115c24..dc3dd3b0d0e 100644 --- a/crates/tracedecay-lcm/src/retention/replay_metadata.rs +++ b/crates/tracedecay-lcm/src/retention/replay_metadata.rs @@ -15,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; @@ -41,7 +43,7 @@ pub struct ReplayMetadataCompactionReport { /// 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 parsing, including ineligible + /// 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. @@ -107,7 +109,7 @@ pub async fn compact_replay_metadata( struct Candidate { store_id: i64, eligible: bool, - metadata_oversized: bool, + oversized: bool, read_bytes: Option, } @@ -118,18 +120,24 @@ async fn candidates( after: i64, limit: usize, ) -> 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, r.storage_kind = 'inline', - 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?; @@ -139,16 +147,10 @@ async fn candidates( let mut read_remaining = MAX_BATCH_BYTES; while let Some(row) = rows.next().await? { let store_id: i64 = row.get(0)?; - let inline = row.get::(1)? != 0; - let metadata: Option = row.get(2)?; - let body: Option = row.get(3)?; - let stored: i64 = row.get(4)?; - let metadata_bytes = if inline { - u64::try_from(metadata.unwrap_or(0)).map_err(db_error)? - } else { - 0 - }; - let metadata_oversized = metadata_bytes > MAX_BATCH_BYTES; + 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)| { @@ -159,32 +161,30 @@ async fn candidates( )) }) .and_then(|(metadata, body, stored)| metadata.checked_add(body)?.checked_add(stored)); - if !metadata_oversized { + 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 every inline row against - // the verification budget so deferred rows are not JSON-parsed - // again on the next tick. Individually oversized rows still - // advance without loading their bodies. - if inline - && let Some(bytes) = read_bytes - && bytes <= MAX_BATCH_BYTES - { + // 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 inline && metadata.is_some() { + if metadata.is_some() { eligibility_ids.push(store_id); } } result.push(Candidate { store_id, eligible: false, - metadata_oversized, + oversized, read_bytes, }); } @@ -238,7 +238,7 @@ async fn step( let mut remaining = MAX_BATCH_BYTES; let mut guard_suspended = false; for candidate in candidates { - if candidate.metadata_oversized { + if candidate.oversized { report.rows_skipped += 1; report.rows_skipped_oversized += 1; } else if candidate.eligible { 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..81ee6bfaefb --- /dev/null +++ b/crates/tracedecay-lcm/src/retention/replay_metadata/read_cost.rs @@ -0,0 +1,195 @@ +//! 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 conn = TestConnection::open(&directory.path().join("replay-read-cost.db")); + 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'); + PRAGMA cache_size = 8; + PRAGMA mmap_size = 0;", + ) + .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(); + + for id in 1..=3 { + conn.execute_batch("PRAGMA shrink_memory;").await.unwrap(); + 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 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(); +} From ee4dc4cb18947177d78862ed4e1c9f400a7a02d2 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sun, 8 Nov 2026 03:53:29 -0800 Subject: [PATCH 11/27] style(retention): format replay read-cost regression Apply the exact Bazel rustfmt_check_2024 correction at read_cost.rs:62. No runtime or test behavior changes. Refs #4787 --- .../tracedecay-lcm/src/retention/replay_metadata/read_cost.rs | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/crates/tracedecay-lcm/src/retention/replay_metadata/read_cost.rs b/crates/tracedecay-lcm/src/retention/replay_metadata/read_cost.rs index 81ee6bfaefb..e490778a241 100644 --- a/crates/tracedecay-lcm/src/retention/replay_metadata/read_cost.rs +++ b/crates/tracedecay-lcm/src/retention/replay_metadata/read_cost.rs @@ -62,8 +62,7 @@ async fn oversized_rows_skip_payload_io() { "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 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); From 33988f731c8d8de09b2fbeca18fe4572e5599a8c Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sun, 8 Nov 2026 06:51:23 -0800 Subject: [PATCH 12/27] test(retention): reopen read-cost fixture without denied PRAGMAs Refs #4787 Use a fresh exact-SQL test connection for each measured replay page, so SQLite cannot reuse a warmed per-connection page cache. Remove the cache/mmap/shrink PRAGMAs rejected by the registered SQL authority. Keep the byte-read thresholds, 48-row semantics, frontier assertions, post-repair replay equivalence, and idle checks unchanged. --- .../src/retention/replay_metadata/read_cost.rs | 13 ++++++++----- 1 file changed, 8 insertions(+), 5 deletions(-) diff --git a/crates/tracedecay-lcm/src/retention/replay_metadata/read_cost.rs b/crates/tracedecay-lcm/src/retention/replay_metadata/read_cost.rs index e490778a241..548348e2cd6 100644 --- a/crates/tracedecay-lcm/src/retention/replay_metadata/read_cost.rs +++ b/crates/tracedecay-lcm/src/retention/replay_metadata/read_cost.rs @@ -42,14 +42,13 @@ async fn oversized_rows_skip_payload_io() { } let directory = tempfile::tempdir().unwrap(); - let conn = TestConnection::open(&directory.path().join("replay-read-cost.db")); + 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'); - PRAGMA cache_size = 8; - PRAGMA mmap_size = 0;", + ); INSERT INTO sessions VALUES ('codex', 'session');", ) .await .unwrap(); @@ -109,9 +108,12 @@ async fn oversized_rows_skip_payload_io() { .unwrap(); } let original = raw::load_raw_message_by_store_id(&conn, 4).await.unwrap(); + drop(conn); for id in 1..=3 { - conn.execute_batch("PRAGMA shrink_memory;").await.unwrap(); + // 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 @@ -134,6 +136,7 @@ async fn oversized_rows_skip_payload_io() { assert!(report.has_more); } + let conn = TestConnection::open(&database_path); let transaction = conn .transaction_with_behavior(TransactionBehavior::Immediate) .await From 370e91d088b9139f6ff9070143169d4e23a90e50 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sun, 8 Nov 2026 12:36:16 -0800 Subject: [PATCH 13/27] fix(retention): verify canonical candidate discharge after triggers Refs #4787 Validate the selected candidate hashes are absent inside the same writer transaction after DELETE. An AFTER DELETE trigger can reinsert all 48 rows while SQLite reports 48 affected deletes, producing false progress and repeat page work. The readback fails closed so the caller rolls back. Add a 52-body database-backed regression proving reinserted pages are refused without losing candidates, then recover as distinct 48+4 pages. --- .../src/observation/canonical_body_gc.rs | 19 +++- .../observation/canonical_body_gc/tests.rs | 93 +++++++++++++++++++ 2 files changed, 111 insertions(+), 1 deletion(-) 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 34dbdf30ad2..771a4054888 100644 --- a/crates/tracedecay-global-db/src/observation/canonical_body_gc.rs +++ b/crates/tracedecay-global-db/src/observation/canonical_body_gc.rs @@ -352,10 +352,27 @@ async fn collect_candidates( .execute( "DELETE FROM canonical_body_gc_candidates WHERE content_hash IN (SELECT value FROM json_each(?1))", - params![hashes], + 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"))?; 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 a81eb82aeef..15cf06540f0 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,15 @@ 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}, @@ -637,3 +646,87 @@ async fn rewritten_bootstrap_checkpoint_refuses_commit_and_recovers_across_48_ro 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); +} From fd2ac3dd2489905a7426fc1e3a1f1c82383b97d0 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sun, 8 Nov 2026 12:37:05 -0800 Subject: [PATCH 14/27] style(retention): keep canonical GC test helpers rustfmt-clean Refs #4787 Remove an extra blank line introduced with the candidate-count helper. --- .../src/observation/canonical_body_gc/tests.rs | 1 - 1 file changed, 1 deletion(-) 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 15cf06540f0..bd71002ad1a 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,7 +92,6 @@ 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", ()) From d21084185df37bd8230545e25cd18524b625980b Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sun, 8 Nov 2026 12:59:10 -0800 Subject: [PATCH 15/27] style(retention): format candidate discharge assertions --- .../src/observation/canonical_body_gc/tests.rs | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) 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 bd71002ad1a..2327c0a9488 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 @@ -670,7 +670,10 @@ async fn reinserted_candidate_page_refuses_false_discharge_and_recovers() { break; } } - assert!(bootstrap_finished, "candidate collection must become eligible"); + 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 @@ -694,7 +697,9 @@ async fn reinserted_candidate_page_refuses_false_discharge_and_recovers() { .unwrap(); let error = step(&transaction, 48).await.unwrap_err(); assert!( - error.to_string().contains("candidate discharge was not persisted"), + error + .to_string() + .contains("candidate discharge was not persisted"), "an apparent 48-row DELETE must not count as committed progress: {error}" ); assert_eq!( From 57bae69e909ec818e9238f26cbf1bfe41dcba69f Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sun, 8 Nov 2026 01:52:34 -0800 Subject: [PATCH 16/27] test(automation): bind scheduler fixtures to backend identities Repair three hosted automation_runner_test failures by supplying the configured executable identity required by the production prefilter. Use the same identity for the prefilter and retained runner, preserving manual-schedule, exact-ledger, live-lock and lock-race assertions. Add a production-prefilter regression for unconfigured refusal, deduplication and same-occurrence recovery after binding, with zero backend calls and no effectful terminal. Do not relax backend admission, change timeouts, or alter the graph publication changes. Refs #4532 --- .../tests/automation_runner_test/jobs.rs | 124 ++++++++++++++++-- 1 file changed, 110 insertions(+), 14 deletions(-) diff --git a/crates/tracedecay/tests/automation_runner_test/jobs.rs b/crates/tracedecay/tests/automation_runner_test/jobs.rs index 740283d56e8..7342786381b 100644 --- a/crates/tracedecay/tests/automation_runner_test/jobs.rs +++ b/crates/tracedecay/tests/automation_runner_test/jobs.rs @@ -86,6 +86,7 @@ fn enabled_job_config() -> AutomationSettingsV1 { struct ContentBackend { calls: AtomicUsize, content: &'static str, + executable: Option, } impl ContentBackend { @@ -93,9 +94,15 @@ impl ContentBackend { Self { calls: AtomicUsize::new(0), content, + executable: None, } } + fn with_executable(mut self, executable: &Path) -> Self { + self.executable = Some(executable.to_path_buf()); + self + } + fn calls(&self) -> usize { self.calls.load(Ordering::SeqCst) } @@ -126,7 +133,7 @@ impl AgentTaskBackend for ContentBackend { } fn executable(&self) -> Option<&std::path::Path> { - None + self.executable.as_deref() } } @@ -988,6 +995,10 @@ async fn scheduler_trigger_skips_repeat_and_respects_lock_discipline() { let mut job = sample_job("sched-job"); job.schedule = Some("manual".to_string()); let config = enabled_job_config(); + // Reach the schedule gate with a bound identity, not the earlier + // backend-unconfigured refusal. This gate never invokes the backend. + let executable = tempfile::NamedTempFile::new().unwrap(); + let backend = ContentBackend::new("unused").with_executable(executable.path()); // Manual schedule: every scheduler tick of the same occurrence reaches // the prefilter, which records the skip once. @@ -996,7 +1007,7 @@ async fn scheduler_trigger_skips_repeat_and_respects_lock_discipline() { let run = evaluate_and_record_scheduler_skip( &dashboard_root, &config, - None, + backend.executable(), &job, "sched-occurrence", None, @@ -1019,7 +1030,7 @@ async fn scheduler_trigger_skips_repeat_and_respects_lock_discipline() { let repeated = evaluate_and_record_scheduler_skip( &dashboard_root, &config, - None, + backend.executable(), &job, "sched-occurrence", None, @@ -1031,6 +1042,76 @@ async fn scheduler_trigger_skips_repeat_and_respects_lock_discipline() { repeated.ledger_record, records[0], "an immediate repeat skip must return the exact durable prior row" ); + assert_eq!(backend.calls(), 0); +} + +#[tokio::test] +async fn scheduler_prefilter_rejects_unconfigured_backend_and_recovers_without_effect() { + let temp = tempdir().unwrap(); + let dashboard_root = temp.path().join("dashboard"); + let job = sample_job("backend-recovery"); + let config = enabled_job_config(); + let occurrence = "backend-recovery-occurrence"; + let backend = ContentBackend::new("unused"); + + let first = evaluate_and_record_scheduler_skip( + &dashboard_root, + &config, + backend.executable(), + &job, + occurrence, + None, + ) + .await + .unwrap() + .expect("a due job still requires a configured backend"); + assert_eq!(first.report["reason"], json!("backend_unconfigured")); + assert_eq!(first.ledger_record.status, AutomationRunStatus::Skipped); + assert_ne!(first.run_id, occurrence); + let repeated = evaluate_and_record_scheduler_skip( + &dashboard_root, + &config, + backend.executable(), + &job, + occurrence, + None, + ) + .await + .unwrap() + .unwrap(); + assert_eq!(repeated.ledger_record, first.ledger_record); + + let executable = tempfile::NamedTempFile::new().unwrap(); + let backend = backend.with_executable(executable.path()); + assert!( + evaluate_and_record_scheduler_skip( + &dashboard_root, + &config, + backend.executable(), + &job, + occurrence, + None, + ) + .await + .unwrap() + .is_none(), + "binding the backend must admit the same due occurrence without cooldown" + ); + assert_eq!(backend.calls(), 0); + assert_eq!( + load_run_records(&dashboard_root, 10).await.unwrap(), + vec![first.ledger_record], + "advisory admission must not append an effectful terminal" + ); + assert!( + tracedecay_automation_runtime::automation::run_ledger::find_run_record_exact_bounded( + &dashboard_root, + occurrence, + ) + .await + .unwrap() + .is_none() + ); } #[tokio::test] @@ -1206,15 +1287,22 @@ async fn scheduler_prefilter_live_lock_wins_over_not_due_summary() { let config = enabled_job_config(); let occurrence = "locked-prefilter-occurrence"; - let prefilter = - evaluate_and_record_scheduler_skip(&dashboard_root, &config, None, &job, occurrence, None) - .await - .unwrap() - .expect("the live job lock must produce a diagnostic"); + let executable = tempfile::NamedTempFile::new().unwrap(); + let backend = ContentBackend::new("unused").with_executable(executable.path()); + let prefilter = evaluate_and_record_scheduler_skip( + &dashboard_root, + &config, + backend.executable(), + &job, + occurrence, + None, + ) + .await + .unwrap() + .expect("the live job lock must produce a diagnostic"); assert_eq!(prefilter.report["reason"], json!("scheduler_lock_active")); assert_ne!(prefilter.run_id, occurrence); - let backend = ContentBackend::new("unused"); let admitted_job = tracedecay_automation_runtime::automation::jobs::admit_job(&job).unwrap(); let retained = run_user_job_with_backend_for_retained_settlement( &dashboard_root, @@ -1262,12 +1350,21 @@ async fn retained_scheduler_runner_reacquires_after_due_prefilter() { job.schedule = Some("every 1h".to_owned()); let config = enabled_job_config(); let occurrence = "prefilter-gap-occurrence"; + let executable = tempfile::NamedTempFile::new().unwrap(); + let backend = ContentBackend::new("unused").with_executable(executable.path()); assert!( - evaluate_and_record_scheduler_skip(&dashboard_root, &config, None, &job, occurrence, None) - .await - .unwrap() - .is_none(), + evaluate_and_record_scheduler_skip( + &dashboard_root, + &config, + backend.executable(), + &job, + occurrence, + None, + ) + .await + .unwrap() + .is_none(), "a new interval job is advisory-due before effect admission" ); let _raced_in = @@ -1276,7 +1373,6 @@ async fn retained_scheduler_runner_reacquires_after_due_prefilter() { .unwrap() .expect("the advisory prefilter must release its reservation lock"); - let backend = ContentBackend::new("unused"); let admitted_job = tracedecay_automation_runtime::automation::jobs::admit_job(&job).unwrap(); let retained = run_user_job_with_backend_for_retained_settlement( &dashboard_root, From 4984a54fc4cf6cd71d2191d60a80639b1f0b183f Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sun, 8 Nov 2026 15:31:24 -0800 Subject: [PATCH 17/27] test(daemon): verify rollback at the store mount boundary --- .../drop_shutdown_journey_test.rs | 19 ++++++++++++++----- 1 file changed, 14 insertions(+), 5 deletions(-) diff --git a/crates/tracedecay/src/daemon/production_harness/drop_shutdown_journey_test.rs b/crates/tracedecay/src/daemon/production_harness/drop_shutdown_journey_test.rs index 7b159cbcf6a..51efc83f65c 100644 --- a/crates/tracedecay/src/daemon/production_harness/drop_shutdown_journey_test.rs +++ b/crates/tracedecay/src/daemon/production_harness/drop_shutdown_journey_test.rs @@ -185,7 +185,7 @@ async fn a_failed_later_mount_releases_the_whole_composition_before_returning() .expect("refused store root"); std::fs::write(&refused_graph, b"not a project graph database").expect("refused graph"); - let error = match ProductionProjectCompositionHarnessV1::open( + let error = match ProductionProjectCompositionHarnessV1::open_for_session_retrieval( isolation.path(), [first.clone(), refused.clone()], ) @@ -207,7 +207,7 @@ async fn a_failed_later_mount_releases_the_whole_composition_before_returning() let stores = [ profile.join("global.db"), profile.join("user-sessions.db"), - first_layout.data_root.join("sessions.db"), + first_layout.sessions_db_path, first_layout.graph_db_path, ]; for store in &stores { @@ -229,14 +229,23 @@ async fn a_failed_later_mount_releases_the_whole_composition_before_returning() } std::fs::remove_file(&refused_graph).expect("remove the fixture refusal"); - let harness = ProductionProjectCompositionHarnessV1::open( + let harness = ProductionProjectCompositionHarnessV1::open_for_session_retrieval( isolation.path(), [first.clone(), refused.clone()], ) .await .expect("both exact projects reopen after failed-open rollback"); - assert!(harness.server(&first).is_ok()); - assert!(harness.server(&refused).is_ok()); + for project in [&first, &refused] { + let server = harness.server(project).expect("reopened project server"); + let store = server + .project_session_db() + .expect("mounted project sessions"); + let layout = default_profile_sharded_layout(project, &profile).unwrap(); + assert_eq!( + store.db_path().canonicalize().unwrap(), + layout.sessions_db_path.canonicalize().unwrap() + ); + } harness.shutdown().await; } From ce112eff1754af4a2fc966acda79014157a599d3 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sun, 8 Nov 2026 17:08:35 -0800 Subject: [PATCH 18/27] fix(retention): report committed canonical bootstrap frontiers Expose the phase-qualified raw, observation and canonical-body SQL frontiers before and after each successful bootstrap page. Populate these only after the same-transaction checkpoint readback succeeds; collection and idle pages cannot masquerade as bootstrap progress. Log the frontiers with the already measured page cost and discharge counts. A real SQLite-backed regression checks three distinct phases, the durable cursor after each tick, retained live content, collection and quiescence. The existing 52-owner rollback/48+4 regression now asserts its committed 0->48->52 frontier values. Refs #4787; not incident acceptance. No CI, dependency, release, or concurrent writer-admission change. --- .../src/observation/canonical_body_gc.rs | 23 +++++- .../observation/canonical_body_gc/tests.rs | 71 +++++++++++++++++++ crates/tracedecay/src/daemon/maintenance.rs | 16 +++++ 3 files changed, 108 insertions(+), 2 deletions(-) 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 771a4054888..3b2725a92ad 100644 --- a/crates/tracedecay-global-db/src/observation/canonical_body_gc.rs +++ b/crates/tracedecay-global-db/src/observation/canonical_body_gc.rs @@ -31,6 +31,11 @@ pub struct CanonicalBodyGcReport { 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. @@ -162,19 +167,33 @@ async fn step(conn: &(impl Executor + ?Sized), max_rows: usize) -> Result 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 rewritten_bootstrap_checkpoint_refuses_commit_and_recovers_across_48_row_pages() { let store = Store::open().await; @@ -630,10 +697,14 @@ async fn rewritten_bootstrap_checkpoint_refuses_commit_and_recovers_across_48_ro .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"); diff --git a/crates/tracedecay/src/daemon/maintenance.rs b/crates/tracedecay/src/daemon/maintenance.rs index 8916ab52886..7c70079ab69 100644 --- a/crates/tracedecay/src/daemon/maintenance.rs +++ b/crates/tracedecay/src/daemon/maintenance.rs @@ -386,6 +386,22 @@ fn record_canonical_body_gc_outcome( &[ ("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 From 1f0446809f5d95ecf0d86c022ec886941c2c9e83 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sun, 8 Nov 2026 17:36:37 -0800 Subject: [PATCH 19/27] fix(retention): report committed empty bootstrap pages --- .../src/observation/canonical_body_gc.rs | 9 +++++++ .../observation/canonical_body_gc/tests.rs | 27 ++++++++++++++++++- crates/tracedecay/src/daemon/maintenance.rs | 2 +- 3 files changed, 36 insertions(+), 2 deletions(-) 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 3b2725a92ad..c69ef98cd98 100644 --- a/crates/tracedecay-global-db/src/observation/canonical_body_gc.rs +++ b/crates/tracedecay-global-db/src/observation/canonical_body_gc.rs @@ -45,6 +45,15 @@ 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, 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 e66c7b7cb10..149f523b8ef 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 @@ -591,7 +591,10 @@ async fn bootstrap_reports_each_committed_sql_frontier_without_claiming_collecti 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)), + ( + "bodies:".to_owned(), + format!("bodies:{}", body.content_hash), + ), ]; for (phase, (before, after)) in frontiers.into_iter().enumerate() { let report = store.tick(1).await; @@ -631,6 +634,28 @@ async fn bootstrap_reports_each_committed_sql_frontier_without_claiming_collecti ); } +#[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; diff --git a/crates/tracedecay/src/daemon/maintenance.rs b/crates/tracedecay/src/daemon/maintenance.rs index 7c70079ab69..97278339b41 100644 --- a/crates/tracedecay/src/daemon/maintenance.rs +++ b/crates/tracedecay/src/daemon/maintenance.rs @@ -380,7 +380,7 @@ 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", &[ From dfd0c6a10600ecdcd93f7e0fa5ed94611e36de44 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sun, 8 Nov 2026 19:59:32 -0800 Subject: [PATCH 20/27] simplify(retention): admit replay body bytes once per page --- .../src/retention/replay_metadata.rs | 56 ++++++++----------- 1 file changed, 23 insertions(+), 33 deletions(-) diff --git a/crates/tracedecay-lcm/src/retention/replay_metadata.rs b/crates/tracedecay-lcm/src/retention/replay_metadata.rs index dc3dd3b0d0e..79849cb70ca 100644 --- a/crates/tracedecay-lcm/src/retention/replay_metadata.rs +++ b/crates/tracedecay-lcm/src/retention/replay_metadata.rs @@ -110,7 +110,7 @@ struct Candidate { store_id: i64, eligible: bool, oversized: bool, - read_bytes: Option, + body_size_known: bool, } /// Admit a prefix bounded by both metadata parsing and body-read estimates @@ -185,7 +185,7 @@ async fn candidates( store_id, eligible: false, oversized, - read_bytes, + body_size_known: read_bytes.is_some(), }); } drop(rows); @@ -235,48 +235,38 @@ async fn step( metadata_bytes_examined, ..ReplayMetadataCompactionReport::default() }; - let mut remaining = MAX_BATCH_BYTES; let mut guard_suspended = false; for candidate in candidates { if candidate.oversized { report.rows_skipped += 1; report.rows_skipped_oversized += 1; } else if candidate.eligible { - let bytes = candidate.read_bytes.ok_or_else(|| { - db_error(format!( + 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; } } report.rows_scanned += 1; From 9a1c8a27d08bba857319950ffb3e4f0678199443 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sun, 8 Nov 2026 21:13:20 -0800 Subject: [PATCH 21/27] fix(retention): account for ineligible replay pages Rows skipped by replay eligibility were not counted as skipped, so the reported 48 scanned / 0 compacted / 0 skipped contradicted what the page actually examined. Include those intact rows in the skip count; keep oversized skips as a subset and leave the durable frontier, eligibility, body reads and scheduling unchanged. Assert the real SQLite-backed 48+4 ineligible pages, five preserved representation kinds and multi-tick metadata-budget pages report true skipped counts while preserving content and quiescence. Refs #4787 --- crates/tracedecay-lcm/src/retention/replay_metadata.rs | 6 +++++- .../src/retention/replay_metadata/tests.rs | 9 ++++++++- 2 files changed, 13 insertions(+), 2 deletions(-) diff --git a/crates/tracedecay-lcm/src/retention/replay_metadata.rs b/crates/tracedecay-lcm/src/retention/replay_metadata.rs index 79849cb70ca..a948c9dfd1b 100644 --- a/crates/tracedecay-lcm/src/retention/replay_metadata.rs +++ b/crates/tracedecay-lcm/src/retention/replay_metadata.rs @@ -38,7 +38,8 @@ pub struct ReplayMetadataCompactionReport { pub frontier_after: Option, pub rows_compacted: u64, /// Rows retained because their representation, sanitizer result, or byte - /// bound could not prove this narrow lossless transformation. + /// 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. @@ -268,6 +269,9 @@ async fn step( } 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; diff --git a/crates/tracedecay-lcm/src/retention/replay_metadata/tests.rs b/crates/tracedecay-lcm/src/retention/replay_metadata/tests.rs index 5c794d2ed99..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); @@ -405,6 +407,8 @@ async fn zero_compaction_pages_advance_distinct_durable_frontiers() { 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); @@ -412,6 +416,8 @@ async fn zero_compaction_pages_advance_distinct_durable_frontiers() { 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); @@ -462,7 +468,8 @@ async fn large_ineligible_metadata_is_bounded_before_eligibility_checks() { ticks += 1; assert!(report.rows_scanned > 0); assert_eq!(report.rows_compacted, 0); - assert_eq!(report.rows_skipped, 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); From bfba4837c1e75340af3370d6ec69c7815d4ba55a Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sun, 8 Nov 2026 22:39:50 -0800 Subject: [PATCH 22/27] fix(retention): scope SQL frontier logs to each store Mounted session stores can commit identical 48-row pages in the same maintenance cycle. Previously their canonical and replay operator lines had no store identity, so interleaved advancing SQL frontiers appeared to be the same stalled page. Include the exact mounted database path in success and refusal events without changing retention or scheduling. A subprocess regression exercises the actual operator stderr with two stores both reporting 0->48, requires four uniquely attributable log lines, and rejects vacuous zero-test success. Refs #4787. --- crates/tracedecay/src/daemon/maintenance.rs | 88 ++++++++++++++++++++- 1 file changed, 84 insertions(+), 4 deletions(-) diff --git a/crates/tracedecay/src/daemon/maintenance.rs b/crates/tracedecay/src/daemon/maintenance.rs index 97278339b41..4eeef0fa4b6 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,12 @@ 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, @@ -384,6 +387,7 @@ fn record_canonical_body_gc_outcome( 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()), ( @@ -433,6 +437,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()), ], ); @@ -442,6 +447,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, @@ -453,6 +459,7 @@ 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", @@ -503,6 +510,7 @@ fn record_replay_metadata_outcome( "retention_degraded", &[ ("pass", "replay_metadata_compaction".to_owned()), + ("store", store.display().to_string()), ("failure", error.diagnostic().to_owned()), ], ); @@ -2069,4 +2077,76 @@ 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}"); + } + } + } } From ddabf4a972f3a89d745204d65af29023e0b4be28 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sun, 8 Nov 2026 23:11:43 -0800 Subject: [PATCH 23/27] style(retention): format store-scoped frontier diagnostics --- crates/tracedecay/src/daemon/maintenance.rs | 20 ++++++++++++++++---- 1 file changed, 16 insertions(+), 4 deletions(-) diff --git a/crates/tracedecay/src/daemon/maintenance.rs b/crates/tracedecay/src/daemon/maintenance.rs index 4eeef0fa4b6..a7c1dce45d1 100644 --- a/crates/tracedecay/src/daemon/maintenance.rs +++ b/crates/tracedecay/src/daemon/maintenance.rs @@ -370,8 +370,14 @@ async fn run_registered_storage_continuation( ) .await; record_session_lcm_outcome(&session_lcm) - .combine(record_canonical_body_gc_outcome(database.db_path(), &canonical_body_gc)) - .combine(record_replay_metadata_outcome(database.db_path(), &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( @@ -2134,7 +2140,10 @@ mod tests { 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_canonical_bodies", + "bootstrap_frontier_after=raw:48", + ), ("retention_replay_metadata", "frontier_after=48"), ] { let count = lines @@ -2145,7 +2154,10 @@ mod tests { && line.contains(frontier) }) .count(); - assert_eq!(count, 1, "missing distinct {event} page for {store}: {stderr}"); + assert_eq!( + count, 1, + "missing distinct {event} page for {store}: {stderr}" + ); } } } From c3f2d94d2bcc4b7c70e074e06e3b7b294c2452aa Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 9 Nov 2026 21:35:47 -0800 Subject: [PATCH 24/27] fix(retention): refuse orphaned canonical bodies on vetoed delete Keep selected unreferenced body candidates queued if the underlying DELETE was ignored. Verify the body is absent before discharging its candidate; otherwise rollback the page instead of silently leaking unreclaimed bodies. Exercise 52 real candidates with a vetoed DELETE, rollback, recovery, 48+4 committed reclamation, and an idle final tick. Refs #4787 --- .../src/observation/canonical_body_gc.rs | 21 ++++++ .../observation/canonical_body_gc/tests.rs | 74 +++++++++++++++++++ 2 files changed, 95 insertions(+) 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 c69ef98cd98..e8513b02a21 100644 --- a/crates/tracedecay-global-db/src/observation/canonical_body_gc.rs +++ b/crates/tracedecay-global-db/src/observation/canonical_body_gc.rs @@ -372,6 +372,27 @@ 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. // Deleting an unreferenced body cascades to its candidate row. The 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 149f523b8ef..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 @@ -515,6 +515,80 @@ 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; From eb9bd17c6eea314364739fb9a57c191a52dcc3ae Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 9 Nov 2026 21:51:36 -0800 Subject: [PATCH 25/27] perf(retention): batch canonical bootstrap writes Replace per-owner and per-body bootstrap INSERT submissions with one indexed INSERT SELECT per bounded page. Keep the selected last identity, sparse-key semantics, eligibility predicate and transactional checkpoint. Add real SQLite executor regressions for bounded write submissions, 52-row sparse frontiers, live-body preservation, eligibility at the page tail, and late statement failure with full rollback and successful retry. Preserve the concurrent vetoed-delete repair without rewriting history. Refs #4787 --- .../src/observation/canonical_body_gc.rs | 50 +-- .../bootstrap_write_tests.rs | 319 ++++++++++++++++++ 2 files changed, 346 insertions(+), 23 deletions(-) create mode 100644 crates/tracedecay-global-db/src/observation/canonical_body_gc/bootstrap_write_tests.rs 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 e8513b02a21..8d86d11440e 100644 --- a/crates/tracedecay-global-db/src/observation/canonical_body_gc.rs +++ b/crates/tracedecay-global-db/src/observation/canonical_body_gc.rs @@ -14,6 +14,8 @@ use tracedecay_runtime_core::db::Database; use tracedecay_runtime_core::db::engine::{Executor, QueryExecutor, params}; use tracedecay_store::collect_body_refs; +#[cfg(test)] +mod bootstrap_write_tests; pub(crate) mod schema; #[cfg(test)] mod tests; @@ -218,33 +220,31 @@ async fn bootstrap_raw( ) -> 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()], - ) - .await - .map_err(db_error)?; - } - progress.raw_through = *owner; + 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)?; + progress.raw_through = last; } if page.len() < limit as usize { progress.raw_through = progress.raw_until; @@ -317,14 +317,18 @@ 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); + progress.bodies_after.clone_from(last); } if page.len() < limit as usize { progress.bodies_after.clone_from(&progress.bodies_until); 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..5630d1426c7 --- /dev/null +++ b/crates/tracedecay-global-db/src/observation/canonical_body_gc/bootstrap_write_tests.rs @@ -0,0 +1,319 @@ +//! 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 { + 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(); + 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; +} From 9b50818107afea5486193022122c3a45f9a5b40d Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 9 Nov 2026 22:26:27 -0800 Subject: [PATCH 26/27] perf(retention): batch observation bootstrap writer submissions Group validated observation reference inserts into bounded storage-worker submissions instead of one asynchronous request per referenced hash. Preserve Rust reference parsing, identity order and transactional rollback. Prove 48+4 pages across the 128-statement boundary, durable SQL frontiers, partial-batch refusal/rollback, reference deduplication and quiescence. Refs #4787 --- .../src/observation/canonical_body_gc.rs | 31 +++- .../bootstrap_write_tests.rs | 143 ++++++++++++++++++ 2 files changed, 166 insertions(+), 8 deletions(-) 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 8d86d11440e..bce70bf46aa 100644 --- a/crates/tracedecay-global-db/src/observation/canonical_body_gc.rs +++ b/crates/tracedecay-global-db/src/observation/canonical_body_gc.rs @@ -11,7 +11,7 @@ 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)] @@ -277,18 +277,33 @@ 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(); 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], - ) - .await - .map_err(db_error)?; + 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 page.len() < limit as usize { progress.observations_through = progress.observations_until; } 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 index 5630d1426c7..56da4607408 100644 --- 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 @@ -52,6 +52,10 @@ struct Fixture { 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(); @@ -96,6 +100,35 @@ impl Fixture { ) .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(); @@ -317,3 +350,113 @@ async fn bootstrap_batch_eligibility_does_not_change_the_page_boundary() { 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; +} From 6edf06957b7bff930d19e6956f6573ac90d22da7 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 9 Nov 2026 23:35:46 -0800 Subject: [PATCH 27/27] fix(retention): verify bootstrap seed integrity before advancing After batched bootstrap writes, verify the selected raw owner references, Rust-validated observation references, and canonical-body candidates exist inside the writer transaction. A suppressed INSERT must not commit a frontier that can drop a live body or permanently abandon an orphan. Exercise RAISE(IGNORE) for all three phases, rollback, 48+4 recovery, and live-data preservation. Refs #4787 --- .../src/observation/canonical_body_gc.rs | 72 ++++++++ .../bootstrap_write_tests.rs | 155 +++++++++++++++++- 2 files changed, 225 insertions(+), 2 deletions(-) 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 bce70bf46aa..a95b950ddab 100644 --- a/crates/tracedecay-global-db/src/observation/canonical_body_gc.rs +++ b/crates/tracedecay-global-db/src/observation/canonical_body_gc.rs @@ -244,6 +244,29 @@ async fn bootstrap_raw( ) .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 = last; } if page.len() < limit as usize { @@ -283,8 +306,10 @@ async fn bootstrap_observations( // 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)? { + expected.push((*owner, hash.clone())); pending.push( WriteStatement::new( "INSERT OR IGNORE INTO canonical_body_gc_refs(owner_kind, owner_id, content_hash) @@ -304,6 +329,32 @@ async fn bootstrap_observations( 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", + )); + } + } if page.len() < limit as usize { progress.observations_through = progress.observations_until; } @@ -343,6 +394,27 @@ async fn bootstrap_bodies( ) .await .map_err(db_error)?; + // 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 { 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 index 56da4607408..5e71811ca92 100644 --- 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 @@ -379,8 +379,14 @@ async fn observation_bootstrap_batches_across_a_bounded_write_submission() { ] { 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!( + 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, @@ -460,3 +466,148 @@ async fn observation_bootstrap_failure_after_a_full_batch_rolls_back_and_recover } 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; +}