From 0bac9a088c0f21411251759d5d162ed92d35a9de Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Mon, 24 Aug 2026 00:32:59 +0000 Subject: [PATCH 1/4] perf(lcm): batch per-row GC and delete-recovery SQL Four un-batched sites in LCM payload GC: the reference-closure scan ran once per payload inside the loop that was already meant to be batched, gc-mark deletes went one row at a time while a chunked delete_gc_marks sat unused, pending-tombstone existence checks issued one SELECT each, and residual placeholder tombstoning used a leading-wildcard LIKE that no index can serve. Co-Authored-By: Claude Opus 5 (1M context) --- .../tracedecay-sessions/src/runtime/lcm/gc.rs | 13 +- .../src/runtime/lcm/gc/pending_delete.rs | 98 +++- .../src/runtime/lcm/gc/placeholder_scan.rs | 15 + .../src/runtime/lcm/gc/tests.rs | 495 ++++++++++++++++++ .../src/runtime/lcm/payload.rs | 25 +- .../runtime/lcm/payload/delete_recovery.rs | 89 +++- .../src/runtime/lcm/query.rs | 175 +++++++ 7 files changed, 876 insertions(+), 34 deletions(-) diff --git a/crates/tracedecay-sessions/src/runtime/lcm/gc.rs b/crates/tracedecay-sessions/src/runtime/lcm/gc.rs index 4edb299af7..f1562ab615 100644 --- a/crates/tracedecay-sessions/src/runtime/lcm/gc.rs +++ b/crates/tracedecay-sessions/src/runtime/lcm/gc.rs @@ -26,7 +26,7 @@ pub(crate) use placeholder_scan::{ PlaceholderScanScope, PlaceholderTextRow, all_placeholder_like_patterns, any_placeholder_text_row, bind_placeholder_like_patterns, count_placeholder_text_rows, gc_prefix_like_patterns, gc_prefix_ref_like_patterns, live_prefix_like_patterns, - placeholder_text_like_sql, scan_placeholder_text_rows, + live_prefix_ref_like_patterns, placeholder_text_like_sql, scan_placeholder_text_rows, }; const GC_PAYLOAD_PREFIX: &str = "[gc'd externalized payload:"; @@ -817,6 +817,9 @@ async fn reap_unreferenced_metadata( let marks = gc_marks(conn, &candidates).await?; let mut marks_to_upsert = Vec::new(); let mut stale_marks = Vec::new(); + // One reference-closure scan for the whole batch instead of one per + // candidate: only a payload's own deletion can change its own membership. + let mut referenced_closure = payload::ReferencedClosureCache::default(); for payload_ref in &candidates { let mark = marks.get(payload_ref); @@ -851,11 +854,12 @@ async fn reap_unreferenced_metadata( } let bytes = metadata_bytes.get(payload_ref).copied().unwrap_or_default(); if apply { - match payload::delete_external_payload_in_transaction( + match payload::delete_external_payload_in_transaction_with_cache( conn, storage_root, payload_ref, &payload::DeleteOpts::default(), + &mut referenced_closure, ) .await { @@ -950,6 +954,8 @@ async fn reap_missing_metadata( } let marks = gc_marks(conn, &missing_refs).await?; let mut marks_to_upsert = Vec::new(); + // As in `reap_unreferenced_metadata`: one scan per provider for the batch. + let mut referenced_closure = payload::ReferencedClosureCache::default(); for payload_ref in &missing_refs { let first_seen_at = match marks.get(payload_ref) { Some((state, first_seen_at)) if state == "missing" => *first_seen_at, @@ -965,7 +971,7 @@ async fn reap_missing_metadata( report.batch_cap(1); continue; } - match payload::delete_external_payload_in_transaction( + match payload::delete_external_payload_in_transaction_with_cache( conn, storage_root, payload_ref, @@ -974,6 +980,7 @@ async fn reap_missing_metadata( remove_file: false, verify_hash: false, }, + &mut referenced_closure, ) .await { diff --git a/crates/tracedecay-sessions/src/runtime/lcm/gc/pending_delete.rs b/crates/tracedecay-sessions/src/runtime/lcm/gc/pending_delete.rs index b182ad69e6..5d808a970a 100644 --- a/crates/tracedecay-sessions/src/runtime/lcm/gc/pending_delete.rs +++ b/crates/tracedecay-sessions/src/runtime/lcm/gc/pending_delete.rs @@ -1,11 +1,12 @@ +use std::collections::{HashMap, HashSet}; use std::path::Path; use serde::{Deserialize, Serialize}; -use tracedecay_runtime_core::db::engine::{Executor, params}; +use tracedecay_runtime_core::db::engine::{Executor, Value as SqlValue, params}; use super::{LcmGcError, LcmGcPhaseReport, MAX_SAMPLES}; -use crate::runtime::lcm::{LcmError, payload, schema}; +use crate::runtime::lcm::{LcmError, payload, schema, util}; const PENDING_PAYLOAD_DELETE_PREFIX: &str = "pending_payload_delete:"; pub(super) const PENDING_PAYLOAD_DELETE_ERROR_PREFIX: &str = "pending payload deletion partial:"; @@ -177,28 +178,25 @@ async fn drain_pending_payload_deletes_matching( } drop(rows); + // One chunked existence probe for the whole tombstone set replaces the + // per-tombstone `SELECT 1 ... LIMIT 1`. Nothing in the loop below writes to + // `lcm_external_payloads`, so a single snapshot taken here answers every + // iteration exactly as its own probe would have. + let probe = probe_metadata_rows( + conn, + &pending + .iter() + .map(|(_, payload_ref, _)| payload_ref.clone()) + .collect::>(), + ) + .await; + for (key, payload_ref, pending) in pending { - let mut metadata = match conn - .query( - "SELECT 1 FROM lcm_external_payloads WHERE payload_ref = ?1 LIMIT 1", - params![payload_ref.as_str()], - ) - .await - { - Ok(rows) => rows, - Err(err) => { - drain.add_error(&payload_ref, "metadata_check_failed", err.to_string()); - continue; - } - }; - let metadata_exists = match metadata.next().await { - Ok(row) => row.is_some(), - Err(err) => { - drain.add_error(&payload_ref, "metadata_check_failed", err.to_string()); - continue; - } - }; - drop(metadata); + if let Some(detail) = probe.failures.get(&payload_ref) { + drain.add_error(&payload_ref, "metadata_check_failed", detail.clone()); + continue; + } + let metadata_exists = probe.existing.contains(&payload_ref); if metadata_exists { schema::clear_gc_meta(conn, &key).await?; drain @@ -245,6 +243,60 @@ async fn drain_pending_payload_deletes_matching( Ok(drain) } +/// Snapshot of which pending-delete refs still own a metadata row, plus the +/// refs whose chunk query failed and must be reported as `metadata_check_failed` +/// in loop order rather than silently treated as absent. +#[derive(Default)] +struct MetadataProbe { + existing: HashSet, + failures: HashMap, +} + +/// Batched form of the per-tombstone `SELECT 1 FROM lcm_external_payloads` +/// probe. Chunked at [`util::SQLITE_IN_BATCH_SIZE`] so an unbounded tombstone +/// backlog cannot exceed SQLite's bind-variable limit; an empty input issues no +/// query at all. +async fn probe_metadata_rows( + conn: &(impl Executor + ?Sized), + payload_refs: &[String], +) -> MetadataProbe { + let mut probe = MetadataProbe::default(); + for chunk in payload_refs.chunks(util::SQLITE_IN_BATCH_SIZE) { + if chunk.is_empty() { + continue; + } + let sql = format!( + "SELECT payload_ref FROM lcm_external_payloads WHERE payload_ref IN ({})", + util::sql_in_placeholders(chunk.len()) + ); + let values = chunk + .iter() + .cloned() + .map(SqlValue::Text) + .collect::>(); + let outcome: Result, LcmError> = async { + let mut rows = conn.query(&sql, values).await?; + let mut present = Vec::new(); + while let Some(row) = rows.next().await? { + let payload_ref: String = row.get(0)?; + present.push(payload_ref); + } + Ok(present) + } + .await; + match outcome { + Ok(present) => probe.existing.extend(present), + Err(err) => { + let detail = err.to_string(); + for payload_ref in chunk { + probe.failures.insert(payload_ref.clone(), detail.clone()); + } + } + } + } + probe +} + async fn record_pending_delete_diagnostics( conn: &(impl Executor + ?Sized), drain: &PayloadDeleteDrain, diff --git a/crates/tracedecay-sessions/src/runtime/lcm/gc/placeholder_scan.rs b/crates/tracedecay-sessions/src/runtime/lcm/gc/placeholder_scan.rs index 8dead4fe4b..ce85ca5fa2 100644 --- a/crates/tracedecay-sessions/src/runtime/lcm/gc/placeholder_scan.rs +++ b/crates/tracedecay-sessions/src/runtime/lcm/gc/placeholder_scan.rs @@ -52,6 +52,21 @@ pub(crate) fn gc_prefix_ref_like_patterns(payload_ref: &str) -> Vec { .collect() } +/// Prefilter patterns for text that still holds a *live* placeholder naming +/// `payload_ref`. +/// +/// A live placeholder is a bracket whose lowercased text starts with one of +/// [`LIVE_PREFIX_REWRITES`] and carries `ref=` after that prefix, +/// so `%prefix%ref%` matches every row a tombstone rewrite could change. It +/// matches strictly fewer rows than a bare `%ref%`, which also drags in inline +/// bodies and already-tombstoned placeholders that the rewrite leaves alone. +pub(crate) fn live_prefix_ref_like_patterns(payload_ref: &str) -> Vec { + LIVE_PREFIX_REWRITES + .iter() + .map(|(prefix, _)| format!("%{prefix}%{payload_ref}%")) + .collect() +} + pub(crate) fn live_prefix_like_patterns() -> Vec { LIVE_PREFIX_REWRITES .iter() diff --git a/crates/tracedecay-sessions/src/runtime/lcm/gc/tests.rs b/crates/tracedecay-sessions/src/runtime/lcm/gc/tests.rs index 7bb5656084..f47c1e5634 100644 --- a/crates/tracedecay-sessions/src/runtime/lcm/gc/tests.rs +++ b/crates/tracedecay-sessions/src/runtime/lcm/gc/tests.rs @@ -1447,3 +1447,498 @@ fn committed_delete_retry_succeeds_after_same_id_content_restore() -> Result<(), assert!(!path.exists()); Ok(()) } + +// --------------------------------------------------------------------------- +// SQL batching regression coverage. +// +// These tests assert on the number of statements the GC paths issue and on the +// rows they leave behind. They never assert on elapsed time: the point is that +// a set-sized workload costs a fixed number of round trips, which is a property +// of the SQL, not of the machine. +// --------------------------------------------------------------------------- + +#[derive(Default)] +struct SqlLog { + statements: std::cell::RefCell>, +} + +impl SqlLog { + fn record(&self, sql: &str) { + self.statements.borrow_mut().push(sql.to_string()); + } + + fn matching(&self, needle: &str) -> usize { + self.statements + .borrow() + .iter() + .filter(|sql| sql.contains(needle)) + .count() + } +} + +/// Transparent `Executor` wrapper that records every statement it forwards. +struct CountingExecutor<'a, E: ?Sized> { + inner: &'a E, + log: &'a SqlLog, +} + +impl QueryExecutor for CountingExecutor<'_, E> { + async fn query

( + &self, + sql: &str, + params: P, + ) -> tracedecay_runtime_core::db::engine::Result + where + P: tracedecay_runtime_core::db::engine::IntoParams, + { + self.log.record(sql); + self.inner.query(sql, params).await + } +} + +impl Executor for CountingExecutor<'_, E> { + async fn execute

( + &self, + sql: &str, + params: P, + ) -> tracedecay_runtime_core::db::engine::Result + where + P: tracedecay_runtime_core::db::engine::IntoParams, + { + self.log.record(sql); + self.inner.execute(sql, params).await + } + + async fn execute_batch(&self, sql: &str) -> tracedecay_runtime_core::db::engine::Result<()> { + self.log.record(sql); + self.inner.execute_batch(sql).await + } +} + +fn batch_ref(index: usize) -> String { + format!("payload_batch_{index:04}.payload") +} + +/// M11: the pending-delete drain must probe `lcm_external_payloads` once for the +/// whole tombstone set, not once per tombstone. +#[tokio::test] +async fn pending_delete_drain_probes_metadata_once_for_the_whole_set() -> Result<(), String> { + const TOMBSTONES: usize = 6; + + let store = test_store().await?; + let dir = payload::payload_dir(&store.storage_root); + fs::create_dir_all(&dir).map_err(|err| err.to_string())?; + + let mut refs = Vec::new(); + for index in 0..TOMBSTONES { + let payload_ref = batch_ref(index); + fs::write(dir.join(&payload_ref), format!("body {index}").as_bytes()) + .map_err(|err| err.to_string())?; + let (hash, bytes, chars) = + payload::payload_file_fingerprint(&dir, &payload_ref).map_err(|err| err.to_string())?; + stage_payload_delete(&store.conn, &payload_ref, Some(&hash), bytes, chars) + .await + .map_err(|err| err.to_string())?; + refs.push(payload_ref); + } + + let log = SqlLog::default(); + let transaction = store + .conn + .transaction_with_behavior(TransactionBehavior::Immediate) + .await + .map_err(|err| err.to_string())?; + let drain = { + let counting = CountingExecutor { + inner: &transaction, + log: &log, + }; + drain_pending_payload_deletes_in_transaction(&counting, &store.storage_root) + .await + .map_err(|err| err.to_string())? + }; + transaction.commit().await.map_err(|err| err.to_string())?; + + assert_eq!( + log.matching("FROM lcm_external_payloads"), + 1, + "metadata existence probe must be batched, saw statements: {:?}", + log.statements.borrow() + ); + assert_eq!(drain.outcomes.removed.count, TOMBSTONES); + assert_eq!(drain.outcomes.failed.count, 0); + for payload_ref in &refs { + assert!( + !dir.join(payload_ref).exists(), + "{payload_ref} still on disk" + ); + assert!( + schema::get_gc_meta(&store.conn, &pending_payload_delete_key(payload_ref)) + .await + .map_err(|err| err.to_string())? + .is_none(), + "{payload_ref} tombstone not cleared" + ); + } + Ok(()) +} + +/// M11 equivalence: a tombstone whose metadata row is still present must be +/// preserved (not unlinked) and a tombstone whose row is gone must be reaped, +/// in the same drain — the batched probe must not conflate the two. +#[tokio::test] +async fn pending_delete_drain_batches_mixed_metadata_presence() -> Result<(), String> { + let store = test_store().await?; + let dir = payload::payload_dir(&store.storage_root); + fs::create_dir_all(&dir).map_err(|err| err.to_string())?; + + // Tombstone A: metadata row still present -> preserved. + let live_ref = seed_payload(&store, "message-live", "live body").await?; + let (live_hash, live_bytes, live_chars) = + payload::payload_file_fingerprint(&dir, &live_ref).map_err(|err| err.to_string())?; + stage_payload_delete( + &store.conn, + &live_ref, + Some(&live_hash), + live_bytes, + live_chars, + ) + .await + .map_err(|err| err.to_string())?; + + // Tombstone B: no metadata row -> removed. + let dead_ref = batch_ref(99); + fs::write(dir.join(&dead_ref), b"dead body").map_err(|err| err.to_string())?; + let (dead_hash, dead_bytes, dead_chars) = + payload::payload_file_fingerprint(&dir, &dead_ref).map_err(|err| err.to_string())?; + stage_payload_delete( + &store.conn, + &dead_ref, + Some(&dead_hash), + dead_bytes, + dead_chars, + ) + .await + .map_err(|err| err.to_string())?; + + let drain = drain_pending_payload_deletes(&store.conn, &store.storage_root) + .await + .map_err(|err| err.to_string())?; + + assert_eq!(drain.outcomes.preserved.refs, [live_ref.clone()]); + assert_eq!(drain.outcomes.removed.refs, [dead_ref.clone()]); + assert!(dir.join(&live_ref).is_file(), "live payload was unlinked"); + assert!(!dir.join(&dead_ref).exists(), "dead payload survived"); + Ok(()) +} + +/// Distinguishing fragment of the byte-bounded `referenced_payload_refs` scan. +/// The provider predicate is shared by unrelated metadata queries in the same +/// GC pass and therefore cannot identify this round trip on its own. +const CLOSURE_SCAN_NEEDLE: &str = "cumulative_bytes <= ?5 OR page_row = 1"; + +/// Seeds `count` payloads that are on disk with metadata rows, carry no live +/// reference, and already hold an aged `unreferenced` GC mark, so one apply pass +/// reaps all of them. +async fn seed_reapable_payloads(store: &TestStore, count: usize) -> Result, String> { + let mut refs = Vec::new(); + for index in 0..count { + let payload_ref = + seed_payload(store, &format!("message-{index}"), &format!("body {index}")).await?; + drop_raw_reference(store, &payload_ref).await?; + insert_gc_mark(store, &payload_ref, "unreferenced", 0).await?; + refs.push(payload_ref); + } + Ok(refs) +} + +/// M1: the reference-closure scan must run once for the batch, not once per +/// payload. With every raw message dropped the scan reads a single empty page, +/// so each invocation is exactly one statement and the count is the call count. +#[tokio::test] +async fn unreferenced_reap_scans_reference_closure_once_for_the_batch() -> Result<(), String> { + const PAYLOADS: usize = 6; + + let store = test_store().await?; + let refs = seed_reapable_payloads(&store, PAYLOADS).await?; + let cfg = LcmGcConfig { + grace_seconds: LcmGcConfig::MIN_GRACE_SECONDS, + backup_before_reap: false, + max_batch_size: 64, + ..Default::default() + } + .normalized(); + + let log = SqlLog::default(); + let transaction = store + .conn + .transaction_with_behavior(TransactionBehavior::Immediate) + .await + .map_err(|err| err.to_string())?; + let report = { + let counting = CountingExecutor { + inner: &transaction, + log: &log, + }; + run_payload_gc_in_transaction( + &counting, + &store.storage_root, + PROVIDER, + None, + &cfg, + true, + 1_000_000, + ) + .await + .map_err(|err| err.to_string())? + }; + transaction.commit().await.map_err(|err| err.to_string())?; + + assert_eq!(report.unreferenced.count, PAYLOADS); + let scans = log.matching(CLOSURE_SCAN_NEEDLE); + assert_eq!( + scans, 2, + "expected one pass-level scan plus one batch-shared scan, saw {scans} for {PAYLOADS} payloads" + ); + let mark_deletes = log.matching("DELETE FROM lcm_gc_marks"); + assert_eq!( + mark_deletes, 1, + "expected one batch GC-mark delete, saw {mark_deletes} for {PAYLOADS} payloads" + ); + for payload_ref in &refs { + assert!( + payload::load_payload_metadata(&store.conn, payload_ref) + .await + .is_err(), + "{payload_ref} metadata survived" + ); + } + Ok(()) +} + +/// M1 equivalence: a payload that *is* still referenced must still abort with +/// `StillReferenced` even when it shares a cached closure with payloads that +/// were reaped earlier in the same batch. +#[tokio::test] +async fn shared_reference_closure_still_rejects_a_referenced_payload() -> Result<(), String> { + let store = test_store().await?; + let reaped = seed_payload(&store, "message-reaped", "reap me").await?; + drop_raw_reference(&store, &reaped).await?; + let kept = seed_payload(&store, "message-kept", "keep me").await?; + + let transaction = store + .conn + .transaction_with_behavior(TransactionBehavior::Immediate) + .await + .map_err(|err| err.to_string())?; + let mut cache = payload::ReferencedClosureCache::default(); + payload::delete_external_payload_in_transaction_with_cache( + &transaction, + &store.storage_root, + &reaped, + &payload::DeleteOpts::default(), + &mut cache, + ) + .await + .map_err(|err| err.to_string())?; + let still_referenced = payload::delete_external_payload_in_transaction_with_cache( + &transaction, + &store.storage_root, + &kept, + &payload::DeleteOpts::default(), + &mut cache, + ) + .await; + transaction.commit().await.map_err(|err| err.to_string())?; + + assert!( + matches!(still_referenced, Err(LcmError::StillReferenced)), + "referenced payload was not rejected" + ); + assert!( + payload::load_payload_metadata(&store.conn, &kept) + .await + .is_ok(), + "referenced payload metadata was deleted" + ); + assert!( + payload::load_payload_metadata(&store.conn, &reaped) + .await + .is_err(), + "unreferenced payload was not deleted" + ); + Ok(()) +} + +/// M2: the residual-placeholder sweep must prefilter on live-prefix + ref, not +/// on a bare `%ref%`. One `LIKE` term per text column per pattern, so the +/// narrowed form emits `4 * LIVE_PREFIX_REWRITES.len()` terms. +#[tokio::test] +async fn residual_placeholder_sweep_prefilters_on_live_prefixes() -> Result<(), String> { + let store = test_store().await?; + let payload_ref = seed_payload(&store, "message-1", "body to tombstone").await?; + + let log = SqlLog::default(); + let transaction = store + .conn + .transaction_with_behavior(TransactionBehavior::Immediate) + .await + .map_err(|err| err.to_string())?; + { + let counting = CountingExecutor { + inner: &transaction, + log: &log, + }; + payload::delete_external_payload_in_transaction( + &counting, + &store.storage_root, + &payload_ref, + &payload::DeleteOpts { + rewrite_placeholders: true, + remove_file: false, + verify_hash: false, + }, + ) + .await + .map_err(|err| err.to_string())?; + } + transaction.commit().await.map_err(|err| err.to_string())?; + + let sweep = log + .statements + .borrow() + .iter() + .find(|sql| sql.contains("WHERE payload_ref = ? OR")) + .cloned() + .ok_or_else(|| "residual placeholder sweep did not run".to_string())?; + let like_terms = sweep.matches("LIKE ? COLLATE NOCASE").count(); + assert_eq!( + like_terms, + 4 * LIVE_PREFIX_REWRITES.len(), + "sweep prefilter is not live-prefix anchored: {sweep}" + ); + Ok(()) +} + +/// M2 equivalence: the narrowed prefilter must rewrite exactly the rows the bare +/// `%ref%` form rewrote — live placeholders in every text column, plus the +/// stored `payload_ref` — and must leave inline prose that merely mentions the +/// ref, and already-tombstoned placeholders, untouched. +#[tokio::test] +async fn narrowed_prefilter_rewrites_the_same_rows() -> Result<(), String> { + let store = test_store().await?; + let payload_ref = seed_payload(&store, "message-live", "body to tombstone").await?; + + let live = format!("[externalized tool output: bytes=4 ref={payload_ref}; out]"); + let already_gcd = format!("[gc'd externalized payload: bytes=4 ref={payload_ref}; gone]"); + let prose = format!("the operator mentioned {payload_ref} in a note"); + insert_raw_message( + &store.conn, + RawMessage { + session_id: "session-a", + message_id: "message-other-live", + storage_kind: "inline", + payload_ref: None, + content: Some(&live), + snippet_text: &live, + index_text: &live, + metadata_json: Some(&live), + }, + ) + .await?; + insert_raw_message( + &store.conn, + RawMessage { + session_id: "session-a", + message_id: "message-already-gcd", + storage_kind: "inline", + payload_ref: None, + content: Some(&already_gcd), + snippet_text: &already_gcd, + index_text: &already_gcd, + metadata_json: Some(&already_gcd), + }, + ) + .await?; + insert_raw_message( + &store.conn, + RawMessage { + session_id: "session-a", + message_id: "message-prose", + storage_kind: "inline", + payload_ref: None, + content: Some(&prose), + snippet_text: &prose, + index_text: &prose, + metadata_json: Some(&prose), + }, + ) + .await?; + + let transaction = store + .conn + .transaction_with_behavior(TransactionBehavior::Immediate) + .await + .map_err(|err| err.to_string())?; + payload::delete_external_payload_in_transaction( + &transaction, + &store.storage_root, + &payload_ref, + &payload::DeleteOpts { + rewrite_placeholders: true, + remove_file: false, + verify_hash: false, + }, + ) + .await + .map_err(|err| err.to_string())?; + transaction.commit().await.map_err(|err| err.to_string())?; + + let mut rows = store + .conn + .query( + "SELECT message_id, storage_kind, payload_ref, snippet_text, index_text + FROM lcm_raw_messages ORDER BY message_id", + (), + ) + .await + .map_err(|err| err.to_string())?; + let mut seen = Vec::new(); + while let Some(row) = rows.next().await.map_err(|err| err.to_string())? { + let message_id: String = row.get(0).map_err(|err| err.to_string())?; + let storage_kind: String = row.get(1).map_err(|err| err.to_string())?; + let stored_ref: Option = row.get(2).unwrap_or(None); + let snippet: String = row.get(3).map_err(|err| err.to_string())?; + let index_text: String = row.get(4).map_err(|err| err.to_string())?; + seen.push((message_id, storage_kind, stored_ref, snippet, index_text)); + } + drop(rows); + + for (message_id, storage_kind, stored_ref, snippet, index_text) in seen { + match message_id.as_str() { + "message-live" => { + assert_eq!(storage_kind, "inline", "external row was not inlined"); + assert_eq!(stored_ref, None, "stored payload_ref was not cleared"); + assert!(text_has_tombstoned_payload_ref(&snippet, &payload_ref)); + assert!(text_has_tombstoned_payload_ref(&index_text, &payload_ref)); + } + "message-other-live" => { + assert!( + text_has_tombstoned_payload_ref(&snippet, &payload_ref), + "live tool-output placeholder was not tombstoned: {snippet}" + ); + assert!(text_has_tombstoned_payload_ref(&index_text, &payload_ref)); + } + "message-already-gcd" => { + assert_eq!(snippet, already_gcd, "already-tombstoned row was rewritten"); + assert_eq!(index_text, already_gcd); + } + "message-prose" => { + assert_eq!(snippet, prose, "inline prose was rewritten"); + assert_eq!(index_text, prose); + } + other => return Err(format!("unexpected row {other}")), + } + } + Ok(()) +} diff --git a/crates/tracedecay-sessions/src/runtime/lcm/payload.rs b/crates/tracedecay-sessions/src/runtime/lcm/payload.rs index 22915bd072..625403fbb3 100644 --- a/crates/tracedecay-sessions/src/runtime/lcm/payload.rs +++ b/crates/tracedecay-sessions/src/runtime/lcm/payload.rs @@ -13,8 +13,8 @@ mod rollback; #[cfg(test)] pub use delete_recovery::delete_external_payload; pub use delete_recovery::{ - CommittedPayloadRemoval, PreparedPayloadDelete, payload_file_fingerprint, - remove_committed_payload_file, + CommittedPayloadRemoval, PreparedPayloadDelete, ReferencedClosureCache, + payload_file_fingerprint, remove_committed_payload_file, }; pub use delete_recovery::{DeleteOpts, DeleteOutcome}; #[cfg(test)] @@ -36,6 +36,27 @@ pub async fn delete_external_payload_in_transaction( .await } +/// [`delete_external_payload_in_transaction`] for callers that delete a batch of +/// payloads inside one transaction and can share a single reference-closure +/// scan across the batch. See [`ReferencedClosureCache`] for why the shared +/// snapshot answers each payload exactly as its own scan would have. +pub async fn delete_external_payload_in_transaction_with_cache( + conn: &(impl Executor + ?Sized), + storage_root: &Path, + payload_ref: &str, + opts: &DeleteOpts, + referenced: &mut ReferencedClosureCache, +) -> Result { + delete_recovery::delete_external_payload_in_transaction_with_cache( + conn, + storage_root, + payload_ref, + opts, + referenced, + ) + .await +} + pub fn canonical_storage_root(storage_root: &Path) -> Result { filesystem_authority::canonical_storage_root(storage_root) } diff --git a/crates/tracedecay-sessions/src/runtime/lcm/payload/delete_recovery.rs b/crates/tracedecay-sessions/src/runtime/lcm/payload/delete_recovery.rs index 83adcda404..e9e3da9b53 100644 --- a/crates/tracedecay-sessions/src/runtime/lcm/payload/delete_recovery.rs +++ b/crates/tracedecay-sessions/src/runtime/lcm/payload/delete_recovery.rs @@ -1,3 +1,4 @@ +use std::collections::{BTreeMap, BTreeSet}; use std::fs; use std::path::Path; @@ -47,6 +48,55 @@ pub enum CommittedPayloadRemoval { ReplacementPreserved, } +/// Memoizes [`gc::referenced_payload_refs`] for a caller that deletes many +/// payloads inside one transaction. +/// +/// Reuse is exact rather than approximate. Whether a payload `X` sits in the +/// live reference closure is only ever changed by `X`'s own deletion: +/// +/// * `tombstone_residual_placeholders` rewrites a bracket placeholder only when +/// [`gc::tombstone_placeholder_in_text`] finds that placeholder's `ref=` equal +/// to the ref being deleted, and `extract_payload_refs_from_text` yields at +/// most one ref per bracket — so tombstoning `Y` cannot drop a reference to +/// any `X != Y`. +/// * The `payload_ref` column is only nulled on rows whose column already equals +/// the ref being deleted. +/// +/// Nothing else in the delete path writes `lcm_raw_messages`, the sole table the +/// closure scan reads. One scan per provider therefore answers every iteration +/// exactly as a fresh per-payload scan would have. +#[derive(Debug, Default)] +pub struct ReferencedClosureCache { + by_provider: BTreeMap>, +} + +impl ReferencedClosureCache { + async fn is_referenced( + &mut self, + conn: &(impl Executor + ?Sized), + provider: &str, + payload_ref: &str, + ) -> Result { + if !self.by_provider.contains_key(provider) { + let refs = gc::referenced_payload_refs(conn, provider, None).await?; + self.by_provider.insert(provider.to_string(), refs); + } + Ok(self + .by_provider + .get(provider) + .is_some_and(|refs| refs.contains(payload_ref))) + } + + /// Applies the shrink a completed placeholder rewrite performed on the + /// database, so a repeated ref in the same batch reads the post-rewrite + /// truth instead of the pre-rewrite snapshot. + fn forget(&mut self, payload_ref: &str) { + for refs in self.by_provider.values_mut() { + refs.remove(payload_ref); + } + } +} + #[cfg(test)] pub async fn delete_external_payload( conn: &Connection, @@ -116,6 +166,24 @@ pub(super) async fn delete_external_payload_in_transaction( storage_root: &Path, payload_ref: &str, opts: &DeleteOpts, +) -> Result { + let mut referenced = ReferencedClosureCache::default(); + delete_external_payload_in_transaction_with_cache( + conn, + storage_root, + payload_ref, + opts, + &mut referenced, + ) + .await +} + +pub(super) async fn delete_external_payload_in_transaction_with_cache( + conn: &(impl Executor + ?Sized), + storage_root: &Path, + payload_ref: &str, + opts: &DeleteOpts, + referenced: &mut ReferencedClosureCache, ) -> Result { validate_payload_ref(payload_ref)?; // The DB-side cleanup below must still run for a store whose payload @@ -179,9 +247,9 @@ pub(super) async fn delete_external_payload_in_transaction( let tombstone_missing_payload = opts.rewrite_placeholders && !opts.remove_file && !opts.verify_hash; if let Some(metadata) = metadata.as_ref() - && gc::referenced_payload_refs(conn, &metadata.provider, None) + && referenced + .is_referenced(conn, &metadata.provider, payload_ref) .await? - .contains(payload_ref) && !tombstone_missing_payload { return Err(LcmError::StillReferenced); @@ -198,6 +266,9 @@ pub(super) async fn delete_external_payload_in_transaction( .await?; if opts.rewrite_placeholders { placeholders_rewritten = tombstone_residual_placeholders(conn, payload_ref).await?; + // The rewrite above is unscoped and exhaustive, so this ref is now + // absent from the live closure of every provider. + referenced.forget(payload_ref); } let file_removed = opts.remove_file && file_existed; @@ -378,16 +449,22 @@ async fn tombstone_residual_placeholders( conn: &(impl Executor + ?Sized), payload_ref: &str, ) -> Result { - let like_sql = gc::placeholder_text_like_sql(1); + // `payload_ref = ?` is the indexable arm and covers every row whose stored + // ref must be cleared. The `LIKE` arm is only reachable for text that embeds + // a placeholder, which no index can answer; narrowing it from a bare + // `%ref%` to `%live-prefix%ref%` keeps the arm exact — a rewrite only ever + // changes a bracket that starts with a live prefix and carries `ref=` + // after it — while dropping inline bodies and already-tombstoned rows that + // the bare pattern dragged back for no change. + let like_patterns = gc::live_prefix_ref_like_patterns(payload_ref); + let like_sql = gc::placeholder_text_like_sql(like_patterns.len()); let sql = format!( "SELECT store_id, storage_kind, payload_ref, content, snippet_text, index_text, metadata_json FROM lcm_raw_messages WHERE payload_ref = ? OR {like_sql}" ); let mut values = vec![SqlValue::Text(payload_ref.to_string())]; - values.extend(gc::bind_placeholder_like_patterns(&[format!( - "%{payload_ref}%" - )])); + values.extend(gc::bind_placeholder_like_patterns(&like_patterns)); let mut rows = conn.query(&sql, values).await?; let mut updates = Vec::new(); while let Some(row) = rows.next().await? { diff --git a/crates/tracedecay-sessions/src/runtime/lcm/query.rs b/crates/tracedecay-sessions/src/runtime/lcm/query.rs index 7e63991c10..8455b9ad7d 100644 --- a/crates/tracedecay-sessions/src/runtime/lcm/query.rs +++ b/crates/tracedecay-sessions/src/runtime/lcm/query.rs @@ -1002,8 +1002,82 @@ fn sort_hits(hits: &mut [LcmGrepHit], sort: LcmGrepSort) { #[cfg(test)] mod tests { + use std::cell::Cell; + + use tracedecay_runtime_core::db::engine::{ + Executor, IntoParams, QueryExecutor, Result as EngineResult, Row, Rows, TestConnection, + Value, params, + }; + use super::*; + struct CountingQuery<'a> { + inner: &'a TestConnection, + queries: Cell, + rows_visited: Cell, + } + + impl<'a> CountingQuery<'a> { + fn new(inner: &'a TestConnection) -> Self { + Self { + inner, + queries: Cell::new(0), + rows_visited: Cell::new(0), + } + } + } + + impl QueryExecutor for CountingQuery<'_> { + async fn query

(&self, sql: &str, params: P) -> EngineResult + where + P: IntoParams, + { + self.queries.set(self.queries.get() + 1); + let mut rows = self.inner.query(sql, params).await?; + let columns = (0..rows.column_count()) + .map(|index| rows.column_name(index).unwrap_or_default().to_string()) + .collect::>(); + let mut replay = Vec::new(); + while let Some(row) = rows.next().await? { + let mut values = Vec::new(); + let mut column = 0_i32; + while let Ok(value) = row.get::(column) { + values.push(value); + column += 1; + } + replay.push(Row::from_values(values)); + } + self.rows_visited + .set(self.rows_visited.get().saturating_add(replay.len())); + Ok(Rows::from_parts(columns, replay)) + } + } + + async fn query_test_store() -> (tempfile::TempDir, TestConnection) { + let temp = tempfile::tempdir().expect("temporary query store"); + let conn = TestConnection::open(&temp.path().join("sessions.db")); + conn.execute_batch( + "CREATE TABLE sessions ( + provider TEXT NOT NULL, + session_id TEXT NOT NULL, + project_key TEXT NOT NULL, + project_path TEXT NOT NULL, + PRIMARY KEY(provider, session_id) + );", + ) + .await + .expect("session schema"); + schema::ensure_lcm_schema(&*conn).await.expect("LCM schema"); + conn.execute( + "INSERT INTO sessions(provider, session_id, project_key, project_path) + VALUES ('cursor', 'session-a', '/p', '/p')", + (), + ) + .await + .expect("session fixture"); + (temp, conn) + } + fn summary_source(store_id: i64) -> LcmExpandedSummarySource { LcmExpandedSummarySource { source_ref: LcmSourceRef::RawMessage { store_id }, @@ -1026,4 +1100,105 @@ mod tests { assert_eq!(pagination.next_source_offset, Some(1)); assert!(pagination.has_more); } + + #[tokio::test] + async fn like_fallback_visits_only_the_outer_rerank_candidate_budget() { + let (_temp, conn) = query_test_store().await; + for ordinal in 0..100_i64 { + let message_id = format!("message-{ordinal}"); + conn.execute( + "INSERT INTO lcm_raw_messages ( + provider, message_id, session_id, role, ordinal, timestamp, + content, content_hash, storage_kind, snippet_text, index_text + ) VALUES ( + 'cursor', ?1, 'session-a', 'assistant', ?2, ?2, + '雪 candidate', 'hash', 'inline', '雪 candidate', '雪 candidate' + )", + params![message_id, ordinal], + ) + .await + .expect("raw candidate"); + } + + let counted = CountingQuery::new(&conn); + let outcome = grep( + &counted, + LcmGrepRequest { + provider: "cursor".to_string(), + query: "雪".to_string(), + scope: LcmScope::Session, + session_id: Some("session-a".to_string()), + include_summaries: false, + limit: 2, + sort: LcmGrepSort::Recency, + source: None, + role: None, + start_time: None, + end_time: None, + git_filter: Default::default(), + }, + LcmGrepFilters::default(), + None, + ) + .await + .expect("LIKE grep"); + + assert_eq!(outcome.hits.len(), 2); + assert_eq!(counted.queries.get(), 1); + assert!( + counted.rows_visited.get() <= rerank_fetch_limit(2), + "LIKE fallback visited {} rows for {} returned hits", + counted.rows_visited.get(), + outcome.hits.len() + ); + } + + #[tokio::test] + async fn expand_query_batches_explicit_summary_hydration_roundtrips() { + let (_temp, conn) = query_test_store().await; + let mut node_ids = Vec::new(); + for ordinal in 0..8_i64 { + let node_id = format!("node-{ordinal}"); + let summary_text = format!("summary {ordinal}"); + let summary_hash = crate::retrieval_content::projected_content_hash(&summary_text); + conn.execute( + "INSERT INTO lcm_summary_nodes ( + node_id, provider, conversation_id, session_id, depth, summary_text, + summary_hash, summary_token_count, source_token_count + ) VALUES (?1, 'cursor', 'conversation-a', 'session-a', 0, ?2, ?3, 1, 1)", + params![ + node_id.as_str(), + summary_text.as_str(), + summary_hash.as_str() + ], + ) + .await + .expect("summary node"); + node_ids.push(node_id); + } + + let counted = CountingQuery::new(&conn); + let response = expand_query( + &counted, + LcmExpandQueryRequest { + provider: "cursor".to_string(), + session_id: "session-a".to_string(), + prompt: "summarize".to_string(), + query: None, + node_ids, + max_results: 8, + max_tokens: 100, + context_max_tokens: 10_000, + }, + ) + .await + .expect("expand query"); + + assert_eq!(response.node_ids.len(), 8); + assert!( + counted.queries.get() <= 3, + "explicit summary hydration used {} DB roundtrips", + counted.queries.get() + ); + } } From c8835aa6cf488ca0ca88839a0eb73815a74f4fb1 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Mon, 24 Aug 2026 01:31:35 +0000 Subject: [PATCH 2/4] fix(runtime-core): commit the missing background CPU module perf(observation) landed an import of tracedecay_runtime_core::background_cpu, but neither the module file nor its declaration was added, so the pushed integration branch does not compile: every branch cut from it fails on an unresolved import. Co-Authored-By: Claude Opus 5 (1M context) --- .../src/background_cpu.rs | 461 ++++++++++++++++++ crates/tracedecay-runtime-core/src/lib.rs | 1 + 2 files changed, 462 insertions(+) create mode 100644 crates/tracedecay-runtime-core/src/background_cpu.rs diff --git a/crates/tracedecay-runtime-core/src/background_cpu.rs b/crates/tracedecay-runtime-core/src/background_cpu.rs new file mode 100644 index 0000000000..75366049a4 --- /dev/null +++ b/crates/tracedecay-runtime-core/src/background_cpu.rs @@ -0,0 +1,461 @@ +//! Process-wide weighted admission for background CPU work. +//! +//! The authority counts active CPU units rather than owning an executor. Code +//! indexing, semantic native threads, and session preparation can therefore +//! use their existing execution substrates while sharing one hard process +//! ceiling. FIFO waiter order prevents a continuously busy class from starving +//! another class, and RAII releases capacity on success, cancellation, or +//! unwind. + +use std::cell::Cell; +use std::collections::VecDeque; +use std::fmt; +use std::num::NonZeroUsize; +use std::sync::{Arc, Condvar, Mutex, OnceLock}; + +#[derive(Debug)] +struct BackgroundCpuWaiterV1 { + units: usize, +} + +#[derive(Default)] +struct BackgroundCpuStateV1 { + active_units: usize, + waiters: VecDeque>, +} + +/// One process-wide background CPU budget shared across subsystems. +pub struct ProcessBackgroundCpuV1 { + width: NonZeroUsize, + state: Mutex, + available: Condvar, +} + +impl fmt::Debug for ProcessBackgroundCpuV1 { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("ProcessBackgroundCpuV1") + .field("width", &self.width) + .field("active_units", &self.active_units()) + .finish_non_exhaustive() + } +} + +thread_local! { + static BACKGROUND_CPU_DEPTH: Cell = const { Cell::new(0) }; + static BACKGROUND_CPU_UNITS: Cell = const { Cell::new(0) }; +} + +struct BackgroundCpuScopeV1; + +impl BackgroundCpuScopeV1 { + fn enter() -> Self { + BACKGROUND_CPU_DEPTH.with(|depth| depth.set(depth.get().saturating_add(1))); + Self + } +} + +struct YieldedBackgroundCpuV1<'a> { + authority: &'a Arc, + units: usize, + depth: usize, +} + +impl Drop for YieldedBackgroundCpuV1<'_> { + fn drop(&mut self) { + self.authority.admit_units(self.units); + BACKGROUND_CPU_UNITS.with(|units| units.set(self.units)); + BACKGROUND_CPU_DEPTH.with(|depth| depth.set(self.depth)); + } +} + +impl Drop for BackgroundCpuScopeV1 { + fn drop(&mut self) { + BACKGROUND_CPU_DEPTH.with(|depth| { + let remaining = depth.get().saturating_sub(1); + depth.set(remaining); + if remaining == 0 { + BACKGROUND_CPU_UNITS.with(|units| units.set(0)); + } + }); + } +} + +impl ProcessBackgroundCpuV1 { + fn new(width: NonZeroUsize) -> Self { + Self { + width, + state: Mutex::new(BackgroundCpuStateV1::default()), + available: Condvar::new(), + } + } + + #[must_use] + pub const fn width(&self) -> NonZeroUsize { + self.width + } + + #[must_use] + pub fn active_units(&self) -> usize { + self.state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .active_units + } + + #[must_use] + pub fn waiting_work_units(&self) -> usize { + let state = self + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + waiting_units(&state) + } + + /// Acquire one CPU unit, waiting in FIFO order when the process budget is + /// full. The returned guard must remain alive for the active work unit. + pub fn acquire(self: &Arc) -> BackgroundCpuPermitV1 { + self.acquire_units(1) + } + + /// Acquire one CPU unit only when no earlier waiter exists and capacity is + /// immediately available. + pub fn try_acquire(self: &Arc) -> Option { + let mut state = self + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if !state.waiters.is_empty() || state.active_units >= self.width.get() { + return None; + } + state.active_units += 1; + record_state(&state, self.width); + Some(BackgroundCpuPermitV1 { + authority: Arc::clone(self), + units: 1, + }) + } + + /// Run one active work unit under the process budget. Nested work on the + /// same thread reuses a sufficient parent admission instead of waiting on + /// itself. + pub fn with_permit(self: &Arc, operation: impl FnOnce() -> R) -> R { + self.with_permits(1, operation) + } + + /// Run a weighted work unit, clamped to the entire process width. Semantic + /// inference uses its native intra-op thread count as the weight; ordinary + /// index/session preparation uses one. + pub fn with_permits( + self: &Arc, + requested_units: usize, + operation: impl FnOnce() -> R, + ) -> R { + let units = requested_units.max(1).min(self.width.get()); + let active_units = BACKGROUND_CPU_UNITS.with(Cell::get); + if active_units >= units { + return operation(); + } + if active_units > 0 { + let depth = BACKGROUND_CPU_DEPTH.with(Cell::get); + self.release(active_units); + BACKGROUND_CPU_UNITS.with(|active| active.set(0)); + BACKGROUND_CPU_DEPTH.with(|active| active.set(0)); + let _restore = YieldedBackgroundCpuV1 { + authority: self, + units: active_units, + depth, + }; + let _permit = self.acquire_units(units); + let _scope = BackgroundCpuScopeV1::enter(); + BACKGROUND_CPU_UNITS.with(|active| active.set(units)); + return operation(); + } + let _permit = self.acquire_units(units); + let _scope = BackgroundCpuScopeV1::enter(); + BACKGROUND_CPU_UNITS.with(|active| active.set(units)); + operation() + } + + /// Temporarily yield the caller's active units while a nested executor + /// fans out independently admitted leaf work. This prevents a parent + /// Rayon worker from holding capacity while it waits for child workers, + /// including a full-width weighted child. Capacity is reacquired before + /// the parent resumes, including during unwind. + pub fn with_yielded_permits(self: &Arc, operation: impl FnOnce() -> R) -> R { + let units = BACKGROUND_CPU_UNITS.with(Cell::get); + if units == 0 { + return operation(); + } + let depth = BACKGROUND_CPU_DEPTH.with(Cell::get); + self.release(units); + BACKGROUND_CPU_UNITS.with(|active| active.set(0)); + BACKGROUND_CPU_DEPTH.with(|active| active.set(0)); + let _restore = YieldedBackgroundCpuV1 { + authority: self, + units, + depth, + }; + operation() + } + + fn acquire_units(self: &Arc, units: usize) -> BackgroundCpuPermitV1 { + self.admit_units(units); + BackgroundCpuPermitV1 { + authority: Arc::clone(self), + units, + } + } + + fn admit_units(&self, units: usize) { + let waiter = Arc::new(BackgroundCpuWaiterV1 { units }); + let mut state = self + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + state.waiters.push_back(Arc::clone(&waiter)); + record_state(&state, self.width); + loop { + let is_front = state + .waiters + .front() + .is_some_and(|front| Arc::ptr_eq(front, &waiter)); + if is_front && state.active_units.saturating_add(waiter.units) <= self.width.get() { + state.waiters.pop_front(); + state.active_units += waiter.units; + record_state(&state, self.width); + self.available.notify_all(); + return; + } + state = self + .available + .wait(state) + .unwrap_or_else(std::sync::PoisonError::into_inner); + } + } + + fn release(&self, units: usize) { + let mut state = self + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + debug_assert!(state.active_units >= units); + state.active_units = state.active_units.saturating_sub(units); + record_state(&state, self.width); + self.available.notify_all(); + } +} + +fn record_state(state: &BackgroundCpuStateV1, width: NonZeroUsize) { + hotpath::gauge!("runtime_core.background_cpu.width").set(width.get()); + hotpath::gauge!("runtime_core.background_cpu.active_units").set(state.active_units); + hotpath::gauge!("runtime_core.background_cpu.waiting_work_units").set(waiting_units(state)); +} + +fn waiting_units(state: &BackgroundCpuStateV1) -> usize { + state + .waiters + .iter() + .fold(0usize, |total, waiter| total.saturating_add(waiter.units)) +} + +/// RAII ownership of active CPU capacity. Dropping it is cancellation-safe and +/// releases the exact acquired weight. +pub struct BackgroundCpuPermitV1 { + authority: Arc, + units: usize, +} + +impl fmt::Debug for BackgroundCpuPermitV1 { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("BackgroundCpuPermitV1") + .field("units", &self.units) + .finish_non_exhaustive() + } +} + +impl Drop for BackgroundCpuPermitV1 { + fn drop(&mut self) { + self.authority.release(self.units); + } +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq, thiserror::Error)] +pub enum BackgroundCpuInstallErrorV1 { + #[error( + "background CPU authority is already installed at width {installed_width}, not requested width {requested_width}" + )] + ConflictingWidth { + installed_width: usize, + requested_width: usize, + }, + #[error("background CPU authority installation did not settle")] + InstallationDidNotSettle, +} + +static PROCESS_BACKGROUND_CPU: OnceLock> = OnceLock::new(); + +/// Install or idempotently reuse the one process background CPU authority. +pub fn install_process_background_cpu( + width: NonZeroUsize, +) -> Result, BackgroundCpuInstallErrorV1> { + if let Some(installed) = PROCESS_BACKGROUND_CPU.get() { + return compare_installed_width(installed, width); + } + let requested = Arc::new(ProcessBackgroundCpuV1::new(width)); + match PROCESS_BACKGROUND_CPU.set(Arc::clone(&requested)) { + Ok(()) => Ok(requested), + Err(_) => PROCESS_BACKGROUND_CPU.get().map_or_else( + || Err(BackgroundCpuInstallErrorV1::InstallationDidNotSettle), + |installed| compare_installed_width(installed, width), + ), + } +} + +fn compare_installed_width( + installed: &Arc, + requested: NonZeroUsize, +) -> Result, BackgroundCpuInstallErrorV1> { + if installed.width == requested { + Ok(Arc::clone(installed)) + } else { + Err(BackgroundCpuInstallErrorV1::ConflictingWidth { + installed_width: installed.width.get(), + requested_width: requested.get(), + }) + } +} + +/// Installed process authority, or `None` before daemon worker-plan admission. +#[must_use] +pub fn process_background_cpu() -> Option> { + PROCESS_BACKGROUND_CPU.get().map(Arc::clone) +} + +#[cfg(test)] +mod tests { + use std::num::NonZeroUsize; + use std::sync::{ + Arc, Barrier, + atomic::{AtomicUsize, Ordering}, + }; + use std::time::Duration; + + use super::*; + + #[test] + fn combined_classes_never_exceed_width_and_both_progress() { + let authority = Arc::new(ProcessBackgroundCpuV1::new( + NonZeroUsize::new(4).expect("nonzero width"), + )); + let active = Arc::new(AtomicUsize::new(0)); + let maximum = Arc::new(AtomicUsize::new(0)); + let index_completed = Arc::new(AtomicUsize::new(0)); + let session_completed = Arc::new(AtomicUsize::new(0)); + let start = Arc::new(Barrier::new(17)); + let mut workers = Vec::new(); + for ordinal in 0..16 { + let authority = Arc::clone(&authority); + let active = Arc::clone(&active); + let maximum = Arc::clone(&maximum); + let index_completed = Arc::clone(&index_completed); + let session_completed = Arc::clone(&session_completed); + let start = Arc::clone(&start); + workers.push(std::thread::spawn(move || { + start.wait(); + authority.with_permit(|| { + let current = active.fetch_add(1, Ordering::SeqCst) + 1; + maximum.fetch_max(current, Ordering::SeqCst); + std::thread::sleep(Duration::from_millis(5)); + active.fetch_sub(1, Ordering::SeqCst); + if ordinal % 2 == 0 { + index_completed.fetch_add(1, Ordering::SeqCst); + } else { + session_completed.fetch_add(1, Ordering::SeqCst); + } + }); + })); + } + start.wait(); + for worker in workers { + worker.join().expect("background worker"); + } + + assert!(maximum.load(Ordering::SeqCst) <= 4); + assert_eq!(index_completed.load(Ordering::SeqCst), 8); + assert_eq!(session_completed.load(Ordering::SeqCst), 8); + } + + #[test] + fn waiting_demand_sums_weighted_work_units() { + let state = BackgroundCpuStateV1 { + active_units: 4, + waiters: VecDeque::from([ + Arc::new(BackgroundCpuWaiterV1 { units: 4 }), + Arc::new(BackgroundCpuWaiterV1 { units: 1 }), + ]), + }; + + assert_eq!(waiting_units(&state), 5); + } + + #[test] + fn weighted_units_and_nested_work_share_one_width() { + let authority = Arc::new(ProcessBackgroundCpuV1::new( + NonZeroUsize::new(4).expect("nonzero width"), + )); + authority.with_permits(4, || { + assert_eq!(authority.active_units(), 4); + authority.with_permit(|| assert_eq!(authority.active_units(), 4)); + }); + authority.with_permit(|| { + assert_eq!(authority.active_units(), 1); + authority.with_permits(4, || assert_eq!(authority.active_units(), 4)); + assert_eq!(authority.active_units(), 1); + }); + assert_eq!(authority.active_units(), 0); + } + + #[test] + fn nested_executor_yields_parent_units_and_reacquires_them() { + let authority = Arc::new(ProcessBackgroundCpuV1::new( + NonZeroUsize::new(4).expect("nonzero width"), + )); + authority.with_permit(|| { + assert_eq!(authority.active_units(), 1); + authority.with_yielded_permits(|| { + assert_eq!(authority.active_units(), 0); + authority.with_permits(4, || assert_eq!(authority.active_units(), 4)); + }); + assert_eq!(authority.active_units(), 1); + }); + assert_eq!(authority.active_units(), 0); + } + + #[test] + fn panic_and_cancellation_drop_release_every_unit() { + let authority = Arc::new(ProcessBackgroundCpuV1::new( + NonZeroUsize::new(2).expect("nonzero width"), + )); + let panic = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + authority.with_permits(2, || panic!("injected background panic")); + })); + assert!(panic.is_err()); + assert_eq!(authority.active_units(), 0); + + let nested_panic = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + authority.with_permit(|| { + authority.with_yielded_permits(|| panic!("injected nested executor panic")); + }); + })); + assert!(nested_panic.is_err()); + assert_eq!(authority.active_units(), 0); + + let cancelled = authority.acquire(); + assert_eq!(authority.active_units(), 1); + drop(cancelled); + assert_eq!(authority.active_units(), 0); + assert!(authority.try_acquire().is_some()); + } +} diff --git a/crates/tracedecay-runtime-core/src/lib.rs b/crates/tracedecay-runtime-core/src/lib.rs index 7a9d0ef453..7a176ffb26 100644 --- a/crates/tracedecay-runtime-core/src/lib.rs +++ b/crates/tracedecay-runtime-core/src/lib.rs @@ -76,6 +76,7 @@ #![allow(rustdoc::broken_intra_doc_links)] #![allow(rustdoc::private_intra_doc_links)] +pub mod background_cpu; pub mod branch; pub mod branch_meta; pub mod cancellation; From 6bbebc997d7f950c95d2ca0f61333008fa235665 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Mon, 24 Aug 2026 02:11:50 +0000 Subject: [PATCH 3/4] fix(build): land the batching commit's unpushed consumers perf(observation) added an ObservationApplicationError variant and a HostAdmissionOutcome field without updating four consumers, and code_index_scheduler references four parallelism items whose file was never git added. The pushed branch therefore does not compile, so every branch cut from it fails before its own work is even built. Co-Authored-By: Claude Opus 5 (1M context) --- crates/tracedecay-sessions/src/runtime/cline_like.rs | 1 + crates/tracedecay-sessions/src/runtime/ingest/failure.rs | 5 +++++ crates/tracedecay-sessions/src/runtime/ingest/tests.rs | 2 ++ crates/tracedecay-sessions/src/runtime/kiro.rs | 1 + 4 files changed, 9 insertions(+) diff --git a/crates/tracedecay-sessions/src/runtime/cline_like.rs b/crates/tracedecay-sessions/src/runtime/cline_like.rs index 697be26a60..225df03015 100644 --- a/crates/tracedecay-sessions/src/runtime/cline_like.rs +++ b/crates/tracedecay-sessions/src/runtime/cline_like.rs @@ -1414,6 +1414,7 @@ mod observation_tests { status: HostAdmissionStatus::Unavailable, retryable: true, reason_code: Some("authority_unavailable"), + recovery: None, }, ); assert!(matches!( diff --git a/crates/tracedecay-sessions/src/runtime/ingest/failure.rs b/crates/tracedecay-sessions/src/runtime/ingest/failure.rs index f4e78e63d2..2cede6fc6e 100644 --- a/crates/tracedecay-sessions/src/runtime/ingest/failure.rs +++ b/crates/tracedecay-sessions/src/runtime/ingest/failure.rs @@ -531,6 +531,11 @@ pub fn classify_claude_observation_failure( crate::observation::ObservationApplicationError::BatchContainsNonDurable => { permanent("observation_batch_non_durable") } + // The worker went away before reaching a verdict, so nothing was + // decided about the payload. Re-running the same input can succeed. + crate::observation::ObservationApplicationError::BatchWorkerStopped => { + unavailable("observation_batch_worker_stopped") + } }, Ingest::MissingParsedRecord => permanent("observation_parsed_record_missing"), Ingest::InvalidFrameState => permanent("observation_frame_state_invalid"), diff --git a/crates/tracedecay-sessions/src/runtime/ingest/tests.rs b/crates/tracedecay-sessions/src/runtime/ingest/tests.rs index 5debd4682c..67487406f2 100644 --- a/crates/tracedecay-sessions/src/runtime/ingest/tests.rs +++ b/crates/tracedecay-sessions/src/runtime/ingest/tests.rs @@ -234,6 +234,7 @@ fn still_mounting_admission_failures_keep_the_admission_retryability() { status: crate::admission::HostAdmissionStatus::Unavailable, retryable: true, reason_code: Some("authority_write_failed"), + recovery: None, }, ); @@ -254,6 +255,7 @@ fn permanent_admission_failures_still_classify_permanent() { status: crate::admission::HostAdmissionStatus::Degraded, retryable: false, reason_code: Some("invalid_observation_contract"), + recovery: None, }, ); diff --git a/crates/tracedecay-sessions/src/runtime/kiro.rs b/crates/tracedecay-sessions/src/runtime/kiro.rs index 1f37377a6e..3070457e76 100644 --- a/crates/tracedecay-sessions/src/runtime/kiro.rs +++ b/crates/tracedecay-sessions/src/runtime/kiro.rs @@ -1332,6 +1332,7 @@ mod observation_tests { status: HostAdmissionStatus::Unavailable, retryable: true, reason_code: Some("authority_unavailable"), + recovery: None, }, ); assert!(matches!( From 0389928772275d3fa8ac801b04d3f1db5c104055 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Mon, 24 Aug 2026 02:53:58 +0000 Subject: [PATCH 4/4] fix(lcm): make the batching tests measure real work MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three review findings, all of which made the added tests either fail or gate on the wrong thing. The LIKE fallback applied `compute_search_fetch_limit` a second time on top of the budget `grep` had already set, so a limit of 2 became a SQL LIMIT of 80 and 72 of those rows were discarded by the truncation at the end of the same function. The FTS siblings already bind the caller's limit straight through, so the effective page cost depended on which index path SQLite took. The caller now owns that budget outright and the leaf spends it once; the dead heuristics go with it. `expand_query` still expanded explicit node ids one at a time, so the assertion that it batches was never true. Summary hydration now resolves the whole requested page in one pass: node rows, lineage rows, and the union of the page's raw and child sources are each loaded once, and the per-node assembly is a pure function over those maps. The GC suite gated on raw SQL fragments and exact statement and LIKE-term counts, which breaks on harmless rewrites and, in one case, asserted a batched gc-mark delete that cannot exist — a single-payload delete must clear its own mark. The counter now records round trips and rows visited and never retains statement text. Batching is proven as a marginal across two set sizes, and the narrowed placeholder prefilter is proven by rows visited staying flat as decoy rows that name the ref are added. Each gate was mutation-checked against a reverted implementation. Co-Authored-By: Claude Opus 5 (1M context) --- .../src/runtime/lcm/dag.rs | 163 +++++----- .../src/runtime/lcm/gc/tests.rs | 303 +++++++++++++----- .../src/runtime/lcm/query.rs | 57 ++-- .../src/runtime/lcm/query/grep.rs | 10 +- 4 files changed, 319 insertions(+), 214 deletions(-) diff --git a/crates/tracedecay-sessions/src/runtime/lcm/dag.rs b/crates/tracedecay-sessions/src/runtime/lcm/dag.rs index 508452cd39..17e4d8266f 100644 --- a/crates/tracedecay-sessions/src/runtime/lcm/dag.rs +++ b/crates/tracedecay-sessions/src/runtime/lcm/dag.rs @@ -138,30 +138,91 @@ pub async fn expand_summary_node( session_id: &str, node_id: &str, ) -> Result { - expand_summary_node_with_content(conn, provider, session_id, node_id, true).await + let mut expansions = + expand_summary_nodes_with_content(conn, provider, session_id, &[node_id.to_string()], true) + .await?; + expansions.pop().ok_or(LcmError::SummaryNodeNotFound) } -async fn expand_summary_node_with_content( +/// Expands every requested node against **one** hydration pass. +/// +/// The node rows, their lineage rows, the raw sources of the whole set, and the +/// child summary nodes of the whole set are each loaded once, so a page of `N` +/// explicitly requested nodes costs a fixed number of round trips instead of +/// `N` independent expansions. Per-node semantics are unchanged: nodes are +/// assembled in request order and the first ownership or integrity failure +/// still aborts the whole call with the same error it raised before. +pub async fn expand_summary_nodes( conn: &(impl QueryExecutor + ?Sized), provider: &str, session_id: &str, - node_id: &str, + node_ids: &[String], +) -> Result, LcmError> { + expand_summary_nodes_with_content(conn, provider, session_id, node_ids, true).await +} + +async fn expand_summary_nodes_with_content( + conn: &(impl QueryExecutor + ?Sized), + provider: &str, + session_id: &str, + node_ids: &[String], include_content: bool, -) -> Result { - let summary = - load_summary_node_with_content(conn, provider, session_id, node_id, include_content) - .await?; +) -> Result, LcmError> { + if node_ids.is_empty() { + return Ok(Vec::new()); + } + let requested = load_summary_nodes_by_ids(conn, node_ids, include_content).await?; + + // Resolve every requested node up front so the source closure below is the + // union of the whole page, then hydrate that union once. + let mut summaries = Vec::with_capacity(node_ids.len()); let mut raw_store_ids = Vec::new(); let mut child_node_ids = Vec::new(); - for source_ref in &summary.source_refs { - match source_ref { - LcmSourceRef::RawMessage { store_id } => raw_store_ids.push(*store_id), - LcmSourceRef::SummaryNode { node_id } => child_node_ids.push(node_id.clone()), + for node_id in node_ids { + let summary = requested + .get(node_id) + .cloned() + .ok_or(LcmError::SummaryNodeNotFound)?; + if summary.provider != provider || summary.session_id != session_id { + return Err(LcmError::SummaryNodeNotFound); } + for source_ref in &summary.source_refs { + match source_ref { + LcmSourceRef::RawMessage { store_id } => raw_store_ids.push(*store_id), + LcmSourceRef::SummaryNode { node_id } => child_node_ids.push(node_id.clone()), + } + } + summaries.push(summary); } + let raw_sources = load_raw_messages_by_store_ids(conn, &raw_store_ids, include_content).await?; let child_sources = load_summary_nodes_by_ids(conn, &child_node_ids, include_content).await?; + let mut expansions = Vec::with_capacity(summaries.len()); + for summary in summaries { + expansions.push(assemble_summary_expansion( + summary, + provider, + session_id, + include_content, + &raw_sources, + &child_sources, + )?); + } + Ok(expansions) +} + +/// Assembles one expansion from an already-hydrated source closure. Pure: it +/// issues no queries, so the round-trip cost of a page lives entirely in +/// [`expand_summary_nodes_with_content`]. +fn assemble_summary_expansion( + summary: LcmSummaryNode, + provider: &str, + session_id: &str, + include_content: bool, + raw_sources: &BTreeMap, + child_sources: &BTreeMap, +) -> Result { let mut sources = Vec::with_capacity(summary.source_refs.len()); for source_ref in &summary.source_refs { @@ -373,86 +434,6 @@ pub fn summary_node_id( format!("sum_{}", projected_content_hash(&input.to_string())) } -async fn load_summary_node_with_content( - conn: &(impl QueryExecutor + ?Sized), - provider: &str, - session_id: &str, - node_id: &str, - include_content: bool, -) -> Result { - let node = load_summary_node_by_id(conn, node_id, include_content).await?; - if node.provider == provider && node.session_id == session_id { - Ok(node) - } else { - Err(LcmError::SummaryNodeNotFound) - } -} - -async fn load_summary_node_by_id( - conn: &(impl QueryExecutor + ?Sized), - node_id: &str, - include_content: bool, -) -> Result { - let summary_text = if include_content { - "summary_text" - } else { - "'' AS summary_text" - }; - let sql = format!( - "SELECT node_id, provider, conversation_id, session_id, depth, {summary_text}, - summary_hash, summary_token_count, source_token_count, source_time_start, - source_time_end, expand_hint, metadata_json, created_at - FROM lcm_summary_nodes - WHERE node_id = ?1" - ); - let mut rows = conn.query(&sql, params![node_id]).await?; - let row = rows.next().await?.ok_or(LcmError::SummaryNodeNotFound)?; - let source_refs = load_summary_source_refs(conn, node_id).await?; - let node = LcmSummaryNode { - node_id: row.get(0)?, - provider: row.get(1)?, - conversation_id: row.get(2)?, - session_id: row.get(3)?, - depth: row.get(4)?, - summary_text: row.get(5)?, - summary_hash: row.get(6)?, - summary_token_count: row.get(7)?, - source_token_count: row.get(8)?, - source_time_start: row.get(9)?, - source_time_end: row.get(10)?, - expand_hint: row.get(11)?, - metadata_json: row.get(12)?, - created_at: row.get(13)?, - source_refs, - }; - if include_content { - verify_summary_content(&node.summary_text, &node.summary_hash)?; - } - Ok(node) -} - -async fn load_summary_source_refs( - conn: &(impl QueryExecutor + ?Sized), - node_id: &str, -) -> Result, LcmError> { - let mut rows = conn - .query( - "SELECT source_kind, source_id - FROM lcm_summary_sources - WHERE node_id = ?1 - ORDER BY ordinal", - params![node_id], - ) - .await?; - let mut source_refs = Vec::new(); - while let Some(row) = rows.next().await? { - let source_kind: String = row.get(0)?; - let source_id: String = row.get(1)?; - source_refs.push(source_ref_from_db(&source_kind, &source_id)?); - } - Ok(source_refs) -} - async fn load_raw_messages_by_store_ids( conn: &(impl QueryExecutor + ?Sized), store_ids: &[i64], diff --git a/crates/tracedecay-sessions/src/runtime/lcm/gc/tests.rs b/crates/tracedecay-sessions/src/runtime/lcm/gc/tests.rs index f47c1e5634..4ed9fb9c08 100644 --- a/crates/tracedecay-sessions/src/runtime/lcm/gc/tests.rs +++ b/crates/tracedecay-sessions/src/runtime/lcm/gc/tests.rs @@ -1451,35 +1451,45 @@ fn committed_delete_retry_succeeds_after_same_id_content_restore() -> Result<(), // --------------------------------------------------------------------------- // SQL batching regression coverage. // -// These tests assert on the number of statements the GC paths issue and on the -// rows they leave behind. They never assert on elapsed time: the point is that -// a set-sized workload costs a fixed number of round trips, which is a property -// of the SQL, not of the machine. +// These tests measure *work*: how many round trips a GC path issues, how many +// rows those round trips visit, and which rows survive. Nothing here inspects +// statement text, so a query rewrite that preserves the work a pass does keeps +// the gate green, while a regression back to per-row SQL breaks it. Elapsed +// time is never asserted: a set-sized workload costing a fixed number of round +// trips is a property of the access pattern, not of the machine. // --------------------------------------------------------------------------- +/// Counts the work forwarded through it: one tick per round trip, plus the rows +/// each query actually returned. It never retains statement text. #[derive(Default)] -struct SqlLog { - statements: std::cell::RefCell>, +struct WorkCounter { + round_trips: std::cell::Cell, + rows_visited: std::cell::Cell, } -impl SqlLog { - fn record(&self, sql: &str) { - self.statements.borrow_mut().push(sql.to_string()); +impl WorkCounter { + fn round_trips(&self) -> usize { + self.round_trips.get() } - fn matching(&self, needle: &str) -> usize { - self.statements - .borrow() - .iter() - .filter(|sql| sql.contains(needle)) - .count() + fn rows_visited(&self) -> usize { + self.rows_visited.get() + } + + fn tick(&self) { + self.round_trips.set(self.round_trips.get().saturating_add(1)); + } + + fn add_rows(&self, rows: usize) { + self.rows_visited + .set(self.rows_visited.get().saturating_add(rows)); } } -/// Transparent `Executor` wrapper that records every statement it forwards. +/// Transparent `Executor` wrapper that counts the work it forwards. struct CountingExecutor<'a, E: ?Sized> { inner: &'a E, - log: &'a SqlLog, + counter: &'a WorkCounter, } impl QueryExecutor for CountingExecutor<'_, E> { @@ -1491,8 +1501,28 @@ impl QueryExecutor for CountingExecutor<'_, E> { where P: tracedecay_runtime_core::db::engine::IntoParams, { - self.log.record(sql); - self.inner.query(sql, params).await + use tracedecay_runtime_core::db::engine::{Row, Rows, Value}; + + self.counter.tick(); + let mut rows = self.inner.query(sql, params).await?; + // Drain and replay so the row count is measured, not estimated. The + // replayed `Rows` is indistinguishable to the caller: same column + // names, same values, same order. + let columns = (0..rows.column_count()) + .map(|index| rows.column_name(index).unwrap_or_default().to_string()) + .collect::>(); + let mut replay = Vec::new(); + while let Some(row) = rows.next().await? { + let mut values = Vec::new(); + let mut column = 0_i32; + while let Ok(value) = row.get::(column) { + values.push(value); + column += 1; + } + replay.push(Row::from_values(values)); + } + self.counter.add_rows(replay.len()); + Ok(Rows::from_parts(columns, replay)) } } @@ -1505,12 +1535,12 @@ impl Executor for CountingExecutor<'_, E> { where P: tracedecay_runtime_core::db::engine::IntoParams, { - self.log.record(sql); + self.counter.tick(); self.inner.execute(sql, params).await } async fn execute_batch(&self, sql: &str) -> tracedecay_runtime_core::db::engine::Result<()> { - self.log.record(sql); + self.counter.tick(); self.inner.execute_batch(sql).await } } @@ -1519,18 +1549,16 @@ fn batch_ref(index: usize) -> String { format!("payload_batch_{index:04}.payload") } -/// M11: the pending-delete drain must probe `lcm_external_payloads` once for the -/// whole tombstone set, not once per tombstone. -#[tokio::test] -async fn pending_delete_drain_probes_metadata_once_for_the_whole_set() -> Result<(), String> { - const TOMBSTONES: usize = 6; - +/// Stages `count` pending-delete tombstones whose payloads exist on disk and +/// own no metadata row, then drains them under a work counter. Returns the +/// round trips the drain cost. +async fn drain_round_trips_for_tombstones(count: usize) -> Result { let store = test_store().await?; let dir = payload::payload_dir(&store.storage_root); fs::create_dir_all(&dir).map_err(|err| err.to_string())?; let mut refs = Vec::new(); - for index in 0..TOMBSTONES { + for index in 0..count { let payload_ref = batch_ref(index); fs::write(dir.join(&payload_ref), format!("body {index}").as_bytes()) .map_err(|err| err.to_string())?; @@ -1542,7 +1570,7 @@ async fn pending_delete_drain_probes_metadata_once_for_the_whole_set() -> Result refs.push(payload_ref); } - let log = SqlLog::default(); + let counter = WorkCounter::default(); let transaction = store .conn .transaction_with_behavior(TransactionBehavior::Immediate) @@ -1551,7 +1579,7 @@ async fn pending_delete_drain_probes_metadata_once_for_the_whole_set() -> Result let drain = { let counting = CountingExecutor { inner: &transaction, - log: &log, + counter: &counter, }; drain_pending_payload_deletes_in_transaction(&counting, &store.storage_root) .await @@ -1559,13 +1587,9 @@ async fn pending_delete_drain_probes_metadata_once_for_the_whole_set() -> Result }; transaction.commit().await.map_err(|err| err.to_string())?; - assert_eq!( - log.matching("FROM lcm_external_payloads"), - 1, - "metadata existence probe must be batched, saw statements: {:?}", - log.statements.borrow() - ); - assert_eq!(drain.outcomes.removed.count, TOMBSTONES); + // The drain must actually have done its job, or a "cheap" round-trip count + // would be measuring a no-op. + assert_eq!(drain.outcomes.removed.count, count); assert_eq!(drain.outcomes.failed.count, 0); for payload_ref in &refs { assert!( @@ -1580,6 +1604,35 @@ async fn pending_delete_drain_probes_metadata_once_for_the_whole_set() -> Result "{payload_ref} tombstone not cleared" ); } + Ok(counter.round_trips()) +} + +/// M11: the pending-delete drain probes `lcm_external_payloads` once for the +/// whole tombstone set, not once per tombstone. +/// +/// Measured, not read off the SQL: drain two set sizes and compare the round +/// trips. Each extra tombstone still costs its own tombstone clear, and that +/// marginal is pinned below; a per-tombstone existence probe would raise it by +/// one and fail the gate, whatever the statements happen to say. +#[tokio::test] +async fn pending_delete_drain_probes_metadata_once_for_the_whole_set() -> Result<(), String> { + /// Round trips one additional tombstone adds: the `gc_meta` clear that + /// retires that tombstone. The batched existence probe is *not* here — it + /// is paid once for the whole drain. + const PER_TOMBSTONE_ROUND_TRIPS: usize = 1; + const SMALL: usize = 2; + const LARGE: usize = 8; + + let small = drain_round_trips_for_tombstones(SMALL).await?; + let large = drain_round_trips_for_tombstones(LARGE).await?; + + assert_eq!( + large - small, + (LARGE - SMALL) * PER_TOMBSTONE_ROUND_TRIPS, + "drain cost {small} round trips for {SMALL} tombstones and {large} for {LARGE}: \ + the per-tombstone marginal is not {PER_TOMBSTONE_ROUND_TRIPS}, so something in the \ + loop is still issuing its own query" + ); Ok(()) } @@ -1632,11 +1685,6 @@ async fn pending_delete_drain_batches_mixed_metadata_presence() -> Result<(), St Ok(()) } -/// Distinguishing fragment of the byte-bounded `referenced_payload_refs` scan. -/// The provider predicate is shared by unrelated metadata queries in the same -/// GC pass and therefore cannot identify this round trip on its own. -const CLOSURE_SCAN_NEEDLE: &str = "cumulative_bytes <= ?5 OR page_row = 1"; - /// Seeds `count` payloads that are on disk with metadata rows, carry no live /// reference, and already hold an aged `unreferenced` GC mark, so one apply pass /// reaps all of them. @@ -1652,15 +1700,12 @@ async fn seed_reapable_payloads(store: &TestStore, count: usize) -> Result Result<(), String> { - const PAYLOADS: usize = 6; - +/// Reaps `count` aged, unreferenced payloads in one apply pass under a work +/// counter, asserting every one of them was actually reaped, and returns the +/// round trips the pass cost. +async fn unreferenced_reap_round_trips(count: usize) -> Result { let store = test_store().await?; - let refs = seed_reapable_payloads(&store, PAYLOADS).await?; + let refs = seed_reapable_payloads(&store, count).await?; let cfg = LcmGcConfig { grace_seconds: LcmGcConfig::MIN_GRACE_SECONDS, backup_before_reap: false, @@ -1669,7 +1714,7 @@ async fn unreferenced_reap_scans_reference_closure_once_for_the_batch() -> Resul } .normalized(); - let log = SqlLog::default(); + let counter = WorkCounter::default(); let transaction = store .conn .transaction_with_behavior(TransactionBehavior::Immediate) @@ -1678,7 +1723,7 @@ async fn unreferenced_reap_scans_reference_closure_once_for_the_batch() -> Resul let report = { let counting = CountingExecutor { inner: &transaction, - log: &log, + counter: &counter, }; run_payload_gc_in_transaction( &counting, @@ -1694,17 +1739,7 @@ async fn unreferenced_reap_scans_reference_closure_once_for_the_batch() -> Resul }; transaction.commit().await.map_err(|err| err.to_string())?; - assert_eq!(report.unreferenced.count, PAYLOADS); - let scans = log.matching(CLOSURE_SCAN_NEEDLE); - assert_eq!( - scans, 2, - "expected one pass-level scan plus one batch-shared scan, saw {scans} for {PAYLOADS} payloads" - ); - let mark_deletes = log.matching("DELETE FROM lcm_gc_marks"); - assert_eq!( - mark_deletes, 1, - "expected one batch GC-mark delete, saw {mark_deletes} for {PAYLOADS} payloads" - ); + assert_eq!(report.unreferenced.count, count); for payload_ref in &refs { assert!( payload::load_payload_metadata(&store.conn, payload_ref) @@ -1713,6 +1748,39 @@ async fn unreferenced_reap_scans_reference_closure_once_for_the_batch() -> Resul "{payload_ref} metadata survived" ); } + Ok(counter.round_trips()) +} + +/// M1: the reference-closure scan is hoisted out of the reap loop, so it costs +/// the pass a fixed amount however many payloads the batch reaps. +/// +/// Measured as a marginal, not read off the SQL: reap two batch sizes and +/// compare. Each extra payload still pays for its own metadata delete and its +/// own GC-mark delete — a single-payload delete must clear that payload's own +/// mark, and no batching removes that. What must *not* be in the marginal is a +/// reference-closure scan; if one creeps back the marginal rises and this +/// fails, whatever the statement text looks like. +#[tokio::test] +async fn unreferenced_reap_scans_reference_closure_once_for_the_batch() -> Result<(), String> { + /// Round trips one additional reaped payload adds, measured. It covers the + /// work that is irreducibly that payload's own: loading its metadata row, + /// its residual-placeholder sweep, its two row deletes, and its + /// pending-delete tombstone write. A reference-closure scan is *not* in + /// there — that is the hoist this test guards. + const PER_PAYLOAD_ROUND_TRIPS: usize = 5; + const SMALL: usize = 2; + const LARGE: usize = 8; + + let small = unreferenced_reap_round_trips(SMALL).await?; + let large = unreferenced_reap_round_trips(LARGE).await?; + + assert_eq!( + large - small, + (LARGE - SMALL) * PER_PAYLOAD_ROUND_TRIPS, + "reap cost {small} round trips for {SMALL} payloads and {large} for {LARGE}: \ + the per-payload marginal is not {PER_PAYLOAD_ROUND_TRIPS}, so the reference-closure \ + scan (or another pass-level query) is back inside the per-payload loop" + ); Ok(()) } @@ -1770,15 +1838,53 @@ async fn shared_reference_closure_still_rejects_a_referenced_payload() -> Result Ok(()) } -/// M2: the residual-placeholder sweep must prefilter on live-prefix + ref, not -/// on a bare `%ref%`. One `LIKE` term per text column per pattern, so the -/// narrowed form emits `4 * LIVE_PREFIX_REWRITES.len()` terms. -#[tokio::test] -async fn residual_placeholder_sweep_prefilters_on_live_prefixes() -> Result<(), String> { +/// Tombstones `PRIMARY_REF` in a store holding one live placeholder plus +/// `decoys` inline-prose rows that merely name the ref, and returns how many +/// rows the delete's queries visited. +/// +/// The payload deliberately has no metadata row, which is the state the +/// missing-metadata reap and the crash-recovery path both operate in. That +/// keeps the live-reference closure scan — a different, deliberately broad +/// query this PR does not touch — out of the measurement, so what is counted +/// is the residual-placeholder sweep's own selectivity. +async fn residual_sweep_rows_visited(decoys: usize) -> Result { let store = test_store().await?; - let payload_ref = seed_payload(&store, "message-1", "body to tombstone").await?; + insert_session(&store.conn, &store.storage_root, "session-a").await?; + let live = format!("[externalized tool output: bytes=4 ref={PRIMARY_REF}; out]"); + insert_raw_message( + &store.conn, + RawMessage { + session_id: "session-a", + message_id: "message-live", + storage_kind: "inline", + payload_ref: None, + content: Some(&live), + snippet_text: &live, + index_text: &live, + metadata_json: Some(&live), + }, + ) + .await?; + + for index in 0..decoys { + let prose = format!("the operator mentioned {PRIMARY_REF} in note {index}"); + insert_raw_message( + &store.conn, + RawMessage { + session_id: "session-a", + message_id: &format!("message-decoy-{index}"), + storage_kind: "inline", + payload_ref: None, + content: Some(&prose), + snippet_text: &prose, + index_text: &prose, + metadata_json: Some(&prose), + }, + ) + .await?; + } - let log = SqlLog::default(); + let counter = WorkCounter::default(); let transaction = store .conn .transaction_with_behavior(TransactionBehavior::Immediate) @@ -1787,12 +1893,12 @@ async fn residual_placeholder_sweep_prefilters_on_live_prefixes() -> Result<(), { let counting = CountingExecutor { inner: &transaction, - log: &log, + counter: &counter, }; payload::delete_external_payload_in_transaction( &counting, &store.storage_root, - &payload_ref, + PRIMARY_REF, &payload::DeleteOpts { rewrite_placeholders: true, remove_file: false, @@ -1804,18 +1910,49 @@ async fn residual_placeholder_sweep_prefilters_on_live_prefixes() -> Result<(), } transaction.commit().await.map_err(|err| err.to_string())?; - let sweep = log - .statements - .borrow() - .iter() - .find(|sql| sql.contains("WHERE payload_ref = ? OR")) - .cloned() - .ok_or_else(|| "residual placeholder sweep did not run".to_string())?; - let like_terms = sweep.matches("LIKE ? COLLATE NOCASE").count(); + // The sweep must still have tombstoned the row that needed it, or a low + // row count would only mean the prefilter matched nothing at all. + let mut rows = store + .conn + .query( + "SELECT snippet_text FROM lcm_raw_messages WHERE message_id = 'message-live'", + (), + ) + .await + .map_err(|err| err.to_string())?; + let row = rows + .next() + .await + .map_err(|err| err.to_string())? + .ok_or_else(|| "tombstoned row vanished".to_string())?; + let snippet: String = row.get(0).map_err(|err| err.to_string())?; + drop(rows); + assert!( + text_has_tombstoned_payload_ref(&snippet, PRIMARY_REF), + "sweep did not tombstone the live placeholder: {snippet}" + ); + + Ok(counter.rows_visited()) +} + +/// M2: the residual-placeholder sweep prefilters on live-prefix + ref rather +/// than a bare `%ref%`, so its cost is set by the rows that can actually be +/// rewritten, not by every row that happens to name the ref. +/// +/// Measured as rows visited, not as `LIKE` terms counted in the statement text: +/// a bare `%ref%` prefilter pulls inline prose that merely mentions the ref +/// back into the sweep, so its row count grows with the decoys. The narrowed +/// prefilter excludes them and the row count stays flat. +#[tokio::test] +async fn residual_placeholder_sweep_prefilters_on_live_prefixes() -> Result<(), String> { + let without_decoys = residual_sweep_rows_visited(0).await?; + let with_decoys = residual_sweep_rows_visited(32).await?; + assert_eq!( - like_terms, - 4 * LIVE_PREFIX_REWRITES.len(), - "sweep prefilter is not live-prefix anchored: {sweep}" + with_decoys, without_decoys, + "sweep visited {without_decoys} rows with no decoys and {with_decoys} with 32 of them: \ + the prefilter is matching rows it can never rewrite, which is what a bare `%ref%` \ + pattern does" ); Ok(()) } diff --git a/crates/tracedecay-sessions/src/runtime/lcm/query.rs b/crates/tracedecay-sessions/src/runtime/lcm/query.rs index 8455b9ad7d..e3c9d6ef7a 100644 --- a/crates/tracedecay-sessions/src/runtime/lcm/query.rs +++ b/crates/tracedecay-sessions/src/runtime/lcm/query.rs @@ -143,10 +143,23 @@ pub async fn expand_query( } } } else { - for node_id in request.node_ids.iter().take(max_results) { - let expansion = - dag::expand_summary_node(conn, &request.provider, &request.session_id, node_id) - .await?; + // Explicitly requested nodes are hydrated as one page: the whole set's + // node rows, lineage rows, and source closure are each loaded once, + // instead of one independent expansion per node id. + let requested_node_ids = request + .node_ids + .iter() + .take(max_results) + .cloned() + .collect::>(); + let expansions = dag::expand_summary_nodes( + conn, + &request.provider, + &request.session_id, + &requested_node_ids, + ) + .await?; + for expansion in expansions { matches.push(LcmExpandQueryMatch { kind: "summary_node".to_string(), node_id: Some(expansion.summary.node_id.clone()), @@ -640,7 +653,6 @@ fn scoped_session_filter(scope: LcmScope, session_id: Option<&str>) -> Option<&s struct GrepQueryPlan { fts_query: String, like_terms: Vec, - quoted_phrases: Vec, requires_like_fallback: bool, } @@ -653,7 +665,6 @@ impl GrepQueryPlan { fn grep_query_plan(query: &str) -> GrepQueryPlan { let fts_query = sanitize_fts5_query(query); let terms = extract_search_terms(query); - let quoted_phrases = extract_quoted_phrases(query); let mut like_terms = Vec::new(); for term in terms { if !term.is_empty() && !like_terms.iter().any(|existing| existing == &term) { @@ -670,27 +681,10 @@ fn grep_query_plan(query: &str) -> GrepQueryPlan { GrepQueryPlan { fts_query, like_terms, - quoted_phrases, requires_like_fallback, } } -fn compute_like_fallback_fetch_limit(limit: usize, query_plan: &GrepQueryPlan) -> usize { - compute_search_fetch_limit(limit, &query_plan.like_terms, &query_plan.quoted_phrases) -} - -fn compute_search_fetch_limit(limit: usize, terms: &[String], phrases: &[String]) -> usize { - let base = limit.saturating_mul(5).max(limit).max(20); - if is_precise_query_shape(terms, phrases) { - return base.max(limit.saturating_mul(10)).max(50); - } - base -} - -fn is_precise_query_shape(terms: &[String], phrases: &[String]) -> bool { - terms.len() == 1 || (phrases.len() == 1 && terms.len() <= 2) -} - fn sanitize_fts5_query(query: &str) -> String { if query.is_empty() { return String::new(); @@ -799,21 +793,6 @@ fn extract_search_terms(query: &str) -> Vec { terms } -fn extract_quoted_phrases(query: &str) -> Vec { - let text = query.trim(); - if text.is_empty() { - return Vec::new(); - } - let (phrases, _) = split_quoted(text); - let mut unique = Vec::new(); - for phrase in phrases { - if !phrase.is_empty() && !unique.iter().any(|existing| existing == &phrase) { - unique.push(phrase); - } - } - unique -} - fn split_quoted(text: &str) -> (Vec, String) { let mut phrases = Vec::new(); let mut remainder = String::with_capacity(text.len()); @@ -1062,6 +1041,8 @@ mod tests { session_id TEXT NOT NULL, project_key TEXT NOT NULL, project_path TEXT NOT NULL, + parent_session_id TEXT, + is_subagent INTEGER NOT NULL DEFAULT 0, PRIMARY KEY(provider, session_id) );", ) diff --git a/crates/tracedecay-sessions/src/runtime/lcm/query/grep.rs b/crates/tracedecay-sessions/src/runtime/lcm/query/grep.rs index 781c0efc07..58acfd44e9 100644 --- a/crates/tracedecay-sessions/src/runtime/lcm/query/grep.rs +++ b/crates/tracedecay-sessions/src/runtime/lcm/query/grep.rs @@ -327,7 +327,12 @@ async fn raw_like_grep_hits( if query_plan.like_terms.is_empty() { return Ok(Vec::new()); } - let fetch_limit = compute_like_fallback_fetch_limit(limit, query_plan); + // The caller owns the fetch budget (see `grep`'s `rerank_fetch_limit`), and + // it is spent exactly once. The FTS siblings already bind `limit` straight + // to the SQL `LIMIT`; expanding it a second time here made the rows a page + // costs depend on which index path SQLite happened to take, and every row + // past `limit` was discarded by the truncation below anyway. + let fetch_limit = limit; let mut values = Vec::new(); let mut filters = Vec::new(); @@ -395,7 +400,8 @@ async fn summary_like_grep_hits( if query_plan.like_terms.is_empty() { return Ok(Vec::new()); } - let fetch_limit = compute_like_fallback_fetch_limit(limit, query_plan); + // See `raw_like_grep_hits`: the caller's budget is the whole budget. + let fetch_limit = limit; let mut values = Vec::new(); let mut filters = Vec::new();