diff --git a/crates/tracedecay-global-db/src/observation_adapter.rs b/crates/tracedecay-global-db/src/observation_adapter.rs index 6993c87b06..1e88baae1a 100644 --- a/crates/tracedecay-global-db/src/observation_adapter.rs +++ b/crates/tracedecay-global-db/src/observation_adapter.rs @@ -1,3 +1,4 @@ +use std::future::Future; use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; use tracedecay_application::clock::now_micros; @@ -21,79 +22,90 @@ use tracedecay_store::{ RepositoryReadOperationV1, RepositoryReadResultV1, RepositoryWritePayloadV1, RuntimeBatchCompatibilityV1, RuntimeCancellationIdV1, RuntimeCancellationIdentityV1, RuntimeDeadlineIdV1, RuntimeDeadlineV1, RuntimeInterruptionV1, RuntimeReadCoverageV1, - RuntimeReadOperationV1, RuntimeReadRequestV1, RuntimeReadResultV1, RuntimeRequestControlV1, - RuntimeRequestProbeV1, RuntimeSubmitOutcomeV1, RuntimeSubmitRequestV1, RuntimeTransactionIdV1, - RuntimeTransactionScopeV1, StoreClientIdV1, StoreIdempotencyKeyV1, StoreOperationIdV1, - StoreOperationMetadataV1, StoredObservation, StoredObservationRowV1, + RuntimeReadOperationV1, RuntimeReadOutcomeV1, RuntimeReadRequestV1, RuntimeReadResultV1, + RuntimeRequestControlV1, RuntimeRequestProbeV1, RuntimeSubmitOutcomeV1, RuntimeSubmitRequestV1, + RuntimeTransactionIdV1, RuntimeTransactionScopeV1, StoreClientIdV1, StoreIdempotencyKeyV1, + StoreOperationIdV1, StoreOperationMetadataV1, StoreRuntimeBindingV1, StoredObservation, + StoredObservationRowV1, }; use tracedecay_runtime_core::db::{Database, DatabaseRuntimeClientV1}; +use tracedecay_runtime_core::store_runtime::registry::StoreRuntimeRegistryFailure; +use tracedecay_rusqlite_runtime::repository::observation_cursor_authority::{ + COMMIT_SOURCE_CURSOR_SQL, READ_CURSOR_ADVANCE_SQL, READ_SOURCE_CURSOR_SQL, + RECORD_CURSOR_ADVANCE_SQL, cursor_advance_ledger_row_matches, +}; -/// Test-only counters over the expensive persist-path work (stored-row -/// decode, collision classification, payload-revision probing, canonical -/// command digesting). A re-admitted terminal identity collision must repeat -/// none of it, and only counters observed at these exact call sites can prove -/// that without editing the domain identity derivation. -#[cfg(test)] -#[derive(Debug, Default)] -pub(crate) struct ObservationPersistProbeV1 { - pub(crate) stored_observation_reads: std::sync::atomic::AtomicU64, - pub(crate) collision_classifications: std::sync::atomic::AtomicU64, - pub(crate) payload_revision_probes: std::sync::atomic::AtomicU64, - pub(crate) canonical_command_digests: std::sync::atomic::AtomicU64, +/// The closed dispatch boundary between this adapter and the authoritative +/// store runtime. Every stored-record read and every runtime write the +/// adapter performs crosses this seam, so an observer wrapped around it sees +/// exactly the record work one persist call dispatches — the collision tests +/// prove the terminal-refusal fast path repeats no stored-row read (and thus +/// no decode, classification, or revision probing) and no submit (and thus no +/// canonical command digest) by counting here. +pub(crate) trait ObservationRuntimeDispatch: Clone + Send + Sync { + fn binding(&self) -> &StoreRuntimeBindingV1; + + fn dispatch_read( + &self, + request: RuntimeReadRequestV1, + probe: &dyn RuntimeRequestProbeV1, + ) -> Result; + + fn dispatch_submit( + &self, + request: RuntimeSubmitRequestV1, + probe: Arc, + ) -> impl Future> + Send; } -#[cfg(test)] -impl ObservationPersistProbeV1 { - pub(crate) fn snapshot(&self) -> (u64, u64, u64, u64) { - use std::sync::atomic::Ordering::Relaxed; - ( - self.stored_observation_reads.load(Relaxed), - self.collision_classifications.load(Relaxed), - self.payload_revision_probes.load(Relaxed), - self.canonical_command_digests.load(Relaxed), - ) +impl ObservationRuntimeDispatch for DatabaseRuntimeClientV1 { + fn binding(&self) -> &StoreRuntimeBindingV1 { + DatabaseRuntimeClientV1::binding(self) } -} -#[cfg(test)] -macro_rules! probe_count { - ($store:expr, $counter:ident) => { - $store - .persist_probe - .$counter - .fetch_add(1, std::sync::atomic::Ordering::Relaxed); - }; -} + fn dispatch_read( + &self, + request: RuntimeReadRequestV1, + probe: &dyn RuntimeRequestProbeV1, + ) -> Result { + DatabaseRuntimeClientV1::dispatch_read(self, request, probe) + } -#[cfg(not(test))] -macro_rules! probe_count { - ($store:expr, $counter:ident) => {}; + async fn dispatch_submit( + &self, + request: RuntimeSubmitRequestV1, + probe: Arc, + ) -> Result { + DatabaseRuntimeClientV1::dispatch_submit(self, request, probe).await + } } /// Observation-store adapter over the already-registered authoritative runtime. +/// +/// The runtime parameter exists so tests can observe the dispatch seam; +/// production only ever constructs the default [`DatabaseRuntimeClientV1`] +/// via [`GlobalDbObservationStore::new`], and the struct shape is identical +/// in every build. #[derive(Clone)] -pub struct GlobalDbObservationStore { +pub struct GlobalDbObservationStore { database: Database, - runtime: DatabaseRuntimeClientV1, - #[cfg(test)] - persist_probe: Arc, + runtime: R, } impl GlobalDbObservationStore { pub fn new(database: Database) -> Self { let runtime = database.runtime_client(); - Self { - database, - runtime, - #[cfg(test)] - persist_probe: Arc::default(), - } + Self { database, runtime } } +} +impl GlobalDbObservationStore { + /// Binds the adapter to an explicit runtime dispatch seam so a test can + /// count the record work a persist path performs. #[cfg(test)] - pub(crate) fn persist_probe(&self) -> Arc { - Arc::clone(&self.persist_probe) + pub(crate) fn with_runtime_dispatch(database: Database, runtime: R) -> Self { + Self { database, runtime } } /// Records a terminal refusal — the marker in @@ -109,20 +121,25 @@ impl GlobalDbObservationStore { /// frontier is re-verified INSIDE the transaction (exact compare-and-set /// against the durable cursor), the advance-ledger row must carry the /// `admission_refused` reason with no receipt, and the cursor moves to - /// the advance's next position — mirroring the runtime cursor-advance - /// authority statement for statement. No record content is decoded, - /// derived, or hashed. + /// the advance's next position — executed through the one canonical + /// cursor-advance statement set + /// (`tracedecay_rusqlite_runtime::repository::observation_cursor_authority`) + /// that the runtime write path also executes. No record content is + /// decoded, derived, or hashed. async fn record_refusal_with_coverage( &self, write: &AnchoredObservationWrite, retained_digest: &PayloadDigestV1, - ) -> ObservationStoreResult<()> { + ) -> ObservationStoreResult<()> + where + R: ObservationRuntimeDispatch, + { const OPERATION: &str = "record refused admission terminal and coverage"; let candidate = write.observation(); let identity = candidate.identity(); let actual_cursor = read_runtime_source_cursor(&self.runtime, identity.source(), identity.scope())?; - let Some(mut advance) = refused_scan_frontier(write, actual_cursor.as_ref()) else { + let Some(mut advance) = refused_scan_frontier(write, actual_cursor.as_ref())? else { return Ok(()); }; match ( @@ -157,8 +174,7 @@ impl GlobalDbObservationStore { // still be the caller's expected frontier. let mut cursor_rows = transaction .query( - "SELECT cursor_json FROM source_cursors - WHERE source_json = ?1 AND scope_json = ?2", + READ_SOURCE_CURSOR_SQL, tracedecay_runtime_core::db::engine::params![ source_json.as_str(), scope_json.as_str() @@ -206,23 +222,20 @@ impl GlobalDbObservationStore { .map_err(|error| runtime_storage_error(OPERATION, error))?; transaction .execute( - "INSERT INTO source_cursor_advances ( - source_json, scope_json, coverage_json, reason, receipt_id - ) VALUES (?1, ?2, ?3, ?4, NULL) - ON CONFLICT(source_json, scope_json, coverage_json) DO NOTHING", + RECORD_CURSOR_ADVANCE_SQL, tracedecay_runtime_core::db::engine::params![ source_json.as_str(), scope_json.as_str(), coverage_json.as_str(), - ObservationCoverageReason::AdmissionRefused.as_str() + ObservationCoverageReason::AdmissionRefused.as_str(), + None::<&str> ], ) .await .map_err(|error| runtime_storage_error(OPERATION, error))?; let mut ledger_rows = transaction .query( - "SELECT reason, receipt_id FROM source_cursor_advances - WHERE source_json = ?1 AND scope_json = ?2 AND coverage_json = ?3", + READ_CURSOR_ADVANCE_SQL, tracedecay_runtime_core::db::engine::params![ source_json.as_str(), scope_json.as_str(), @@ -248,14 +261,11 @@ impl GlobalDbObservationStore { // A coverage row that names any other reason or a receipt is a real // cursor-advance failure: roll the WHOLE transaction back so the // marker is not visible either — no orphan, by construction. - if ledger - != Some(( - ObservationCoverageReason::AdmissionRefused - .as_str() - .to_owned(), - None, - )) - { + if !cursor_advance_ledger_row_matches( + ledger.as_ref(), + ObservationCoverageReason::AdmissionRefused.as_str(), + None, + ) { transaction .rollback() .await @@ -264,10 +274,7 @@ impl GlobalDbObservationStore { } transaction .execute( - "INSERT INTO source_cursors (source_json, scope_json, cursor_json) - VALUES (?1, ?2, ?3) - ON CONFLICT(source_json, scope_json) DO UPDATE SET - cursor_json = excluded.cursor_json", + COMMIT_SOURCE_CURSOR_SQL, tracedecay_runtime_core::db::engine::params![ source_json.as_str(), scope_json.as_str(), @@ -290,7 +297,7 @@ impl GlobalDbObservationStore { } } -impl ObservationStore for GlobalDbObservationStore { +impl ObservationStore for GlobalDbObservationStore { async fn persist_observation( &self, write: AnchoredObservationWrite, @@ -332,14 +339,11 @@ impl ObservationStore for GlobalDbObservationStore { outcome: ObservationCollisionOutcomeV1::IdentityCollision, }); } - probe_count!(self, stored_observation_reads); let existing = read_runtime_stored_observation(runtime, &observation_id)?; - let collision = existing.as_ref().map(|existing| { - probe_count!(self, collision_classifications); - classify_observation_collision(existing.observation(), &candidate) - }); + let collision = existing + .as_ref() + .map(|existing| classify_observation_collision(existing.observation(), &candidate)); let canonical_payload_revision = existing.as_ref().is_some_and(|existing| { - probe_count!(self, payload_revision_probes); is_canonical_payload_revision_replay(existing.observation(), &candidate) }); if collision == Some(ObservationCollisionOutcomeV1::IdentityCollision) @@ -517,7 +521,6 @@ impl ObservationStore for GlobalDbObservationStore { existing.commit_receipt().clone(), )); } - probe_count!(self, canonical_command_digests); let idempotency_key = format!( "observation.{}", canonical_runtime_digest(&runtime_observation_command(&write))? @@ -538,7 +541,6 @@ impl ObservationStore for GlobalDbObservationStore { candidate.source().session_id().as_str(), ) .map_err(|(operation, detail)| runtime_storage_error(operation, detail))?; - probe_count!(self, stored_observation_reads); let stored = read_runtime_stored_observation(runtime, &observation_id)?.ok_or_else(|| { runtime_storage_error("read committed observation", "row unavailable") @@ -609,7 +611,6 @@ impl ObservationStore for GlobalDbObservationStore { "scope": advance.next_cursor().scope(), "coverage": advance.coverage(), }); - probe_count!(self, canonical_command_digests); let key = format!("cursor.{}", canonical_runtime_digest(&identity)?); let outcome = submit_runtime_write( runtime, @@ -718,7 +719,7 @@ impl RuntimeRequestProbeV1 for RuntimeObservationProbe { } fn dispatch_runtime_observation_read( - runtime: &DatabaseRuntimeClientV1, + runtime: &impl ObservationRuntimeDispatch, operation: ObservationReadOperationV1, ) -> ObservationStoreResult { let command_digest = canonical_sha256(&operation) @@ -819,7 +820,7 @@ fn stored_observation_from_runtime_row( } fn read_runtime_source_cursor( - runtime: &DatabaseRuntimeClientV1, + runtime: &impl ObservationRuntimeDispatch, source: &ClaudeSourceIdentityV1, scope: &ObservationScopeV1, ) -> ObservationStoreResult> { @@ -839,7 +840,7 @@ fn read_runtime_source_cursor( } fn read_runtime_retrieval_anchor_by_alias( - runtime: &DatabaseRuntimeClientV1, + runtime: &impl ObservationRuntimeDispatch, scope: &ObservationScopeV1, alias: &tracedecay_domain::NativeAliasV2, ) -> ObservationStoreResult> { @@ -858,18 +859,24 @@ fn read_runtime_retrieval_anchor_by_alias( } } -/// Whether a refused candidate stands at the sequential scan frontier: the -/// durable cursor has NOT covered its range, the caller's expected cursor -/// matches the durable one, and the record either continues the current -/// generation contiguously or restarts a replacement generation from position -/// zero. Generation values are opaque source identities, not ordered counters. -/// Coverage is -/// recorded only for this shape — the one production ingest actually loops -/// on; gaps and stale views prove the caller is not the scan frontier. +/// The typed cursor advance for a refused candidate standing at the +/// sequential scan frontier: the durable cursor has NOT covered its range and +/// the caller's expected cursor matches the durable one. Generation values +/// are opaque source identities, not ordered counters. +/// +/// `Ok(None)` is the not-at-frontier verdict — a covered replay or a stale +/// expected view — and leaves every ledger untouched. Gaps and generation +/// jumps never reach the advance constructor here: `ObservationWrite::new` +/// already validated that this write's expected→next cursor transition +/// covers the candidate range, which is exactly the transition the advance +/// re-derives from the same identity, expected cursor, and range. A +/// construction failure is therefore a contract violation, and it surfaces +/// as the typed store error — silently answering "not at the scan frontier" +/// would record no coverage and leave the refused record re-read forever. fn refused_scan_frontier( write: &AnchoredObservationWrite, actual_cursor: Option<&ClaudeSourceCursorV1>, -) -> Option { +) -> ObservationStoreResult> { let identity = write.observation().identity(); let candidate_covered = actual_cursor.is_some_and(|cursor| { cursor.generation() == identity.generation() @@ -877,7 +884,7 @@ fn refused_scan_frontier( && cursor.position() >= identity.position().end() }); if candidate_covered || actual_cursor != write.expected_cursor() { - return None; + return Ok(None); } ObservationCursorAdvance::for_ordering( identity.source().clone(), @@ -888,7 +895,7 @@ fn refused_scan_frontier( identity.position(), ObservationCoverageReason::AdmissionRefused, ) - .ok() + .map(Some) } /// Durable terminal marker for a previously refused identity collision. @@ -938,7 +945,7 @@ async fn read_admission_refusal( } fn read_runtime_stored_observation( - runtime: &DatabaseRuntimeClientV1, + runtime: &impl ObservationRuntimeDispatch, observation_id: &CanonicalObservationIdV1, ) -> ObservationStoreResult> { match dispatch_runtime_observation_read( @@ -958,7 +965,7 @@ fn read_runtime_stored_observation( } async fn submit_runtime_write( - runtime: &DatabaseRuntimeClientV1, + runtime: &impl ObservationRuntimeDispatch, payload: RepositoryWritePayloadV1, idempotency_key: String, operation: &'static str, @@ -1101,7 +1108,7 @@ fn runtime_storage_error( } } -impl ObservationProjectionStore for GlobalDbObservationStore { +impl ObservationProjectionStore for GlobalDbObservationStore { async fn next_queued_observation( &self, ) -> ProjectionStoreResult> { @@ -1172,7 +1179,6 @@ mod tests { let GlobalDbObservationStore { database: _, runtime: _, - persist_probe: _, } = store; } diff --git a/crates/tracedecay-global-db/src/observation_collision_tests.rs b/crates/tracedecay-global-db/src/observation_collision_tests.rs index e2d962667a..4d47258d3e 100644 --- a/crates/tracedecay-global-db/src/observation_collision_tests.rs +++ b/crates/tracedecay-global-db/src/observation_collision_tests.rs @@ -50,7 +50,10 @@ use tracedecay_store::{ ProjectionPersistOutcome, ProjectionSkipReason, SESSION_MESSAGE_PROJECTOR_VERSION, }; -use crate::tests::harness::{HostAdmissionScope, HostAdmissionTestRuntimeV1}; +use crate::tests::harness::{ + CountingObservationRuntime, HostAdmissionScope, HostAdmissionTestRuntimeV1, + ObservationDispatchCounts, +}; use tracedecay_runtime_core::db::engine::params; const COLLISION_PROVIDER: &str = "collision-test"; @@ -538,8 +541,8 @@ async fn re_admitted_identity_collision_short_circuits_without_decode_or_hash() let runtime = HostAdmissionTestRuntimeV1::profile(tmp.path()) .await .unwrap(); - let store = runtime - .observation_store(HostAdmissionScope::Profile) + let (store, counts) = runtime + .counting_observation_store(HostAdmissionScope::Profile) .unwrap(); let session_id = SessionId::new("session.identity-collision.readmitted").unwrap(); let (original, original_write) = collision_candidate( @@ -566,6 +569,7 @@ async fn re_admitted_identity_collision_short_circuits_without_decode_or_hash() "receipt.identity-collision.readmitted.rewritten", committed_cursor, ); + let before_first = counts.snapshot(); let first = store .persist_observation(rewritten_write.clone()) .await @@ -580,9 +584,14 @@ async fn re_admitted_identity_collision_short_circuits_without_decode_or_hash() ), "{first:?}" ); + let before = counts.snapshot(); + // Non-vacuity anchor: the first collision classifies against the stored + // row, so the counting seam must have observed its stored-row read. + assert!( + before.stored_observation_reads > before_first.stored_observation_reads, + "the first collision must read the stored row through the counted dispatch seam" + ); - let probe = store.persist_probe(); - let (reads, classifications, revision_probes, digests) = probe.snapshot(); // A later catch-up pass or temporal trigger re-presents the exact same // candidate with its now-stale expected cursor. let second = store @@ -600,27 +609,25 @@ async fn re_admitted_identity_collision_short_circuits_without_decode_or_hash() "{second:?}" ); - let (reads_after, classifications_after, revision_probes_after, digests_after) = - probe.snapshot(); - assert_eq!( - reads_after - reads, - 0, - "re-admitted terminal collision must not decode the stored observation row again" - ); + let after = counts.snapshot(); assert_eq!( - classifications_after - classifications, + after.stored_observation_reads - before.stored_observation_reads, 0, - "re-admitted terminal collision must not re-classify the collision" + "re-admitted terminal collision must not read the stored observation row again — \ + without that read it cannot decode, re-classify the collision, or re-probe the \ + payload revision" ); assert_eq!( - revision_probes_after - revision_probes, + after.submits - before.submits, 0, - "re-admitted terminal collision must not re-probe the payload revision" + "re-admitted terminal collision must not submit a runtime write, so it never \ + canonicalizes or hashes a command again" ); - assert_eq!( - digests_after - digests, - 0, - "re-admitted terminal collision must not canonicalize or hash again" + // The seam stayed live inside the asserted window: the re-admission's own + // frontier check dispatches exactly one typed source-cursor read. + assert!( + after.source_cursor_reads > before.source_cursor_reads, + "the counting seam must observe the re-admission's frontier cursor read" ); // The terminal coverage stays single-row and the cursor stays put. assert_eq!( @@ -647,8 +654,8 @@ async fn replacement_domain_collision_records_terminal_coverage_without_rework() let runtime = HostAdmissionTestRuntimeV1::profile(tmp.path()) .await .unwrap(); - let store = runtime - .observation_store(HostAdmissionScope::Profile) + let (store, counts) = runtime + .counting_observation_store(HostAdmissionScope::Profile) .unwrap(); let session_id = SessionId::new("session.identity-collision.domain-replacement").unwrap(); let (original, original_write) = collision_candidate( @@ -712,8 +719,7 @@ async fn replacement_domain_collision_records_terminal_coverage_without_rework() ); assert_eq!(admission_refusal_rows(&runtime).await.len(), 1); - let probe = store.persist_probe(); - let before = probe.snapshot(); + let before = counts.snapshot(); let second = store .persist_observation(replacement_write) .await @@ -725,16 +731,16 @@ async fn replacement_domain_collision_records_terminal_coverage_without_rework() .. } )); - let after = probe.snapshot(); + let after = counts.snapshot(); assert_eq!( ( - after.0 - before.0, - after.1 - before.1, - after.2 - before.2, - after.3 - before.3, + after.stored_observation_reads - before.stored_observation_reads, + after.submits - before.submits, ), - (0, 0, 0, 0), - "re-admission must not read, classify, probe revisions, or hash the record" + (0, 0), + "re-admission must not read the stored row (the only route to decoding, \ + classifying, or revision-probing it) and must not submit (so nothing is \ + canonicalized or hashed)" ); } @@ -1054,10 +1060,13 @@ async fn raw_observation_json( /// Boundary accounting for one record a catch-up pass decoded and persisted. struct CatchUpRecordReceipt { result: Result, - /// Adapter-probe deltas across the persist call: stored-observation - /// reads, collision classifications, payload-revision probes, canonical - /// command digests. - persist_probe_deltas: (u64, u64, u64, u64), + /// Runtime-dispatch deltas across the persist call, observed at the + /// adapter's real dispatch seam: `(stored-observation reads, runtime + /// submits)`. A stored-observation read is the only route to decoding, + /// collision-classifying, or revision-probing the retained row, and + /// every submit is keyed by exactly one canonical command digest — so + /// `(0, 0)` proves the persist call repeated none of that record work. + dispatch_deltas: (u64, u64), } /// One real catch-up pass over raw persisted source input: read the durable @@ -1066,7 +1075,8 @@ struct CatchUpRecordReceipt { /// ABORTING the pass on a persist error — an identity collision ends the /// pass, it does not skip to the next record. async fn run_catch_up_pass( - store: &crate::GlobalDbObservationStore, + store: &crate::GlobalDbObservationStore, + counts: &ObservationDispatchCounts, session_id: &SessionId, generation: u64, raw_lines: &[((u64, u64), String)], @@ -1076,7 +1086,6 @@ async fn run_catch_up_pass( let source = ObservationSourceIdentityV1::for_provider(provider, session_id.clone()).unwrap(); let scope = ObservationScopeV1::Profile; let scan_generation = ObservationSourceGenerationV1::new(generation).unwrap(); - let probe = store.persist_probe(); let mut decoded = 0; let mut receipts = Vec::new(); for (index, (range, raw_line)) in raw_lines.iter().enumerate() { @@ -1098,18 +1107,15 @@ async fn run_catch_up_pass( &format!("receipt.catch-up.{pass_label}.{index}"), ); let write = anchored_write_for(observation, cursor); - let (reads, classifications, revision_probes, command_digests) = probe.snapshot(); + let before = counts.snapshot(); let result = store.persist_observation(write).await; - let (reads_after, classifications_after, revision_probes_after, command_digests_after) = - probe.snapshot(); + let after = counts.snapshot(); let aborted = result.is_err(); receipts.push(CatchUpRecordReceipt { result, - persist_probe_deltas: ( - reads_after - reads, - classifications_after - classifications, - revision_probes_after - revision_probes, - command_digests_after - command_digests, + dispatch_deltas: ( + after.stored_observation_reads - before.stored_observation_reads, + after.submits - before.submits, ), }); if aborted { @@ -1633,8 +1639,8 @@ async fn terminal_refusal_survives_retention_and_catch_up_never_reopens_the_reco let runtime = HostAdmissionTestRuntimeV1::profile(tmp.path()) .await .unwrap(); - let store = runtime - .observation_store(HostAdmissionScope::Profile) + let (store, counts) = runtime + .counting_observation_store(HostAdmissionScope::Profile) .unwrap(); let session_id = SessionId::new("session.terminal-refusal.retention").unwrap(); @@ -1671,7 +1677,7 @@ async fn terminal_refusal_survives_retention_and_catch_up_never_reopens_the_reco // Pass 0: gen-1 ingest of the original file. let (decoded, receipts) = - run_catch_up_pass(&store, &session_id, 1, &original_lines, "gen1").await; + run_catch_up_pass(&store, &counts, &session_id, 1, &original_lines, "gen1").await; assert_eq!(decoded, 1); assert!(matches!( receipts[0].result, @@ -1683,7 +1689,7 @@ async fn terminal_refusal_survives_retention_and_catch_up_never_reopens_the_reco // refusal's own coverage advance lets the follow-up pass move on to // record one and converge. let (decoded, receipts) = - run_catch_up_pass(&store, &session_id, 2, &rewritten_lines, "gen2").await; + run_catch_up_pass(&store, &counts, &session_id, 2, &rewritten_lines, "gen2").await; assert_eq!(decoded, 1, "the collision aborts the pass"); assert!(matches!( receipts[0].result, @@ -1692,8 +1698,15 @@ async fn terminal_refusal_survives_retention_and_catch_up_never_reopens_the_reco .. }) )); - let (decoded, receipts) = - run_catch_up_pass(&store, &session_id, 2, &rewritten_lines, "gen2-resume").await; + let (decoded, receipts) = run_catch_up_pass( + &store, + &counts, + &session_id, + 2, + &rewritten_lines, + "gen2-resume", + ) + .await; assert_eq!(decoded, 1, "the resumed pass skips the refused coverage"); assert!(matches!( receipts[0].result, @@ -1710,7 +1723,8 @@ async fn terminal_refusal_survives_retention_and_catch_up_never_reopens_the_reco assert_eq!(admission_refusal_rows(&runtime).await.len(), 1); // Pass 2: a later catch-up pass reopens nothing. - let (decoded, _) = run_catch_up_pass(&store, &session_id, 2, &rewritten_lines, "gen2-b").await; + let (decoded, _) = + run_catch_up_pass(&store, &counts, &session_id, 2, &rewritten_lines, "gen2-b").await; assert_eq!( decoded, 0, "catch-up must not reopen covered source records" @@ -1754,8 +1768,7 @@ async fn terminal_refusal_survives_retention_and_catch_up_never_reopens_the_reco // refused candidate without a current frontier view) still terminates // with zero decode/derive/hash work. let stale_replay = anchored_write_for(refused.clone(), None); - let probe = store.persist_probe(); - let (reads, classifications, revision_probes, command_digests) = probe.snapshot(); + let before = counts.snapshot(); let error = store.persist_observation(stale_replay).await.unwrap_err(); assert!( matches!( @@ -1767,12 +1780,18 @@ async fn terminal_refusal_survives_retention_and_catch_up_never_reopens_the_reco ), "{error:?}" ); - let (reads_after, classifications_after, revision_probes_after, command_digests_after) = - probe.snapshot(); - assert_eq!(reads_after - reads, 0); - assert_eq!(classifications_after - classifications, 0); - assert_eq!(revision_probes_after - revision_probes, 0); - assert_eq!(command_digests_after - command_digests, 0); + let after = counts.snapshot(); + assert_eq!( + after.stored_observation_reads - before.stored_observation_reads, + 0, + "the stale re-admission must not read the stored row, so it cannot decode, \ + classify, or revision-probe it" + ); + assert_eq!( + after.submits - before.submits, + 0, + "the stale re-admission must not submit a runtime write, so nothing is hashed" + ); // Restart: the terminal and coverage are durable, catch-up still reopens // nothing, and the retained row is byte-identical. @@ -1781,11 +1800,18 @@ async fn terminal_refusal_survives_retention_and_catch_up_never_reopens_the_reco let reopened = HostAdmissionTestRuntimeV1::profile(tmp.path()) .await .unwrap(); - let reopened_store = reopened - .observation_store(HostAdmissionScope::Profile) + let (reopened_store, reopened_counts) = reopened + .counting_observation_store(HostAdmissionScope::Profile) .unwrap(); - let (decoded, _) = - run_catch_up_pass(&reopened_store, &session_id, 2, &rewritten_lines, "gen2-c").await; + let (decoded, _) = run_catch_up_pass( + &reopened_store, + &reopened_counts, + &session_id, + 2, + &rewritten_lines, + "gen2-c", + ) + .await; assert_eq!(decoded, 0); assert_eq!(admission_refusal_rows(&reopened).await.len(), 1); assert_eq!( @@ -1803,10 +1829,11 @@ async fn terminal_refusal_survives_retention_and_catch_up_never_reopens_the_reco /// re-admits the refused record and must be suppressed by the retained /// terminal with ZERO store-side decode/canonicalize/SHA work. /// -/// Adapter dispatch counts prove the retained row is not decoded, collision -/// classified, revision-probed, or command-digested again. The real Vibe -/// journey below separately proves the production source boundary performs -/// no subsequent frame materialization. +/// Runtime-dispatch counts at the adapter's dispatch seam prove the retained +/// row is not read again (so never decoded, collision classified, or +/// revision-probed) and nothing is submitted (so never command-digested). The +/// real Vibe journey below separately proves the production source boundary +/// performs no subsequent frame materialization. #[tokio::test] async fn post_retention_rescan_re_admits_from_raw_source_without_terminal_rework() { use crate::observation::retention::{ObservationRetentionConfig, RetentionMode}; @@ -1815,8 +1842,8 @@ async fn post_retention_rescan_re_admits_from_raw_source_without_terminal_rework let runtime = HostAdmissionTestRuntimeV1::profile(tmp.path()) .await .unwrap(); - let store = runtime - .observation_store(HostAdmissionScope::Profile) + let (store, counts) = runtime + .counting_observation_store(HostAdmissionScope::Profile) .unwrap(); let session_id = SessionId::new("session.terminal-refusal.rescan").unwrap(); let original_lines = vec![( @@ -1853,14 +1880,14 @@ async fn post_retention_rescan_re_admits_from_raw_source_without_terminal_rework // record terminally and commits the appended record, advancing the cursor // strictly past the refused coverage. let (decoded, receipts) = - run_catch_up_pass(&store, &session_id, 1, &original_lines, "gen1").await; + run_catch_up_pass(&store, &counts, &session_id, 1, &original_lines, "gen1").await; assert_eq!(decoded, 1); assert!(matches!( receipts[0].result, Ok(ObservationPersistOutcome::Committed(_)) )); let (decoded, receipts) = - run_catch_up_pass(&store, &session_id, 2, &rewritten_lines, "gen2").await; + run_catch_up_pass(&store, &counts, &session_id, 2, &rewritten_lines, "gen2").await; assert_eq!(decoded, 1, "the collision aborts the pass like production"); assert!(matches!( receipts[0].result, @@ -1869,8 +1896,15 @@ async fn post_retention_rescan_re_admits_from_raw_source_without_terminal_rework .. }) )); - let (decoded, receipts) = - run_catch_up_pass(&store, &session_id, 2, &rewritten_lines, "gen2-resume").await; + let (decoded, receipts) = run_catch_up_pass( + &store, + &counts, + &session_id, + 2, + &rewritten_lines, + "gen2-resume", + ) + .await; assert_eq!(decoded, 1, "the resumed pass skips the refused coverage"); assert!(matches!( receipts[0].result, @@ -1910,7 +1944,7 @@ async fn post_retention_rescan_re_admits_from_raw_source_without_terminal_rework // fast path answers from the terminal, converges coverage with one typed // cursor-advance write, and aborts the pass like production. let (decoded, receipts) = - run_catch_up_pass(&store, &session_id, 3, &rewritten_lines, "gen3").await; + run_catch_up_pass(&store, &counts, &session_id, 3, &rewritten_lines, "gen3").await; assert_eq!( decoded, 1, "a rescan after a real file change re-reads the raw source and aborts on the collision" @@ -1928,11 +1962,11 @@ async fn post_retention_rescan_re_admits_from_raw_source_without_terminal_rework refused_readmit.result ); assert_eq!( - refused_readmit.persist_probe_deltas, - (0, 0, 0, 0), - "the re-admit must not read the stored row, classify, probe revisions, or \ - digest commands; coverage converges inside one direct authority \ - transaction with no record work" + refused_readmit.dispatch_deltas, + (0, 0), + "the re-admit must not read the stored row (so it cannot classify or probe \ + revisions) and must not submit (so it digests nothing); coverage converges \ + inside one direct authority transaction with no record work" ); // The suppression above was answered by the retained refusal terminal: // it must have survived cursor-advance retention. @@ -1943,11 +1977,19 @@ async fn post_retention_rescan_re_admits_from_raw_source_without_terminal_rework ); // The resumed pass commits the appended record past the converged // coverage, and the NEXT pass reopens zero source records. - let (decoded, receipts) = - run_catch_up_pass(&store, &session_id, 3, &rewritten_lines, "gen3-resume").await; + let (decoded, receipts) = run_catch_up_pass( + &store, + &counts, + &session_id, + 3, + &rewritten_lines, + "gen3-resume", + ) + .await; assert_eq!(decoded, 1, "the resumed pass skips the refused coverage"); assert!(receipts[0].result.is_ok(), "{:?}", receipts[0].result); - let (decoded, _) = run_catch_up_pass(&store, &session_id, 3, &rewritten_lines, "gen3-b").await; + let (decoded, _) = + run_catch_up_pass(&store, &counts, &session_id, 3, &rewritten_lines, "gen3-b").await; assert_eq!(decoded, 0, "the converged rescan reopens no source records"); // Immutable old row: byte-identical after every pass. @@ -1972,8 +2014,8 @@ async fn eof_refusal_converges_new_generation_rescans_without_reopening() { let runtime = HostAdmissionTestRuntimeV1::profile(tmp.path()) .await .unwrap(); - let store = runtime - .observation_store(HostAdmissionScope::Profile) + let (store, counts) = runtime + .counting_observation_store(HostAdmissionScope::Profile) .unwrap(); let session_id = SessionId::new("session.terminal-refusal.eof").unwrap(); // The refused record is the ONLY record: nothing follows it, ever. @@ -1987,7 +2029,7 @@ async fn eof_refusal_converges_new_generation_rescans_without_reopening() { )]; let (decoded, receipts) = - run_catch_up_pass(&store, &session_id, 1, &original_lines, "gen1").await; + run_catch_up_pass(&store, &counts, &session_id, 1, &original_lines, "gen1").await; assert_eq!(decoded, 1); assert!(matches!( receipts[0].result, @@ -1997,7 +2039,7 @@ async fn eof_refusal_converges_new_generation_rescans_without_reopening() { // Gen-2 rescan: the EOF record collides and the pass aborts. The refusal // records terminal + coverage, so the SAME generation never reopens it. let (decoded, receipts) = - run_catch_up_pass(&store, &session_id, 2, &rewritten_lines, "gen2").await; + run_catch_up_pass(&store, &counts, &session_id, 2, &rewritten_lines, "gen2").await; assert_eq!(decoded, 1); assert!(matches!( receipts[0].result, @@ -2014,7 +2056,8 @@ async fn eof_refusal_converges_new_generation_rescans_without_reopening() { "receipt.catch-up.gen2.0", ); let retained_row = raw_observation_json(&runtime, refused.observation_id().as_str()).await; - let (decoded, _) = run_catch_up_pass(&store, &session_id, 2, &rewritten_lines, "gen2-b").await; + let (decoded, _) = + run_catch_up_pass(&store, &counts, &session_id, 2, &rewritten_lines, "gen2-b").await; assert_eq!( decoded, 0, "the refused EOF coverage holds within its generation" @@ -2039,7 +2082,7 @@ async fn eof_refusal_converges_new_generation_rescans_without_reopening() { // terminal AND converges the new generation's coverage, so this exact // decode happens once per real file change — never again for gen 3. let (decoded, receipts) = - run_catch_up_pass(&store, &session_id, 3, &rewritten_lines, "gen3").await; + run_catch_up_pass(&store, &counts, &session_id, 3, &rewritten_lines, "gen3").await; assert_eq!(decoded, 1); let readmit = &receipts[0]; assert!( @@ -2054,11 +2097,12 @@ async fn eof_refusal_converges_new_generation_rescans_without_reopening() { readmit.result ); assert_eq!( - readmit.persist_probe_deltas, - (0, 0, 0, 0), + readmit.dispatch_deltas, + (0, 0), "the EOF re-admit converges coverage atomically with no record work" ); - let (decoded, _) = run_catch_up_pass(&store, &session_id, 3, &rewritten_lines, "gen3-b").await; + let (decoded, _) = + run_catch_up_pass(&store, &counts, &session_id, 3, &rewritten_lines, "gen3-b").await; assert_eq!( decoded, 0, "later gen-3 passes must never reopen the refused EOF record" @@ -2083,11 +2127,18 @@ async fn eof_refusal_converges_new_generation_rescans_without_reopening() { let reopened = HostAdmissionTestRuntimeV1::profile(tmp.path()) .await .unwrap(); - let reopened_store = reopened - .observation_store(HostAdmissionScope::Profile) + let (reopened_store, reopened_counts) = reopened + .counting_observation_store(HostAdmissionScope::Profile) .unwrap(); - let (decoded, _) = - run_catch_up_pass(&reopened_store, &session_id, 3, &rewritten_lines, "gen3-c").await; + let (decoded, _) = run_catch_up_pass( + &reopened_store, + &reopened_counts, + &session_id, + 3, + &rewritten_lines, + "gen3-c", + ) + .await; assert_eq!( decoded, 0, "restarted rescans must never reopen the refused EOF record" @@ -2111,8 +2162,8 @@ async fn orphaned_refusal_marker_repairs_coverage_on_the_next_frontier_pass() { let runtime = HostAdmissionTestRuntimeV1::profile(tmp.path()) .await .unwrap(); - let store = runtime - .observation_store(HostAdmissionScope::Profile) + let (store, counts) = runtime + .counting_observation_store(HostAdmissionScope::Profile) .unwrap(); let session_id = SessionId::new("session.terminal-refusal.orphan").unwrap(); let original_lines = vec![( @@ -2124,7 +2175,7 @@ async fn orphaned_refusal_marker_repairs_coverage_on_the_next_frontier_pass() { raw_source_line(&session_id, "record.orphan.0", (0, 1), "rewritten record"), )]; let (decoded, receipts) = - run_catch_up_pass(&store, &session_id, 1, &original_lines, "gen1").await; + run_catch_up_pass(&store, &counts, &session_id, 1, &original_lines, "gen1").await; assert_eq!(decoded, 1); assert!(matches!( receipts[0].result, @@ -2172,7 +2223,7 @@ async fn orphaned_refusal_marker_repairs_coverage_on_the_next_frontier_pass() { // The next frontier pass re-admits the record from raw source: the // orphaned marker must answer it AND repair the missing coverage. let (decoded, receipts) = - run_catch_up_pass(&store, &session_id, 2, &rewritten_lines, "gen2").await; + run_catch_up_pass(&store, &counts, &session_id, 2, &rewritten_lines, "gen2").await; assert_eq!(decoded, 1); let repair = &receipts[0]; assert!( @@ -2187,14 +2238,15 @@ async fn orphaned_refusal_marker_repairs_coverage_on_the_next_frontier_pass() { repair.result ); assert_eq!( - repair.persist_probe_deltas, - (0, 0, 0, 0), + repair.dispatch_deltas, + (0, 0), "the orphan-marker re-admit repairs coverage atomically with no record work" ); // Coverage is repaired: later passes never reopen the record, even after // a restart, and no duplicate marker rows appear. - let (decoded, _) = run_catch_up_pass(&store, &session_id, 2, &rewritten_lines, "gen2-b").await; + let (decoded, _) = + run_catch_up_pass(&store, &counts, &session_id, 2, &rewritten_lines, "gen2-b").await; assert_eq!(decoded, 0, "repaired coverage must not reopen the record"); assert_eq!(admission_refusal_rows(&runtime).await.len(), 1); drop(store); @@ -2202,11 +2254,18 @@ async fn orphaned_refusal_marker_repairs_coverage_on_the_next_frontier_pass() { let reopened = HostAdmissionTestRuntimeV1::profile(tmp.path()) .await .unwrap(); - let reopened_store = reopened - .observation_store(HostAdmissionScope::Profile) + let (reopened_store, reopened_counts) = reopened + .counting_observation_store(HostAdmissionScope::Profile) .unwrap(); - let (decoded, _) = - run_catch_up_pass(&reopened_store, &session_id, 2, &rewritten_lines, "gen2-c").await; + let (decoded, _) = run_catch_up_pass( + &reopened_store, + &reopened_counts, + &session_id, + 2, + &rewritten_lines, + "gen2-c", + ) + .await; assert_eq!(decoded, 0); assert_eq!( raw_observation_json(&reopened, refused.observation_id().as_str()).await, diff --git a/crates/tracedecay-global-db/src/observation_projection/rebuild.rs b/crates/tracedecay-global-db/src/observation_projection/rebuild.rs index 7b164d0fd9..1e3a154398 100644 --- a/crates/tracedecay-global-db/src/observation_projection/rebuild.rs +++ b/crates/tracedecay-global-db/src/observation_projection/rebuild.rs @@ -27,8 +27,8 @@ use super::apply::{ use super::state::{ consume_projection_queue_item, decode_observation_row, decode_sequence, ensure_projection_output_state_cache, projection_retry_state, queued_sequence, read_checkpoint, - read_message, read_observation, read_session, schedule_projection_retry, storage, - storage_message, write_checkpoint, + read_message, read_observation, read_session, reaggregate_output_state_for_output, + schedule_projection_retry, storage, storage_message, write_checkpoint, }; use super::transition::{ MessageTransition, MessageTransitionState, WorkflowFactTarget, WorkflowFactTransition, @@ -1128,86 +1128,7 @@ async fn reconcile_collided_observation_provenance( .await .map_err(|error| storage("remove collided projection provenance", error))?; for (output_provider, output_message_id) in affected { - conn.execute( - "DELETE FROM temp.observation_projection_output_state - WHERE projector_version = ?1 - AND output_provider = ?2 AND output_message_id = ?3", - params![ - SESSION_MESSAGE_PROJECTOR_VERSION, - output_provider.as_str(), - output_message_id.as_str(), - ], - ) - .await - .map_err(|error| storage("reset collided projection output state", error))?; - conn.execute( - "INSERT INTO temp.observation_projection_output_state ( - projector_version, output_provider, output_message_id, - canonical_observation_id, latest_observation_id, latest_sequence, - projector_owned, owner_count - ) - SELECT groups.projector_version, groups.output_provider, groups.output_message_id, - CASE WHEN groups.projector_owned = 1 THEN ( - SELECT provenance.observation_id - FROM observation_projection_provenance AS provenance - JOIN observations AS observation - ON observation.observation_id = provenance.observation_id - WHERE provenance.projector_version = groups.projector_version - AND provenance.output_provider = groups.output_provider - AND provenance.output_message_id = groups.output_message_id - ORDER BY observation.sequence DESC, provenance.observation_id DESC - LIMIT 1 - ) ELSE ( - SELECT provenance.observation_id - FROM observation_projection_provenance AS provenance - JOIN observations AS observation - ON observation.observation_id = provenance.observation_id - WHERE provenance.projector_version = groups.projector_version - AND provenance.output_provider = groups.output_provider - AND provenance.output_message_id = groups.output_message_id - ORDER BY observation.sequence ASC, provenance.observation_id ASC - LIMIT 1 - ) END, - ( - SELECT provenance.observation_id - FROM observation_projection_provenance AS provenance - JOIN observations AS observation - ON observation.observation_id = provenance.observation_id - WHERE provenance.projector_version = groups.projector_version - AND provenance.output_provider = groups.output_provider - AND provenance.output_message_id = groups.output_message_id - ORDER BY observation.sequence DESC, provenance.observation_id DESC - LIMIT 1 - ), - ( - SELECT observation.sequence - FROM observation_projection_provenance AS provenance - JOIN observations AS observation - ON observation.observation_id = provenance.observation_id - WHERE provenance.projector_version = groups.projector_version - AND provenance.output_provider = groups.output_provider - AND provenance.output_message_id = groups.output_message_id - ORDER BY observation.sequence DESC, provenance.observation_id DESC - LIMIT 1 - ), - groups.projector_owned, groups.owner_count - FROM ( - SELECT projector_version, output_provider, output_message_id, - MAX(message_created) AS projector_owned, - COUNT(*) AS owner_count - FROM observation_projection_provenance - WHERE projector_version = ?1 - AND output_provider = ?2 AND output_message_id = ?3 - GROUP BY projector_version, output_provider, output_message_id - ) AS groups", - params![ - SESSION_MESSAGE_PROJECTOR_VERSION, - output_provider.as_str(), - output_message_id.as_str(), - ], - ) - .await - .map_err(|error| storage("reaggregate collided projection output state", error))?; + reaggregate_output_state_for_output(conn, &output_provider, &output_message_id).await?; } Ok(()) } diff --git a/crates/tracedecay-global-db/src/observation_projection/state.rs b/crates/tracedecay-global-db/src/observation_projection/state.rs index 4dbe1eaff1..9c0a401382 100644 --- a/crates/tracedecay-global-db/src/observation_projection/state.rs +++ b/crates/tracedecay-global-db/src/observation_projection/state.rs @@ -360,6 +360,117 @@ pub(super) async fn read_message( })) } +/// The one definition of the projected-output ownership aggregation that +/// populates `temp.observation_projection_output_state`: for every +/// `(projector_version, output_provider, output_message_id)` group in the +/// (optionally filtered) provenance authority it derives the canonical owner +/// (newest row when the projector owns the output, oldest otherwise), the +/// newest owner row and its sequence, and the group's ownership counts. +/// Whole-cache initialization and per-output re-aggregation both render +/// their statement from this single spelling so the aggregation cannot +/// drift; `provenance_filter` scopes only the grouped rows (the correlated +/// owner lookups constrain themselves to each group's exact key). +fn output_state_aggregation_sql(provenance_filter: &str) -> String { + format!( + "INSERT INTO temp.observation_projection_output_state ( + projector_version, output_provider, output_message_id, + canonical_observation_id, latest_observation_id, latest_sequence, + projector_owned, owner_count + ) + SELECT groups.projector_version, groups.output_provider, groups.output_message_id, + CASE WHEN groups.projector_owned = 1 THEN ( + SELECT provenance.observation_id + FROM observation_projection_provenance AS provenance + JOIN observations AS observation + ON observation.observation_id = provenance.observation_id + WHERE provenance.projector_version = groups.projector_version + AND provenance.output_provider = groups.output_provider + AND provenance.output_message_id = groups.output_message_id + ORDER BY observation.sequence DESC, provenance.observation_id DESC + LIMIT 1 + ) ELSE ( + SELECT provenance.observation_id + FROM observation_projection_provenance AS provenance + JOIN observations AS observation + ON observation.observation_id = provenance.observation_id + WHERE provenance.projector_version = groups.projector_version + AND provenance.output_provider = groups.output_provider + AND provenance.output_message_id = groups.output_message_id + ORDER BY observation.sequence ASC, provenance.observation_id ASC + LIMIT 1 + ) END, + ( + SELECT provenance.observation_id + FROM observation_projection_provenance AS provenance + JOIN observations AS observation + ON observation.observation_id = provenance.observation_id + WHERE provenance.projector_version = groups.projector_version + AND provenance.output_provider = groups.output_provider + AND provenance.output_message_id = groups.output_message_id + ORDER BY observation.sequence DESC, provenance.observation_id DESC + LIMIT 1 + ), + ( + SELECT observation.sequence + FROM observation_projection_provenance AS provenance + JOIN observations AS observation + ON observation.observation_id = provenance.observation_id + WHERE provenance.projector_version = groups.projector_version + AND provenance.output_provider = groups.output_provider + AND provenance.output_message_id = groups.output_message_id + ORDER BY observation.sequence DESC, provenance.observation_id DESC + LIMIT 1 + ), + groups.projector_owned, groups.owner_count + FROM ( + SELECT projector_version, output_provider, output_message_id, + MAX(message_created) AS projector_owned, + COUNT(*) AS owner_count + FROM observation_projection_provenance + {provenance_filter} + GROUP BY projector_version, output_provider, output_message_id + ) AS groups" + ) +} + +/// Re-aggregates the ownership cache for one exact output from the +/// provenance authority: the output's cached row is removed and rebuilt +/// through the canonical aggregation ([`output_state_aggregation_sql`]), so +/// convergence paths (e.g. collided-provenance reconciliation) share the +/// initialization's single definition. +pub(super) async fn reaggregate_output_state_for_output( + conn: &impl Executor, + output_provider: &str, + output_message_id: &str, +) -> ProjectionStoreResult<()> { + conn.execute( + "DELETE FROM temp.observation_projection_output_state + WHERE projector_version = ?1 + AND output_provider = ?2 AND output_message_id = ?3", + params![ + SESSION_MESSAGE_PROJECTOR_VERSION, + output_provider, + output_message_id, + ], + ) + .await + .map_err(|error| storage("reset collided projection output state", error))?; + conn.execute( + &output_state_aggregation_sql( + "WHERE projector_version = ?1 + AND output_provider = ?2 AND output_message_id = ?3", + ), + params![ + SESSION_MESSAGE_PROJECTOR_VERSION, + output_provider, + output_message_id, + ], + ) + .await + .map_err(|error| storage("reaggregate collided projection output state", error))?; + Ok(()) +} + pub(super) struct ProjectionOutputOwner { pub(super) sequence: u64, pub(super) observation: DurableObservationV1, @@ -427,68 +538,13 @@ pub(super) async fn ensure_projection_output_state_cache( conn.execute_batch( "DELETE FROM temp.observation_projection_output_state; - DELETE FROM temp.observation_projection_output_state_meta; - WITH owner_groups AS ( - SELECT projector_version, output_provider, output_message_id, - MAX(message_created) AS projector_owned, - COUNT(*) AS owner_count - FROM observation_projection_provenance - GROUP BY projector_version, output_provider, output_message_id - ) - INSERT INTO temp.observation_projection_output_state ( - projector_version, output_provider, output_message_id, - canonical_observation_id, latest_observation_id, latest_sequence, - projector_owned, owner_count - ) - SELECT groups.projector_version, groups.output_provider, groups.output_message_id, - CASE WHEN groups.projector_owned = 1 THEN ( - SELECT provenance.observation_id - FROM observation_projection_provenance AS provenance - JOIN observations AS observation - ON observation.observation_id = provenance.observation_id - WHERE provenance.projector_version = groups.projector_version - AND provenance.output_provider = groups.output_provider - AND provenance.output_message_id = groups.output_message_id - ORDER BY observation.sequence DESC, provenance.observation_id DESC - LIMIT 1 - ) ELSE ( - SELECT provenance.observation_id - FROM observation_projection_provenance AS provenance - JOIN observations AS observation - ON observation.observation_id = provenance.observation_id - WHERE provenance.projector_version = groups.projector_version - AND provenance.output_provider = groups.output_provider - AND provenance.output_message_id = groups.output_message_id - ORDER BY observation.sequence ASC, provenance.observation_id ASC - LIMIT 1 - ) END, - ( - SELECT provenance.observation_id - FROM observation_projection_provenance AS provenance - JOIN observations AS observation - ON observation.observation_id = provenance.observation_id - WHERE provenance.projector_version = groups.projector_version - AND provenance.output_provider = groups.output_provider - AND provenance.output_message_id = groups.output_message_id - ORDER BY observation.sequence DESC, provenance.observation_id DESC - LIMIT 1 - ), - ( - SELECT observation.sequence - FROM observation_projection_provenance AS provenance - JOIN observations AS observation - ON observation.observation_id = provenance.observation_id - WHERE provenance.projector_version = groups.projector_version - AND provenance.output_provider = groups.output_provider - AND provenance.output_message_id = groups.output_message_id - ORDER BY observation.sequence DESC, provenance.observation_id DESC - LIMIT 1 - ), - groups.projector_owned, groups.owner_count - FROM owner_groups AS groups;", + DELETE FROM temp.observation_projection_output_state_meta;", ) .await .map_err(|error| storage("initialize projection output state cache", error))?; + conn.execute(&output_state_aggregation_sql(""), ()) + .await + .map_err(|error| storage("initialize projection output state cache", error))?; conn.execute( "INSERT INTO temp.observation_projection_output_state_meta(initialized, data_version) VALUES (1, ?1)", diff --git a/crates/tracedecay-global-db/src/registered.rs b/crates/tracedecay-global-db/src/registered.rs index b9ad7352f3..28bc9772aa 100644 --- a/crates/tracedecay-global-db/src/registered.rs +++ b/crates/tracedecay-global-db/src/registered.rs @@ -581,6 +581,22 @@ impl RegisteredGlobalDb { crate::GlobalDbObservationStore::new(self.database.clone()) } + /// Test-only [`Self::observation_store`] variant that binds the adapter + /// to an explicit runtime dispatch seam (e.g. a counting wrapper over + /// this client's runtime client), so tests can observe the record work a + /// persist call dispatches without changing the adapter's production + /// shape. + #[cfg(test)] + pub(crate) fn observation_store_with_runtime_dispatch( + &self, + runtime: R, + ) -> crate::GlobalDbObservationStore + where + R: crate::observation_adapter::ObservationRuntimeDispatch, + { + crate::GlobalDbObservationStore::with_runtime_dispatch(self.database.clone(), runtime) + } + /// Retains this exact client for closed runtime read/submit requests. /// /// The returned capability has no raw Store runtime, connection, or diff --git a/crates/tracedecay-global-db/src/tests/harness.rs b/crates/tracedecay-global-db/src/tests/harness.rs index 504862e60a..68bc86e033 100644 --- a/crates/tracedecay-global-db/src/tests/harness.rs +++ b/crates/tracedecay-global-db/src/tests/harness.rs @@ -476,6 +476,107 @@ pub(crate) enum SessionTemporalFixtureCountV1 { RefreshProgress, } +/// Runtime-dispatch counts observed by [`CountingObservationRuntime`]. +/// +/// `stored_observation_reads` counts stored-record runtime reads — the only +/// way a persist path can obtain, decode, collision-classify, or +/// revision-probe a retained observation row. `submits` counts runtime +/// writes — every submit is keyed by exactly one canonical command digest +/// computed immediately before dispatch, and the digest call sites are +/// unreachable without either a stored-record read or a submit. A window +/// with zero deltas on both therefore proves the persist call repeated none +/// of the expensive record work, while `source_cursor_reads` shows the seam +/// stayed live inside that same window. +#[cfg(test)] +#[derive(Debug, Default)] +pub(crate) struct ObservationDispatchCounts { + stored_observation_reads: AtomicU64, + source_cursor_reads: AtomicU64, + submits: AtomicU64, +} + +/// One consistent view of [`ObservationDispatchCounts`]. +#[cfg(test)] +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) struct ObservationDispatchSnapshot { + pub(crate) stored_observation_reads: u64, + pub(crate) source_cursor_reads: u64, + pub(crate) submits: u64, +} + +#[cfg(test)] +impl ObservationDispatchCounts { + pub(crate) fn snapshot(&self) -> ObservationDispatchSnapshot { + ObservationDispatchSnapshot { + stored_observation_reads: self.stored_observation_reads.load(Ordering::Relaxed), + source_cursor_reads: self.source_cursor_reads.load(Ordering::Relaxed), + submits: self.submits.load(Ordering::Relaxed), + } + } +} + +/// Counting test double over the production runtime client at the adapter's +/// real dispatch seam: every read and submit is forwarded unchanged to the +/// wrapped [`DatabaseRuntimeClientV1`] after classifying it into +/// [`ObservationDispatchCounts`]. +#[cfg(test)] +#[derive(Clone)] +pub(crate) struct CountingObservationRuntime { + inner: tracedecay_runtime_core::db::DatabaseRuntimeClientV1, + counts: Arc, +} + +#[cfg(test)] +impl crate::observation_adapter::ObservationRuntimeDispatch for CountingObservationRuntime { + fn binding(&self) -> &tracedecay_store::StoreRuntimeBindingV1 { + self.inner.binding() + } + + fn dispatch_read( + &self, + request: tracedecay_store::RuntimeReadRequestV1, + probe: &dyn tracedecay_store::RuntimeRequestProbeV1, + ) -> Result< + tracedecay_store::RuntimeReadOutcomeV1, + tracedecay_runtime_core::store_runtime::registry::StoreRuntimeRegistryFailure, + > { + if let tracedecay_store::RuntimeReadOperationV1::Repository { + op: + tracedecay_store::RepositoryReadOperationV1::Project( + tracedecay_store::ProjectReadOperationV1::Observation(operation), + ), + } = request.operation() + { + match operation { + tracedecay_store::ObservationReadOperationV1::Observation { .. } => { + self.counts + .stored_observation_reads + .fetch_add(1, Ordering::Relaxed); + } + tracedecay_store::ObservationReadOperationV1::SourceCursor { .. } => { + self.counts + .source_cursor_reads + .fetch_add(1, Ordering::Relaxed); + } + _ => {} + } + } + self.inner.dispatch_read(request, probe) + } + + async fn dispatch_submit( + &self, + request: tracedecay_store::RuntimeSubmitRequestV1, + probe: Arc, + ) -> Result< + tracedecay_store::RuntimeSubmitOutcomeV1, + tracedecay_runtime_core::store_runtime::registry::StoreRuntimeRegistryFailure, + > { + self.counts.submits.fetch_add(1, Ordering::Relaxed); + self.inner.dispatch_submit(request, probe).await + } +} + /// Test-only registered database fixture retained below the use-case layer. #[doc(hidden)] pub struct HostAdmissionTestRuntimeV1 { @@ -635,6 +736,29 @@ impl HostAdmissionTestRuntimeV1 { Ok(database.observation_store()) } + /// An observation store whose runtime dispatch seam is wrapped in + /// [`CountingObservationRuntime`], plus the shared counters, so a test + /// can prove exactly which record work a persist call dispatched. + #[cfg(test)] + pub(crate) fn counting_observation_store( + &self, + scope: HostAdmissionScope, + ) -> tracedecay_runtime_core::errors::Result<( + crate::GlobalDbObservationStore, + Arc, + )> { + let database = self.session_database_for_test(scope)?; + let counts = Arc::new(ObservationDispatchCounts::default()); + let runtime = CountingObservationRuntime { + inner: database.runtime_client(), + counts: Arc::clone(&counts), + }; + Ok(( + database.observation_store_with_runtime_dispatch(runtime), + counts, + )) + } + pub async fn upsert_session_for_test( &self, scope: HostAdmissionScope, diff --git a/crates/tracedecay-rusqlite-runtime/src/repository/mod.rs b/crates/tracedecay-rusqlite-runtime/src/repository/mod.rs index 7845ccb311..68a104318e 100644 --- a/crates/tracedecay-rusqlite-runtime/src/repository/mod.rs +++ b/crates/tracedecay-rusqlite-runtime/src/repository/mod.rs @@ -65,6 +65,7 @@ pub use external_source::{EXTERNAL_SOURCE_SCHEMA_V1, ExternalSourceExecutor}; pub use fact::FactExecutor; pub use graph_publication::{GRAPH_PUBLICATION_SCHEMA_V1, GraphPublicationExactSqlStorage}; pub use observation::ObservationExecutor; +pub use observation::cursor_authority as observation_cursor_authority; pub use project::ProjectExecutor; pub use retained_exact_sql::RetainedExactSqlCapability; pub use retrieval_anchor::RetrievalAnchorExecutor; diff --git a/crates/tracedecay-rusqlite-runtime/src/repository/observation/authority.rs b/crates/tracedecay-rusqlite-runtime/src/repository/observation/authority.rs index e31d210e1f..4d258d06cf 100644 --- a/crates/tracedecay-rusqlite-runtime/src/repository/observation/authority.rs +++ b/crates/tracedecay-rusqlite-runtime/src/repository/observation/authority.rs @@ -11,6 +11,9 @@ use tracedecay_store::{ }; use super::super::support::{decode, encode, invalid}; +use super::cursor_authority::{ + READ_CURSOR_ADVANCE_SQL, READ_SOURCE_CURSOR_SQL, cursor_advance_ledger_row_matches, +}; pub(super) fn persist_sanitization_receipt( connection: &rusqlite::Connection, @@ -51,8 +54,7 @@ pub(super) fn cursor_advance_receipt_matches( ) -> rusqlite::Result { let stored = connection .query_row( - "SELECT reason, receipt_id FROM source_cursor_advances - WHERE source_json = ?1 AND scope_json = ?2 AND coverage_json = ?3", + READ_CURSOR_ADVANCE_SQL, params![source_json, scope_json, encode(&advance.coverage())?], |row| Ok((row.get::<_, String>(0)?, row.get::<_, Option>(1)?)), ) @@ -60,9 +62,11 @@ pub(super) fn cursor_advance_receipt_matches( let expected_receipt_id = advance .sanitization_receipt() .map(|receipt| receipt.receipt().receipt_id().as_str()); - if stored.as_ref().is_none_or(|(reason, receipt_id)| { - reason != advance.reason().as_str() || receipt_id.as_deref() != expected_receipt_id - }) { + if !cursor_advance_ledger_row_matches( + stored.as_ref(), + advance.reason().as_str(), + expected_receipt_id, + ) { return Ok(false); } if let Some(receipt) = advance.sanitization_receipt() { @@ -269,8 +273,7 @@ pub(super) fn read_cursor( ) -> rusqlite::Result> { connection .query_row( - "SELECT cursor_json FROM source_cursors - WHERE source_json = ?1 AND scope_json = ?2", + READ_SOURCE_CURSOR_SQL, params![source_json, scope_json], |row| row.get::<_, String>(0), ) diff --git a/crates/tracedecay-rusqlite-runtime/src/repository/observation/cursor_authority.rs b/crates/tracedecay-rusqlite-runtime/src/repository/observation/cursor_authority.rs new file mode 100644 index 0000000000..60f70673a5 --- /dev/null +++ b/crates/tracedecay-rusqlite-runtime/src/repository/observation/cursor_authority.rs @@ -0,0 +1,52 @@ +//! The one canonical spelling of "advance an observation source cursor". +//! +//! Every executor of the cursor-advance authority — the runtime write +//! command path ([`super::ObservationExecutor`]) and the global-db +//! observation adapter's atomic refusal-marker + coverage transaction — +//! reads, records, verifies, and commits through exactly this statement set, +//! so the authority cannot drift into parallel spellings. The statements are +//! transport-neutral text: one caller binds them on the runtime writer's +//! rusqlite savepoint, the other on the engine's guarded write transaction. + +/// Durable cursor for one source: params `(source_json, scope_json)`, +/// column `cursor_json`. +pub const READ_SOURCE_CURSOR_SQL: &str = "SELECT cursor_json FROM source_cursors + WHERE source_json = ?1 AND scope_json = ?2"; + +/// Idempotent advance-ledger insert: params `(source_json, scope_json, +/// coverage_json, reason, receipt_id)`. A replay of the same coverage key is +/// a no-op; the read-back verification decides whether the retained row is +/// this advance or a collision. +pub const RECORD_CURSOR_ADVANCE_SQL: &str = "INSERT INTO source_cursor_advances ( + source_json, scope_json, coverage_json, reason, receipt_id + ) VALUES (?1, ?2, ?3, ?4, ?5) + ON CONFLICT(source_json, scope_json, coverage_json) DO NOTHING"; + +/// Advance-ledger read-back for in-transaction verification: params +/// `(source_json, scope_json, coverage_json)`, columns `(reason, +/// receipt_id)`. +pub const READ_CURSOR_ADVANCE_SQL: &str = "SELECT reason, receipt_id FROM source_cursor_advances + WHERE source_json = ?1 AND scope_json = ?2 AND coverage_json = ?3"; + +/// Moves the durable cursor to the advance's next position: params +/// `(source_json, scope_json, cursor_json)`. +pub const COMMIT_SOURCE_CURSOR_SQL: &str = + "INSERT INTO source_cursors (source_json, scope_json, cursor_json) + VALUES (?1, ?2, ?3) + ON CONFLICT(source_json, scope_json) DO UPDATE SET + cursor_json = excluded.cursor_json"; + +/// Whether one [`READ_CURSOR_ADVANCE_SQL`] row is exactly this advance's +/// row — the same reason and the same (possibly absent) sanitization receipt +/// id. Any other row retained under the coverage key is a cursor-advance +/// collision. +#[must_use] +pub fn cursor_advance_ledger_row_matches( + stored: Option<&(String, Option)>, + reason: &str, + receipt_id: Option<&str>, +) -> bool { + stored.is_some_and(|(stored_reason, stored_receipt)| { + stored_reason == reason && stored_receipt.as_deref() == receipt_id + }) +} diff --git a/crates/tracedecay-rusqlite-runtime/src/repository/observation/mod.rs b/crates/tracedecay-rusqlite-runtime/src/repository/observation/mod.rs index 34fac67fa8..c94f433874 100644 --- a/crates/tracedecay-rusqlite-runtime/src/repository/observation/mod.rs +++ b/crates/tracedecay-rusqlite-runtime/src/repository/observation/mod.rs @@ -19,12 +19,14 @@ use tracedecay_store::{ use super::support::{decode, encode, invalid}; mod authority; +pub mod cursor_authority; mod rows; use authority::{ cursor_advance_receipt_matches, persist_repository_provenance, persist_retrieval_anchor, persist_sanitization_receipt, read_cursor, verify_observation_authority, }; +use cursor_authority::{COMMIT_SOURCE_CURSOR_SQL, RECORD_CURSOR_ADVANCE_SQL}; use rows::{ OBSERVATION_ROW_PROJECTION, decode_nonnegative, decode_observation_row, encoded_observation_row, }; @@ -146,10 +148,7 @@ impl ObservationExecutor { write.repository_provenance_attachment(), )?; savepoint.execute( - "INSERT INTO source_cursors (source_json, scope_json, cursor_json) - VALUES (?1, ?2, ?3) - ON CONFLICT(source_json, scope_json) DO UPDATE SET - cursor_json = excluded.cursor_json", + COMMIT_SOURCE_CURSOR_SQL, params![source_json, scope_json, committed_cursor_json], )?; savepoint.execute( @@ -182,10 +181,7 @@ impl ObservationExecutor { } let coverage_json = encode(&advance.coverage())?; savepoint.execute( - "INSERT INTO source_cursor_advances ( - source_json, scope_json, coverage_json, reason, receipt_id - ) VALUES (?1, ?2, ?3, ?4, ?5) - ON CONFLICT(source_json, scope_json, coverage_json) DO NOTHING", + RECORD_CURSOR_ADVANCE_SQL, params![ source_json, scope_json, @@ -200,10 +196,7 @@ impl ObservationExecutor { return Err(invalid("source cursor advance identity collision")); } savepoint.execute( - "INSERT INTO source_cursors (source_json, scope_json, cursor_json) - VALUES (?1, ?2, ?3) - ON CONFLICT(source_json, scope_json) DO UPDATE SET - cursor_json = excluded.cursor_json", + COMMIT_SOURCE_CURSOR_SQL, params![source_json, scope_json, encode(advance.next_cursor())?], )?; Ok(())