From 7bd29d693ce5e5df8aa1a58edf66375c4a81fbbb Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Sat, 19 Sep 2026 06:49:09 +0000 Subject: [PATCH 01/15] fix(extraction): bind a unique trait bound as the callee A type parameter with one trait bound names Trait::method. Two bounds still abstain. Dotted calls do not regain the bare method name #1814 removed, and self still names the enclosing type. Co-authored-by: Zack Jackson --- .../src/rust_extractor.rs | 256 +++++++++++++++++- .../tests/main/rust.rs | 87 ++++++ crates/tracedecay-code-index/src/chunks.rs | 65 +++++ 3 files changed, 405 insertions(+), 3 deletions(-) diff --git a/crates/tracedecay-code-extraction/src/rust_extractor.rs b/crates/tracedecay-code-extraction/src/rust_extractor.rs index 17261a1a99..baf5aadbd5 100644 --- a/crates/tracedecay-code-extraction/src/rust_extractor.rs +++ b/crates/tracedecay-code-extraction/src/rust_extractor.rs @@ -61,6 +61,70 @@ impl ReceiverTypes { } } +/// One type parameter's trait bounds, as a method owner. +/// +/// `T: Processor` makes `value.process()` the callee `Processor::process`. +/// Two bounds, or a bound the syntax does not name, stay unresolved so the +/// call is not attached to both traits. This does not rename `self`: #1814 +/// still records that binding from the enclosing type. +enum ParamBound { + Unbound, + Unique(String), + Ambiguous, +} + +#[derive(Default)] +struct TraitBounds { + parameters: BTreeMap, +} + +struct BoundClause { + paths: Vec, + ambiguous: bool, +} + +impl TraitBounds { + fn declare(&mut self, name: String) { + self.parameters.insert(name, ParamBound::Unbound); + } + + fn knows(&self, name: &str) -> bool { + self.parameters.contains_key(name) + } + + fn constrain(&mut self, name: &str, clause: BoundClause) { + if clause.paths.is_empty() && !clause.ambiguous { + return; + } + let Some(slot) = self.parameters.get_mut(name) else { + return; + }; + if clause.ambiguous || clause.paths.len() != 1 { + *slot = ParamBound::Ambiguous; + return; + } + let Some(path) = clause.paths.into_iter().next() else { + *slot = ParamBound::Ambiguous; + return; + }; + match slot { + ParamBound::Unbound => *slot = ParamBound::Unique(path), + ParamBound::Unique(existing) if existing == &path => {} + ParamBound::Unique(_) | ParamBound::Ambiguous => *slot = ParamBound::Ambiguous, + } + } + + /// Replace a written type-parameter name with its unique trait. Any other + /// path, including the enclosing type recorded for `self`, is unchanged. + fn resolve(&self, path: String) -> Option { + match self.parameters.get(path.as_str()) { + Some(ParamBound::Unique(bound)) => Some(bound.clone()), + Some(ParamBound::Ambiguous) => None, + Some(ParamBound::Unbound) | None => Some(path), + } + } +} + /// Internal state used during AST traversal. /// /// Borrows the caller's source for the lifetime of the walk: copying the @@ -1704,6 +1768,17 @@ impl RustExtractor { node: TsNode<'_>, function: TsNode<'_>, receivers: &mut ReceiverTypes, + ) { + let bounds = Self::trait_bounds_for(state, function); + Self::collect_receiver_bindings(state, node, function, receivers, &bounds); + } + + fn collect_receiver_bindings( + state: &ExtractionState<'_>, + node: TsNode<'_>, + function: TsNode<'_>, + receivers: &mut ReceiverTypes, + bounds: &TraitBounds, ) { match node.kind() { "self_parameter" => { @@ -1715,14 +1790,14 @@ impl RustExtractor { if let Some(pattern) = node.child_by_field_name("pattern") { let type_path = node .child_by_field_name("type") - .and_then(|ty| Self::stated_type_path(state, ty)); + .and_then(|ty| Self::receiver_type_path(state, ty, bounds)); Self::record_receiver_pattern(state, pattern, type_path, receivers); } } "let_declaration" => { if let Some(pattern) = node.child_by_field_name("pattern") { let type_path = match node.child_by_field_name("type") { - Some(ty) => Self::stated_type_path(state, ty), + Some(ty) => Self::receiver_type_path(state, ty, bounds), None => node .child_by_field_name("value") .and_then(|value| Self::stated_initializer_type_path(state, value)), @@ -1759,7 +1834,7 @@ impl RustExtractor { let mut cursor = node.walk(); if cursor.goto_first_child() { loop { - Self::collect_receiver_types(state, cursor.node(), function, receivers); + Self::collect_receiver_bindings(state, cursor.node(), function, receivers, bounds); if !cursor.goto_next_sibling() { break; } @@ -1812,10 +1887,185 @@ impl RustExtractor { "dynamic_type" | "abstract_type" => ty .child_by_field_name("trait") .and_then(|inner| Self::stated_type_path(state, inner)), + "higher_ranked_trait_bound" => ty + .child_by_field_name("type") + .and_then(|inner| Self::stated_type_path(state, inner)), + // `impl Trait + 'a` still names that trait. Two nominals do not. + "bounded_type" => Self::unique_sum_type_path(state, ty), _ => None, } } + /// A parameter type, with a type parameter replaced by its unique trait + /// bound. `Self` is left as #1814 mapped it: the enclosing type, not the + /// trait the parameter happens to implement. + fn receiver_type_path( + state: &ExtractionState<'_>, + ty: TsNode<'_>, + bounds: &TraitBounds, + ) -> Option { + if Self::annotation_is_self(state, ty) { + return Self::enclosing_receiver_type(state); + } + bounds.resolve(Self::stated_type_path(state, ty)?) + } + + fn annotation_is_self(state: &ExtractionState<'_>, ty: TsNode<'_>) -> bool { + match ty.kind() { + "type_identifier" => state.node_text(ty) == "Self", + "reference_type" => ty + .child_by_field_name("type") + .is_some_and(|inner| Self::annotation_is_self(state, inner)), + _ => false, + } + } + + fn unique_sum_type_path(state: &ExtractionState<'_>, ty: TsNode<'_>) -> Option { + let mut found = None; + let mut cursor = ty.walk(); + if !cursor.goto_first_child() { + return None; + } + loop { + let child = cursor.node(); + if child.is_named() { + let path = match child.kind() { + "lifetime" | "use_bounds" => None, + "bounded_type" => Self::unique_sum_type_path(state, child), + _ => Self::stated_type_path(state, child), + }; + match path { + None if matches!(child.kind(), "lifetime" | "use_bounds") => {} + None => return None, + Some(path) => { + if found.replace(path).is_some() { + return None; + } + } + } + } + if !cursor.goto_next_sibling() { + break; + } + } + found + } + + fn trait_bounds_for(state: &ExtractionState<'_>, function: TsNode<'_>) -> TraitBounds { + let mut ancestors = Vec::new(); + let mut current = function.parent(); + while let Some(node) = current { + if matches!(node.kind(), "function_item" | "function_signature_item") { + break; + } + if matches!(node.kind(), "impl_item" | "trait_item") { + ancestors.push(node); + } + current = node.parent(); + } + ancestors.reverse(); + let mut bounds = TraitBounds::default(); + for item in ancestors { + Self::absorb_generic_bounds(state, item, &mut bounds); + } + Self::absorb_generic_bounds(state, function, &mut bounds); + bounds + } + + fn absorb_generic_bounds( + state: &ExtractionState<'_>, + item: TsNode<'_>, + bounds: &mut TraitBounds, + ) { + if let Some(parameters) = item.child_by_field_name("type_parameters") { + let mut cursor = parameters.walk(); + if cursor.goto_first_child() { + loop { + let child = cursor.node(); + if child.kind() == "type_parameter" + && let Some(name_node) = child.child_by_field_name("name") + { + let name = state.node_text(name_node).to_owned(); + bounds.declare(name.clone()); + if let Some(clause) = child.child_by_field_name("bounds") { + bounds.constrain(&name, Self::trait_bound_clause(state, clause)); + } + } + if !cursor.goto_next_sibling() { + break; + } + } + } + } + let Some(where_clause) = Self::child_of_kind(item, "where_clause") else { + return; + }; + let mut cursor = where_clause.walk(); + if !cursor.goto_first_child() { + return; + } + loop { + let child = cursor.node(); + if child.kind() == "where_predicate" + && let Some(left) = child.child_by_field_name("left") + && left.kind() == "type_identifier" + { + let name = state.node_text(left); + if bounds.knows(name) + && let Some(clause) = child.child_by_field_name("bounds") + { + bounds.constrain(name, Self::trait_bound_clause(state, clause)); + } + } + if !cursor.goto_next_sibling() { + break; + } + } + } + + fn trait_bound_clause(state: &ExtractionState<'_>, bounds: TsNode<'_>) -> BoundClause { + let mut clause = BoundClause { + paths: Vec::new(), + ambiguous: false, + }; + let mut cursor = bounds.walk(); + if !cursor.goto_first_child() { + return clause; + } + loop { + let child = cursor.node(); + if child.is_named() { + match child.kind() { + "lifetime" | "use_bounds" | "removed_trait_bound" => {} + _ => match Self::stated_type_path(state, child) { + Some(path) => clause.paths.push(path), + None => clause.ambiguous = true, + }, + } + } + if !cursor.goto_next_sibling() { + break; + } + } + clause + } + + fn child_of_kind<'t>(node: TsNode<'t>, kind: &str) -> Option> { + let mut cursor = node.walk(); + if !cursor.goto_first_child() { + return None; + } + loop { + let child = cursor.node(); + if child.kind() == kind { + return Some(child); + } + if !cursor.goto_next_sibling() { + return None; + } + } + } + /// The type a `let` initialiser states in syntax: a `T { .. }` literal, /// optionally behind `?`. Method names are never return-type evidence. /// Abstain rather than fabricate a receiver type. diff --git a/crates/tracedecay-code-extraction/tests/main/rust.rs b/crates/tracedecay-code-extraction/tests/main/rust.rs index 7e74529f99..becfee81dc 100644 --- a/crates/tracedecay-code-extraction/tests/main/rust.rs +++ b/crates/tracedecay-code-extraction/tests/main/rust.rs @@ -1237,6 +1237,93 @@ fn make() -> Vec { Vec::new() } ); } +#[test] +fn trait_bound_calls_name_the_trait_without_a_bare_method() { + let source = r#" +trait Processor { + fn process(&self, input: u32) -> u32; + fn via_self(&self, input: u32) -> u32 { + self.process(input) + } +} +trait Other { + fn process(&self, input: u32) -> u32; +} +struct Doubler; +impl Doubler { + fn kick(&self, input: u32) -> u32 { + self.process(input) + } +} +impl Processor for Doubler { + fn process(&self, input: u32) -> u32 { + input * 2 + } +} +fn via_dyn(processor: &dyn Processor, input: u32) -> u32 { + processor.process(input) +} +fn via_impl(processor: impl Processor + 'static, input: u32) -> u32 { + processor.process(input) +} +fn via_bound(processor: &T, input: u32) -> u32 { + processor.process(input) +} +fn via_where(processor: &T, input: u32) -> u32 +where + T: Processor, +{ + processor.process(input) +} +fn ambiguous(processor: &T, input: u32) -> u32 { + processor.process(input) +} +"#; + let result = RustExtractor.extract("src/lib.rs", source); + assert!(result.errors.is_empty(), "{:?}", result.errors); + let from = |name: &str| { + let function = result + .nodes + .iter() + .find(|node| { + matches!(node.kind, NodeKind::Function | NodeKind::Method) && node.name == name + }) + .unwrap_or_else(|| panic!("{name} is extracted")); + result + .unresolved_refs + .iter() + .filter(|reference| { + reference.reference_kind == EdgeKind::Calls && reference.from_node_id == function.id + }) + .map(|reference| reference.reference_name.as_str()) + .collect::>() + }; + + for owner in ["via_self", "via_dyn", "via_impl", "via_bound", "via_where"] { + let names = from(owner); + assert!( + names.contains(&"Processor::process"), + "{owner} must name the trait callee: {names:?}" + ); + assert!( + !names.contains(&"process"), + "{owner} must not reintroduce the bare method name: {names:?}" + ); + } + let kick = from("kick"); + assert!( + kick.contains(&"Doubler::process") && !kick.contains(&"Processor::process"), + "self in an inherent impl stays the type, not the trait: {kick:?}" + ); + assert!(!kick.contains(&"process"), "{kick:?}"); + let ambiguous = from("ambiguous"); + assert!( + !ambiguous.iter().any(|name| name.contains("::process")), + "two trait bounds must not pick a callee: {ambiguous:?}" + ); + assert!(!ambiguous.contains(&"process"), "{ambiguous:?}"); +} + #[test] fn wildcard_imports_retain_unresolved_dependencies_alongside_named_bindings() { let result = RustExtractor.extract( diff --git a/crates/tracedecay-code-index/src/chunks.rs b/crates/tracedecay-code-index/src/chunks.rs index 85f27ed90c..3b2896c71a 100644 --- a/crates/tracedecay-code-index/src/chunks.rs +++ b/crates/tracedecay-code-index/src/chunks.rs @@ -4452,6 +4452,71 @@ pub fn real_symbol() {} ); } + #[test] + fn trait_bound_method_call_binds_the_trait_callee() { + let source = concat!( + "pub trait Processor {\n", + " fn process(&self, input: u32) -> u32;\n", + " fn via_self(&self, input: u32) -> u32 { self.process(input) }\n", + "}\n", + "pub trait Other { fn process(&self, input: u32) -> u32; }\n", + "pub struct Doubler;\n", + "impl Doubler { fn kick(&self, input: u32) -> u32 { self.process(input) } }\n", + "impl Processor for Doubler { fn process(&self, input: u32) -> u32 { input * 2 } }\n", + "pub fn via_dyn(processor: &dyn Processor, input: u32) -> u32 { processor.process(input) }\n", + "pub fn via_impl(processor: impl Processor + 'static, input: u32) -> u32 { processor.process(input) }\n", + "pub fn via_bound(processor: &T, input: u32) -> u32 { processor.process(input) }\n", + "pub fn via_where(processor: &T, input: u32) -> u32 where T: Processor { processor.process(input) }\n", + "pub fn ambiguous(processor: &T, input: u32) -> u32 { processor.process(input) }\n", + ); + let file = validated_file("src/lib.rs", source.as_bytes()); + let batch = batch_for(&file, ParseOutcomeV1::Complete); + let artifacts = chunker() + .index_file(&file, &batch, &rust_descriptor(), &NeverCancelled) + .expect("indexing succeeds"); + let qualified = |occurrence: &SymbolOccurrenceId| { + artifacts + .symbols + .iter() + .find(|symbol| &symbol.occurrence == occurrence) + .map(|symbol| symbol.qualified_name.as_str()) + .unwrap_or("") + }; + let mut calls = artifacts + .edges + .iter() + .filter(|edge| edge.kind == RelationEdgeKindV1::Calls) + .map(|edge| { + ( + qualified(&edge.from_occurrence).to_owned(), + qualified(&edge.to_occurrence).to_owned(), + ) + }) + .collect::>(); + calls.sort(); + + let trait_method = "src/lib.rs::Processor::process"; + let impl_method = "src/lib.rs::::process"; + assert_eq!( + calls, + vec![ + ( + "src/lib.rs::Doubler::kick".to_owned(), + impl_method.to_owned() + ), + ( + "src/lib.rs::Processor::via_self".to_owned(), + trait_method.to_owned() + ), + ("src/lib.rs::via_bound".to_owned(), trait_method.to_owned()), + ("src/lib.rs::via_dyn".to_owned(), trait_method.to_owned()), + ("src/lib.rs::via_impl".to_owned(), trait_method.to_owned()), + ("src/lib.rs::via_where".to_owned(), trait_method.to_owned()), + ], + "a unique trait bound is the callee; two bounds and a bare method name are not: {calls:?}" + ); + } + #[test] fn rust_type_path_alias_parses_ufcs_trait_impl_methods() { assert_eq!( From 3a862976df4b901aa9ff5d69b0c8961854521413 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Sat, 19 Sep 2026 07:01:23 +0000 Subject: [PATCH 02/15] fix(global-db): replay an applied cursor coverage as a duplicate A history rescan reuses a coverage-only idempotency key with a new resume checkpoint. That conflict is an already-applied advance when the durable cursor is already at next, not a permanent collision that blocks ingest. --- .../src/observation_adapter.rs | 11 ++ .../src/observation_collision_tests.rs | 127 +++++++++++++++++- 2 files changed, 137 insertions(+), 1 deletion(-) diff --git a/crates/tracedecay-global-db/src/observation_adapter.rs b/crates/tracedecay-global-db/src/observation_adapter.rs index 8148f6b9e1..571f4c6fbc 100644 --- a/crates/tracedecay-global-db/src/observation_adapter.rs +++ b/crates/tracedecay-global-db/src/observation_adapter.rs @@ -1526,6 +1526,17 @@ impl ObservationStore for GlobalDbObservationStore { Ok(CursorAdvanceOutcome::Committed) } RuntimeSubmitOutcomeV1::ExactReplay { .. } => Ok(CursorAdvanceOutcome::ExactDuplicate), + // The idempotency key covers the advanced coverage, not the whole + // command, so a re-scan of already-admitted history reuses the key + // with different bytes (a fresh `expected_cursor` or resume + // checkpoint) and the writer reports a conflict against the earlier + // committed receipt. When the durable cursor is already exactly + // `next_cursor`, that earlier commit is this advance: the coverage + // is applied and the replay is a duplicate. Only a conflict that + // left the cursor somewhere else is an unresolved collision. + RuntimeSubmitOutcomeV1::IdempotencyConflict { .. } if existed_at_next => { + Ok(CursorAdvanceOutcome::ExactDuplicate) + } RuntimeSubmitOutcomeV1::IdempotencyConflict { .. } => { Err(ObservationStoreError::CursorAdvanceCollision) } diff --git a/crates/tracedecay-global-db/src/observation_collision_tests.rs b/crates/tracedecay-global-db/src/observation_collision_tests.rs index 31cd26d4db..c02f99ca30 100644 --- a/crates/tracedecay-global-db/src/observation_collision_tests.rs +++ b/crates/tracedecay-global-db/src/observation_collision_tests.rs @@ -55,7 +55,8 @@ use tracedecay_domain::{ use tracedecay_store::observation::ObservationIdentityCollisionDispositionV1; use tracedecay_store::{ AnchoredObservationWrite, CursorAdvanceLedgerReasonV1, CursorAdvanceLedgerReceiptIdV1, - ObservationCoverageReason, ObservationCursorAdvance, ObservationPersistOutcome, + CursorAdvanceOutcome, ObservationCoverageReason, ObservationCursorAdvance, + ObservationPersistOutcome, ObservationProjectionStore, ObservationStore, ObservationStoreError, ObservationWrite, ProjectionPersistOutcome, ProjectionSkipReason, SESSION_MESSAGE_PROJECTOR_VERSION, }; @@ -3073,6 +3074,130 @@ async fn runtime_cursor_replay_without_a_ledger_row_keeps_generic_collision_sema )); } +/// Overwrites the durable cursor for one source, the shape a retained-history +/// rescan sees when it resumes behind the admitted frontier. +async fn rewind_source_cursor( + runtime: &HostAdmissionTestRuntimeV1, + cursor: &ObservationSourceCursorV1, +) { + let database = runtime + .registered_database(HostAdmissionScope::Profile) + .expect("registered profile database"); + let transaction = database.begin_write_transaction().await.unwrap(); + transaction + .execute( + COMMIT_SOURCE_CURSOR_SQL, + params![ + serde_json::to_string(cursor.source()).unwrap().as_str(), + serde_json::to_string(cursor.scope()).unwrap().as_str(), + serde_json::to_string(cursor).unwrap().as_str() + ], + ) + .await + .unwrap(); + transaction.commit().await.unwrap(); +} + +/// A cursor advance is idempotency-keyed by its coverage, not by the whole +/// command, so a retained-history rescan replays an already-admitted coverage +/// with different command bytes: the `expected_cursor` it resumed from carries +/// a freshly computed resume checkpoint. The writer answers with a conflict +/// against the earlier receipt, and treating that as a permanent collision is +/// what wedges retained ingest. The durable cursor already sits exactly at +/// `next_cursor`, so the coverage is applied and the replay is a duplicate. +/// A conflict that leaves the cursor somewhere else is still a collision. +#[tokio::test] +async fn already_positioned_cursor_replay_with_new_command_bytes_is_a_duplicate() { + const FILE_IDENTITY: u64 = 41; + + let tmp = TempDir::new().unwrap(); + let runtime = HostAdmissionTestRuntimeV1::profile(tmp.path()) + .await + .unwrap(); + let store = runtime + .observation_store(HostAdmissionScope::Profile) + .unwrap(); + let source = ObservationSourceIdentityV1::for_provider( + ProviderId::new(COLLISION_PROVIDER).unwrap(), + SessionId::new("session.cursor-replay-new-command-bytes").unwrap(), + ) + .unwrap(); + let generation = ObservationSourceGenerationV1::new(7).unwrap(); + let cursor_at = |offset: u64, resume_fingerprint: u64| { + ObservationSourceCursorV1::new( + source.clone(), + ObservationScopeV1::Profile, + generation, + offset, + ) + .unwrap() + .with_resume_checkpoint(FILE_IDENTITY, resume_fingerprint) + }; + let advance_over = |expected: Option, + covered: (u64, u64), + resume_fingerprint: u64| { + ObservationCursorAdvance::new( + source.clone(), + ObservationScopeV1::Profile, + generation, + expected, + ObservationSourceRangeV1::new(covered.0, covered.1).unwrap(), + ObservationCoverageReason::BlankFrame, + ) + .unwrap() + .with_resume_checkpoint(FILE_IDENTITY, resume_fingerprint) + }; + + assert_eq!( + store + .advance_source_cursor(advance_over(None, (0, 5), 11)) + .await + .unwrap(), + CursorAdvanceOutcome::Committed + ); + let admitted = advance_over(Some(cursor_at(5, 11)), (5, 10), 22); + assert_eq!( + store.advance_source_cursor(admitted.clone()).await.unwrap(), + CursorAdvanceOutcome::Committed + ); + + let replay = advance_over(Some(cursor_at(5, 33)), (5, 10), 22); + assert_eq!( + replay.coverage(), + admitted.coverage(), + "the replay must reuse the admitted coverage idempotency key" + ); + assert_ne!( + replay, admitted, + "the replay must carry different command bytes" + ); + assert_eq!( + store.advance_source_cursor(replay).await.unwrap(), + CursorAdvanceOutcome::ExactDuplicate, + "a coverage replay whose cursor is already at next must not wedge history" + ); + assert_eq!( + store + .get_source_cursor(&source, &ObservationScopeV1::Profile) + .await + .unwrap(), + Some(cursor_at(10, 22)), + "a duplicate advance must leave the admitted frontier untouched" + ); + + rewind_source_cursor(&runtime, &cursor_at(5, 11)).await; + assert!( + matches!( + store + .advance_source_cursor(advance_over(Some(cursor_at(5, 11)), (5, 10), 44)) + .await + .unwrap_err(), + ObservationStoreError::CursorAdvanceCollision + ), + "a conflicting replay that does not leave the cursor at next stays a collision" + ); +} + #[tokio::test] async fn runtime_cursor_replay_preserves_storage_failure() { let tmp = TempDir::new().unwrap(); From d1da6710d2cc8af016de0ea638a5b8c03f3717ca Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Sat, 19 Sep 2026 07:01:36 +0000 Subject: [PATCH 03/15] fix(session-temporal): page reads that were refusing stored results Whole-generation projection and doctor probes materialized past the exact SQL ceiling, so persist retried forever and doctor refused an oversized store before any finding. Page those reads and return the counts seen. --- .../src/doctor_health.rs | 1037 ++++++++++------- .../src/projection/derived.rs | 265 +++-- .../src/projection/receipts.rs | 137 ++- 3 files changed, 948 insertions(+), 491 deletions(-) diff --git a/crates/tracedecay-session-temporal-store/src/doctor_health.rs b/crates/tracedecay-session-temporal-store/src/doctor_health.rs index 9dbe978c52..8a76c5a39a 100644 --- a/crates/tracedecay-session-temporal-store/src/doctor_health.rs +++ b/crates/tracedecay-session-temporal-store/src/doctor_health.rs @@ -22,7 +22,8 @@ const MAX_FINDING_COUNT: u64 = 1_000_000; const SQLITE_CORRUPT_VTAB: i32 = 267; const SESSION_TEMPORAL_HEALTH_CACHE_TTL: Duration = Duration::from_secs(2); const MAX_CACHED_SESSION_TEMPORAL_STORES: usize = 64; -const MAX_SYNCHRONOUS_SESSION_TEMPORAL_HEALTH_BYTES: u64 = 64 * 1024 * 1024; +const HEALTH_PROBE_PAGE_SIZE: i64 = 512; +const HEALTH_PROBE_QUERY_LIMIT: i64 = HEALTH_PROBE_PAGE_SIZE + 1; #[derive(Clone, Copy, Debug, Eq, PartialEq)] struct SessionTemporalStoreFileFingerprint { @@ -119,77 +120,6 @@ fn session_temporal_store_fingerprint( }) } -fn session_temporal_store_family_bytes(database_path: &Path) -> std::io::Result { - hotpath::measure_block!("session_temporal.doctor.stat", { - let database = std::fs::metadata(database_path)?.len(); - let mut wal_path = database_path.as_os_str().to_os_string(); - wal_path.push("-wal"); - match std::fs::metadata(PathBuf::from(wal_path)) { - Ok(wal) => database.checked_add(wal.len()).ok_or_else(|| { - std::io::Error::new( - std::io::ErrorKind::InvalidData, - "session temporal store size overflowed", - ) - }), - Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(database), - Err(error) => Err(error), - } - }) -} - -fn permits_synchronous_session_temporal_health(database_path: &Path) -> bool { - session_temporal_store_family_bytes(database_path) - .is_ok_and(|bytes| bytes <= MAX_SYNCHRONOUS_SESSION_TEMPORAL_HEALTH_BYTES) -} - -// Occurrence and summary FTS integrity share the same content/docsize -// EXCEPT + probe-token shape; only the table names differ. -macro_rules! fts_integrity_check_sql { - ($content:literal, $fts:literal, $docsize:literal) => { - concat!( - "SELECT - (SELECT COUNT(*) FROM ( - SELECT rowid AS id FROM ", - $content, - " - EXCEPT SELECT id FROM ", - $docsize, - " - LIMIT 1000001 - )) - + (SELECT COUNT(*) FROM ( - SELECT id FROM ", - $docsize, - " - EXCEPT SELECT rowid AS id FROM ", - $content, - " - LIMIT 1000001 - )) - + COALESCE(( - SELECT 0 FROM ", - $fts, - " - WHERE ", - $fts, - " MATCH 'tracedecay_health_probe_token' - LIMIT 1 - ), 0)" - ) - }; -} - -const OCCURRENCE_FTS_CHECK_SQL: &str = fts_integrity_check_sql!( - "session_occurrences", - "session_occurrences_fts", - "session_occurrences_fts_docsize" -); -const SUMMARY_FTS_CHECK_SQL: &str = fts_integrity_check_sql!( - "session_summary_nodes", - "session_summary_nodes_fts", - "session_summary_nodes_fts_docsize" -); - const REQUIRED_BASE_TABLES: &[&str] = &[ "lcm_summary_nodes", "lcm_summary_sources", @@ -305,226 +235,428 @@ const REQUIRED_TRIGGERS: &[(&str, &str)] = &[ ), ]; +const INVALID_GENERATION_TAIL: &str = "WHERE candidate.generation <= 0 + OR json_valid(candidate.frozen_watermarks_json) = 0 + OR CASE WHEN json_valid(candidate.frozen_watermarks_json) = 1 THEN ( + json_type(candidate.frozen_watermarks_json, '$.active_generation') IS NOT 'integer' + OR CAST(json_extract( + candidate.frozen_watermarks_json, '$.active_generation' + ) AS INTEGER) <= 0 + OR CAST(json_extract( + candidate.frozen_watermarks_json, '$.active_generation' + ) AS INTEGER) > candidate.generation + OR json_type(candidate.frozen_watermarks_json, '$.source_frontier') IS NOT 'integer' + OR CAST(json_extract( + candidate.frozen_watermarks_json, '$.source_frontier' + ) AS INTEGER) < 0 + OR json_type(candidate.frozen_watermarks_json, '$.projection_frontier') IS NOT 'integer' + OR CAST(json_extract( + candidate.frozen_watermarks_json, '$.projection_frontier' + ) AS INTEGER) < 0 + OR json_type(candidate.frozen_watermarks_json, '$.summary_frontier') IS NOT 'integer' + OR CAST(json_extract( + candidate.frozen_watermarks_json, '$.summary_frontier' + ) AS INTEGER) < 0 + OR NOT ( + (candidate.state = 'building' AND candidate.ready_at IS NULL + AND candidate.activated_at IS NULL AND candidate.completed_at IS NULL) + OR (candidate.state = 'ready' AND candidate.ready_at IS NOT NULL + AND candidate.activated_at IS NULL AND candidate.completed_at IS NULL) + OR (candidate.state = 'active' AND candidate.ready_at IS NOT NULL + AND candidate.activated_at IS NOT NULL AND candidate.completed_at IS NULL) + OR (candidate.state = 'superseded' AND candidate.ready_at IS NOT NULL + AND candidate.activated_at IS NOT NULL + AND candidate.completed_at IS NOT NULL) + OR (candidate.state IN ('failed', 'cancelled') + AND candidate.completed_at IS NOT NULL) + ) + ) ELSE 0 END"; + +const MULTI_ACTIVE_GENERATION_TAIL: &str = "WHERE candidate.state = 'active' + AND NOT EXISTS ( + SELECT 1 + FROM session_temporal_generations AS earlier + WHERE earlier.session_id = candidate.session_id + AND earlier.state = 'active' + AND earlier.rowid < candidate.source_rowid + ) + AND EXISTS ( + SELECT 1 + FROM session_temporal_generations AS later + WHERE later.session_id = candidate.session_id + AND later.state = 'active' + AND later.rowid > candidate.source_rowid + )"; + +const CURSOR_KEY_ABSENT_TAIL: &str = "LEFT JOIN session_query_cursor_keys AS key + ON key.key_id = json_extract(candidate.frozen_watermarks_json, '$.cursor_key.key_id') + AND key.key_version = CAST(json_extract( + candidate.frozen_watermarks_json, '$.cursor_key.version' + ) AS INTEGER) + AND key.retired_at IS NULL + WHERE candidate.state = 'active' + AND ( + json_type(candidate.frozen_watermarks_json, '$.cursor_key') IS NOT 'object' + OR key.key_id IS NULL + )"; + +const STUCK_BINDING_TAIL: &str = "LEFT JOIN session_refresh_bindings AS binding + ON binding.session_id = candidate.session_id + AND binding.operation_id = candidate.operation_id + LEFT JOIN session_temporal_generations AS generation + ON generation.session_id = binding.session_id + AND generation.generation = binding.generation + WHERE candidate.state = 'running' + AND ( + binding.operation_id IS NULL + OR generation.session_id IS NULL + OR generation.state <> 'building' + )"; + +const STUCK_PROGRESS_SQL: &str = "WITH operation_source AS MATERIALIZED ( + SELECT rowid AS source_rowid, session_id, operation_id, state, updated_at + FROM session_refresh_operations + ORDER BY rowid + LIMIT ?1 + ), + operation_page AS MATERIALIZED ( + SELECT * FROM operation_source ORDER BY source_rowid LIMIT ?2 + ), + operation_progress AS MATERIALIZED ( + SELECT operation.*, + ( + SELECT MAX(progress.recorded_at) + FROM session_refresh_progress AS progress + WHERE progress.session_id = operation.session_id + AND progress.operation_id = operation.operation_id + AND progress.progress_ordinal < ?2 + ) AS latest_progress + FROM operation_page AS operation + ) + SELECT + (SELECT COUNT(*) + FROM operation_progress AS operation + JOIN session_refresh_bindings AS binding + ON binding.session_id = operation.session_id + AND binding.operation_id = operation.operation_id + WHERE operation.state = 'running' + AND NOT EXISTS( + SELECT 1 + FROM session_refresh_progress AS progress + WHERE progress.session_id = operation.session_id + AND progress.operation_id = operation.operation_id + AND progress.progress_ordinal >= ?2 + ) + AND ( + (operation.latest_progress IS NULL + AND operation.updated_at + < CAST(strftime('%s', 'now') AS INTEGER) * 1000000 - 900000000) + OR operation.latest_progress + < CAST(strftime('%s', 'now') AS INTEGER) * 1000000 - 900000000 + )), + EXISTS(SELECT 1 FROM operation_source LIMIT 1 OFFSET ?2) + OR EXISTS( + SELECT 1 + FROM operation_page AS operation + WHERE EXISTS( + SELECT 1 + FROM session_refresh_progress AS progress + WHERE progress.session_id = operation.session_id + AND progress.operation_id = operation.operation_id + AND progress.progress_ordinal >= ?2 + ) + )"; + +const STUCK_RECEIPT_TAIL: &str = "LEFT JOIN session_refresh_receipts AS receipt + ON receipt.session_id = candidate.session_id + AND receipt.operation_id = candidate.operation_id + WHERE (candidate.state = 'running' AND receipt.operation_id IS NOT NULL) + OR (candidate.state <> 'running' AND receipt.operation_id IS NULL) + OR (receipt.operation_id IS NOT NULL + AND ( + receipt.terminal_state <> candidate.state + OR receipt.terminal_at IS NOT candidate.terminal_at + OR receipt.failure_code IS NOT candidate.failure_code + ))"; + +const COMPATIBILITY_DRIFT_TAIL: &str = "LEFT JOIN lcm_summary_nodes AS compatibility + ON compatibility.node_id = candidate.summary_id + WHERE compatibility.node_id IS NULL + OR candidate.publication_json IS NULL + OR json_extract(candidate.publication_json, '$.summary_hash') IS NULL + OR compatibility.session_id <> candidate.session_id + OR compatibility.summary_text <> candidate.summary_text + OR compatibility.summary_hash + <> json_extract(candidate.publication_json, '$.summary_hash')"; + +macro_rules! row_health_check { + ( + $kind:ident, + $tables:expr, + $source_table:literal, + $source_columns:literal, + $count:literal, + $tail:expr + ) => { + HealthCheck { + kind: SessionTemporalHealthFindingKind::$kind, + tables: $tables, + probe: HealthProbe::Rows { + source_table: $source_table, + source_columns: $source_columns, + count: $count, + tail: $tail, + }, + } + }; +} + const CHECKS: &[HealthCheck] = &[ + row_health_check!( + OccurrenceFtsCorruption, + &["session_occurrences", "session_occurrences_fts_docsize"], + "session_occurrences", + "", + "COUNT(*)", + "LEFT JOIN session_occurrences_fts_docsize AS docsize + ON docsize.id = candidate.source_rowid + WHERE docsize.id IS NULL" + ), + row_health_check!( + OccurrenceFtsCorruption, + &["session_occurrences", "session_occurrences_fts_docsize"], + "session_occurrences_fts_docsize", + ", id", + "COUNT(*)", + "LEFT JOIN session_occurrences AS occurrence + ON occurrence.rowid = candidate.id + WHERE occurrence.rowid IS NULL" + ), HealthCheck { kind: SessionTemporalHealthFindingKind::OccurrenceFtsCorruption, - tables: &[ - "session_occurrences", - "session_occurrences_fts", - "session_occurrences_fts_docsize", - ], - sql: OCCURRENCE_FTS_CHECK_SQL, + tables: &["session_occurrences_fts"], + probe: HealthProbe::Sql( + "SELECT COALESCE(( + SELECT 0 FROM session_occurrences_fts + WHERE session_occurrences_fts MATCH 'tracedecay_health_probe_token' + LIMIT 1 + ), 0), (?1 - ?1) + (?2 - ?2)", + ), }, + row_health_check!( + SummaryFtsCorruption, + &["session_summary_nodes", "session_summary_nodes_fts_docsize"], + "session_summary_nodes", + "", + "COUNT(*)", + "LEFT JOIN session_summary_nodes_fts_docsize AS docsize + ON docsize.id = candidate.source_rowid + WHERE docsize.id IS NULL" + ), + row_health_check!( + SummaryFtsCorruption, + &["session_summary_nodes", "session_summary_nodes_fts_docsize"], + "session_summary_nodes_fts_docsize", + ", id", + "COUNT(*)", + "LEFT JOIN session_summary_nodes AS summary + ON summary.rowid = candidate.id + WHERE summary.rowid IS NULL" + ), HealthCheck { kind: SessionTemporalHealthFindingKind::SummaryFtsCorruption, - tables: &[ - "session_summary_nodes", - "session_summary_nodes_fts", - "session_summary_nodes_fts_docsize", - ], - sql: SUMMARY_FTS_CHECK_SQL, + tables: &["session_summary_nodes_fts"], + probe: HealthProbe::Sql( + "SELECT COALESCE(( + SELECT 0 FROM session_summary_nodes_fts + WHERE session_summary_nodes_fts MATCH 'tracedecay_health_probe_token' + LIMIT 1 + ), 0), (?1 - ?1) + (?2 - ?2)", + ), }, - HealthCheck { - kind: SessionTemporalHealthFindingKind::MissingAnchor, - tables: &[ - "retrieval_anchors", - "session_assertions", - "session_occurrences", - "session_summary_nodes", + row_health_check!( + MissingAnchor, + &["retrieval_anchors", "session_summary_nodes"], + "session_summary_nodes", + ", summary_anchor_id", + "COUNT(*)", + "LEFT JOIN retrieval_anchors AS anchor + ON anchor.anchor_id = candidate.summary_anchor_id + WHERE anchor.anchor_id IS NULL" + ), + row_health_check!( + MissingAnchor, + &["retrieval_anchors", "session_occurrences"], + "session_occurrences", + ", retrieval_anchor_id", + "COUNT(*)", + "LEFT JOIN retrieval_anchors AS anchor + ON anchor.anchor_id = candidate.retrieval_anchor_id + WHERE anchor.anchor_id IS NULL" + ), + row_health_check!( + MissingAnchor, + &["retrieval_anchors", "session_assertions"], + "session_assertions", + ", subject_anchor_id, object_anchor_id", + "COUNT(*)", + "LEFT JOIN retrieval_anchors AS subject + ON subject.anchor_id = candidate.subject_anchor_id + LEFT JOIN retrieval_anchors AS object + ON object.anchor_id = candidate.object_anchor_id + WHERE subject.anchor_id IS NULL OR object.anchor_id IS NULL" + ), + row_health_check!( + MissingReceipt, + &[ + "sanitization_receipts", + "session_external_payload_manifests" ], - sql: "SELECT - (SELECT COUNT(*) FROM session_summary_nodes AS node - LEFT JOIN retrieval_anchors AS anchor - ON anchor.anchor_id = node.summary_anchor_id - WHERE anchor.anchor_id IS NULL) - + (SELECT COUNT(*) FROM session_occurrences AS occurrence - LEFT JOIN retrieval_anchors AS anchor - ON anchor.anchor_id = occurrence.retrieval_anchor_id - WHERE anchor.anchor_id IS NULL) - + (SELECT COUNT(*) FROM session_assertions AS assertion - LEFT JOIN retrieval_anchors AS subject - ON subject.anchor_id = assertion.subject_anchor_id - LEFT JOIN retrieval_anchors AS object - ON object.anchor_id = assertion.object_anchor_id - WHERE subject.anchor_id IS NULL OR object.anchor_id IS NULL)", - }, - HealthCheck { - kind: SessionTemporalHealthFindingKind::MissingReceipt, - tables: &[ + "session_external_payload_manifests", + ", receipt_id", + "COUNT(*)", + "LEFT JOIN sanitization_receipts AS receipt + ON receipt.receipt_id = candidate.receipt_id + WHERE receipt.receipt_id IS NULL" + ), + row_health_check!( + MissingReceipt, + &[ "sanitization_receipts", - "session_external_payload_manifests", - "session_refresh_batch_bindings", - "session_summary_nodes", - "session_temporal_observation_effects", - "session_temporal_projection_receipts", + "session_temporal_observation_effects" ], - sql: "SELECT - (SELECT COUNT(*) FROM session_external_payload_manifests AS manifest - LEFT JOIN sanitization_receipts AS receipt - ON receipt.receipt_id = manifest.receipt_id - WHERE receipt.receipt_id IS NULL) - + (SELECT COUNT(*) FROM session_temporal_observation_effects AS effect - LEFT JOIN sanitization_receipts AS receipt - ON receipt.receipt_id = effect.receipt_id - WHERE receipt.receipt_id IS NULL) - + (SELECT COUNT(*) FROM session_summary_nodes AS summary - LEFT JOIN sanitization_receipts AS receipt - ON receipt.receipt_id = json_extract(summary.publication_json, '$.receipt_id') - WHERE summary.publication_json IS NULL OR receipt.receipt_id IS NULL) - + (SELECT COUNT(*) FROM session_refresh_batch_bindings AS binding - LEFT JOIN session_temporal_projection_receipts AS receipt - ON receipt.session_id = binding.session_id - AND receipt.generation = binding.generation - AND receipt.batch_ordinal = binding.batch_ordinal - WHERE receipt.session_id IS NULL)", - }, - HealthCheck { - kind: SessionTemporalHealthFindingKind::InvalidGeneration, - tables: &["session_temporal_generations"], - sql: "SELECT COUNT(*) FROM session_temporal_generations - WHERE generation <= 0 - OR json_valid(frozen_watermarks_json) = 0 - OR CASE WHEN json_valid(frozen_watermarks_json) = 1 THEN ( - json_type(frozen_watermarks_json, '$.active_generation') IS NOT 'integer' - OR CAST(json_extract( - frozen_watermarks_json, '$.active_generation' - ) AS INTEGER) <= 0 - OR CAST(json_extract( - frozen_watermarks_json, '$.active_generation' - ) AS INTEGER) > generation - OR json_type( - frozen_watermarks_json, '$.source_frontier' - ) IS NOT 'integer' - OR CAST(json_extract( - frozen_watermarks_json, '$.source_frontier' - ) AS INTEGER) < 0 - OR json_type( - frozen_watermarks_json, '$.projection_frontier' - ) IS NOT 'integer' - OR CAST(json_extract( - frozen_watermarks_json, '$.projection_frontier' - ) AS INTEGER) < 0 - OR json_type( - frozen_watermarks_json, '$.summary_frontier' - ) IS NOT 'integer' - OR CAST(json_extract( - frozen_watermarks_json, '$.summary_frontier' - ) AS INTEGER) < 0 - OR NOT ( - (state = 'building' AND ready_at IS NULL - AND activated_at IS NULL AND completed_at IS NULL) - OR (state = 'ready' AND ready_at IS NOT NULL - AND activated_at IS NULL AND completed_at IS NULL) - OR (state = 'active' AND ready_at IS NOT NULL - AND activated_at IS NOT NULL AND completed_at IS NULL) - OR (state = 'superseded' AND ready_at IS NOT NULL - AND activated_at IS NOT NULL AND completed_at IS NOT NULL) - OR (state IN ('failed', 'cancelled') AND completed_at IS NOT NULL) - ) - ) ELSE 0 END", - }, - HealthCheck { - kind: SessionTemporalHealthFindingKind::MultiActiveGeneration, - tables: &["session_temporal_generations"], - sql: "SELECT COUNT(*) FROM ( - SELECT session_id - FROM session_temporal_generations - WHERE state = 'active' - GROUP BY session_id - HAVING COUNT(*) > 1 - )", - }, - HealthCheck { - kind: SessionTemporalHealthFindingKind::CursorChainAbsent, - tables: &["session_query_cursor_keys", "session_temporal_generations"], - sql: "SELECT - (SELECT COUNT(*) FROM session_query_cursor_keys AS key - WHERE key.key_version > 1 - AND NOT EXISTS ( - SELECT 1 FROM session_query_cursor_keys AS predecessor - WHERE predecessor.key_version = key.key_version - 1 - )) - + (SELECT CASE - WHEN EXISTS( - SELECT 1 FROM session_temporal_generations - WHERE state = 'active' - ) AND ( - SELECT COUNT(*) FROM session_query_cursor_keys - WHERE retired_at IS NULL - ) <> 1 - THEN 1 ELSE 0 END)", - }, - HealthCheck { - kind: SessionTemporalHealthFindingKind::CursorKeyAbsent, - tables: &["session_query_cursor_keys", "session_temporal_generations"], - sql: "SELECT COUNT(*) - FROM session_temporal_generations AS generation - LEFT JOIN session_query_cursor_keys AS key - ON key.key_id = json_extract( - generation.frozen_watermarks_json, '$.cursor_key.key_id' - ) - AND key.key_version = CAST(json_extract( - generation.frozen_watermarks_json, '$.cursor_key.version' - ) AS INTEGER) - AND key.retired_at IS NULL - WHERE generation.state = 'active' - AND ( - json_type(generation.frozen_watermarks_json, '$.cursor_key') IS NOT 'object' - OR key.key_id IS NULL - )", - }, - HealthCheck { - kind: SessionTemporalHealthFindingKind::OwnershipDrift, - tables: &[ + "session_temporal_observation_effects", + ", receipt_id", + "COUNT(*)", + "LEFT JOIN sanitization_receipts AS receipt + ON receipt.receipt_id = candidate.receipt_id + WHERE receipt.receipt_id IS NULL" + ), + row_health_check!( + MissingReceipt, + &["sanitization_receipts", "session_summary_nodes"], + "session_summary_nodes", + ", publication_json", + "COUNT(*)", + "LEFT JOIN sanitization_receipts AS receipt + ON receipt.receipt_id = + json_extract(candidate.publication_json, '$.receipt_id') + WHERE candidate.publication_json IS NULL OR receipt.receipt_id IS NULL" + ), + row_health_check!( + MissingReceipt, + &[ "session_refresh_batch_bindings", - "session_refresh_bindings", - "session_summary_availability", - "session_summary_nodes", + "session_temporal_projection_receipts" ], - sql: "SELECT - (SELECT COUNT(*) - FROM session_summary_availability AS availability - LEFT JOIN session_summary_nodes AS summary - ON summary.summary_id = availability.summary_id - WHERE summary.summary_id IS NULL - OR availability.session_id IS NOT summary.session_id) - + (SELECT COUNT(*) - FROM session_refresh_batch_bindings AS batch - LEFT JOIN session_refresh_bindings AS binding - ON binding.session_id = batch.session_id - AND binding.operation_id = batch.operation_id - WHERE binding.operation_id IS NULL - OR batch.generation IS NOT binding.generation)", - }, - HealthCheck { - kind: SessionTemporalHealthFindingKind::StuckRefresh, - tables: &["session_refresh_operations"], - sql: "SELECT COUNT(*) FROM session_refresh_operations - WHERE state = 'running' - AND updated_at < CAST(strftime('%s', 'now') AS INTEGER) * 1000000 - 900000000", - }, - HealthCheck { - kind: SessionTemporalHealthFindingKind::StuckBinding, - tables: &[ + "session_refresh_batch_bindings", + ", session_id, generation, batch_ordinal", + "COUNT(*)", + "LEFT JOIN session_temporal_projection_receipts AS receipt + ON receipt.session_id = candidate.session_id + AND receipt.generation = candidate.generation + AND receipt.batch_ordinal = candidate.batch_ordinal + WHERE receipt.session_id IS NULL" + ), + row_health_check!( + InvalidGeneration, + &["session_temporal_generations"], + "session_temporal_generations", + ", generation, state, frozen_watermarks_json, ready_at, activated_at, completed_at", + "COUNT(*)", + INVALID_GENERATION_TAIL + ), + row_health_check!( + MultiActiveGeneration, + &["session_temporal_generations"], + "session_temporal_generations", + ", session_id, state", + "COUNT(*)", + MULTI_ACTIVE_GENERATION_TAIL + ), + row_health_check!( + CursorChainAbsent, + &["session_query_cursor_keys"], + "session_query_cursor_keys", + ", key_version", + "COUNT(*)", + "WHERE candidate.key_version > 1 + AND NOT EXISTS ( + SELECT 1 + FROM session_query_cursor_keys AS predecessor + WHERE predecessor.key_version = candidate.key_version - 1 + )" + ), + row_health_check!( + CursorChainAbsent, + &["session_query_cursor_keys", "session_temporal_generations"], + "session_temporal_generations", + ", state", + "CASE WHEN COUNT(*) > 0 THEN 1 ELSE 0 END", + "WHERE candidate.state = 'active' + AND ( + SELECT COUNT(*) + FROM ( + SELECT 1 + FROM session_query_cursor_keys + WHERE retired_at IS NULL + LIMIT 2 + ) + ) <> 1" + ), + row_health_check!( + CursorKeyAbsent, + &["session_query_cursor_keys", "session_temporal_generations"], + "session_temporal_generations", + ", state, frozen_watermarks_json", + "COUNT(*)", + CURSOR_KEY_ABSENT_TAIL + ), + row_health_check!( + OwnershipDrift, + &["session_summary_availability", "session_summary_nodes"], + "session_summary_availability", + ", session_id, summary_id", + "COUNT(*)", + "LEFT JOIN session_summary_nodes AS summary + ON summary.summary_id = candidate.summary_id + WHERE summary.summary_id IS NULL + OR candidate.session_id IS NOT summary.session_id" + ), + row_health_check!( + OwnershipDrift, + &["session_refresh_batch_bindings", "session_refresh_bindings"], + "session_refresh_batch_bindings", + ", session_id, operation_id, generation", + "COUNT(*)", + "LEFT JOIN session_refresh_bindings AS binding + ON binding.session_id = candidate.session_id + AND binding.operation_id = candidate.operation_id + WHERE binding.operation_id IS NULL + OR candidate.generation IS NOT binding.generation" + ), + row_health_check!( + StuckRefresh, + &["session_refresh_operations"], + "session_refresh_operations", + ", state, updated_at", + "COUNT(*)", + "WHERE candidate.state = 'running' + AND candidate.updated_at + < CAST(strftime('%s', 'now') AS INTEGER) * 1000000 - 900000000" + ), + row_health_check!( + StuckBinding, + &[ "session_refresh_bindings", "session_refresh_operations", - "session_temporal_generations", + "session_temporal_generations" ], - sql: "SELECT COUNT(*) - FROM session_refresh_operations AS operation - LEFT JOIN session_refresh_bindings AS binding - ON binding.session_id = operation.session_id - AND binding.operation_id = operation.operation_id - LEFT JOIN session_temporal_generations AS generation - ON generation.session_id = binding.session_id - AND generation.generation = binding.generation - WHERE operation.state = 'running' - AND ( - binding.operation_id IS NULL - OR generation.session_id IS NULL - OR generation.state <> 'building' - )", - }, + "session_refresh_operations", + ", session_id, operation_id, state", + "COUNT(*)", + STUCK_BINDING_TAIL + ), HealthCheck { kind: SessionTemporalHealthFindingKind::StuckProgress, tables: &[ @@ -532,56 +664,24 @@ const CHECKS: &[HealthCheck] = &[ "session_refresh_operations", "session_refresh_progress", ], - sql: "SELECT COUNT(*) FROM ( - SELECT operation.session_id, operation.operation_id - FROM session_refresh_operations AS operation - JOIN session_refresh_bindings AS binding - ON binding.session_id = operation.session_id - AND binding.operation_id = operation.operation_id - LEFT JOIN session_refresh_progress AS progress - ON progress.session_id = operation.session_id - AND progress.operation_id = operation.operation_id - WHERE operation.state = 'running' - GROUP BY operation.session_id, operation.operation_id - HAVING (MAX(progress.recorded_at) IS NULL - AND MAX(operation.updated_at) - < CAST(strftime('%s', 'now') AS INTEGER) * 1000000 - 900000000) - OR MAX(progress.recorded_at) - < CAST(strftime('%s', 'now') AS INTEGER) * 1000000 - 900000000 - )", - }, - HealthCheck { - kind: SessionTemporalHealthFindingKind::StuckReceipt, - tables: &["session_refresh_operations", "session_refresh_receipts"], - sql: "SELECT COUNT(*) - FROM session_refresh_operations AS operation - LEFT JOIN session_refresh_receipts AS receipt - ON receipt.session_id = operation.session_id - AND receipt.operation_id = operation.operation_id - WHERE (operation.state = 'running' AND receipt.operation_id IS NOT NULL) - OR (operation.state <> 'running' AND receipt.operation_id IS NULL) - OR (receipt.operation_id IS NOT NULL - AND ( - receipt.terminal_state <> operation.state - OR receipt.terminal_at IS NOT operation.terminal_at - OR receipt.failure_code IS NOT operation.failure_code - ))", - }, - HealthCheck { - kind: SessionTemporalHealthFindingKind::CompatibilityDrift, - tables: &["lcm_summary_nodes", "session_summary_nodes"], - sql: "SELECT COUNT(*) - FROM session_summary_nodes AS canonical - LEFT JOIN lcm_summary_nodes AS compatibility - ON compatibility.node_id = canonical.summary_id - WHERE compatibility.node_id IS NULL - OR canonical.publication_json IS NULL - OR json_extract(canonical.publication_json, '$.summary_hash') IS NULL - OR compatibility.session_id <> canonical.session_id - OR compatibility.summary_text <> canonical.summary_text - OR compatibility.summary_hash - <> json_extract(canonical.publication_json, '$.summary_hash')", + probe: HealthProbe::Sql(STUCK_PROGRESS_SQL), }, + row_health_check!( + StuckReceipt, + &["session_refresh_operations", "session_refresh_receipts"], + "session_refresh_operations", + ", session_id, operation_id, state, terminal_at, failure_code", + "COUNT(*)", + STUCK_RECEIPT_TAIL + ), + row_health_check!( + CompatibilityDrift, + &["lcm_summary_nodes", "session_summary_nodes"], + "session_summary_nodes", + ", summary_id, session_id, summary_text, publication_json", + "COUNT(*)", + COMPATIBILITY_DRIFT_TAIL + ), ]; #[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)] @@ -640,9 +740,8 @@ impl SessionTemporalHealthFinding { pub struct SessionTemporalHealthReport { status: SessionTemporalHealthStatus, findings: Vec, - /// Why diagnosis could not complete: a fixed machine reason for path-API - /// unavailability (for example `synchronous_diagnosis_size_budget_exceeded`) - /// or a `: ` detail naming the read that failed. + /// Why diagnosis could not complete: a fixed bounded-probe reason or a + /// `: ` detail naming the read that failed. /// Omitted when diagnosis ran to completion against an immutable snapshot. #[serde(default, skip_serializing_if = "Option::is_none")] reason: Option, @@ -666,7 +765,17 @@ impl SessionTemporalHealthReport { struct HealthCheck { kind: SessionTemporalHealthFindingKind, tables: &'static [&'static str], - sql: &'static str, + probe: HealthProbe, +} + +enum HealthProbe { + Rows { + source_table: &'static str, + source_columns: &'static str, + count: &'static str, + tail: &'static str, + }, + Sql(&'static str), } impl SessionTemporalAccess<'_, D> { @@ -677,12 +786,6 @@ impl SessionTemporalAccess<'_, D> { #[hotpath::measure(future = true, label = "session_temporal.doctor.query")] pub async fn session_temporal_doctor_health(&self) -> SessionTemporalHealthReport { let database_path = self.db_path(); - if !permits_synchronous_session_temporal_health(database_path) { - return unavailable_report_with_reason( - SessionTemporalHealthStatus::Unavailable, - Some("synchronous_diagnosis_size_budget_exceeded"), - ); - } let cache = session_temporal_health_cache_cell(database_path); let mut cached = cache.lock().await; let before = session_temporal_store_fingerprint(database_path).ok(); @@ -753,6 +856,7 @@ async fn diagnose_snapshot( let mut status = SessionTemporalHealthStatus::Complete; let mut findings = Vec::new(); + let mut partial_reasons = BTreeSet::new(); let missing_tables = required_table_names() .filter(|table| !inventory.tables.contains(*table)) .count() as u64; @@ -835,33 +939,61 @@ async fn diagnose_snapshot( continue; } record_session_doctor_check(); - match snapshot_count(conn, check.sql).await { - Ok(0) => {} - Ok(value) => merge_finding(&mut findings, check.kind, value), - Err(error) if is_fts_finding(check.kind) && is_fts_virtual_table_corruption(&error) => { - merge_finding(&mut findings, check.kind, 1); - } - Err(error) if is_engine_locked(&error) => { - return SessionTemporalHealthReport { - status: SessionTemporalHealthStatus::Locked, - findings, - reason: None, - }; - } - Err(error) => { - return unavailable_report_with_detail( - SessionTemporalHealthStatus::Unavailable, - check_probe_name(check.kind), - &error, - ); - } + if diagnose_health_check( + conn, + check, + &mut status, + &mut findings, + &mut partial_reasons, + ) + .await + { + return SessionTemporalHealthReport { + status: SessionTemporalHealthStatus::Locked, + findings, + reason: None, + }; } } findings.sort_by_key(SessionTemporalHealthFinding::kind); SessionTemporalHealthReport { status, findings, - reason: None, + reason: (!partial_reasons.is_empty()) + .then(|| partial_reasons.into_iter().collect::>().join("; ")), + } +} + +async fn diagnose_health_check( + conn: &impl crate::handle::SessionTemporalQuery, + check: &HealthCheck, + status: &mut SessionTemporalHealthStatus, + findings: &mut Vec, + partial_reasons: &mut BTreeSet, +) -> bool { + let probe_name = check_probe_name(check.kind); + match snapshot_count(conn, check).await { + Ok(outcome) => { + if outcome.count > 0 || !outcome.complete { + merge_finding(findings, check.kind, outcome.count); + } + if !outcome.complete { + *status = SessionTemporalHealthStatus::Partial; + partial_reasons.insert(format!("{probe_name}: bounded_row_probe_incomplete")); + } + false + } + Err(error) if is_fts_finding(check.kind) && is_fts_virtual_table_corruption(&error) => { + merge_finding(findings, check.kind, 1); + false + } + Err(error) if is_engine_locked(&error) => true, + Err(error) => { + *status = SessionTemporalHealthStatus::Partial; + merge_finding(findings, check.kind, 0); + partial_reasons.insert(format!("{probe_name}: {error}")); + false + } } } @@ -959,19 +1091,54 @@ async fn snapshot_schema_version( #[hotpath::measure(future = true, label = "session_temporal.doctor.query.count")] async fn snapshot_count( conn: &impl crate::handle::SessionTemporalQuery, - sql: &str, -) -> tracedecay_runtime_core::db::engine::Result { - let mut rows = conn.query(sql, ()).await?; - let value = rows - .next() - .await? - .map(|row| row.get::>(0)) - .transpose()? - .flatten(); - Ok(value - .and_then(|value| u64::try_from(value).ok()) - .unwrap_or(0) - .min(MAX_FINDING_COUNT)) + check: &HealthCheck, +) -> tracedecay_runtime_core::db::engine::Result { + let sql = match &check.probe { + HealthProbe::Rows { + source_table, + source_columns, + count, + tail, + } => format!( + "WITH source AS MATERIALIZED ( + SELECT rowid AS source_rowid{source_columns} + FROM {source_table} + ORDER BY rowid + LIMIT ?1 + ), + page AS MATERIALIZED ( + SELECT * FROM source ORDER BY source_rowid LIMIT ?2 + ) + SELECT {count}, + EXISTS(SELECT 1 FROM source LIMIT 1 OFFSET ?2) + FROM page AS candidate + {tail}" + ), + HealthProbe::Sql(sql) => (*sql).to_owned(), + }; + let mut rows = conn + .query(&sql, [HEALTH_PROBE_QUERY_LIMIT, HEALTH_PROBE_PAGE_SIZE]) + .await?; + let Some(row) = rows.next().await? else { + return Ok(HealthProbeOutcome { + count: 0, + complete: true, + }); + }; + let value = row.get::>(0)?; + let incomplete = row.get::(1)? != 0; + Ok(HealthProbeOutcome { + count: value + .and_then(|value| u64::try_from(value).ok()) + .unwrap_or(0) + .min(MAX_FINDING_COUNT), + complete: !incomplete, + }) +} + +struct HealthProbeOutcome { + count: u64, + complete: bool, } fn finding(kind: SessionTemporalHealthFindingKind, count: u64) -> SessionTemporalHealthFinding { @@ -1070,17 +1237,6 @@ fn unavailable_report_with_detail( } } -fn unavailable_report_with_reason( - status: SessionTemporalHealthStatus, - reason: Option<&'static str>, -) -> SessionTemporalHealthReport { - SessionTemporalHealthReport { - status, - findings: Vec::new(), - reason: reason.map(str::to_string), - } -} - #[inline(always)] fn record_session_doctor_cache_hit() { #[cfg(feature = "hotpath")] @@ -1102,8 +1258,6 @@ fn record_session_doctor_check() { #[cfg(test)] mod cache_tests { use super::*; - use crate::handle::SessionTemporalAccess; - use tracedecay_global_db::tests::harness::RegisteredGlobalDbHarness; #[test] fn session_temporal_fingerprint_tracks_database_and_wal_changes() { @@ -1121,20 +1275,115 @@ mod cache_tests { let expanded = session_temporal_store_fingerprint(&database).expect("expanded fingerprint"); assert_ne!(with_wal, expanded); } +} - #[test] - fn session_temporal_size_budget_includes_wal_bytes() { +#[cfg(test)] +mod probe_tests { + use super::*; + use crate::handle::SessionTemporalExec; + use tracedecay_runtime_core::db::engine::TestConnection; + + #[tokio::test] + async fn row_probe_reports_observed_findings_when_its_page_is_incomplete() { let tmp = tempfile::TempDir::new().expect("tempdir"); - let database = tmp.path().join("sessions.db"); - std::fs::write(&database, b"database").expect("database"); - assert!(permits_synchronous_session_temporal_health(&database)); + let connection = TestConnection::open(&tmp.path().join("doctor-page.db")); + SessionTemporalExec::execute_batch( + &connection, + &format!( + "CREATE TABLE retrieval_anchors (anchor_id TEXT PRIMARY KEY); + CREATE TABLE session_summary_nodes (summary_anchor_id TEXT NOT NULL); + CREATE TABLE session_occurrences (retrieval_anchor_id TEXT NOT NULL); + CREATE TABLE session_assertions ( + subject_anchor_id TEXT NOT NULL, + object_anchor_id TEXT NOT NULL + ); + WITH RECURSIVE sequence(value) AS ( + VALUES(0) + UNION ALL + SELECT value + 1 FROM sequence WHERE value < {} + ) + INSERT INTO session_summary_nodes (summary_anchor_id) + SELECT printf('missing-%d', value) FROM sequence;", + HEALTH_PROBE_PAGE_SIZE + ), + ) + .await + .expect("seed oversized health probe"); + let check = CHECKS + .iter() + .find(|check| check.kind == SessionTemporalHealthFindingKind::MissingAnchor) + .expect("missing-anchor check"); + + let mut status = SessionTemporalHealthStatus::Complete; + let mut findings = Vec::new(); + let mut partial_reasons = BTreeSet::new(); + + assert!( + !diagnose_health_check( + &connection, + check, + &mut status, + &mut findings, + &mut partial_reasons, + ) + .await + ); + assert_eq!(status, SessionTemporalHealthStatus::Partial); + assert_eq!( + findings, + vec![finding( + SessionTemporalHealthFindingKind::MissingAnchor, + HEALTH_PROBE_PAGE_SIZE as u64, + )] + ); + assert_eq!( + partial_reasons, + BTreeSet::from(["missing_anchor: bounded_row_probe_incomplete".to_owned()]) + ); + } +} - let wal = tmp.path().join("sessions.db-wal"); - std::fs::File::create(wal) - .expect("wal") - .set_len(MAX_SYNCHRONOUS_SESSION_TEMPORAL_HEALTH_BYTES) - .expect("wal size"); - assert!(!permits_synchronous_session_temporal_health(&database)); +#[cfg(test)] +mod registered_tests { + use super::*; + use crate::handle::{SessionTemporalAccess, SessionTemporalExec, SessionTemporalRegisteredDb}; + use tracedecay_global_db::tests::harness::RegisteredGlobalDbHarness; + + #[tokio::test] + async fn oversized_session_temporal_store_still_reports_schema_findings() { + const OLD_SYNCHRONOUS_HEALTH_BUDGET_BYTES: u64 = 64 * 1024 * 1024; + let harness = + RegisteredGlobalDbHarness::open_without_relation_graph("doctor-oversized-store").await; + let writer = harness.registered.writer_connection().expect("writer"); + SessionTemporalExec::execute( + &writer, + "DROP INDEX idx_session_occurrences_generation_order", + (), + ) + .await + .expect("drop required index"); + drop(writer); + + let database = SessionTemporalRegisteredDb::db_path(&harness.registered); + std::fs::OpenOptions::new() + .write(true) + .open(database) + .expect("open session database") + .set_len(OLD_SYNCHRONOUS_HEALTH_BUDGET_BYTES + 4096) + .expect("grow session database past the former admission budget"); + + let report = SessionTemporalAccess::new(&harness.registered) + .session_temporal_doctor_health() + .await; + + assert_eq!(report.status(), SessionTemporalHealthStatus::Partial); + assert_ne!( + report.reason(), + Some("synchronous_diagnosis_size_budget_exceeded") + ); + assert!(report.findings().iter().any(|finding| { + finding.kind() == SessionTemporalHealthFindingKind::MigrationGap && finding.count() >= 1 + })); } #[tokio::test] diff --git a/crates/tracedecay-session-temporal-store/src/projection/derived.rs b/crates/tracedecay-session-temporal-store/src/projection/derived.rs index fbc84ea164..df18db3bf5 100644 --- a/crates/tracedecay-session-temporal-store/src/projection/derived.rs +++ b/crates/tracedecay-session-temporal-store/src/projection/derived.rs @@ -11,6 +11,10 @@ use tracedecay_temporal_query::ports::ExecutionControl; use super::super::query::{PERSIST_OPERATION, generation_i64, storage, storage_message}; use super::super::rebuild::checkpoint_relation_rebuild_control; +/// Keep one occurrence-ref read under the exact-SQL materialization ceiling +/// (10_000 rows / 64 MiB). A terminal generation is larger than that ceiling. +const OCCURRENCE_REF_PAGE_ROWS: i64 = 512; + #[hotpath::measure(future = true, label = "session_temporal.projection.rebuild_derived")] pub(super) async fn rebuild_derived_evidence( conn: &impl crate::handle::SessionTemporalExec, @@ -63,79 +67,138 @@ async fn load_occurrence_refs( generation: i64, control: &ExecutionControl, ) -> SessionStoreResult> { - checkpoint_relation_rebuild_control(control)?; - let mut rows = conn - .query( - "SELECT occurrence.occurrence_id, - occurrence.retrieval_anchor_id, - occurrence.thread_id, - occurrence.message_id, - occurrence.knowledge_at, - effect.observation_sequence, - occurrence.projection_output_ordinal - FROM session_occurrences AS occurrence - JOIN session_temporal_observation_effects AS effect - ON effect.observation_id = occurrence.source_observation_id - AND effect.session_id = occurrence.session_id - WHERE occurrence.session_id = ?1 AND occurrence.generation = ?2 - ORDER BY effect.observation_sequence ASC, - occurrence.projection_output_ordinal ASC, - occurrence.occurrence_id ASC", - params![session_id.as_str(), generation], - ) - .await - .map_err(|error| storage(PERSIST_OPERATION, error))?; let mut occurrences = Vec::new(); - while let Some(row) = rows - .next() - .await - .map_err(|error| storage(PERSIST_OPERATION, error))? - { + let mut cursor: Option<(i64, i64, String)> = None; + loop { checkpoint_relation_rebuild_control(control)?; - let occurrence_id = row - .get::(0) - .map_err(|error| storage(PERSIST_OPERATION, error))?; - let retrieval_anchor_id = row - .get::(1) - .map_err(|error| storage(PERSIST_OPERATION, error))?; - let thread_id = row - .get::>(2) - .map_err(|error| storage(PERSIST_OPERATION, error))?; - let message_id = row - .get::>(3) - .map_err(|error| storage(PERSIST_OPERATION, error))?; - let knowledge_at = row - .get::(4) - .map_err(|error| storage(PERSIST_OPERATION, error))?; - let observation_sequence = row - .get::(5) - .map_err(|error| storage(PERSIST_OPERATION, error))?; - let projection_output_ordinal = row - .get::(6) - .map_err(|error| storage(PERSIST_OPERATION, error))?; - occurrences.push(DerivedEvidenceOccurrenceRefV1 { - occurrence_id: MessageOccurrenceIdV1::new(occurrence_id) - .map_err(|error| storage(PERSIST_OPERATION, error))?, - retrieval_anchor_id: RetrievalAnchorId::new(retrieval_anchor_id) - .map_err(|error| storage_message(PERSIST_OPERATION, error.to_string()))?, - thread_id: thread_id - .map(|value| { - ThreadId::new(value) - .map_err(|error| storage_message(PERSIST_OPERATION, error.to_string())) - }) - .transpose()?, - message_id: message_id - .map(|value| { - MessageId::new(value) - .map_err(|error| storage_message(PERSIST_OPERATION, error.to_string())) - }) - .transpose()?, - knowledge_at: UtcMicros(knowledge_at), - observation_sequence: u64::try_from(observation_sequence) + let mut rows = match &cursor { + None => conn + .query( + "SELECT occurrence.occurrence_id, + occurrence.retrieval_anchor_id, + occurrence.thread_id, + occurrence.message_id, + occurrence.knowledge_at, + effect.observation_sequence, + occurrence.projection_output_ordinal + FROM session_occurrences AS occurrence + JOIN session_temporal_observation_effects AS effect + ON effect.observation_id = occurrence.source_observation_id + AND effect.session_id = occurrence.session_id + WHERE occurrence.session_id = ?1 AND occurrence.generation = ?2 + ORDER BY effect.observation_sequence ASC, + occurrence.projection_output_ordinal ASC, + occurrence.occurrence_id ASC + LIMIT ?3", + params![session_id.as_str(), generation, OCCURRENCE_REF_PAGE_ROWS], + ) + .await .map_err(|error| storage(PERSIST_OPERATION, error))?, - projection_output_ordinal: u32::try_from(projection_output_ordinal) + Some((sequence, ordinal, occurrence_id)) => conn + .query( + "SELECT occurrence.occurrence_id, + occurrence.retrieval_anchor_id, + occurrence.thread_id, + occurrence.message_id, + occurrence.knowledge_at, + effect.observation_sequence, + occurrence.projection_output_ordinal + FROM session_occurrences AS occurrence + JOIN session_temporal_observation_effects AS effect + ON effect.observation_id = occurrence.source_observation_id + AND effect.session_id = occurrence.session_id + WHERE occurrence.session_id = ?1 AND occurrence.generation = ?2 + AND ( + effect.observation_sequence > ?3 + OR ( + effect.observation_sequence = ?3 + AND occurrence.projection_output_ordinal > ?4 + ) + OR ( + effect.observation_sequence = ?3 + AND occurrence.projection_output_ordinal = ?4 + AND occurrence.occurrence_id > ?5 + ) + ) + ORDER BY effect.observation_sequence ASC, + occurrence.projection_output_ordinal ASC, + occurrence.occurrence_id ASC + LIMIT ?6", + params![ + session_id.as_str(), + generation, + *sequence, + *ordinal, + occurrence_id.as_str(), + OCCURRENCE_REF_PAGE_ROWS + ], + ) + .await .map_err(|error| storage(PERSIST_OPERATION, error))?, - }); + }; + let mut page_rows = 0_i64; + let mut page_cursor = None; + while let Some(row) = rows + .next() + .await + .map_err(|error| storage(PERSIST_OPERATION, error))? + { + checkpoint_relation_rebuild_control(control)?; + let occurrence_id = row + .get::(0) + .map_err(|error| storage(PERSIST_OPERATION, error))?; + let retrieval_anchor_id = row + .get::(1) + .map_err(|error| storage(PERSIST_OPERATION, error))?; + let thread_id = row + .get::>(2) + .map_err(|error| storage(PERSIST_OPERATION, error))?; + let message_id = row + .get::>(3) + .map_err(|error| storage(PERSIST_OPERATION, error))?; + let knowledge_at = row + .get::(4) + .map_err(|error| storage(PERSIST_OPERATION, error))?; + let observation_sequence = row + .get::(5) + .map_err(|error| storage(PERSIST_OPERATION, error))?; + let projection_output_ordinal = row + .get::(6) + .map_err(|error| storage(PERSIST_OPERATION, error))?; + page_cursor = Some(( + observation_sequence, + projection_output_ordinal, + occurrence_id.clone(), + )); + page_rows += 1; + occurrences.push(DerivedEvidenceOccurrenceRefV1 { + occurrence_id: MessageOccurrenceIdV1::new(occurrence_id) + .map_err(|error| storage(PERSIST_OPERATION, error))?, + retrieval_anchor_id: RetrievalAnchorId::new(retrieval_anchor_id) + .map_err(|error| storage_message(PERSIST_OPERATION, error.to_string()))?, + thread_id: thread_id + .map(|value| { + ThreadId::new(value) + .map_err(|error| storage_message(PERSIST_OPERATION, error.to_string())) + }) + .transpose()?, + message_id: message_id + .map(|value| { + MessageId::new(value) + .map_err(|error| storage_message(PERSIST_OPERATION, error.to_string())) + }) + .transpose()?, + knowledge_at: UtcMicros(knowledge_at), + observation_sequence: u64::try_from(observation_sequence) + .map_err(|error| storage(PERSIST_OPERATION, error))?, + projection_output_ordinal: u32::try_from(projection_output_ordinal) + .map_err(|error| storage(PERSIST_OPERATION, error))?, + }); + } + if page_rows < OCCURRENCE_REF_PAGE_ROWS { + break; + } + cursor = page_cursor; } Ok(occurrences) } @@ -260,3 +323,67 @@ async fn ensure_derived_anchor( .map_err(|error| storage(PERSIST_OPERATION, error))?; Ok(()) } + +#[cfg(test)] +mod tests { + use super::*; + use tracedecay_runtime_core::db::engine::Executor; + + #[tokio::test] + async fn occurrence_refs_span_more_than_one_exact_sql_page() { + let dir = tempfile::TempDir::new().expect("occurrence dir"); + let conn = tracedecay_runtime_core::db::engine::TestConnection::open( + &dir.path().join("occurrences.db"), + ); + Executor::execute_batch( + &conn, + "CREATE TABLE session_occurrences ( + session_id TEXT, generation INTEGER, occurrence_id TEXT, + retrieval_anchor_id TEXT, thread_id TEXT, message_id TEXT, + knowledge_at INTEGER, projection_output_ordinal INTEGER, + source_observation_id TEXT + ); + CREATE TABLE session_temporal_observation_effects ( + observation_id TEXT, session_id TEXT, observation_sequence INTEGER + );", + ) + .await + .expect("schema"); + let total = OCCURRENCE_REF_PAGE_ROWS + 3; + for index in 0..total { + let occurrence_id = format!("sha256:{index:064x}"); + let observation_id = format!("obs-{index}"); + Executor::execute( + &conn, + "INSERT INTO session_occurrences ( + session_id, generation, occurrence_id, retrieval_anchor_id, + knowledge_at, projection_output_ordinal, source_observation_id + ) VALUES ('page-session', 1, ?1, 'anchor.page', ?2, 0, ?3)", + params![occurrence_id, index, observation_id.clone()], + ) + .await + .expect("occurrence"); + Executor::execute( + &conn, + "INSERT INTO session_temporal_observation_effects ( + observation_id, session_id, observation_sequence + ) VALUES (?1, 'page-session', ?2)", + params![observation_id, index], + ) + .await + .expect("effect"); + } + + let session_id = SessionId::new("page-session").expect("session"); + let refs = load_occurrence_refs(&conn, &session_id, 1, &ExecutionControl::default()) + .await + .expect("paged refs"); + + assert_eq!(refs.len(), usize::try_from(total).expect("total")); + assert_eq!(refs[0].observation_sequence, 0); + assert_eq!( + refs.last().expect("last").observation_sequence, + u64::try_from(total - 1).expect("last sequence") + ); + } +} diff --git a/crates/tracedecay-session-temporal-store/src/projection/receipts.rs b/crates/tracedecay-session-temporal-store/src/projection/receipts.rs index d73162675e..07f34b3533 100644 --- a/crates/tracedecay-session-temporal-store/src/projection/receipts.rs +++ b/crates/tracedecay-session-temporal-store/src/projection/receipts.rs @@ -22,6 +22,11 @@ use super::super::relations::{LogicalCopyRelation, SessionRelationProjection}; use super::persist::*; const MAX_RECEIPT_COPY_ENTITIES: usize = 100_000; +/// One coverage digest page stays under the exact-SQL materialization ceiling. +/// `snippet_text` and `index_text` are the row bulk, so this is row-count-small +/// on purpose. OFFSET is O(n²) row visits on a terminal batch; keyset on the +/// ORDER BY column is the upgrade if a generation makes this the hot path. +const COVERAGE_DIGEST_PAGE_ROWS: i64 = 32; #[hotpath::measure(future = true, label = "session_temporal.projection.validate_receipt")] pub async fn validate_final_projection_receipt( @@ -642,40 +647,69 @@ pub(super) async fn digest_query_rows( batch: &SessionTemporalProjectionBatchV1, control: Option<&ExecutionControl>, ) -> SessionStoreResult<(usize, String)> { - if let Some(control) = control { - checkpoint_relation_rebuild_control(control)?; - } - record_coverage_query_probe(); - let mut rows = conn - .query( - sql, - params![ - batch.session_id().as_str(), - generation_i64(batch.generation(), PERSIST_OPERATION)?, - ], - ) - .await - .map_err(|error| storage(PERSIST_OPERATION, error))?; + digest_paged_rows( + conn, + sql, + batch.session_id().as_str(), + generation_i64(batch.generation(), PERSIST_OPERATION)?, + control, + ) + .await +} + +async fn digest_paged_rows( + conn: &impl crate::handle::SessionTemporalExec, + sql: &str, + session_id: &str, + generation: i64, + control: Option<&ExecutionControl>, +) -> SessionStoreResult<(usize, String)> { + // SQLite drops ORDER BY inside a subquery unless that subquery has LIMIT. + // The inner LIMIT -1 keeps digest order; the outer page is what exact SQL + // materializes. + let paged = format!("SELECT * FROM ({sql} LIMIT -1) LIMIT ?3 OFFSET ?4"); + let mut offset = 0_i64; let mut digest = Sha256::new(); let mut count = 0usize; - while let Some(row) = rows - .next() - .await - .map_err(|error| storage(PERSIST_OPERATION, error))? - { + loop { if let Some(control) = control { checkpoint_relation_rebuild_control(control)?; } - let value = row - .get::(0) + record_coverage_query_probe(); + let mut rows = conn + .query( + &paged, + params![session_id, generation, COVERAGE_DIGEST_PAGE_ROWS, offset], + ) + .await .map_err(|error| storage(PERSIST_OPERATION, error))?; - record_coverage_row( - u64::try_from(value.len()).map_err(|error| storage(PERSIST_OPERATION, error))?, - ); - update_ordered_row_digest(&mut digest, count, value.as_bytes()); - count = count - .checked_add(1) - .ok_or_else(|| storage_message(PERSIST_OPERATION, "coverage row count overflow"))?; + let mut page_rows = 0_i64; + while let Some(row) = rows + .next() + .await + .map_err(|error| storage(PERSIST_OPERATION, error))? + { + if let Some(control) = control { + checkpoint_relation_rebuild_control(control)?; + } + let value = row + .get::(0) + .map_err(|error| storage(PERSIST_OPERATION, error))?; + record_coverage_row( + u64::try_from(value.len()).map_err(|error| storage(PERSIST_OPERATION, error))?, + ); + update_ordered_row_digest(&mut digest, count, value.as_bytes()); + count = count + .checked_add(1) + .ok_or_else(|| storage_message(PERSIST_OPERATION, "coverage row count overflow"))?; + page_rows += 1; + } + if page_rows < COVERAGE_DIGEST_PAGE_ROWS { + break; + } + offset = offset.checked_add(page_rows).ok_or_else(|| { + storage_message(PERSIST_OPERATION, "coverage digest page offset overflow") + })?; } Ok(( count, @@ -1029,4 +1063,51 @@ mod digest_tests { digest_bytes(b"alpha\n\nomega") ); } + + #[tokio::test] + async fn coverage_digest_reads_every_ordered_row_across_pages() { + let dir = tempfile::TempDir::new().expect("digest dir"); + let conn = tracedecay_runtime_core::db::engine::TestConnection::open( + &dir.path().join("digest.db"), + ); + tracedecay_runtime_core::db::engine::Executor::execute_batch( + &conn, + "CREATE TABLE coverage_rows(session_id TEXT, generation INTEGER, encoded TEXT)", + ) + .await + .expect("create coverage rows"); + let total = COVERAGE_DIGEST_PAGE_ROWS + 8; + let mut expected = Sha256::new(); + for index in 0..total { + let encoded = format!("row-{index:04}"); + update_ordered_row_digest( + &mut expected, + usize::try_from(index).expect("index"), + encoded.as_bytes(), + ); + tracedecay_runtime_core::db::engine::Executor::execute( + &conn, + "INSERT INTO coverage_rows(session_id, generation, encoded) VALUES ('session', 7, ?1)", + params![encoded], + ) + .await + .expect("insert coverage row"); + } + + let (count, digest) = digest_paged_rows( + &conn, + "SELECT encoded FROM coverage_rows WHERE session_id = ?1 AND generation = ?2 ORDER BY encoded", + "session", + 7, + None, + ) + .await + .expect("paged digest"); + + assert_eq!(count, usize::try_from(total).expect("total")); + assert_eq!( + digest, + encode_tagged_lowercase_hex("sha256:", &expected.finalize()) + ); + } } From fc4b8486f652d4f913bf50df7de5d5898cee9391 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Sat, 19 Sep 2026 07:20:58 +0000 Subject: [PATCH 04/15] style(global-db): format the observation collision tests #1842 merged with a rustfmt diff in observation_collision_tests.rs, so the master push run failed its formatting gate. Co-Authored-By: Claude Fable 5.1 --- .../tracedecay-global-db/src/observation_collision_tests.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/crates/tracedecay-global-db/src/observation_collision_tests.rs b/crates/tracedecay-global-db/src/observation_collision_tests.rs index c02f99ca30..4fd33d97cc 100644 --- a/crates/tracedecay-global-db/src/observation_collision_tests.rs +++ b/crates/tracedecay-global-db/src/observation_collision_tests.rs @@ -56,9 +56,9 @@ use tracedecay_store::observation::ObservationIdentityCollisionDispositionV1; use tracedecay_store::{ AnchoredObservationWrite, CursorAdvanceLedgerReasonV1, CursorAdvanceLedgerReceiptIdV1, CursorAdvanceOutcome, ObservationCoverageReason, ObservationCursorAdvance, - ObservationPersistOutcome, - ObservationProjectionStore, ObservationStore, ObservationStoreError, ObservationWrite, - ProjectionPersistOutcome, ProjectionSkipReason, SESSION_MESSAGE_PROJECTOR_VERSION, + ObservationPersistOutcome, ObservationProjectionStore, ObservationStore, ObservationStoreError, + ObservationWrite, ProjectionPersistOutcome, ProjectionSkipReason, + SESSION_MESSAGE_PROJECTOR_VERSION, }; use tracing::field::{Field, Visit}; use tracing::span::{Attributes, Id, Record}; From f550e771c38c54f520a4e09513affa48c26f66c3 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Sat, 19 Sep 2026 05:43:07 +0000 Subject: [PATCH 05/15] fix(hermes): converge past refused rows instead of eternal skip A Hermes capture that the admission refuses deterministically (privacy boundary, identity collision, receipt collision) re-fails identically on every sweep. The snapshot admit loop had no durable skip for it, unlike the shared JSONL path, so the offending row pinned the source cursor and the whole `state.db` was abandoned once per sweep pass, forever: on a live daemon that was 66 WARN lines in six minutes across both Hermes profile stores with no recovery path at all. Cover past a deterministic refusal with the same typed coverage reason the JSONL admission already writes, so the source converges. Two further defects made it undiagnosable and unbearable: - host_admission_error reduced the outcome to its status family, so every refusal, cursor mismatch and contract violation surfaced as the single sentence "Hermes observation admission was degraded". Carry the reason code, retryability and storage cause the outcome already holds. - the sweep re-logged an identical WARN for a source whose state had not changed. Report a source failure when it is new or its reason changed. Co-Authored-By: Claude Fable 5.1 (cherry picked from commit 39a0e8538888e7c3669099da85543283ecda431d) --- .../tracedecay-sessions/src/admission/mod.rs | 18 +++++ .../src/runtime/hosts/hermes/coverage.rs | 67 +++++++++++++++++-- .../src/runtime/hosts/hermes/ingest.rs | 66 ++++++++++++++---- .../src/runtime/hosts/hermes/tests.rs | 56 ++++++++++++++++ .../jsonl_observation_admission.rs | 2 +- 5 files changed, 190 insertions(+), 19 deletions(-) diff --git a/crates/tracedecay-sessions/src/admission/mod.rs b/crates/tracedecay-sessions/src/admission/mod.rs index dc51b835a1..092d05457c 100644 --- a/crates/tracedecay-sessions/src/admission/mod.rs +++ b/crates/tracedecay-sessions/src/admission/mod.rs @@ -924,6 +924,7 @@ pub(crate) mod test_support { projection_failure: Arc>>, cancel_on_discovery_queue_read: Arc>>, session_backfill_page_pause: Arc>>, + deterministic_capture_refusal: Arc>>, } impl MemoryHostAdmission { @@ -940,6 +941,16 @@ pub(crate) mod test_support { self.store.state().capture_failures_remaining = 1; } + /// Refuse every capture the way a deterministic content refusal does: + /// the same record fails identically on every retry, so callers must + /// converge past it rather than re-attempt the source forever. + pub(crate) fn refuse_captures_deterministically(&self, reason: &'static str) { + *self + .deterministic_capture_refusal + .lock() + .unwrap_or_else(|error| error.into_inner()) = Some(reason); + } + /// Make the next `count` session-message lookups report the store as /// unavailable, the way reader-pool saturation does. pub(crate) fn fail_next_session_message_lookups(&self, count: usize) { @@ -1042,6 +1053,13 @@ pub(crate) mod test_support { request: CaptureObservationRequest, ) -> AdmissionFuture<'a, CaptureObservationOutcome> { Box::pin(async move { + if let Some(reason) = *self + .deterministic_capture_refusal + .lock() + .unwrap_or_else(|error| error.into_inner()) + { + return Err(HostAdmissionOutcome::deterministic_content_refusal(reason)); + } { let mut state = self.store.state(); state.scalar_capture_calls = state.scalar_capture_calls.saturating_add(1); diff --git a/crates/tracedecay-sessions/src/runtime/hosts/hermes/coverage.rs b/crates/tracedecay-sessions/src/runtime/hosts/hermes/coverage.rs index 117e0d70ca..8982f1a1e2 100644 --- a/crates/tracedecay-sessions/src/runtime/hosts/hermes/coverage.rs +++ b/crates/tracedecay-sessions/src/runtime/hosts/hermes/coverage.rs @@ -15,6 +15,7 @@ use tracedecay_store::observation::{ObservationCoverageReason, ObservationCursor use crate::admission::{HostAdmission, HostAdmissionOutcome}; use crate::observation::{CaptureObservationOutcome, ObservationCancellation}; +use crate::runtime::jsonl_observation_admission::is_deterministic_content_refusal; use crate::runtime::shared::TranscriptIngestStats; use tracedecay_runtime_core::db::{SqliteFileIdentityOperation, sqlite_generation_identity}; @@ -89,8 +90,29 @@ async fn advance_coverage( .map_err(host_admission_error) } +/// The admission's own verdict, verbatim. +/// +/// The status alone names a family ("degraded"), not a cause: every +/// deterministic refusal, cursor mismatch and contract violation collapsed +/// into one indistinguishable sentence, so a sweep that skipped the same +/// `state.db` every five seconds forever gave an operator nothing to act on. +/// Carry the reason code, retryability and storage cause the outcome already +/// holds. fn host_admission_error(outcome: HostAdmissionOutcome) -> String { - crate::runtime::snapshot_observation::host_admission_status_message("Hermes", outcome.status) + let mut message = crate::runtime::snapshot_observation::host_admission_status_message( + "Hermes", + outcome.status, + ); + if let Some(reason) = outcome.reason_code { + message.push_str(&format!( + " (reason_code={reason}, retryable={})", + outcome.retryable + )); + } + if let Some(cause) = outcome.storage_cause { + message.push_str(&format!(": {cause}")); + } + message } pub(super) async fn drain_hermes_projections_with_admission( @@ -221,11 +243,44 @@ pub(super) async fn admit_rows_with_admission_and_cancellation( .await?; } HermesAdmissionAction::Capture(request) => { - match facade - .capture_observation(*request) - .await - .map_err(host_admission_error)? - { + let captured = match facade.capture_observation(*request).await { + Ok(captured) => captured, + // A deterministic content refusal re-fails identically on + // every pass. Without a durable skip the source's cursor + // never clears the offending row, so the whole `state.db` + // is abandoned every sweep, forever, with one WARN each + // time. Cover past it with a typed reason exactly as the + // shared JSONL path does so the stream converges. + Err(outcome) if is_deterministic_content_refusal(&outcome) => { + tracing::warn!( + provider = PROVIDER, + row = row.id, + reason = outcome.reason_code.unwrap_or("host_admission_refused"), + "admission refused a Hermes row; covering past it" + ); + advance_coverage( + facade, + source, + range, + expected_cursor, + scope.clone(), + generation, + if outcome.reason_code == Some("observation_identity_collision") { + ObservationCoverageReason::ObservationIdentityCollision + } else { + ObservationCoverageReason::AdmissionRefused + }, + None, + file_identity, + resume_fingerprint, + cancellation, + ) + .await?; + continue; + } + Err(outcome) => return Err(host_admission_error(outcome)), + }; + match captured { CaptureObservationOutcome::Persisted { outcome, .. } | CaptureObservationOutcome::AcceptedForReplay { outcome, .. } => { if matches!(*outcome, ObservationPersistOutcome::Committed(_)) { diff --git a/crates/tracedecay-sessions/src/runtime/hosts/hermes/ingest.rs b/crates/tracedecay-sessions/src/runtime/hosts/hermes/ingest.rs index 88476f6bf4..3d2ca7642f 100644 --- a/crates/tracedecay-sessions/src/runtime/hosts/hermes/ingest.rs +++ b/crates/tracedecay-sessions/src/runtime/hosts/hermes/ingest.rs @@ -26,6 +26,38 @@ fn new_sweep_budget(max_new_bytes: Option) -> IngestByteBudget { IngestByteBudget::bounded(max_new_bytes.unwrap_or(DEFAULT_HERMES_SWEEP_BYTES)) } +/// Whether this sweep pass should report a source outcome, given what the last +/// pass reported for the same `state.db`. +/// +/// A source that cannot be admitted stays unadmittable until something about +/// the store or the file changes, and the sweep runs every few seconds. Logging +/// the identical line each pass buried every other daemon warning without +/// telling an operator anything the first line did not. Report a failure when +/// it is new or its reason changed; `None` records a recovered source so its +/// next failure is reported again. The map is keyed by discovered Hermes +/// profile, so it is bounded by the number of profiles on disk. +fn hermes_source_outcome_is_new(state_db: &Path, error: Option<&str>) -> bool { + use std::collections::BTreeMap; + use std::sync::{LazyLock, Mutex, PoisonError}; + + static REPORTED: LazyLock>> = + LazyLock::new(|| Mutex::new(BTreeMap::new())); + let mut reported = REPORTED.lock().unwrap_or_else(PoisonError::into_inner); + match error { + Some(error) => { + if reported.get(state_db).is_some_and(|last| last == error) { + return false; + } + reported.insert(state_db.to_path_buf(), error.to_owned()); + true + } + None => { + reported.remove(state_db); + false + } + } +} + /// Default Hermes profile homes under the resolved user home. /// /// Missing home is a typed absence (`None`), never an empty successful sweep. @@ -283,14 +315,19 @@ pub(super) async fn ingest_homes_capped_with_admission_and_cancellation( ) .await { - Ok(source_stats) => outcome.stats = outcome.stats.merge(source_stats), + Ok(source_stats) => { + hermes_source_outcome_is_new(&source.state_db, None); + outcome.stats = outcome.stats.merge(source_stats); + } Err(error) => { outcome.source_failures = outcome.source_failures.saturating_add(1); - tracing::warn!( - state_db = %source.state_db.display(), - error, - "skipping Hermes transcript source" - ); + if hermes_source_outcome_is_new(&source.state_db, Some(&error)) { + tracing::warn!( + state_db = %source.state_db.display(), + error, + "skipping Hermes transcript source" + ); + } } } } @@ -403,14 +440,19 @@ async fn ingest_user_homes_capped_with_admission( ) .await { - Ok(source_stats) => outcome.stats = outcome.stats.merge(source_stats), + Ok(source_stats) => { + hermes_source_outcome_is_new(&source.state_db, None); + outcome.stats = outcome.stats.merge(source_stats); + } Err(error) => { outcome.source_failures = outcome.source_failures.saturating_add(1); - tracing::warn!( - state_db = %source.state_db.display(), - error, - "skipping projectless Hermes transcript source" - ); + if hermes_source_outcome_is_new(&source.state_db, Some(&error)) { + tracing::warn!( + state_db = %source.state_db.display(), + error, + "skipping projectless Hermes transcript source" + ); + } } } } diff --git a/crates/tracedecay-sessions/src/runtime/hosts/hermes/tests.rs b/crates/tracedecay-sessions/src/runtime/hosts/hermes/tests.rs index 7d5f30d72c..d69d457c28 100644 --- a/crates/tracedecay-sessions/src/runtime/hosts/hermes/tests.rs +++ b/crates/tracedecay-sessions/src/runtime/hosts/hermes/tests.rs @@ -1688,3 +1688,59 @@ async fn unreadable_state_db_is_a_counted_source_failure_not_a_clean_sweep() { ); assert!(admission.observations().is_empty()); } + +/// A deterministic admission refusal is permanent: the same row fails the same +/// way on every sweep. Without a durable skip the source cursor never clears +/// it, so the whole profile `state.db` is abandoned every pass forever, which +/// is what produced an endless "skipping projectless Hermes transcript source" +/// WARN on a live daemon. Cover past it, exactly as the shared JSONL path +/// does, so the source converges. +mod deterministic_refusal_recovery { + use super::*; + + async fn admit_one_refused_row(reason: &'static str) -> MemoryHostAdmission { + let admission = MemoryHostAdmission::default(); + admission.refuse_captures_deterministically(reason); + let stats = admit_rows_with_admission_and_cancellation( + &admission, + &[fixture(1)], + ObservationScopeV1::Profile, + ObservationSourceGenerationV1::new(1).unwrap(), + 1, + 1, + |_| Some(fixture_projection()), + &ObservationCancellation::default(), + ) + .await + .expect("a permanently refused row must not abandon the whole source"); + assert_eq!(stats.messages_upserted, 0); + admission + } + + #[tokio::test] + async fn refused_row_is_covered_past_instead_of_skipping_the_source() { + let admission = admit_one_refused_row("privacy_boundary_failed").await; + + let advances = admission.non_durable_advances(); + assert_eq!( + advances.len(), + 1, + "the refused row must be covered exactly once" + ); + assert_eq!( + advances[0].reason(), + ObservationCoverageReason::AdmissionRefused + ); + assert_eq!(advances[0].next_cursor().position(), 1); + } + + #[tokio::test] + async fn identity_collision_keeps_its_own_coverage_reason() { + let admission = admit_one_refused_row("observation_identity_collision").await; + + assert_eq!( + admission.non_durable_advances()[0].reason(), + ObservationCoverageReason::ObservationIdentityCollision + ); + } +} diff --git a/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission.rs b/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission.rs index 468022fd84..8ed43b7405 100644 --- a/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission.rs +++ b/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission.rs @@ -2852,7 +2852,7 @@ pub(in crate::runtime) async fn admit_jsonl_observations( /// unbound authorities, retryable races, says nothing about the record and /// must surface as a typed block instead of writing coverage over a commit /// that never landed (or one that already landed and advanced the cursor). -fn is_deterministic_content_refusal(outcome: &HostAdmissionOutcome) -> bool { +pub(in crate::runtime) fn is_deterministic_content_refusal(outcome: &HostAdmissionOutcome) -> bool { matches!( outcome.recovery, Some(HostAdmissionRecovery::DeterministicContentRefusal) From 278d18faf0b912e3490cdf69507b578f9df4b0d4 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Sat, 19 Sep 2026 05:48:08 +0000 Subject: [PATCH 06/15] fix(ingest): treat a peer-covered cursor CAS loss as a no-op Live hook ingest and the scheduled catch-up sweep own the same (source, scope) observation cursor and routinely read the same transcript at once. The store's compare-and-swap keeps them honest, so one loses and gets `cursor_conflict`. The JSONL admission seam turned that into a typed block for the whole source pass, which the provider reported as a catch-up failure: on a live daemon that was 33 "Cursor transcript catch-up failed" WARN lines in six minutes for ranges the winner had already committed. The store already returns enough to decide: re-read the source cursor on a lost CAS. When the winner is on this generation and already past the frame (or, for an atomic batch, past the window's last frame), the range is durable, so adopt the winner's frontier and count the frames as skipped instead of failing. A cursor short of the frame, a different generation or an unreadable cursor all keep the existing typed block, so a frontier the winner never reached is never adopted. No store contract changes. Co-Authored-By: Claude Fable 5.1 (cherry picked from commit 9a7810c6dcfa7d39f70ceb009ec5d8c245b52a77) --- .../jsonl_observation_admission.rs | 79 ++++++++++++++++- .../jsonl_observation_admission/tests.rs | 86 +++++++++++++++++++ 2 files changed, 162 insertions(+), 3 deletions(-) diff --git a/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission.rs b/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission.rs index 8ed43b7405..2c18933c7c 100644 --- a/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission.rs +++ b/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission.rs @@ -1849,10 +1849,19 @@ impl ActiveAdmission<'_> { .with_resume_checkpoint(self.file_identity, checkpoint.resume_fingerprint); hotpath::gauge!("jsonl_admission_coverage_frames").inc(1.0); hotpath::gauge!("jsonl_admission_writer_submits").inc(1.0); - self.admission + if let Err(outcome) = self + .admission .advance_non_durable_source_cursor(advance, self.cancellation.clone()) .await - .map_err(|outcome| { + { + if is_lost_cursor_cas(&outcome) + && self + .peer_already_covered(expected_cursor, checkpoint.end_offset) + .await + { + return Ok(()); + } + return Err({ if is_admission_cancellation(&outcome, &self.cancellation) { TranscriptIngestError::Cancelled { provider: self.provider, @@ -1874,12 +1883,47 @@ impl ActiveAdmission<'_> { .unwrap_or("non_durable_cursor_advance_failed"), } } - })?; + }); + } *expected_cursor = Some(self.cursor_at(checkpoint.end_offset, checkpoint.resume_fingerprint)?); Ok(()) } + /// Whether the peer that won a cursor CAS already covered this range. + /// + /// Live hook ingest and the catch-up sweep own the same `(source, scope)` + /// cursor and routinely read the same transcript at once; the store's + /// compare-and-swap is what keeps them honest, so one of them loses. The + /// loser's frames are almost always already durable behind the winner's + /// cursor, and re-reading that cursor is enough to prove it. Adopt the + /// winner's cursor and let the pass continue instead of failing the whole + /// source over work that is already committed. + /// + /// A read failure, a different generation, or a cursor short of this frame + /// all answer "not covered", which keeps the caller's typed block. + #[hotpath::skip] + async fn peer_already_covered( + &self, + expected_cursor: &mut Option, + end_offset: u64, + ) -> bool { + let Ok(actual) = self + .admission + .get_source_cursor(&self.source, &self.scope) + .await + else { + return false; + }; + let covered = actual.as_ref().is_some_and(|cursor| { + cursor.generation() == self.generation && cursor.position() >= end_offset + }); + if covered { + *expected_cursor = actual; + } + covered + } + fn capture_request( &self, expected_cursor: Option, @@ -1987,6 +2031,13 @@ impl ActiveAdmission<'_> { if outcome.status == HostAdmissionStatus::Backpressured { hotpath::gauge!("jsonl_admission_backpressure_writer").inc(1.0); } + if is_lost_cursor_cas(&outcome) + && self + .peer_already_covered(expected_cursor, checkpoint.end_offset) + .await + { + return Ok(DurableFrameDisposition::AlreadyDurable); + } if is_admission_cancellation(&outcome, &self.cancellation) { Err(TranscriptIngestError::Cancelled { provider: self.provider, @@ -2172,6 +2223,21 @@ impl ActiveAdmission<'_> { } } } + // The batch is atomic: nothing in this window committed. When + // the peer that won the CAS is already past the window's last + // frame, every frame in it is durable behind the winner's + // cursor, so this is a no-op rather than a failed source pass. + if is_lost_cursor_cas(&outcome) + && let Some(last) = checkpoints.last() + && self + .peer_already_covered(expected_cursor, last.end_offset) + .await + { + progress.frames_skipped = progress + .frames_skipped + .saturating_add(checkpoints.len() as u64); + return Ok(()); + } if is_admission_cancellation(&outcome, &self.cancellation) { Err(CaptureWindowError::Ingest( TranscriptIngestError::Cancelled { @@ -2852,6 +2918,13 @@ pub(in crate::runtime) async fn admit_jsonl_observations( /// unbound authorities, retryable races, says nothing about the record and /// must surface as a typed block instead of writing coverage over a commit /// that never landed (or one that already landed and advanced the cursor). +/// A cursor compare-and-swap lost to a peer that owns the same +/// `(source, scope)` cursor. Retryable by construction; whether it is a +/// failure at all depends on what the winner already covered. +fn is_lost_cursor_cas(outcome: &HostAdmissionOutcome) -> bool { + outcome.reason_code == Some("cursor_conflict") +} + pub(in crate::runtime) fn is_deterministic_content_refusal(outcome: &HostAdmissionOutcome) -> bool { matches!( outcome.recovery, diff --git a/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission/tests.rs b/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission/tests.rs index cb9b053e7d..c1ed937220 100644 --- a/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission/tests.rs +++ b/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission/tests.rs @@ -59,6 +59,10 @@ struct SeamSpyAdmission { capture_calls: AtomicU64, capture_collision_dispositions: Mutex>, cover_past_advances: Mutex>, + /// Commit the next capture through the shared store and then report the + /// cursor CAS as lost, the way a live hook ingest wins the race a sweep + /// was still trying to write. + peer_wins_next_cursor_cas: AtomicBool, } #[tokio::test] @@ -651,6 +655,14 @@ impl SeamSpyAdmission { *self.scripted_capture_error.lock().unwrap() = Some(outcome); } + fn script_peer_wins_next_cursor_cas(&self) { + self.peer_wins_next_cursor_cas.store(true, Ordering::SeqCst); + } + + fn peer_won_cursor_cas(&self) -> bool { + self.peer_wins_next_cursor_cas.swap(false, Ordering::SeqCst) + } + fn script_batch_error(&self, outcome: HostAdmissionOutcome) { *self.scripted_batch_error.lock().unwrap() = Some(outcome); } @@ -683,6 +695,12 @@ impl HostAdmission for SeamSpyAdmission { .lock() .unwrap() .push(request.identity_collision_disposition()); + if self.peer_won_cursor_cas() { + let _ = self.inner.capture_observation(request).await; + return Err(HostAdmissionOutcome::retained_backpressured( + "cursor_conflict", + )); + } if let Some(outcome) = self.scripted_capture_error_once.lock().unwrap().take() { return Err(outcome); } @@ -703,6 +721,12 @@ impl HostAdmission for SeamSpyAdmission { .iter() .map(CaptureObservationRequest::identity_collision_disposition), ); + if self.peer_won_cursor_cas() { + let _ = self.inner.capture_observations(requests).await; + return Err(HostAdmissionOutcome::retained_backpressured( + "cursor_conflict", + )); + } if let Some(outcome) = self.scripted_batch_error.lock().unwrap().take() { return Err(outcome); } @@ -898,6 +922,68 @@ async fn retryable_admission_failures_keep_their_own_verdict() { assert!(stored_cursor(&spy).await.is_none()); } +/// Live hook ingest and the catch-up sweep own the same `(source, scope)` +/// cursor, so one of them loses the store's compare-and-swap. When the winner +/// already covered the range the loser was writing, the loser's work is +/// durable and the pass is a no-op, not a failed source: reporting it as a +/// failure produced a "Cursor transcript catch-up failed" WARN roughly every +/// ten seconds on a live daemon for work that was already committed. +#[tokio::test] +async fn cursor_cas_lost_to_a_peer_that_covered_the_range_is_a_no_op() { + let (_temp, path, len) = rollout_fixture(); + let spy = SeamSpyAdmission::default(); + spy.script_peer_wins_next_cursor_cas(); + + let stats = + try_admit_codex_jsonl_observations_for_profile_with_admission(&path, None, &[], &spy, None) + .await + .expect("a CAS the peer already covered must not fail the source pass"); + + assert_eq!( + stored_cursor(&spy).await.map(|cursor| cursor.position()), + Some(len), + "the pass must adopt the winner's frontier" + ); + assert!( + !spy.inner.observations().is_empty(), + "the peer's commit is the durable record this pass stopped duplicating" + ); + assert_eq!( + stats.frames_accepted, 0, + "the loser accepts nothing of its own" + ); + assert!( + stats.frames_skipped > 0, + "the covered frames are counted as skipped, not lost" + ); +} + +/// The same lost CAS with nothing behind it stays a typed retryable block: +/// adopting a frontier the winner never reached would skip real records. +#[tokio::test] +async fn cursor_cas_lost_without_peer_coverage_stays_a_typed_block() { + let (_temp, path, _len) = rollout_fixture(); + let spy = SeamSpyAdmission::default(); + spy.script_capture_error(HostAdmissionOutcome::retained_backpressured( + "cursor_conflict", + )); + + let error = + try_admit_codex_jsonl_observations_for_profile_with_admission(&path, None, &[], &spy, None) + .await + .expect_err("an uncovered race must surface for another pass"); + + assert!(matches!( + error, + TranscriptIngestError::HostAdmission { + reason: "cursor_conflict", + retryable: true, + .. + } + )); + assert!(stored_cursor(&spy).await.is_none()); +} + #[tokio::test] async fn eligible_identity_collision_retries_once_with_normalizer_fallback() { super::install_test_shared_jsonl_preparation_authority(); From 19ba2b60c34404e47eaef8a951034d3a60916e8c Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Sat, 19 Sep 2026 06:38:10 +0000 Subject: [PATCH 07/15] test(codex): count replay index visits per thread `indexed_replay_starts_at_the_acknowledged_btree_position` resets a counter, runs one indexed replay pass and asserts it visited at most 9 B-tree entries. The counter was a process-global `AtomicU64`, so every other test replaying an index on one of the harness's other threads added to the number this test read: it measured the suite's traversal, not its own pass, and failed intermittently with no bound that held. `indexed_replay_pass` runs entirely on its caller's thread, so a thread-local `Cell` is exactly the scope the assertion means. The bound stays at 9; only whose traversal it counts changes. Co-Authored-By: Claude Fable 5.1 (cherry picked from commit 34d40d792041ac2d0a5311f2c033fdec3de8d1fc) --- .../src/runtime/hosts/codex.rs | 17 ++++++++++++----- 1 file changed, 12 insertions(+), 5 deletions(-) diff --git a/crates/tracedecay-sessions/src/runtime/hosts/codex.rs b/crates/tracedecay-sessions/src/runtime/hosts/codex.rs index 67439f9621..b663217469 100644 --- a/crates/tracedecay-sessions/src/runtime/hosts/codex.rs +++ b/crates/tracedecay-sessions/src/runtime/hosts/codex.rs @@ -269,17 +269,23 @@ impl Default for CodexReplayIndex { } #[cfg(test)] -static CODEX_REPLAY_INDEX_ENTRIES_VISITED: std::sync::atomic::AtomicU64 = - std::sync::atomic::AtomicU64::new(0); +thread_local! { + /// Per-thread, because `indexed_replay_pass` runs entirely on its caller's + /// thread and the one test that measures B-tree traversal shares the + /// process with every other test replaying an index in parallel. A global + /// counter measures the whole suite's traversal, not this pass's. + static CODEX_REPLAY_INDEX_ENTRIES_VISITED: std::cell::Cell = + const { std::cell::Cell::new(0) }; +} #[cfg(test)] fn reset_replay_index_entries_visited_for_test() { - CODEX_REPLAY_INDEX_ENTRIES_VISITED.store(0, std::sync::atomic::Ordering::Release); + CODEX_REPLAY_INDEX_ENTRIES_VISITED.with(|visited| visited.set(0)); } #[cfg(test)] fn replay_index_entries_visited_for_test() -> u64 { - CODEX_REPLAY_INDEX_ENTRIES_VISITED.load(std::sync::atomic::Ordering::Acquire) + CODEX_REPLAY_INDEX_ENTRIES_VISITED.with(std::cell::Cell::get) } fn indexed_replay_pass( @@ -295,7 +301,8 @@ fn indexed_replay_pass( let lower = position.map_or(Bound::Unbounded, Bound::Excluded); for indexed in index.paths.range((lower, Bound::Unbounded)) { #[cfg(test)] - CODEX_REPLAY_INDEX_ENTRIES_VISITED.fetch_add(1, std::sync::atomic::Ordering::AcqRel); + CODEX_REPLAY_INDEX_ENTRIES_VISITED + .with(|visited| visited.set(visited.get().saturating_add(1))); let path_bytes = u64::try_from(crate::runtime::source::path_byte_len(&indexed.path)).unwrap_or(u64::MAX); if paths.len() >= bounds.max_files.max(1) From 9f8092d123e0bd4ff76821611e5cac678ed23763 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Sat, 19 Sep 2026 06:38:20 +0000 Subject: [PATCH 08/15] test(ingest): keep the shared meta cache off its degraded mode `codex_session_meta_prefix_is_decoded_once_across_consumers` asserts two profile consumers of one rollout share a single prefix decode, which holds only while the shared metadata cache still retains the entry between them. It failed intermittently for two compounding reasons. The test never installed the preparation authority its siblings install, and without one `shared_jsonl_preparation_capacity` returns the degraded fallback of a single entry, so the next publish from any parallel test evicted this path before the second consumer looked it up. Installing it was not enough. The test authority metered the whole 96-thread harness against one 32 GiB budget, and each in-flight page holds a 544 MiB reservation, so bursts drove the derived capacity down to two entries and a couple of peer publishes still evicted the entry. That ceiling is an artifact of the harness, not the product: a production process meters one ingest workload against the machine. Size the test budget past what the harness itself can reserve so capacity stays CPU-bound. No assertion or production rule changes; the eviction and capacity logic is untouched. Co-Authored-By: Claude Fable 5.1 (cherry picked from commit 0e1cb031afacdb27f0e144e4339e9638a01b7b37) --- .../runtime/observation/jsonl_observation_admission.rs | 10 +++++++++- .../observation/jsonl_observation_admission/tests.rs | 5 +++++ 2 files changed, 14 insertions(+), 1 deletion(-) diff --git a/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission.rs b/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission.rs index 2c18933c7c..2f40a44a2f 100644 --- a/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission.rs +++ b/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission.rs @@ -483,10 +483,18 @@ pub(in crate::runtime) fn install_test_shared_jsonl_preparation_authority() { use std::num::NonZeroUsize; use tracedecay_runtime_core::resident_memory::ProcessResidentMemoryV1; + // One process-wide budget serves the whole suite, so every test thread + // holding a `SHARED_JSONL_WORKER_RESERVATION_BYTES` page charges it at + // once. At 32 GiB a wide harness drove the derived preparation capacity + // down to two entries, which is the shared metadata cache's degraded mode, + // not the product's: a production process meters one ingest workload + // against the machine. Size the budget past what the harness's own + // parallelism can reserve so capacity stays CPU-bound, the way the + // composition root installs it. static MEMORY: OnceLock> = OnceLock::new(); let memory = Arc::clone(MEMORY.get_or_init(|| { Arc::new(ProcessResidentMemoryV1::new( - NonZeroU64::new(32 * 1024 * 1024 * 1024).unwrap(), + NonZeroU64::new(1024 * 1024 * 1024 * 1024).unwrap(), )) })); let background_cpu = Arc::new(ProcessBackgroundCpuV1::new(NonZeroUsize::new(48).unwrap())); diff --git a/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission/tests.rs b/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission/tests.rs index c1ed937220..3915ac089b 100644 --- a/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission/tests.rs +++ b/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission/tests.rs @@ -1261,6 +1261,11 @@ async fn content_refusals_cover_past_so_the_stream_converges() { #[tokio::test] async fn codex_session_meta_prefix_is_decoded_once_across_consumers() { + // The shared metadata cache retains entries up to + // `shared_jsonl_preparation_capacity()`, so this test only observes the + // shared decode once the preparation authority is installed: without it the + // capacity is the degraded fallback of one entry. + super::install_test_shared_jsonl_preparation_authority(); let (_temp, path, _) = rollout_fixture(); let first = SeamSpyAdmission::default(); let second = SeamSpyAdmission::default(); From 822b18198351b24c56a39c80da61337ef91badb0 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Sat, 19 Sep 2026 07:21:29 +0000 Subject: [PATCH 09/15] fix(session-temporal): drop the no-op drop in the doctor test #1842 merged with a `drop` of a writer handle that does not implement Drop, which clippy refuses under CI's `-D warnings` lens, so the master push run failed its Clippy job. Co-Authored-By: Claude Fable 5.1 --- crates/tracedecay-session-temporal-store/src/doctor_health.rs | 1 - 1 file changed, 1 deletion(-) diff --git a/crates/tracedecay-session-temporal-store/src/doctor_health.rs b/crates/tracedecay-session-temporal-store/src/doctor_health.rs index 8a76c5a39a..9bdf989170 100644 --- a/crates/tracedecay-session-temporal-store/src/doctor_health.rs +++ b/crates/tracedecay-session-temporal-store/src/doctor_health.rs @@ -1362,7 +1362,6 @@ mod registered_tests { ) .await .expect("drop required index"); - drop(writer); let database = SessionTemporalRegisteredDb::db_path(&harness.registered); std::fs::OpenOptions::new() From 259256db07ad4d217862613263c2f3478cf121d2 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Sat, 19 Sep 2026 07:31:48 +0000 Subject: [PATCH 10/15] fix(session-temporal): count parent copies without loading the graph Refresh baseline asked for 100k relations in one read. That exceeds the exact-SQL page, so materialize failed as storage busy and the running refresh never recorded progress. Page the copy count instead. --- .../src/projection.rs | 9 +-- .../src/projection/tests.rs | 11 ++++ .../src/relations/projection_read.rs | 57 +++++++++++++++++++ 3 files changed, 70 insertions(+), 7 deletions(-) diff --git a/crates/tracedecay-session-temporal-store/src/projection.rs b/crates/tracedecay-session-temporal-store/src/projection.rs index c5ebf7ecc4..3cc6bf03fa 100644 --- a/crates/tracedecay-session-temporal-store/src/projection.rs +++ b/crates/tracedecay-session-temporal-store/src/projection.rs @@ -36,8 +36,6 @@ pub(super) use receipts::validate_final_projection_receipt; const DISCOVER_REFRESH: &str = "discover session temporal refresh"; const MATERIALIZE_REFRESH: &str = "materialize session temporal refresh"; -const MAX_BASELINE_RELATION_ITEMS: usize = 100_000; - pub struct SessionTemporalRefreshDiscoveryPage { requests: Vec, active_scanned_through: Option, @@ -279,16 +277,13 @@ impl SessionTemporalAccess<'_, D> { let (scope, relation_store) = self .session_relation_store() .map_err(|error| storage(MATERIALIZE_REFRESH, error))?; - match relation_store.load_projection( + match relation_store.logical_copy_count( &scope, recovery.session_id(), recovery.frozen_watermarks().active_generation().value(), - MAX_BASELINE_RELATION_ITEMS, - MAX_BASELINE_RELATION_ITEMS, Arc::new(NeverCancelled), ) { - Ok(projection) => u64::try_from(projection.logical_copies.len()) - .map_err(|error| storage(MATERIALIZE_REFRESH, error))?, + Ok(copies) => copies, Err(SessionRelationError::NotFound) => { let mut rows = snapshot .query( diff --git a/crates/tracedecay-session-temporal-store/src/projection/tests.rs b/crates/tracedecay-session-temporal-store/src/projection/tests.rs index 8ea0fa39c2..595a63953d 100644 --- a/crates/tracedecay-session-temporal-store/src/projection/tests.rs +++ b/crates/tracedecay-session-temporal-store/src/projection/tests.rs @@ -1582,6 +1582,17 @@ async fn explicit_copy_survives_reconstruction_in_the_native_relation_graph() { Arc::new(NeverCancelled), ) .unwrap(); + assert_eq!( + relation_store + .logical_copy_count( + &scope, + &session_id, + batch.generation().value(), + Arc::new(NeverCancelled), + ) + .expect("paged logical copy count"), + loaded.logical_copies.len() as u64 + ); assert_eq!( loaded.logical_copies, vec![crate::relations::LogicalCopyRelation { diff --git a/crates/tracedecay-session-temporal-store/src/relations/projection_read.rs b/crates/tracedecay-session-temporal-store/src/relations/projection_read.rs index 0cd64ebb53..d9f435fa6a 100644 --- a/crates/tracedecay-session-temporal-store/src/relations/projection_read.rs +++ b/crates/tracedecay-session-temporal-store/src/relations/projection_read.rs @@ -70,6 +70,63 @@ impl SessionRelationGraphStore { } decode_projection(scope, session_id, generation, page.entities, page.relations) } + + /// Counts logical-copy relations without loading the generation. + /// + /// A single `load_projection` asks for up to 100_000 relations. That read + /// is one exact-SQL page, whose ceiling is 10_000 rows, so a large parent + /// generation fails and refresh retries it as storage busy. The successor + /// baseline only needs the count. + pub fn logical_copy_count( + &self, + scope: &super::SessionRelationScope, + session_id: &SessionId, + generation: u64, + cancellation: Arc, + ) -> Result { + // Stay under the exact-SQL materialization ceiling (10_000 rows). + const RELATION_PAGE: usize = 4_096; + let namespace = namespace(scope)?; + let projection_id = projection(session_id, generation)?; + if self + .database + .projection_telemetry(GraphProjectionTelemetryRequest { + namespace: namespace.clone(), + projection: projection_id.clone(), + cancellation: Arc::clone(&cancellation), + }) + .map_err(map_graph_error)? + .is_none() + { + return Err(SessionRelationError::NotFound); + } + let mut after_relation = None; + let mut copies = 0_u64; + loop { + let page = self + .database + .read_projection(GraphProjectionReadRequest { + namespace: namespace.clone(), + projection: projection_id.clone(), + after_entity: None, + after_relation, + max_entities: 1, + max_relations: RELATION_PAGE, + cancellation: Arc::clone(&cancellation), + }) + .map_err(map_graph_error)?; + for relation in &page.relations { + if relation.kind.as_str() == LOGICAL_COPY_KIND { + copies = copies.saturating_add(1); + } + } + match page.next_relation { + Some(next) => after_relation = Some(next), + None => break, + } + } + Ok(copies) + } } fn decode_projection( From 8f33ad6206ce850d51c7a1a1fd71f52d9613fdbe Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Sat, 19 Sep 2026 07:51:08 +0000 Subject: [PATCH 11/15] test(sessions): re-reasoned cursor replay is a duplicate #1842 made an idempotency conflict on a cursor already at its `next_cursor` an `ExactDuplicate` (the coverage is applied; only a conflict that left the cursor elsewhere is a collision). Two tests still expected `CursorAdvanceCollision` for the same range under a different coverage reason and failed both tries on the master push run 35428222386. Assert the new contract and keep the committed cursor. Co-Authored-By: Claude Fable 5.1 --- .../src/daemon/store_runtime_tests.rs | 21 +++++++++++-------- .../session_suite/observation_store/mod.rs | 15 ++++++++----- 2 files changed, 22 insertions(+), 14 deletions(-) diff --git a/crates/tracedecay/src/daemon/store_runtime_tests.rs b/crates/tracedecay/src/daemon/store_runtime_tests.rs index 3b58371f8a..8799d13b61 100644 --- a/crates/tracedecay/src/daemon/store_runtime_tests.rs +++ b/crates/tracedecay/src/daemon/store_runtime_tests.rs @@ -30,9 +30,9 @@ use tracedecay_session_memory::memory::{ }; use tracedecay_store::{ CursorAdvanceOutcome, FactReadControl, FactWriteControl, ObservationCoverageReason, - ObservationCursorAdvance, ObservationStore, ObservationStoreError, ProjectId, - ProjectMemoryFactHistoryQueryV1, ProjectMemoryFactIdV1, ProjectMemoryFactProjectionV1, - RetainedGraphStoreLeaseV1, StoreShardIdV1, + ObservationCursorAdvance, ObservationStore, ProjectId, ProjectMemoryFactHistoryQueryV1, + ProjectMemoryFactIdV1, ProjectMemoryFactProjectionV1, RetainedGraphStoreLeaseV1, + StoreShardIdV1, }; use tracedecay_store_runtime::{ DaemonSessionRuntimeRegistryV1, RegisteredSchemaConvergenceStatus, process_runtime_generation, @@ -1193,19 +1193,22 @@ async fn retained_runtime_ledger_replays_during_bounded_background_convergence() .expect("replay retained cursor while convergence is pending"), CursorAdvanceOutcome::ExactDuplicate ); - let conflicting_advance = runtime_cursor_advance( + // The same range under a different coverage reason finds the retained + // cursor already at its `next_cursor`, so it is a duplicate of the + // applied coverage rather than a collision (#1842). + let rereasoned_advance = runtime_cursor_advance( &project_id, "retired", ObservationCoverageReason::BlankFrame, ); - assert!(matches!( + assert_eq!( database .observation_store() - .advance_source_cursor(conflicting_advance) + .advance_source_cursor(rereasoned_advance) .await - .expect_err("classify retained cursor collision while convergence is pending"), - ObservationStoreError::CursorAdvanceCollision - )); + .expect("classify a re-reasoned retained cursor while convergence is pending"), + CursorAdvanceOutcome::ExactDuplicate + ); let fresh_advance = runtime_cursor_advance(&project_id, "fresh", ObservationCoverageReason::OutOfScope); diff --git a/crates/tracedecay/tests/session_suite/observation_store/mod.rs b/crates/tracedecay/tests/session_suite/observation_store/mod.rs index c4d27abd67..eeeb421313 100644 --- a/crates/tracedecay/tests/session_suite/observation_store/mod.rs +++ b/crates/tracedecay/tests/session_suite/observation_store/mod.rs @@ -1476,7 +1476,11 @@ async fn cursor_only_progress_persists_non_payload_receipt_and_retries_idempoten } #[tokio::test] -async fn cursor_only_retry_rejects_same_cursor_with_different_reason() { +/// A second owner replaying the same range with a different coverage reason +/// finds the durable cursor already at its `next_cursor`: the coverage it +/// wanted to record is applied, so the replay is a duplicate, not a +/// collision that blocks ingest (#1842). The committed reason stays. +async fn cursor_only_retry_with_same_cursor_and_different_reason_is_a_duplicate() { let tmp = TempDir::new().unwrap(); let runtime = profile_runtime(&tmp).await; let store = runtime @@ -1493,7 +1497,7 @@ async fn cursor_only_retry_rejects_same_cursor_with_different_reason() { .await .unwrap(); - assert!(matches!( + assert_eq!( store .advance_source_cursor(cursor_advance( None, @@ -1501,9 +1505,10 @@ async fn cursor_only_retry_rejects_same_cursor_with_different_reason() { 10, NonDurableFrameReason::OutOfScope, )) - .await, - Err(ObservationStoreError::CursorAdvanceCollision) - )); + .await + .unwrap(), + CursorAdvanceOutcome::ExactDuplicate + ); assert_eq!( store.get_source_cursor(&source(), &scope()).await.unwrap(), Some(cursor(10)) From c39a2eddf6df8b160e152b1c6799209a995aceef Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Sat, 19 Sep 2026 07:56:16 +0000 Subject: [PATCH 12/15] fix(session-temporal): split observation prefetches that exceed a page A refresh batch loaded 500 observation bodies in one query. Past the exact-SQL ceiling that failed as storage, and the projector reported only source_busy, so the same pass retried forever. Split the page and keep the storage error on the retry. --- .../projector.rs | 4 +- .../src/query.rs | 75 +++++++++++++++++-- 2 files changed, 72 insertions(+), 7 deletions(-) diff --git a/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/projector.rs b/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/projector.rs index e988993e7c..10818733c1 100644 --- a/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/projector.rs +++ b/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/projector.rs @@ -120,7 +120,9 @@ impl SessionTemporalRefreshProjector for CanonicalSessionTemporalProjector { // empty complete progress batch instead of deferring forever. Ok(None) => canonical_noop_complete_effect(&recovery), Err(error) if error.is_storage() => Err( - SessionTemporalRefreshProjectorError::retryable("source_busy"), + SessionTemporalRefreshProjectorError::retryable(format!( + "source_busy: {error}" + )), ), Err(_) => Err(SessionTemporalRefreshProjectorError::terminal( "projector_failed", diff --git a/crates/tracedecay-session-temporal-store/src/query.rs b/crates/tracedecay-session-temporal-store/src/query.rs index 8dd5a56501..367b14bb79 100644 --- a/crates/tracedecay-session-temporal-store/src/query.rs +++ b/crates/tracedecay-session-temporal-store/src/query.rs @@ -185,9 +185,10 @@ pub(super) async fn read_observation( Ok((sequence, observation)) } -/// The largest `observation_id IN (...)` batch one observation prefetch binds, -/// kept clear of `SQLite`'s default variable ceiling. -const OBSERVATION_READ_BATCH: usize = 500; +/// First prefetch width. A batch of full `observation_json` bodies is bounded +/// by the exact-SQL materialization ceiling (64 MiB), not by SQLite's variable +/// limit, so this starts small and splits when a page still does not fit. +const OBSERVATION_READ_BATCH: usize = 32; /// Prefetches the observations a projection pass is about to decode. /// @@ -211,7 +212,20 @@ pub(super) async fn read_observations( if unique.is_empty() { return Ok(observations); } - for chunk in unique.chunks(OBSERVATION_READ_BATCH) { + let mut pending = Vec::new(); + let mut offset = 0usize; + while offset < unique.len() { + let end = offset + .saturating_add(OBSERVATION_READ_BATCH) + .min(unique.len()); + pending.push((offset, end)); + offset = end; + } + while let Some((start, end)) = pending.pop() { + if start >= end { + continue; + } + let chunk = &unique[start..end]; let placeholders = (1..=chunk.len()) .map(|index| format!("?{index}")) .collect::>() @@ -221,10 +235,27 @@ pub(super) async fn read_observations( FROM observations WHERE observation_id IN ({placeholders})" ); - let mut rows = conn + let mut rows = match conn .query(&sql, params_from_iter(chunk.iter().copied())) .await - .map_err(|error| storage(PERSIST_OPERATION, error))?; + { + Ok(rows) => rows, + Err(error) => { + let error = storage(PERSIST_OPERATION, error); + // One materialized page of bodies exceeded the exact-SQL + // ceiling. Split until a single observation remains; that + // observation is then a typed storage failure, not a retry + // that looks like a busy source. + if observation_prefetch_exceeded_materialization_limit(&error) && chunk.len() > 1 + { + let mid = start + chunk.len() / 2; + pending.push((mid, end)); + pending.push((start, mid)); + continue; + } + return Err(error); + } + }; while let Some(row) = rows .next() .await @@ -249,6 +280,38 @@ pub(super) async fn read_observations( Ok(observations) } +fn observation_prefetch_exceeded_materialization_limit(error: &SessionStoreError) -> bool { + match error { + SessionStoreError::Storage { source, .. } => source + .to_string() + .contains("materialization exceeded its limit"), + _ => false, + } +} + +#[cfg(test)] +mod tests { + use super::{ + PERSIST_OPERATION, observation_prefetch_exceeded_materialization_limit, storage, + }; + + #[test] + fn materialization_limit_is_the_prefetch_split_signal() { + let exceeded = storage( + PERSIST_OPERATION, + std::io::Error::other("exact SQL query materialization exceeded its limit"), + ); + let locked = storage( + PERSIST_OPERATION, + std::io::Error::other("database is locked"), + ); + assert!(observation_prefetch_exceeded_materialization_limit( + &exceeded + )); + assert!(!observation_prefetch_exceeded_materialization_limit(&locked)); + } +} + /// The error `read_observation` raises for an id the store does not hold, reused /// by callers that resolve prefetched observations out of a batch map. pub(super) fn missing_observation(observation_id: &CanonicalObservationIdV1) -> SessionStoreError { From 8babdf32a24744695bd759851212db934a37209b Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Sat, 19 Sep 2026 08:20:17 +0000 Subject: [PATCH 13/15] fix(session-temporal): reconstruct copies without a relation graph Refresh refused a parent generation that had rows but no applied native graph, then retried that absence as source_busy. Count canonical logical copies from the sealed rows instead. --- .../src/projection.rs | 46 ++---- .../src/relation_projection.rs | 133 ++++++++++++++++++ 2 files changed, 143 insertions(+), 36 deletions(-) diff --git a/crates/tracedecay-session-temporal-store/src/projection.rs b/crates/tracedecay-session-temporal-store/src/projection.rs index 3cc6bf03fa..037e5cbed7 100644 --- a/crates/tracedecay-session-temporal-store/src/projection.rs +++ b/crates/tracedecay-session-temporal-store/src/projection.rs @@ -9,7 +9,7 @@ use tracedecay_store::{ }; use tracedecay_temporal_query::ports::ExecutionControl; -use super::query::{PERSIST_OPERATION, storage, storage_message}; +use super::query::{PERSIST_OPERATION, storage}; use super::refresh::{SessionRefreshRecoveryV1, SessionRefreshRestartStateV1}; use super::relations::SessionRelationError; use crate::handle::{SessionTemporalAccess, SessionTemporalRegisteredDb, SessionTemporalWriteTxn}; @@ -285,41 +285,15 @@ impl SessionTemporalAccess<'_, D> { ) { Ok(copies) => copies, Err(SessionRelationError::NotFound) => { - let mut rows = snapshot - .query( - "SELECT COUNT(*) - FROM session_occurrences - WHERE session_id = ?1 AND generation = ?2", - params![ - recovery.session_id().as_str(), - i64::try_from( - recovery.frozen_watermarks().active_generation().value() - ) - .map_err(|error| storage(MATERIALIZE_REFRESH, error))?, - ], - ) - .await - .map_err(|error| storage(MATERIALIZE_REFRESH, error))?; - let retained: i64 = rows - .next() - .await - .map_err(|error| storage(MATERIALIZE_REFRESH, error))? - .ok_or_else(|| { - storage_message( - MATERIALIZE_REFRESH, - "active projection count returned no row", - ) - })? - .get(0) - .map_err(|error| storage(MATERIALIZE_REFRESH, error))?; - if retained == 0 { - 0 - } else { - return Err(storage_message( - MATERIALIZE_REFRESH, - "active native relation projection is unavailable", - )); - } + // No native graph was applied for this generation. Reconstruct + // the copy count from the sealed rows instead of retrying the + // absence as a busy source. + crate::relation_projection::count_canonical_logical_copies( + &snapshot, + recovery.session_id(), + recovery.frozen_watermarks().active_generation(), + ) + .await? } Err(error) => return Err(storage(MATERIALIZE_REFRESH, error)), } diff --git a/crates/tracedecay-session-temporal-store/src/relation_projection.rs b/crates/tracedecay-session-temporal-store/src/relation_projection.rs index d881616136..520304afb8 100644 --- a/crates/tracedecay-session-temporal-store/src/relation_projection.rs +++ b/crates/tracedecay-session-temporal-store/src/relation_projection.rs @@ -533,6 +533,139 @@ pub(crate) async fn reconstruct_session_relation_projection( Ok(projection) } +/// Counts logical copies for a generation whose native graph was never applied. +/// +/// The refresh baseline used to refuse that state. The graph comment says the +/// caller must reconstruct instead. This walks occurrences in precedence order +/// and keeps only the latest predecessor per message, so a large generation is +/// not one exact-SQL page of `observation_json`. +pub(crate) async fn count_canonical_logical_copies( + conn: &impl crate::handle::SessionTemporalQuery, + session_id: &SessionId, + generation: SessionProjectionGenerationV1, +) -> SessionStoreResult { + const PAGE: i64 = 8; + let generation = generation_i64(generation, RECONSTRUCT_OPERATION)?; + let mut cursor: Option<(i64, i64, String)> = None; + let mut predecessors: BTreeMap = BTreeMap::new(); + let mut copies = 0_u64; + loop { + let mut rows = match &cursor { + None => conn + .query( + "SELECT occurrence.occurrence_id, occurrence.message_id, + occurrence.projection_output_ordinal, occurrence.knowledge_at, + observation.observation_json + FROM session_occurrences AS occurrence + JOIN observations AS observation + ON observation.observation_id = occurrence.source_observation_id + WHERE occurrence.session_id = ?1 AND occurrence.generation = ?2 + ORDER BY occurrence.knowledge_at, occurrence.projection_output_ordinal, + occurrence.occurrence_id + LIMIT ?3", + params![session_id.as_str(), generation, PAGE], + ) + .await + .map_err(|error| storage(RECONSTRUCT_OPERATION, error))?, + Some((knowledge_at, ordinal, occurrence_id)) => conn + .query( + "SELECT occurrence.occurrence_id, occurrence.message_id, + occurrence.projection_output_ordinal, occurrence.knowledge_at, + observation.observation_json + FROM session_occurrences AS occurrence + JOIN observations AS observation + ON observation.observation_id = occurrence.source_observation_id + WHERE occurrence.session_id = ?1 AND occurrence.generation = ?2 + AND ( + occurrence.knowledge_at > ?3 + OR ( + occurrence.knowledge_at = ?3 + AND occurrence.projection_output_ordinal > ?4 + ) + OR ( + occurrence.knowledge_at = ?3 + AND occurrence.projection_output_ordinal = ?4 + AND occurrence.occurrence_id > ?5 + ) + ) + ORDER BY occurrence.knowledge_at, occurrence.projection_output_ordinal, + occurrence.occurrence_id + LIMIT ?6", + params![ + session_id.as_str(), + generation, + *knowledge_at, + *ordinal, + occurrence_id.as_str(), + PAGE + ], + ) + .await + .map_err(|error| storage(RECONSTRUCT_OPERATION, error))?, + }; + let mut page_rows = 0_i64; + let mut page_cursor = None; + while let Some(row) = rows + .next() + .await + .map_err(|error| storage(RECONSTRUCT_OPERATION, error))? + { + let occurrence_id: String = row + .get(0) + .map_err(|error| storage(RECONSTRUCT_OPERATION, error))?; + let message_id: Option = row + .get(1) + .map_err(|error| storage(RECONSTRUCT_OPERATION, error))?; + let ordinal = u32::try_from( + row.get::(2) + .map_err(|error| storage(RECONSTRUCT_OPERATION, error))?, + ) + .map_err(|error| storage(RECONSTRUCT_OPERATION, error))?; + let knowledge_at = UtcMicros( + row.get(3) + .map_err(|error| storage(RECONSTRUCT_OPERATION, error))?, + ); + let observation: DurableObservationV1 = serde_json::from_str( + &row.get::(4) + .map_err(|error| storage(RECONSTRUCT_OPERATION, error))?, + ) + .map_err(|error| storage(RECONSTRUCT_OPERATION, error))?; + let parent_message_id = observation_envelope_from_payload(observation.payload()) + .map_err(|error| storage(RECONSTRUCT_OPERATION, error))? + .relations() + .parent_message_id() + .map(|id| id.as_str().to_owned()); + page_cursor = Some((knowledge_at.0, i64::from(ordinal), occurrence_id.clone())); + page_rows += 1; + if let (Some(message_id), Some(parent_message_id)) = (&message_id, &parent_message_id) + && message_id == parent_message_id + && predecessors.get(parent_message_id).is_some_and(|source| { + (source.0, source.1, source.2.as_str()) + < (knowledge_at, ordinal, occurrence_id.as_str()) + }) + { + copies = copies.saturating_add(1); + } + if let Some(message_id) = message_id { + let candidate = (knowledge_at, ordinal, occurrence_id); + match predecessors.get(&message_id) { + Some(existing) + if (existing.0, existing.1, existing.2.as_str()) + >= (candidate.0, candidate.1, candidate.2.as_str()) => {} + _ => { + predecessors.insert(message_id, candidate); + } + } + } + } + if page_rows < PAGE { + break; + } + cursor = page_cursor; + } + Ok(copies) +} + pub(crate) async fn reconstruct_logical_copy_relations( conn: &impl crate::handle::SessionTemporalQuery, session_id: &SessionId, From 250e318e8ebfc8272366a50586bb8bf2d0c4bd41 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Sat, 19 Sep 2026 09:15:34 +0000 Subject: [PATCH 14/15] chore(code-index): move Rust extraction to extractor.rust.v12 A receiver typed by a type parameter with one trait bound now names `Trait::method`, so the same bytes produce different reference rows than extractor.rust.v11 does. `generation_language_revisions_match` compares a sealed generation's extractor revisions against the registry, so without a bump every generation already sealed at v11 keeps the missing trait callee this change exists to add until some unrelated edit forces re-extraction. Re-pin the fixtures that carry the revision string, exactly as 1ebadc81f9 did for v11: - `canonical_rows_digest_matches_pinned_identity`: the revision is part of the batch identity, so the pinned rows digest moves to sha256:4e483806df. - `partitioned_codec_has_stable_bytes_and_round_trips`: the revision sits in every sealed file segment. v11 and v12 are the same length, so all four segment sizes are unchanged (11_071, 5_171, 6_279, 6_837) and only the digests move: an identity change, not container drift. - The two worker tests and the reconcile test that assert the current revision after a forced re-extraction. Co-Authored-By: Claude Fable 5.1 --- .../src/code_index_scheduler/tests/reconcile.rs | 2 +- crates/tracedecay-code-index/src/extract.rs | 10 ++++++---- crates/tracedecay-code-index/src/languages.rs | 8 +++++--- .../src/production/worker_tests.rs | 4 ++-- .../tests/code_index_suite/production_orchestration.rs | 10 +++++----- 5 files changed, 19 insertions(+), 15 deletions(-) diff --git a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/reconcile.rs b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/reconcile.rs index 44e4d9f5b1..5f19b178a6 100644 --- a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/reconcile.rs +++ b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/reconcile.rs @@ -851,7 +851,7 @@ fn retained_stale_rust_extractor_generation_is_refused_and_rebuilt() { .iter() .find(|(language, _)| language.as_str() == "rust") .map(|(_, revision)| revision.as_str()), - Some("extractor.rust.v11") + Some("extractor.rust.v12") ); } diff --git a/crates/tracedecay-code-index/src/extract.rs b/crates/tracedecay-code-index/src/extract.rs index cd4f39eb1a..3aedf93cb1 100644 --- a/crates/tracedecay-code-index/src/extract.rs +++ b/crates/tracedecay-code-index/src/extract.rs @@ -777,12 +777,14 @@ mod tests { // typed value and no longer fabricates receiver types for method // initializers; v9 adds the clone-body token bound and v10 the byte // bound; v11 drops the bare method name of a dotted call and types - // `self` from the enclosing impl or trait. The revision is part of the - // batch identity, so the pinned digest moves with it. - assert_eq!(descriptor.extractor_revision.as_str(), "extractor.rust.v11"); + // `self` from the enclosing impl or trait; v12 binds a receiver typed + // by a type parameter with one trait bound to `Trait::method`. The + // revision is part of the batch identity, so the pinned digest moves + // with it. + assert_eq!(descriptor.extractor_revision.as_str(), "extractor.rust.v12"); assert_eq!( extraction.batch().rows_digest.as_str(), - "sha256:e92b7ad8f93e3576c70adafb0690d064bd996c207a4ecd7b855c96e3ad3959ad" + "sha256:4e483806dfce308dfc97ec460c65f674b28f2dc8bad0e57c28a3b1519c8e495c" ); } diff --git a/crates/tracedecay-code-index/src/languages.rs b/crates/tracedecay-code-index/src/languages.rs index 5d7ea779ed..fdcf1392e6 100644 --- a/crates/tracedecay-code-index/src/languages.rs +++ b/crates/tracedecay-code-index/src/languages.rs @@ -214,9 +214,11 @@ impl StaticLanguageRegistry { // stopped emitting the bare method name of a dotted call, so an // unrelated same-file callable sharing that name is no longer a // caller, and types `self` through the enclosing impl or trait. - // Only re-extraction removes the poisoned record. + // Rust v12 resolves a receiver whose type is a type parameter with + // one trait bound to `Trait::method`, so that call now has a + // callee. Only re-extraction removes the poisoned record. let extractor_revision = if language == "rust" { - 11 + 12 } else if matches!(language.as_str(), "typescript" | "protobuf" | "sql") { 6 } else { @@ -421,7 +423,7 @@ mod tests { assert!(rust.stable_member_spans); assert!(rust.capabilities.extraction); assert_eq!(rust.root_markers, vec!["Cargo.toml".to_owned()]); - assert_eq!(rust.extractor_revision.as_str(), "extractor.rust.v11"); + assert_eq!(rust.extractor_revision.as_str(), "extractor.rust.v12"); assert_eq!( registry diff --git a/crates/tracedecay-code-index/src/production/worker_tests.rs b/crates/tracedecay-code-index/src/production/worker_tests.rs index 7531f94588..ba58d2d4d2 100644 --- a/crates/tracedecay-code-index/src/production/worker_tests.rs +++ b/crates/tracedecay-code-index/src/production/worker_tests.rs @@ -357,7 +357,7 @@ fn extractor_revision_change_reextracts_before_validating_retained_import_rows() assert_eq!( rebuilt.files[0].extraction.extractor_revision.as_str(), - "extractor.rust.v11" + "extractor.rust.v12" ); assert_ne!( rebuilt.files[0].extraction.parser_import_rows_digest, @@ -425,7 +425,7 @@ fn physical_artifact_reuse_rejects_a_stale_extractor_revision() { assert_eq!( rebuilt.files[0].extraction.extractor_revision.as_str(), - "extractor.rust.v11" + "extractor.rust.v12" ); assert!( rebuilt.files[0] diff --git a/crates/tracedecay-code-index/tests/code_index_suite/production_orchestration.rs b/crates/tracedecay-code-index/tests/code_index_suite/production_orchestration.rs index 6e7d5f7894..f4f4d73f23 100644 --- a/crates/tracedecay-code-index/tests/code_index_suite/production_orchestration.rs +++ b/crates/tracedecay-code-index/tests/code_index_suite/production_orchestration.rs @@ -3244,22 +3244,22 @@ fn partitioned_codec_fixture() -> ( } const PARTITIONED_FORMAT_STATE_DIGEST: &str = - "sha256:4f78e1a1b0a4ea366f748e4699d4c28d9913782912809bbe2afafba4eac25268"; + "sha256:89fa34dded4c696b44119ee9b658cea4d627fcef4b37b98360ca2298e36d1df3"; const PARTITIONED_FORMAT_SEGMENTS: &[(&str, u64)] = &[ ( - "sha256:7cd13a44df02cc2dc2e13d5a867aaab5410594eaa398fe535ed59450ddc63b36", + "sha256:8caabc1a8e5b468ba6ccd6826e81efa2fa2627763e7217df64ea0f7313a383a6", 11_071, ), ( - "sha256:64b5c4d5c08f363d66c1dc3922fb61df7c8df0b022dab6b72507e61e6e1bf403", + "sha256:c44d7072ba9f13b9e5c734950f94bec9bb646ba4f5b9d6c691d43f822fd657c8", 5_171, ), ( - "sha256:782d1321bdc39aac9efcff85a44f8d48da66e24212d28150020cb04309d23880", + "sha256:7709e7304333ac9c4a94e0e02508029f09712072e1606a707854f42c2c1f9ce3", 6_279, ), ( - "sha256:3a307f49e46059b54a86dac6921287afbe0b25cadc581840b5143ce2bd5a04d1", + "sha256:2da2f3ca63181ae45a17db21cfeb68b8a42ae6e104820cd91a0b5c0673c5d249", 6_837, ), ]; From e3b140e26e47fe5cf25ebb3ad8af72fbec7328de Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Sat, 19 Sep 2026 08:22:00 +0000 Subject: [PATCH 15/15] style(session-temporal): keep the test module last and rustfmt master Master run 35431539771 failed Check formatting (projector.rs, query.rs) and Clippy (items_after_test_module in query.rs) after #1844/#1845 merged without CI. Co-Authored-By: Claude Fable 5.1 --- .../projector.rs | 8 ++--- .../src/query.rs | 29 +++++++++---------- 2 files changed, 18 insertions(+), 19 deletions(-) diff --git a/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/projector.rs b/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/projector.rs index 10818733c1..632898ca24 100644 --- a/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/projector.rs +++ b/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/projector.rs @@ -119,11 +119,11 @@ impl SessionTemporalRefreshProjector for CanonicalSessionTemporalProjector { // Empty remaining range is a durable no-op: terminalize with an // empty complete progress batch instead of deferring forever. Ok(None) => canonical_noop_complete_effect(&recovery), - Err(error) if error.is_storage() => Err( - SessionTemporalRefreshProjectorError::retryable(format!( + Err(error) if error.is_storage() => { + Err(SessionTemporalRefreshProjectorError::retryable(format!( "source_busy: {error}" - )), - ), + ))) + } Err(_) => Err(SessionTemporalRefreshProjectorError::terminal( "projector_failed", )), diff --git a/crates/tracedecay-session-temporal-store/src/query.rs b/crates/tracedecay-session-temporal-store/src/query.rs index 367b14bb79..ae0ab6b4de 100644 --- a/crates/tracedecay-session-temporal-store/src/query.rs +++ b/crates/tracedecay-session-temporal-store/src/query.rs @@ -246,8 +246,7 @@ pub(super) async fn read_observations( // ceiling. Split until a single observation remains; that // observation is then a typed storage failure, not a retry // that looks like a busy source. - if observation_prefetch_exceeded_materialization_limit(&error) && chunk.len() > 1 - { + if observation_prefetch_exceeded_materialization_limit(&error) && chunk.len() > 1 { let mid = start + chunk.len() / 2; pending.push((mid, end)); pending.push((start, mid)); @@ -289,11 +288,18 @@ fn observation_prefetch_exceeded_materialization_limit(error: &SessionStoreError } } +/// The error `read_observation` raises for an id the store does not hold, reused +/// by callers that resolve prefetched observations out of a batch map. +pub(super) fn missing_observation(observation_id: &CanonicalObservationIdV1) -> SessionStoreError { + storage_message( + PERSIST_OPERATION, + format!("source observation {} is missing", observation_id.as_str()), + ) +} + #[cfg(test)] mod tests { - use super::{ - PERSIST_OPERATION, observation_prefetch_exceeded_materialization_limit, storage, - }; + use super::{PERSIST_OPERATION, observation_prefetch_exceeded_materialization_limit, storage}; #[test] fn materialization_limit_is_the_prefetch_split_signal() { @@ -308,15 +314,8 @@ mod tests { assert!(observation_prefetch_exceeded_materialization_limit( &exceeded )); - assert!(!observation_prefetch_exceeded_materialization_limit(&locked)); + assert!(!observation_prefetch_exceeded_materialization_limit( + &locked + )); } } - -/// The error `read_observation` raises for an id the store does not hold, reused -/// by callers that resolve prefetched observations out of a batch map. -pub(super) fn missing_observation(observation_id: &CanonicalObservationIdV1) -> SessionStoreError { - storage_message( - PERSIST_OPERATION, - format!("source observation {} is missing", observation_id.as_str()), - ) -}