diff --git a/crates/tracedecay-global-db/src/observation/schema.rs b/crates/tracedecay-global-db/src/observation/schema.rs index 1b9b2a8867..e5144f8712 100644 --- a/crates/tracedecay-global-db/src/observation/schema.rs +++ b/crates/tracedecay-global-db/src/observation/schema.rs @@ -237,6 +237,12 @@ const OBSERVATION_AUTHORITY_SCHEMA_SQL: &str = "CREATE TABLE IF NOT EXISTS sanit attempt_count INTEGER NOT NULL DEFAULT 0 CHECK(attempt_count >= 0), next_retry_at_micros INTEGER NOT NULL DEFAULT 0 CHECK(next_retry_at_micros >= 0), last_error TEXT, + revival_generation TEXT, + revival_start_offset INTEGER CHECK( + (revival_generation IS NULL AND revival_start_offset IS NULL) + OR (revival_generation IS NOT NULL AND revival_start_offset IS NOT NULL + AND revival_start_offset >= 0) + ), FOREIGN KEY(observation_id) REFERENCES observations(observation_id) ); CREATE INDEX IF NOT EXISTS idx_retrieval_anchor_dispositions_release_due @@ -263,6 +269,39 @@ pub async fn ensure_observation_schema( conn.execute_batch(OBSERVATION_AUTHORITY_SCHEMA_SQL) .await .map_err(|error| global_db_operation_error(OBSERVATION_SCHEMA_OPERATION, error))?; + for (column, alteration) in [ + ( + "revival_generation", + "ALTER TABLE projection_queue ADD COLUMN revival_generation TEXT", + ), + ( + "revival_start_offset", + "ALTER TABLE projection_queue ADD COLUMN revival_start_offset INTEGER CHECK( + (revival_generation IS NULL AND revival_start_offset IS NULL) + OR (revival_generation IS NOT NULL AND revival_start_offset IS NOT NULL + AND revival_start_offset >= 0) + )", + ), + ] { + let mut rows = conn + .query( + "SELECT 1 FROM pragma_table_xinfo('projection_queue') WHERE name = ?1", + [column], + ) + .await + .map_err(|error| global_db_operation_error(OBSERVATION_SCHEMA_OPERATION, error))?; + let present = rows + .next() + .await + .map_err(|error| global_db_operation_error(OBSERVATION_SCHEMA_OPERATION, error))? + .is_some(); + drop(rows); + if !present { + conn.execute(alteration, ()) + .await + .map_err(|error| global_db_operation_error(OBSERVATION_SCHEMA_OPERATION, error))?; + } + } if !table_preexisted { for migration in [ OBSERVATION_SCHEMA_MIGRATION, diff --git a/crates/tracedecay-global-db/src/observation_adapter.rs b/crates/tracedecay-global-db/src/observation_adapter.rs index 648f47944a..1c792d359b 100644 --- a/crates/tracedecay-global-db/src/observation_adapter.rs +++ b/crates/tracedecay-global-db/src/observation_adapter.rs @@ -459,82 +459,7 @@ impl GlobalDbObservationStore { )); } if canonical_payload_revision { - let Some(existing) = existing.as_ref() else { - return Err(runtime_storage_error( - "persist canonical payload revision", - "classified revision replay has no retained observation", - )); - }; - let identity = observation.identity(); - let actual_cursor = actual_cursor(); - let revision_covered = actual_cursor.as_ref().is_some_and(|cursor| { - cursor.generation() == identity.generation() - && cursor.ordering_domain() == identity.ordering_domain() - && cursor.position() >= identity.position().end() - }); - if revision_covered { - return Ok(PreparedObservationPersist::ready( - ObservationPersistOutcome::CoveredDuplicate( - ObservationCommitReceipt::new( - existing.sequence(), - existing.observation().clone(), - write.next_cursor().clone(), - existing.retrieval_anchor().clone(), - existing.projection_generation().clone(), - )? - .with_repository_provenance_attachment( - existing.repository_provenance_attachment().clone(), - )?, - ), - existing, - )); - } - if known_cursor.is_some() { - return Ok(PreparedObservationPersist::AwaitsDurablePredecessor( - Box::new(write), - )); - } - let mut advance = ObservationCursorAdvance::for_ordering_with_sanitization_receipt( - identity.source().clone(), - identity.scope().clone(), - identity.generation(), - identity.ordering_domain(), - write.expected_cursor().cloned(), - identity.position(), - ObservationCoverageReason::CanonicalPayloadRevision, - observation.receipt().clone(), - )?; - match ( - write.next_cursor().file_identity(), - write.next_cursor().resume_fingerprint(), - ) { - (Some(file_identity), Some(resume_fingerprint)) => { - advance = advance.with_resume_checkpoint(file_identity, resume_fingerprint); - } - (None, None) => {} - _ => { - return Err(runtime_storage_error( - "persist canonical payload revision", - "cursor resume checkpoint is incomplete", - )); - } - } - self.advance_source_cursor(advance).await?; - return Ok(PreparedObservationPersist::ready( - ObservationPersistOutcome::CoveredDuplicate( - ObservationCommitReceipt::new( - existing.sequence(), - existing.observation().clone(), - write.next_cursor().clone(), - existing.retrieval_anchor().clone(), - existing.projection_generation().clone(), - )? - .with_repository_provenance_attachment( - existing.repository_provenance_attachment().clone(), - )?, - ), - existing, - )); + return Ok(PreparedObservationPersist::Submit(Box::new(write))); } let same_identity = pending .as_ref() @@ -607,23 +532,33 @@ impl GlobalDbObservationStore { retained.observation().receipt() == observation.receipt() })); if existed_exact { - if pending.is_some() { - return Ok(PreparedObservationPersist::DeferredExactDuplicate( - Box::new(write), + // An exact retry retains its first accepted derived authority; + // later projection clocks or Git captures cannot replace it. The + // writer still serializes source revival against retirement. + let (anchor, provenance) = if let Some(retained) = pending.as_ref() { + (&retained.retrieval_anchor, &retained.repository_provenance) + } else if let Some(retained) = existing { + ( + retained.retrieval_anchor(), + retained.repository_provenance_attachment(), + ) + } else { + return Err(runtime_storage_error( + "persist exact observation retry", + "exact duplicate has no retained observation authority", )); - } - let Some(existing) = existing.as_ref() else { - return Err(ObservationStoreError::Storage { - operation: "persist_observation", - source: Box::new(std::io::Error::other( - "exact duplicate classification requires a stored observation", - )), - }); }; - return Ok(PreparedObservationPersist::ready( - ObservationPersistOutcome::ExactDuplicate(existing.commit_receipt().clone()), - existing, - )); + let replay = AnchoredObservationWrite::new( + write.write().clone(), + anchor.clone(), + anchor.projection_generation().clone(), + )? + .with_repository_provenance_attachment( + provenance.availability().clone(), + provenance.anchor().cloned(), + )? + .with_identity_collision_disposition(write.identity_collision_disposition()); + return Ok(PreparedObservationPersist::ExactDuplicate(Box::new(replay))); } if pending.is_none() && existing.is_none() { batch_state.register_pending(&write)?; @@ -667,7 +602,7 @@ impl GlobalDbObservationStore { let mut outcomes: Vec> = Vec::with_capacity(writes.len()); let mut submits = Vec::new(); - let mut deferred_exact_duplicates = Vec::new(); + let mut exact_duplicates = HashSet::new(); let mut awaiting = Vec::new(); let mut writes = writes.into_iter(); while let Some(write) = writes.next() { @@ -681,13 +616,13 @@ impl GlobalDbObservationStore { .prepare_observation_persist(write, &preflight, &mut batch_state, known_cursor) .await? { - PreparedObservationPersist::Ready(outcome) => outcomes.push(Some(*outcome)), PreparedObservationPersist::Submit(write) => { submits.push((outcomes.len(), *write)); outcomes.push(None); } - PreparedObservationPersist::DeferredExactDuplicate(write) => { - deferred_exact_duplicates.push((outcomes.len(), *write)); + PreparedObservationPersist::ExactDuplicate(write) => { + exact_duplicates.insert(outcomes.len()); + submits.push((outcomes.len(), *write)); outcomes.push(None); } PreparedObservationPersist::AwaitsDurablePredecessor(write) => { @@ -705,21 +640,19 @@ impl GlobalDbObservationStore { published_cursors.insert(key, next_cursor); } if !submits.is_empty() { - let submitted = submit_observation_writes( - &self.database, - &self.runtime, - submits, - deferred_exact_duplicates, - ) - .await?; + let submitted = + submit_observation_writes(&self.database, &self.runtime, submits).await?; for (slot, outcome) in submitted { - outcomes[slot] = Some(outcome); + outcomes[slot] = Some(if exact_duplicates.contains(&slot) { + let (outcome, stored) = outcome.into_parts(); + ObservationBatchPersistOutcome::new( + ObservationPersistOutcome::ExactDuplicate(outcome.receipt().clone()), + stored, + ) + } else { + outcome + }); } - } else if !deferred_exact_duplicates.is_empty() { - return Err(runtime_storage_error( - "persist_observations", - "deferred duplicate has no preceding batch submission", - )); } let outcomes = outcomes .into_iter() @@ -857,6 +790,8 @@ struct PendingObservationAuthority { payload_reference: PayloadReferenceV1, identity: ObservationIdentityMaterialV1, receipt: SanitizationReceiptV1, + retrieval_anchor: RetrievalAnchorRecord, + repository_provenance: RepositoryProvenanceAttachmentV1, } #[derive(Clone)] @@ -923,6 +858,8 @@ impl ObservationBatchState { payload_reference: write.observation().payload_reference().clone(), identity: write.observation().identity().clone(), receipt: write.observation().receipt().clone(), + retrieval_anchor: write.retrieval_anchor().clone(), + repository_provenance: write.repository_provenance_attachment().clone(), }, ); self.pending_receipts.insert( @@ -1522,28 +1459,13 @@ fn decode_json( } enum PreparedObservationPersist { - // Both payloads are large and only one is ever live, so both are boxed: - // leaving either inline makes every value of this enum carry that variant's - // footprint. - Ready(Box), Submit(Box), - DeferredExactDuplicate(Box), - /// The write collides with, or must compare-and-set a source cursor - /// published by, an earlier member of this batch that is not durable yet. - /// The batch commits its prefix first and re-prepares this write against - /// the durable result, so it settles exactly as it would alone. + ExactDuplicate(Box), + /// The write collides with an earlier member of this batch that is not + /// durable yet. Commit the prefix before classifying the collision. AwaitsDurablePredecessor(Box), } -impl PreparedObservationPersist { - fn ready(outcome: ObservationPersistOutcome, stored: &StoredObservation) -> Self { - Self::Ready(Box::new(ObservationBatchPersistOutcome::new( - outcome, - Some(stored.clone()), - ))) - } -} - impl ObservationStore for GlobalDbObservationStore { async fn persist_observation( &self, @@ -2041,19 +1963,10 @@ async fn submit_observation_writes( database: &Database, runtime: &DatabaseRuntimeClientV1, writes: Vec<(usize, AnchoredObservationWrite)>, - deferred_exact_duplicates: Vec<(usize, AnchoredObservationWrite)>, ) -> ObservationStoreResult> { - let batches = partition_observation_writes(writes)?; - let mut deferred_exact_duplicates = Some(deferred_exact_duplicates); let mut outcomes = Vec::new(); - let batch_count = batches.len(); - for (index, batch) in batches.into_iter().enumerate() { - let deferred = if index + 1 == batch_count { - deferred_exact_duplicates.take().unwrap_or_default() - } else { - Vec::new() - }; - outcomes.extend(submit_observation_write_batch(database, runtime, batch, deferred).await?); + for batch in partition_observation_writes(writes)? { + outcomes.extend(submit_observation_write_batch(database, runtime, batch).await?); } Ok(outcomes) } @@ -2115,7 +2028,6 @@ async fn submit_observation_write_batch( database: &Database, runtime: &DatabaseRuntimeClientV1, writes: Vec<(usize, AnchoredObservationWrite)>, - deferred_exact_duplicates: Vec<(usize, AnchoredObservationWrite)>, ) -> ObservationStoreResult> { let admitted_at = now_micros(); let command = serde_json::json!({ @@ -2191,7 +2103,6 @@ async fn submit_observation_write_batch( const READBACK_OPERATION: &str = "read committed observation batch"; let observation_ids = writes .iter() - .chain(deferred_exact_duplicates.iter()) .map(|(_, write)| write.observation().observation_id().clone()) .collect::>(); let snapshot = begin_observation_read_snapshot(database, READBACK_OPERATION).await?; @@ -2202,7 +2113,7 @@ async fn submit_observation_write_batch( .commit() .await .map_err(|error| runtime_storage_error(READBACK_OPERATION, error))?; - let mut outcomes = writes + let outcomes = writes .into_iter() .map(|(slot, write)| { let retained = stored @@ -2224,20 +2135,6 @@ async fn submit_observation_write_batch( }) }) .collect::>>()?; - for (slot, write) in deferred_exact_duplicates { - let retained = stored - .get(write.observation().observation_id().as_str()) - .ok_or_else(|| { - runtime_storage_error("read committed observation", "row unavailable") - })?; - outcomes.push(( - slot, - ObservationBatchPersistOutcome::new( - ObservationPersistOutcome::ExactDuplicate(retained.commit_receipt().clone()), - Some(retained.clone()), - ), - )); - } Ok(outcomes) } @@ -2258,9 +2155,10 @@ fn persist_outcome_from_submit( RuntimeSubmitOutcomeV1::Committed { .. } | RuntimeSubmitOutcomeV1::CommittedAfterCancellation { .. } | RuntimeSubmitOutcomeV1::ExactReplay { .. } - if stored.observation().identity() != candidate.identity() - && classify_observation_collision(stored.observation(), candidate) - == ObservationCollisionOutcomeV1::ExactDuplicate => + if is_canonical_payload_revision_replay(stored.observation(), candidate) + || (stored.observation().identity() != candidate.identity() + && classify_observation_collision(stored.observation(), candidate) + == ObservationCollisionOutcomeV1::ExactDuplicate) => { Ok(ObservationPersistOutcome::CoveredDuplicate( ObservationCommitReceipt::new( diff --git a/crates/tracedecay-global-db/src/observation_batch_tests.rs b/crates/tracedecay-global-db/src/observation_batch_tests.rs index 3a6aae3899..b5972b6807 100644 --- a/crates/tracedecay-global-db/src/observation_batch_tests.rs +++ b/crates/tracedecay-global-db/src/observation_batch_tests.rs @@ -251,11 +251,14 @@ fn anchored_write_under( AnchoredObservationWrite::new(write, anchor, projection_generation).unwrap() } -fn sequential_writes(session_id: &SessionId, count: usize) -> Vec { +pub(crate) fn sequential_writes( + session_id: &SessionId, + count: usize, +) -> Vec { sequential_writes_with_text(session_id, count, |ordinal| format!("frame {ordinal}")) } -fn sequential_writes_with_text( +pub(crate) fn sequential_writes_with_text( session_id: &SessionId, count: usize, text: impl Fn(u64) -> String, @@ -431,10 +434,21 @@ async fn intra_batch_exact_duplicate_is_hydrated_as_a_duplicate() { let first = sequential_writes(&session_id, 1) .pop() .expect("intra-batch exact observation"); + let retry_projection = ProjectionGenerationId::new("projection.retry.batch.v2").unwrap(); + let retry_anchor = tracedecay_store::build_observation_retrieval_anchor( + first.observation(), + retry_projection.clone(), + UtcMicros(2), + first.retrieval_anchor().authorization().clone(), + ) + .unwrap(); + let retry = + AnchoredObservationWrite::new(first.write().clone(), retry_anchor, retry_projection) + .unwrap(); let before = writer_txn_census(&runtime).await; let outcomes = store - .persist_observations(vec![first.clone(), first.clone()]) + .persist_observations(vec![first.clone(), retry, first.clone()]) .await .unwrap(); @@ -446,7 +460,12 @@ async fn intra_batch_exact_duplicate_is_hydrated_as_a_duplicate() { outcomes[1].outcome(), ObservationPersistOutcome::ExactDuplicate(_) )); + assert!(matches!( + outcomes[2].outcome(), + ObservationPersistOutcome::ExactDuplicate(_) + )); assert_eq!(outcomes[0].stored(), outcomes[1].stored()); + assert_eq!(outcomes[0].stored(), outcomes[2].stored()); let after = writer_txn_census(&runtime).await; assert_eq!(after.operations - before.operations, 1); } diff --git a/crates/tracedecay-global-db/src/observation_collision_tests.rs b/crates/tracedecay-global-db/src/observation_collision_tests.rs index b248d6e83e..7bf3fd307b 100644 --- a/crates/tracedecay-global-db/src/observation_collision_tests.rs +++ b/crates/tracedecay-global-db/src/observation_collision_tests.rs @@ -64,6 +64,7 @@ use tracing::field::{Field, Visit}; use tracing::span::{Attributes, Id, Record}; use tracing::{Event, Metadata, Subscriber}; +use crate::ObservationSourcePresenceV1; use crate::schema_contract::invariants::SOURCE_CURSOR_ADVANCE_DELETE_GUARD_SQL; use crate::tests::harness::{HostAdmissionScope, HostAdmissionTestRuntimeV1}; use tracedecay_runtime_core::db::engine::params; @@ -2255,6 +2256,117 @@ async fn canonical_payload_revision_replay_survives_an_earlier_refusal() { assert_eq!(stored.observation().payload(), legacy.payload()); } +#[asupersync::test] +async fn an_accepted_canonical_payload_revision_revives_a_retired_record() { + let tmp = TempDir::new().unwrap(); + let runtime = HostAdmissionTestRuntimeV1::profile(tmp.path()) + .await + .unwrap(); + let registered = runtime + .registered_database(HostAdmissionScope::Profile) + .unwrap(); + let store = registered.observation_store(); + let session_id = SessionId::new("session.codex.revision-revival").unwrap(); + let legacy = codex_revision_observation( + &session_id, + 1, + "receipt.codex.revival.legacy", + true, + "stable authored content", + ); + registered + .record_observation_source_presence( + legacy.source(), + legacy.scope(), + &[ObservationSourcePresenceV1 { + observation_id: legacy.observation_id().clone(), + generation: legacy.identity().generation(), + start_offset: 0, + }], + ) + .await + .unwrap(); + store + .persist_observation(anchored_write_for(legacy.clone(), None)) + .await + .unwrap(); + store + .project_observation(legacy.observation_id()) + .await + .unwrap(); + assert_eq!(table_count(&runtime, "lcm_raw_messages").await, 1); + let retired_generation = ObservationSourceGenerationV1::new(2).unwrap(); + registered + .begin_observation_source_rewrite( + legacy.source(), + legacy.scope(), + legacy.identity().generation(), + retired_generation, + 0, + ) + .await + .unwrap(); + assert_eq!( + registered + .complete_observation_source_rewrite( + legacy.source(), + legacy.scope(), + retired_generation, + ) + .await + .unwrap(), + 1 + ); + assert_eq!(table_count(&runtime, "lcm_raw_messages").await, 0); + + let revision = codex_revision_observation( + &session_id, + 3, + "receipt.codex.revival.current", + false, + "stable authored content", + ); + registered + .record_observation_source_presence( + revision.source(), + revision.scope(), + &[ObservationSourcePresenceV1 { + observation_id: revision.observation_id().clone(), + generation: revision.identity().generation(), + start_offset: 0, + }], + ) + .await + .unwrap(); + let expected = store + .get_source_cursor(legacy.source(), legacy.scope()) + .await + .unwrap(); + let outcome = store + .persist_observation(anchored_write_for(revision.clone(), expected)) + .await + .unwrap(); + assert!(matches!( + outcome, + ObservationPersistOutcome::CoveredDuplicate(_) + )); + assert_eq!(table_count(&runtime, "lcm_raw_messages").await, 0); + store + .project_observation(legacy.observation_id()) + .await + .unwrap(); + assert_eq!(table_count(&runtime, "lcm_raw_messages").await, 1); + assert_eq!( + store + .get_observation(legacy.observation_id()) + .await + .unwrap() + .unwrap() + .observation(), + &legacy + ); +} + /// Atomicity gate: the refusal marker commits before its cursor advance, so a /// failure between the two, the injected cursor-advance failure state, here /// seeded durably as exactly what such a crash leaves behind, produces a diff --git a/crates/tracedecay-global-db/src/observation_projection/apply.rs b/crates/tracedecay-global-db/src/observation_projection/apply.rs index b30fdaadb7..55f91b7189 100644 --- a/crates/tracedecay-global-db/src/observation_projection/apply.rs +++ b/crates/tracedecay-global-db/src/observation_projection/apply.rs @@ -1954,6 +1954,14 @@ pub(in super::super) async fn retire_source_record( conn: &(impl Executor + SendProjectionDerive), observation_id: &str, ) -> ProjectionStoreResult { + conn.execute( + "UPDATE projection_queue + SET revival_generation = NULL, revival_start_offset = NULL + WHERE observation_id = ?1", + params![observation_id], + ) + .await + .map_err(|error| storage("cancel accepted source offer", error))?; let mut rows = conn .query( "SELECT observation.receipt_id, @@ -1985,6 +1993,16 @@ pub(in super::super) async fn retire_source_record( if settled { return Ok(false); } + // Activation must not restore a rendering this retirement removes. + conn.execute( + "DELETE FROM observation_projection_rebuilds + WHERE projector_version = ?1 AND staged_through >= ( + SELECT sequence FROM observations WHERE observation_id = ?2 + )", + params![SESSION_MESSAGE_PROJECTOR_VERSION, observation_id], + ) + .await + .map_err(|error| storage("invalidate retired source rebuild", error))?; let mut rows = conn .query( "SELECT DISTINCT output_provider, output_message_id diff --git a/crates/tracedecay-global-db/src/observation_projection/rebuild.rs b/crates/tracedecay-global-db/src/observation_projection/rebuild.rs index e225c63ca3..63203391f9 100644 --- a/crates/tracedecay-global-db/src/observation_projection/rebuild.rs +++ b/crates/tracedecay-global-db/src/observation_projection/rebuild.rs @@ -23,18 +23,19 @@ use super::apply::{ derive_projection_with_alias, stage_provider_usage_effects, verify_effect, }; use super::state::{ - canonical_message_location, canonicalize_session_project_paths, consume_projection_queue_item, - decode_observation_row, decode_sequence, projected_stored_message, projection_retry_state, - queued_sequence, read_checkpoint, read_message, read_observation, read_session, - reconcile_session_rows_detailed, schedule_projection_retry, storage, storage_message, - stored_output_digest, write_checkpoint, + AcceptedSourceOffer, canonical_message_location, canonicalize_session_project_paths, + consume_projection_queue_item, decode_observation_row, decode_sequence, + projected_stored_message, projection_retry_state, queued_projection, read_checkpoint, + read_message, read_observation, read_session, reconcile_session_rows_detailed, + schedule_projection_retry, storage, storage_message, stored_output_digest, write_checkpoint, }; use super::transition::{ MessageTransition, MessageTransitionState, WorkflowFactTarget, WorkflowFactTransition, message_transition, write_workflow_fact_transition, }; use tracedecay_session_temporal_store::{ - record_canonical_observation_effect, request_session_temporal_reset, + record_canonical_observation_effect, + record_live_collision_preserving_canonical_observation_effect, request_session_temporal_reset, }; pub(super) const REBUILD_PAGE_SIZE: i64 = 128; @@ -376,7 +377,11 @@ async fn persist_projection_rejection_on_database( actual: sequence, }); } - if queued_sequence(&transaction, observation_id).await? != Some(sequence) { + if queued_projection(&transaction, observation_id) + .await? + .map(|queued| queued.sequence) + != Some(sequence) + { return Err(ProjectionStoreError::NotQueued); } apply_skip_disposition(&transaction, &observation, reason).await?; @@ -533,51 +538,73 @@ async fn advance_projection_rebuild( frontier_sequence: u64, cancelled: &AtomicBool, ) -> ProjectionStoreResult { - let job = read_rebuild_job(&database.read_connection()).await?; - match job.state { - RebuildState::Aliasing => { - let transaction = database - .begin_write_transaction("begin projection alias staging") - .await - .map_err(|error| storage("begin projection alias staging", error))?; - stage_projection_alias_batch_transaction(&transaction).await?; - transaction - .commit() - .await - .map_err(|error| storage("commit projection alias batch", error))?; - Ok(RebuildAdvance::Pending) + let transaction = database + .begin_write_transaction("begin projection rebuild advance") + .await + .map_err(|error| storage("begin projection rebuild advance", error))?; + let mut rows = transaction + .query( + "SELECT observation_id FROM projection_queue + WHERE revival_generation IS NOT NULL + ORDER BY observation_sequence LIMIT ?1", + params![REBUILD_PAGE_SIZE], + ) + .await + .map_err(|error| storage("read accepted source offers for rebuild", error))?; + let mut revivals = Vec::new(); + while let Some(row) = rows + .next() + .await + .map_err(|error| storage("read accepted source offers for rebuild", error))? + { + revivals.push( + CanonicalObservationIdV1::new( + row.get::(0) + .map_err(|error| storage("read accepted source offer identity", error))?, + ) + .map_err(ProjectionStoreError::Contract)?, + ); + } + drop(rows); + for observation_id in &revivals { + if cancelled.load(Ordering::Acquire) { + break; } - RebuildState::Building => { - let transaction = database - .begin_write_transaction("begin projection rebuild batch") - .await - .map_err(|error| storage("begin projection rebuild batch", error))?; - let outcome = + let queued = queued_projection(&transaction, observation_id) + .await? + .ok_or(ProjectionStoreError::NotQueued)?; + let offer = queued.revival.ok_or_else(|| { + storage_message("consume accepted source offer", "queued offer has no fence") + })?; + let (_, observation) = read_observation(&transaction, observation_id) + .await? + .ok_or(ProjectionStoreError::ObservationNotFound)?; + Box::pin(consume_source_revival(&transaction, &observation, &offer)).await?; + } + let outcome = if revivals.len() == REBUILD_PAGE_SIZE as usize + || cancelled.load(Ordering::Acquire) + { + RebuildAdvance::Pending + } else { + match read_rebuild_job(&transaction).await?.state { + RebuildState::Aliasing => { + stage_projection_alias_batch_transaction(&transaction).await?; + RebuildAdvance::Pending + } + RebuildState::Building => { stage_projection_rebuild_batch_transaction(&transaction, cancelled).await?; - let commit_operation = match outcome { - RebuildBatchStage::AlreadyReady => "commit completed projection rebuild batch", - RebuildBatchStage::Advanced => "commit projection rebuild batch", - }; - transaction - .commit() - .await - .map_err(|error| storage(commit_operation, error))?; - Ok(RebuildAdvance::Pending) - } - RebuildState::Ready => { - let transaction = database - .begin_write_transaction("begin projection rebuild activation") - .await - .map_err(|error| storage("begin projection rebuild activation", error))?; - let outcome = - activate_projection_rebuild_transaction(&transaction, frontier_sequence).await?; - transaction - .commit() - .await - .map_err(|error| storage("commit projection rebuild activation", error))?; - Ok(RebuildAdvance::Complete(outcome)) + RebuildAdvance::Pending + } + RebuildState::Ready => RebuildAdvance::Complete( + activate_projection_rebuild_transaction(&transaction, frontier_sequence).await?, + ), } - } + }; + transaction + .commit() + .await + .map_err(|error| storage("commit projection rebuild advance", error))?; + Ok(outcome) } async fn project_observation_in_transaction( @@ -712,21 +739,17 @@ async fn project_observation_in_transaction_with_session( return Err(ProjectionStoreError::ObservationNotFound); }; let session_id = observation.source().session_id().as_str().to_owned(); + let queued = queued_projection(transaction, observation_id).await?; + if let Some(offer) = queued.as_ref().and_then(|queued| queued.revival.as_ref()) { + Box::pin(consume_source_revival(transaction, &observation, offer)).await?; + } let mut effect = derive_projection_with_alias(transaction, &observation).await?; if sequence <= checkpoint.last_sequence() { verify_effect(transaction, &observation, &effect).await?; - if !matches!( - effect, - ObservationProjection::Skipped( - ProjectionSkipReason::InvalidContract - | ProjectionSkipReason::NativeSourceSuperseded - | ProjectionSkipReason::SourceRecordRetired - ) - ) { - record_canonical_observation_effect(transaction, sequence, &observation, &effect) - .await?; + record_canonical_effect(transaction, sequence, &observation, &effect).await?; + if !queue_row_is_terminal_collision(transaction, observation_id).await? { + consume_projection_queue_item(transaction, observation_id).await?; } - consume_projection_queue_item(transaction, observation_id).await?; return Ok(( ProjectionPersistOutcome::ExactDuplicate(checkpoint), session_id, @@ -739,7 +762,7 @@ async fn project_observation_in_transaction_with_session( actual: sequence, }); } - if queued_sequence(transaction, observation_id).await? != Some(sequence) { + if queued.as_ref().map(|queued| queued.sequence) != Some(sequence) { return Err(ProjectionStoreError::NotQueued); } @@ -753,16 +776,7 @@ async fn project_observation_in_transaction_with_session( &mut effect, )) .await?; - if !matches!( - effect, - ObservationProjection::Skipped( - ProjectionSkipReason::InvalidContract - | ProjectionSkipReason::NativeSourceSuperseded - | ProjectionSkipReason::SourceRecordRetired - ) - ) { - record_canonical_observation_effect(transaction, sequence, &observation, &effect).await?; - } + record_canonical_effect(transaction, sequence, &observation, &effect).await?; // A genuine session collision stays on its queue row with last_error set // and a terminal deadline. The checkpoint still advances, and head // selection skips that row, so later observations project. @@ -782,6 +796,195 @@ async fn project_observation_in_transaction_with_session( Ok((outcome, session_id)) } +async fn consume_source_revival( + transaction: &DatabaseWriteTransaction<'_>, + observation: &DurableObservationV1, + offer: &AcceptedSourceOffer, +) -> ProjectionStoreResult<()> { + let source_key = serde_json::to_string(&(observation.source(), observation.scope())) + .map_err(|error| storage("encode accepted source offer", error))?; + let mut rows = transaction + .query( + "SELECT 1 FROM observation_source_presence + WHERE source_key = ?1 AND observation_id = ?2 + AND generation = ?3 AND start_offset = ?4", + params![ + source_key, + observation.observation_id().as_str(), + offer.generation.generation_id().to_string(), + sequence_i64(offer.start_offset)?, + ], + ) + .await + .map_err(|error| storage("validate accepted source offer", error))?; + let current = rows + .next() + .await + .map_err(|error| storage("validate accepted source offer", error))? + .is_some(); + drop(rows); + if current { + revive_source_record(transaction, observation.observation_id()).await?; + } + transaction + .execute( + "UPDATE projection_queue + SET revival_generation = NULL, revival_start_offset = NULL + WHERE observation_id = ?1", + params![observation.observation_id().as_str()], + ) + .await + .map_err(|error| storage("consume accepted source offer", error))?; + Ok(()) +} + +/// Restores the projection of an observation a source rewrite retired, once +/// admission accepts its source's current offer. An observation still queued past the +/// checkpoint only needs its disposition removed, since the drain projects it +/// next; one the drain already passed has its effect applied here. A rebuild +/// that already staged it as retired restages it, or activation would bring +/// the retirement back. +async fn revive_source_record( + transaction: &DatabaseWriteTransaction<'_>, + observation_id: &CanonicalObservationIdV1, +) -> ProjectionStoreResult<()> { + let retired = ProjectionSkipReason::SourceRecordRetired.as_str(); + let revived = transaction + .execute( + "DELETE FROM observation_projection_dispositions + WHERE projector_version = ?1 AND observation_id = ?2 AND reason = ?3", + params![ + SESSION_MESSAGE_PROJECTOR_VERSION, + observation_id.as_str(), + retired + ], + ) + .await + .map_err(|error| storage("revive source record", error))?; + if revived == 0 { + return Ok(()); + } + let (sequence, observation) = read_observation(transaction, observation_id) + .await? + .ok_or(ProjectionStoreError::ObservationNotFound)?; + if sequence <= read_checkpoint(transaction).await?.last_sequence() { + let mut effect = derive_projection_with_alias(transaction, &observation).await?; + Box::pin(write_effect_converging_collisions( + transaction, + &CollisionGuardedWrite::Drain, + sequence, + &observation, + &mut effect, + )) + .await?; + record_canonical_effect(transaction, sequence, &observation, &effect).await?; + request_session_temporal_reset(transaction, observation_id.as_str()).await?; + } + let Some(job) = read_optional_rebuild_job(transaction).await? else { + return Ok(()); + }; + let unstaged = transaction + .execute( + "DELETE FROM observation_projection_rebuild_dispositions + WHERE projector_version = ?1 AND generation = ?2 + AND observation_id = ?3 AND reason = ?4", + params![ + SESSION_MESSAGE_PROJECTOR_VERSION, + job.generation.as_str(), + observation_id.as_str(), + retired + ], + ) + .await + .map_err(|error| storage("revive staged source record", error))?; + if unstaged == 0 { + return Ok(()); + } + let effect = Box::pin(stage_rebuild_observation( + transaction, + &job.generation, + sequence, + &observation, + )) + .await?; + if matches!(effect, ObservationProjection::Skipped(_)) { + return Ok(()); + } + let skipped_observations = job.skipped_observations.checked_sub(1).ok_or_else(|| { + storage_message( + "revive staged source record", + "staged retirement was not counted as skipped", + ) + })?; + transaction + .execute( + "UPDATE observation_projection_rebuilds + SET projected_rows = ?3, skipped_observations = ?4 + WHERE projector_version = ?1 AND generation = ?2", + params![ + SESSION_MESSAGE_PROJECTOR_VERSION, + job.generation.as_str(), + usize_i64(job.projected_rows.saturating_add(effect.output_count()))?, + usize_i64(skipped_observations)?, + ], + ) + .await + .map_err(|error| storage("revive staged source record", error))?; + Ok(()) +} + +async fn stage_rebuild_observation( + transaction: &DatabaseWriteTransaction<'_>, + generation: &str, + sequence: u64, + observation: &DurableObservationV1, +) -> ProjectionStoreResult { + let mut effect = derive_projection_for_rebuild(transaction, observation, generation).await?; + write_effect_converging_collisions( + transaction, + &CollisionGuardedWrite::Stage { generation }, + sequence, + observation, + &mut effect, + ) + .await?; + record_canonical_effect(transaction, sequence, observation, &effect).await?; + Ok(effect) +} + +/// The skips that leave an observation without a canonical rendering record +/// no temporal effect. +async fn record_canonical_effect( + transaction: &DatabaseWriteTransaction<'_>, + sequence: u64, + observation: &DurableObservationV1, + effect: &ObservationProjection, +) -> ProjectionStoreResult<()> { + if matches!( + effect, + ObservationProjection::Skipped(ProjectionSkipReason::OutputCollision) + ) { + return record_live_collision_preserving_canonical_observation_effect( + transaction, + sequence, + observation, + effect, + ) + .await; + } + if matches!( + effect, + ObservationProjection::Skipped( + ProjectionSkipReason::InvalidContract + | ProjectionSkipReason::NativeSourceSuperseded + | ProjectionSkipReason::SourceRecordRetired + ) + ) { + return Ok(()); + } + record_canonical_observation_effect(transaction, sequence, observation, effect).await +} + async fn start_or_resume_projection_rebuild_transaction( transaction: &impl Executor, frontier_sequence: u64, @@ -923,10 +1126,10 @@ async fn stage_projection_alias_batch_transaction( async fn stage_projection_rebuild_batch_transaction( transaction: &DatabaseWriteTransaction<'_>, cancelled: &AtomicBool, -) -> ProjectionStoreResult { +) -> ProjectionStoreResult<()> { let job = read_rebuild_job(transaction).await?; if job.state == RebuildState::Ready { - return Ok(RebuildBatchStage::AlreadyReady); + return Ok(()); } if job.state != RebuildState::Building || job.aliases_staged_through != job.frontier { return Err(storage_message( @@ -972,29 +1175,8 @@ async fn stage_projection_rebuild_batch_transaction( "observation sequence gap before rebuild frontier", )); } - let mut effect = - derive_projection_for_rebuild(transaction, &observation, &job.generation).await?; - write_effect_converging_collisions( - transaction, - &CollisionGuardedWrite::Stage { - generation: &job.generation, - }, - sequence, - &observation, - &mut effect, - ) - .await?; - if !matches!( - effect, - ObservationProjection::Skipped( - ProjectionSkipReason::InvalidContract - | ProjectionSkipReason::NativeSourceSuperseded - | ProjectionSkipReason::SourceRecordRetired - ) - ) { - record_canonical_observation_effect(transaction, sequence, &observation, &effect) - .await?; - } + let effect = + stage_rebuild_observation(transaction, &job.generation, sequence, &observation).await?; match &effect { ObservationProjection::Message(_) | ObservationProjection::Composite { .. } => { projected_rows = projected_rows.saturating_add(effect.output_count()); @@ -1036,13 +1218,7 @@ async fn stage_projection_rebuild_batch_transaction( ) .await .map_err(|error| storage("advance projection rebuild batch", error))?; - Ok(RebuildBatchStage::Advanced) -} - -#[derive(Clone, Copy, Debug, PartialEq, Eq)] -enum RebuildBatchStage { - AlreadyReady, - Advanced, + Ok(()) } async fn activate_projection_rebuild_transaction( @@ -1087,7 +1263,8 @@ async fn activate_projection_rebuild_transaction( transaction .execute( "DELETE FROM projection_queue - WHERE observation_sequence <= ?1 AND next_retry_at_micros < ?2", + WHERE observation_sequence <= ?1 AND next_retry_at_micros < ?2 + AND revival_generation IS NULL", params![job.frontier, PROJECTION_TERMINAL_RETRY_MICROS], ) .await diff --git a/crates/tracedecay-global-db/src/observation_projection/state.rs b/crates/tracedecay-global-db/src/observation_projection/state.rs index 29336c0809..cf9333f022 100644 --- a/crates/tracedecay-global-db/src/observation_projection/state.rs +++ b/crates/tracedecay-global-db/src/observation_projection/state.rs @@ -1,7 +1,9 @@ use std::borrow::Cow; use std::collections::{BTreeSet, HashMap}; -use tracedecay_domain::{CanonicalObservationIdV1, DurableObservationV1, PayloadDigestV1}; +use tracedecay_domain::{ + CanonicalObservationIdV1, DurableObservationV1, ObservationSourceGenerationV1, PayloadDigestV1, +}; use tracedecay_store::{ EDITED_FILES_KEY, ObservationProjection, PROJECTION_TERMINAL_RETRY_MICROS, ProjectionCheckpoint, ProjectionStoreError, ProjectionStoreResult, @@ -127,13 +129,24 @@ pub(super) async fn write_checkpoint( Ok(ProjectionCheckpoint::new(sequence)) } -pub(super) async fn queued_sequence( +pub(super) struct AcceptedSourceOffer { + pub(super) generation: ObservationSourceGenerationV1, + pub(super) start_offset: u64, +} + +pub(super) struct QueuedProjection { + pub(super) sequence: u64, + pub(super) revival: Option, +} + +pub(super) async fn queued_projection( conn: &impl QueryExecutor, observation_id: &CanonicalObservationIdV1, -) -> ProjectionStoreResult> { +) -> ProjectionStoreResult> { let mut rows = conn .query( - "SELECT observation_sequence FROM projection_queue WHERE observation_id = ?1", + "SELECT observation_sequence, revival_generation, revival_start_offset + FROM projection_queue WHERE observation_id = ?1", params![observation_id.as_str()], ) .await @@ -145,12 +158,36 @@ pub(super) async fn queued_sequence( else { return Ok(None); }; - decode_sequence( + let sequence = decode_sequence( row.get::(0) .map_err(|error| storage("read projection queue", error))?, "read projection queue", - ) - .map(Some) + )?; + let generation = row + .get::>(1) + .map_err(|error| storage("read accepted source offer", error))?; + let start_offset = row + .get::>(2) + .map_err(|error| storage("read accepted source offer", error))?; + let revival = match (generation, start_offset) { + (None, None) => None, + (Some(generation), Some(start_offset)) => Some(AcceptedSourceOffer { + generation: ObservationSourceGenerationV1::new( + generation + .parse() + .map_err(|error| storage("decode accepted source generation", error))?, + ) + .map_err(ProjectionStoreError::Contract)?, + start_offset: decode_sequence(start_offset, "decode accepted source offset")?, + }), + _ => { + return Err(storage_message( + "read accepted source offer", + "incomplete offer fence", + )); + } + }; + Ok(Some(QueuedProjection { sequence, revival })) } #[derive(Clone, Debug, PartialEq, Eq)] diff --git a/crates/tracedecay-global-db/src/registered_source_rewrite.rs b/crates/tracedecay-global-db/src/registered_source_rewrite.rs index e36a2eab5b..c6d0cc566e 100644 --- a/crates/tracedecay-global-db/src/registered_source_rewrite.rs +++ b/crates/tracedecay-global-db/src/registered_source_rewrite.rs @@ -3,7 +3,8 @@ //! Admission records where each record sits in the newest layout that offered //! it. A rewrite keeps the byte prefix it retained and re-offers the rest under //! a new generation; once that generation has been read to the end of the -//! file, every record it did not offer again is retired from projection. +//! file, every record it did not offer again is retired from projection. A +//! later layout's accepted offer queues the retired record for projection. use std::future::Future; use std::pin::Pin; @@ -88,42 +89,45 @@ impl RegisteredGlobalDb { return Ok(()); } let key = source_key(source, scope)?; - let rows = records - .iter() - .map(|record| { - Ok(( - record.observation_id.as_str().to_owned(), - record.generation.generation_id().to_string(), - offset(record.start_offset)?, - )) - }) - .collect::>>()?; - self.transact_or( - |error| storage("run source presence transaction", error), - async move |transaction| { - for (observation_id, generation, start_offset) in &rows { - transaction - .execute( - "INSERT INTO observation_source_presence ( - source_key, observation_id, generation, start_offset - ) VALUES (?1, ?2, ?3, ?4) - ON CONFLICT(source_key, observation_id) DO UPDATE SET - generation = excluded.generation, - start_offset = excluded.start_offset", - params![ - key.as_str(), - observation_id.as_str(), - generation.as_str(), - *start_offset, - ], - ) - .await - .map_err(|error| storage("record source presence", error))?; - } - Ok(()) - }, - ) - .await + let transaction = self + .begin_write_transaction() + .await + .map_err(|error| storage("begin source presence transaction", error))?; + let recorded = async { + for record in records { + transaction + .execute( + "INSERT INTO observation_source_presence ( + source_key, observation_id, generation, start_offset + ) VALUES (?1, ?2, ?3, ?4) + ON CONFLICT(source_key, observation_id) DO UPDATE SET + generation = excluded.generation, + start_offset = excluded.start_offset", + params![ + key.as_str(), + record.observation_id.as_str(), + record.generation.generation_id().to_string(), + offset(record.start_offset)?, + ], + ) + .await + .map_err(|error| storage("record source presence", error))?; + } + Ok(()) + } + .await; + match recorded { + Ok(()) => transaction + .commit() + .await + .map_err(|error| storage("commit source presence transaction", error)), + Err(error) => { + transaction.rollback().await.map_err(|rollback| { + storage("roll back source presence transaction", rollback) + })?; + Err(error) + } + } } /// Starts a rewrite of the source from `previous` to `generation`. The @@ -156,6 +160,28 @@ impl RegisteredGlobalDb { // A rewrite the previous generation never finished still names its // own layout: its retained prefix was relabeled to it. let unfinished = pending_rewrite(&transaction, &key).await?; + transaction + .execute( + "UPDATE projection_queue SET revival_generation = ?2 + WHERE EXISTS ( + SELECT 1 FROM observation_source_presence AS presence + WHERE presence.source_key = ?1 + AND presence.observation_id = projection_queue.observation_id + AND presence.start_offset < ?3 + AND presence.generation IN (?4, ?5) + AND presence.generation = projection_queue.revival_generation + AND presence.start_offset = projection_queue.revival_start_offset + )", + params![ + key.as_str(), + generation.as_str(), + retained_through, + previous.as_str(), + unfinished.as_deref(), + ], + ) + .await + .map_err(|error| storage("retain accepted source offer", error))?; transaction .execute( "UPDATE observation_source_presence SET generation = ?2 diff --git a/crates/tracedecay-global-db/src/schema_contract/definitions.rs b/crates/tracedecay-global-db/src/schema_contract/definitions.rs index 9b4f017883..e18d02a976 100644 --- a/crates/tracedecay-global-db/src/schema_contract/definitions.rs +++ b/crates/tracedecay-global-db/src/schema_contract/definitions.rs @@ -501,6 +501,8 @@ pub(super) const TABLES: &[Table] = &[ column("attempt_count", "INTEGER", true, Some("0"), 0), column("next_retry_at_micros", "INTEGER", true, Some("0"), 0), column("last_error", "TEXT", false, None, 0), + column("revival_generation", "TEXT", false, None, 0), + column("revival_start_offset", "INTEGER", false, None, 0), ], [foreign_key( "observation_id", diff --git a/crates/tracedecay-global-db/src/schema_contract/invariants/repair.rs b/crates/tracedecay-global-db/src/schema_contract/invariants/repair.rs index 97358fa9aa..0d8ec8f31a 100644 --- a/crates/tracedecay-global-db/src/schema_contract/invariants/repair.rs +++ b/crates/tracedecay-global-db/src/schema_contract/invariants/repair.rs @@ -151,7 +151,8 @@ pub(super) async fn repair_projection_frontier( .map_err(audit_read_error)?; conn.execute( "DELETE FROM projection_queue - WHERE observation_sequence <= ?1 AND next_retry_at_micros < ?2", + WHERE observation_sequence <= ?1 AND next_retry_at_micros < ?2 + AND revival_generation IS NULL", params![repaired_checkpoint, PROJECTION_TERMINAL_RETRY_MICROS], ) .await diff --git a/crates/tracedecay-global-db/src/session_project_rebind_tests.rs b/crates/tracedecay-global-db/src/session_project_rebind_tests.rs index 6ce5b42fca..f90416009e 100644 --- a/crates/tracedecay-global-db/src/session_project_rebind_tests.rs +++ b/crates/tracedecay-global-db/src/session_project_rebind_tests.rs @@ -28,7 +28,7 @@ use crate::tests::harness::{ HostAdmissionScope, HostAdmissionTestRuntimeV1, RegisteredGlobalDbHarness, bind_test_session_relation_graph, reopen_project_sessions_database, }; -use crate::{GlobalDbObservationStore, RegisteredGlobalDb}; +use crate::{GlobalDbObservationStore, ObservationSourcePresenceV1, RegisteredGlobalDb}; const CODEX_SESSION: &str = "01a0da2f-45db-7572-bd15-e0b83c7a09d3"; const CURSOR_SESSION: &str = "session.cursor-catch-up"; @@ -670,6 +670,118 @@ fn session_fact_with_transcript(cwd: &str, transcript_path: &str) -> CanonicalOb } } +#[asupersync::test] +async fn a_retired_reoffer_keeps_its_terminal_session_collision() { + let tmp = TempDir::new().unwrap(); + let project_root = tmp.path().join("repo"); + std::fs::create_dir_all(&project_root).unwrap(); + let cwd = project_root.to_string_lossy().into_owned(); + let runtime = HostAdmissionTestRuntimeV1::project( + &tmp.path().join("profile"), + &project_root, + ProjectId::new("project.reoffer-collision").unwrap(), + ) + .await + .unwrap(); + let database = runtime + .registered_database(HostAdmissionScope::Project) + .unwrap(); + let store = database.observation_store(); + let write = anchored(observation( + "codex", + CODEX_SESSION, + project_scope("project.reoffer-collision"), + "record.codex.retired-collision", + ORIGINAL_TEXT, + "receipt.codex.retired-collision", + Some(session_fact_with_transcript( + &cwd, + "/private/original.jsonl", + )), + )); + let observation = write.observation(); + let presence = [ObservationSourcePresenceV1 { + observation_id: observation.observation_id().clone(), + generation: observation.identity().generation(), + start_offset: observation.identity().position().start(), + }]; + database + .record_observation_source_presence(observation.source(), observation.scope(), &presence) + .await + .unwrap(); + store.persist_observation(write.clone()).await.unwrap(); + drain_projection_queue(&store).await; + database + .begin_observation_source_rewrite( + observation.source(), + observation.scope(), + observation.identity().generation(), + ObservationSourceGenerationV1::new(2).unwrap(), + 0, + ) + .await + .unwrap(); + assert_eq!( + database + .complete_observation_source_rewrite( + observation.source(), + observation.scope(), + ObservationSourceGenerationV1::new(2).unwrap(), + ) + .await + .unwrap(), + 1 + ); + assert!(!projected_text_contains(database, "codex", CODEX_SESSION, ORIGINAL_TEXT).await); + database + .writer_connection() + .unwrap() + .execute( + "UPDATE sessions SET transcript_path = ?3 WHERE provider = ?1 AND session_id = ?2", + params!["codex", CODEX_SESSION, "/private/current.jsonl"], + ) + .await + .unwrap(); + database + .record_observation_source_presence(observation.source(), observation.scope(), &presence) + .await + .unwrap(); + assert!(matches!( + store.persist_observation(write.clone()).await.unwrap(), + ObservationPersistOutcome::ExactDuplicate(_) + )); + persist( + &store, + observation_from_source( + "cursor", + CURSOR_SESSION, + None, + project_scope("project.reoffer-collision"), + "record.cursor.after-retired-collision", + CURSOR_TEXT, + "receipt.cursor.after-retired-collision", + None, + ), + ) + .await; + let window = store.project_queued_observations(8).await.unwrap().unwrap(); + assert_eq!(window.items.len(), 2); + assert!(!window.has_more); + let retained = queue_attempts(database, observation.observation_id().as_str()).await; + assert!( + retained.as_ref().is_some_and(|(attempts, error)| { + *attempts == 1 + && error + .as_deref() + .is_some_and(|detail| detail.contains("transcript_path")) + }), + "the retired reoffer must retain its terminal collision: {retained:?}" + ); + assert!(!projected_text_contains(database, "codex", CODEX_SESSION, ORIGINAL_TEXT).await); + assert!(projected_text_contains(database, "cursor", CURSOR_SESSION, CURSOR_TEXT).await); + assert!(store.next_queued_observation().await.unwrap().is_none()); +} + /// beta.88 (950bf21f85) parked a session collision with a durable skip, an /// advanced checkpoint, and a zero-output temporal effect. Exercise that full /// state through a registered daemon remount, not just a queued retry deadline. diff --git a/crates/tracedecay-global-db/src/tests.rs b/crates/tracedecay-global-db/src/tests.rs index 2320902c97..335a5185e4 100644 --- a/crates/tracedecay-global-db/src/tests.rs +++ b/crates/tracedecay-global-db/src/tests.rs @@ -22,6 +22,8 @@ mod schema_install_cancellation; #[cfg(test)] mod session_sync; #[cfg(test)] +mod source_revival; +#[cfg(test)] mod source_rewrite; #[cfg(test)] diff --git a/crates/tracedecay-global-db/src/tests/source_revival.rs b/crates/tracedecay-global-db/src/tests/source_revival.rs new file mode 100644 index 0000000000..82bda20f20 --- /dev/null +++ b/crates/tracedecay-global-db/src/tests/source_revival.rs @@ -0,0 +1,996 @@ +//! A record a source rewrite retired projects again only after admission +//! accepts an offer from a later layout of the same source. + +use tempfile::TempDir; + +use tracedecay_domain::{ + DurableObservationV1, ObservationIdentityMaterialV1, ObservationScopeV1, + ObservationSourceCursorV1, ObservationSourceGenerationV1, ObservationSourceIdentityV1, + SessionId, UtcMicros, +}; +use tracedecay_runtime_core::db::engine::params; +use tracedecay_store::{ + AnchoredObservationWrite, ObservationPersistOutcome, ObservationProjectionStore, + ObservationStore, ObservationStoreError, ObservationWrite, ProjectionPersistOutcome, + ProjectionRebuildOutcome, ProjectionSkipReason, build_observation_resolution_authorization_v1, + build_observation_retrieval_anchor, +}; + +use crate::observation_batch_tests::{sequential_writes, sequential_writes_with_text}; +use crate::tests::harness::{ + HostAdmissionScope, HostAdmissionTestRuntimeV1, RegisteredGlobalDbTestRuntime, +}; +use crate::{GlobalDbObservationStore, ObservationSourcePresenceV1, RegisteredGlobalDb}; + +/// One file-byte source, laid out the way admission lays out a rewrite. +struct Source<'a> { + registered: &'a RegisteredGlobalDb, + identity: ObservationSourceIdentityV1, + scope: ObservationScopeV1, +} + +impl<'a> Source<'a> { + fn of(registered: &'a RegisteredGlobalDb, writes: &[AnchoredObservationWrite]) -> Self { + Self { + registered, + identity: writes[0].observation().source().clone(), + scope: writes[0].observation().scope().clone(), + } + } + + async fn offer(&self, generation: u64, offered: &[&AnchoredObservationWrite]) { + let presence: Vec<_> = offered + .iter() + .map(|write| ObservationSourcePresenceV1 { + observation_id: write.observation().observation_id().clone(), + generation: ObservationSourceGenerationV1::new(generation).unwrap(), + start_offset: write.observation().identity().position().start(), + }) + .collect(); + self.registered + .record_observation_source_presence(&self.identity, &self.scope, &presence) + .await + .unwrap(); + } + + async fn accept(&self, generation: u64, write: &AnchoredObservationWrite) { + let store = self.registered.observation_store(); + let expected = store + .get_source_cursor(&self.identity, &self.scope) + .await + .unwrap(); + let candidate = reoffer(write, generation, expected); + assert!(matches!( + store.persist_observation(candidate).await.unwrap(), + ObservationPersistOutcome::CoveredDuplicate(_) + | ObservationPersistOutcome::ExactDuplicate(_) + )); + } + + /// Rewrites the source into a layout that retains the previous layout's + /// records before `retained_through` and offers `offered` after it. + /// Returns how many records the completed rewrite retired. + async fn rewrite( + &self, + previous: u64, + generation: u64, + retained_through: u64, + offered: &[&AnchoredObservationWrite], + ) -> u64 { + self.registered + .begin_observation_source_rewrite( + &self.identity, + &self.scope, + ObservationSourceGenerationV1::new(previous).unwrap(), + ObservationSourceGenerationV1::new(generation).unwrap(), + retained_through, + ) + .await + .unwrap(); + self.offer(generation, offered).await; + self.registered + .complete_observation_source_rewrite( + &self.identity, + &self.scope, + ObservationSourceGenerationV1::new(generation).unwrap(), + ) + .await + .unwrap() + } +} + +fn reoffer( + write: &AnchoredObservationWrite, + generation: u64, + expected: Option, +) -> AnchoredObservationWrite { + let original = write.observation(); + let identity = original.identity(); + let generation = ObservationSourceGenerationV1::new(generation).unwrap(); + let observation = DurableObservationV1::new( + ObservationIdentityMaterialV1::for_native_record( + original.source().clone(), + original.scope().clone(), + generation, + identity.position(), + identity.ordering_domain(), + identity.native_record_id().unwrap().clone(), + ) + .unwrap(), + original.receipt().clone(), + original.retention_class().clone(), + original.payload().clone(), + ) + .unwrap(); + let next = ObservationSourceCursorV1::for_ordering( + original.source().clone(), + original.scope().clone(), + generation, + identity.ordering_domain(), + identity.position().end(), + ) + .unwrap(); + let expected = if expected + .as_ref() + .is_some_and(|cursor| cursor.reached(&next)) + { + write.expected_cursor().cloned() + } else { + expected + }; + let anchor = build_observation_retrieval_anchor( + &observation, + write.projection_generation().clone(), + UtcMicros(1), + build_observation_resolution_authorization_v1(&observation, "observation-batch-test") + .unwrap(), + ) + .unwrap(); + AnchoredObservationWrite::new( + ObservationWrite::new(observation, expected, next).unwrap(), + anchor, + write.projection_generation().clone(), + ) + .unwrap() +} + +async fn persist(store: &GlobalDbObservationStore, writes: &[AnchoredObservationWrite]) { + for chunk in writes.chunks(100) { + store.persist_observations(chunk.to_vec()).await.unwrap(); + } +} + +async fn drain(store: &GlobalDbObservationStore) -> Vec { + let mut outcomes = Vec::new(); + loop { + let batch = store + .project_queued_observations(128) + .await + .unwrap() + .unwrap(); + outcomes.extend(batch.items.into_iter().map(|item| item.outcome)); + if !batch.has_more { + return outcomes; + } + } +} + +async fn rebuild(registered: &RegisteredGlobalDb) -> ProjectionRebuildOutcome { + let snapshot = registered.read_snapshot().await.unwrap(); + let mut rows = snapshot + .query("SELECT MAX(sequence) FROM observations", ()) + .await + .unwrap(); + let frontier = rows.next().await.unwrap().unwrap().get::(0).unwrap(); + drop(rows); + drop(snapshot); + registered + .rebuild_observation_projection(u64::try_from(frontier).unwrap()) + .await + .unwrap() +} + +async fn rebuild_to_completion(registered: &RegisteredGlobalDb) -> ProjectionRebuildOutcome { + loop { + let outcome = rebuild(registered).await; + if outcome.is_complete() { + return outcome; + } + } +} + +async fn projected_messages(registered: &RegisteredGlobalDb) -> Vec { + let snapshot = registered.read_snapshot().await.unwrap(); + let mut rows = snapshot + .query( + "SELECT message_id FROM lcm_raw_messages ORDER BY message_id", + (), + ) + .await + .unwrap(); + let mut messages = Vec::new(); + while let Some(row) = rows.next().await.unwrap() { + messages.push(row.get::(0).unwrap()); + } + messages +} + +async fn projected_message_content(registered: &RegisteredGlobalDb, message_id: &str) -> String { + let snapshot = registered.read_snapshot().await.unwrap(); + let mut rows = snapshot + .query( + "SELECT COALESCE(content, placeholder_text, '') + FROM lcm_raw_messages WHERE message_id = ?1", + params![message_id], + ) + .await + .unwrap(); + rows.next() + .await + .unwrap() + .unwrap() + .get::(0) + .unwrap() +} + +async fn temporal_effect( + registered: &RegisteredGlobalDb, + write: &AnchoredObservationWrite, +) -> (String, i64) { + let snapshot = registered.read_snapshot().await.unwrap(); + let mut rows = snapshot + .query( + "SELECT effect_digest, output_count + FROM session_temporal_observation_effects WHERE observation_id = ?1", + params![write.observation().observation_id().as_str()], + ) + .await + .unwrap(); + let row = rows.next().await.unwrap().unwrap(); + (row.get::(0).unwrap(), row.get::(1).unwrap()) +} + +async fn projection_disposition( + registered: &RegisteredGlobalDb, + write: &AnchoredObservationWrite, +) -> String { + let snapshot = registered.read_snapshot().await.unwrap(); + let mut rows = snapshot + .query( + "SELECT reason FROM observation_projection_dispositions + WHERE observation_id = ?1", + params![write.observation().observation_id().as_str()], + ) + .await + .unwrap(); + rows.next() + .await + .unwrap() + .unwrap() + .get::(0) + .unwrap() +} + +async fn temporal_reset_requested( + registered: &RegisteredGlobalDb, + write: &AnchoredObservationWrite, +) -> bool { + let snapshot = registered.read_snapshot().await.unwrap(); + let mut rows = snapshot + .query( + "SELECT EXISTS ( + SELECT 1 + FROM session_temporal_resets AS reset + JOIN session_temporal_observation_effects AS effect + ON effect.session_id = reset.session_id + WHERE effect.observation_id = ?1 + )", + params![write.observation().observation_id().as_str()], + ) + .await + .unwrap(); + rows.next().await.unwrap().unwrap().get::(0).unwrap() == 1 +} + +async fn source_presence_generation( + registered: &RegisteredGlobalDb, + write: &AnchoredObservationWrite, +) -> u64 { + let snapshot = registered.read_snapshot().await.unwrap(); + let mut rows = snapshot + .query( + "SELECT generation FROM observation_source_presence + WHERE observation_id = ?1", + params![write.observation().observation_id().as_str()], + ) + .await + .unwrap(); + rows.next() + .await + .unwrap() + .unwrap() + .get::(0) + .unwrap() + .parse() + .unwrap() +} + +async fn clear_temporal_reset(registered: &RegisteredGlobalDb, write: &AnchoredObservationWrite) { + registered + .writer_connection() + .unwrap() + .execute( + "DELETE FROM session_temporal_resets + WHERE session_id = ( + SELECT session_id FROM session_temporal_observation_effects + WHERE observation_id = ?1 + )", + params![write.observation().observation_id().as_str()], + ) + .await + .unwrap(); +} + +#[asupersync::test] +async fn a_reoffered_retired_record_projects_again_across_rebuild_and_restart() { + let tmp = TempDir::new().unwrap(); + let runtime = RegisteredGlobalDbTestRuntime::profile(tmp.path()) + .await + .unwrap(); + let writes = sequential_writes(&SessionId::new("session.source-reoffer").unwrap(), 2); + { + let registered = runtime.profile_database(); + let store = registered.observation_store(); + let source = Source::of(registered, &writes); + source.offer(1, &[&writes[0], &writes[1]]).await; + persist(&store, &writes).await; + drain(&store).await; + assert_eq!( + projected_messages(registered).await, + ["message.batch.0", "message.batch.1"] + ); + assert_eq!(source.rewrite(1, 2, 0, &[]).await, 2); + assert!(projected_messages(registered).await.is_empty()); + + assert!(matches!( + store.persist_observation(writes[0].clone()).await.unwrap(), + ObservationPersistOutcome::ExactDuplicate(_) + )); + drain(&store).await; + assert!( + projected_messages(registered).await.is_empty(), + "a replayed durable write is not an offer from its source" + ); + rebuild_to_completion(registered).await; + assert!( + projected_messages(registered).await.is_empty(), + "a rebuild keeps the retirement" + ); + } + + let runtime = runtime.reopen_profile_database_for_test().await.unwrap(); + { + let registered = runtime.profile_database(); + let source = Source::of(registered, &writes); + assert!(projected_messages(registered).await.is_empty()); + assert_eq!(source.rewrite(2, 3, 0, &[&writes[0]]).await, 0); + assert!( + projected_messages(registered).await.is_empty(), + "presence is not acceptance" + ); + let store = registered.observation_store(); + store.persist_observation(writes[0].clone()).await.unwrap(); + drain(&store).await; + assert!( + projected_messages(registered).await.is_empty(), + "old coordinates do not accept the current offer" + ); + source.accept(3, &writes[0]).await; + assert!( + projected_messages(registered).await.is_empty(), + "projection resumes after acceptance" + ); + } + + let runtime = runtime.reopen_profile_database_for_test().await.unwrap(); + { + let registered = runtime.profile_database(); + let source = Source::of(registered, &writes); + drain(®istered.observation_store()).await; + assert_eq!( + projected_messages(registered).await, + ["message.batch.0"], + "the re-offered record projects again; the other stays retired" + ); + source.offer(3, &[&writes[0]]).await; + assert_eq!(projected_messages(registered).await, ["message.batch.0"]); + rebuild_to_completion(registered).await; + assert_eq!( + projected_messages(registered).await, + ["message.batch.0"], + "a rebuild keeps the revival" + ); + } + + let runtime = runtime.reopen_profile_database_for_test().await.unwrap(); + let registered = runtime.profile_database(); + assert_eq!(projected_messages(registered).await, ["message.batch.0"]); +} + +#[asupersync::test] +async fn a_newer_layout_retires_a_revived_record_again() { + let tmp = TempDir::new().unwrap(); + let runtime = HostAdmissionTestRuntimeV1::profile(tmp.path()) + .await + .unwrap(); + let store = runtime + .observation_store(HostAdmissionScope::Profile) + .unwrap(); + let registered = runtime + .registered_database(HostAdmissionScope::Profile) + .unwrap(); + let writes = sequential_writes(&SessionId::new("session.source-relayout").unwrap(), 1); + let source = Source::of(registered, &writes); + source.offer(1, &[&writes[0]]).await; + persist(&store, &writes).await; + drain(&store).await; + + assert_eq!(source.rewrite(1, 2, 0, &[]).await, 1); + assert_eq!(source.rewrite(2, 3, 0, &[&writes[0]]).await, 0); + source.accept(3, &writes[0]).await; + drain(&store).await; + assert_eq!(projected_messages(registered).await, ["message.batch.0"]); + assert_eq!(source.rewrite(3, 4, 0, &[]).await, 1); + assert!(projected_messages(registered).await.is_empty()); + assert_eq!(source.rewrite(4, 5, 0, &[&writes[0]]).await, 0); + source.accept(5, &writes[0]).await; + drain(&store).await; + assert_eq!(projected_messages(registered).await, ["message.batch.0"]); +} + +#[asupersync::test] +async fn a_reoffered_record_settles_if_another_source_claimed_its_retired_output() { + let tmp = TempDir::new().unwrap(); + let runtime = RegisteredGlobalDbTestRuntime::profile(tmp.path()) + .await + .unwrap(); + let original = sequential_writes_with_text( + &SessionId::new("session.source-collision.original").unwrap(), + 1, + |_| "original source body".to_owned(), + ); + let claimant = sequential_writes_with_text( + &SessionId::new("session.source-collision.claimant").unwrap(), + 1, + |_| "claimant source body".to_owned(), + ); + + { + let registered = runtime.profile_database(); + let store = registered.observation_store(); + let original_source = Source::of(registered, &original); + original_source.offer(1, &[&original[0]]).await; + persist(&store, &original).await; + drain(&store).await; + let original_effect = temporal_effect(registered, &original[0]).await; + assert_eq!(original_effect.1, 1); + + assert_eq!(original_source.rewrite(1, 2, 0, &[]).await, 1); + let claimant_source = Source::of(registered, &claimant); + claimant_source.offer(1, &[&claimant[0]]).await; + persist(&store, &claimant).await; + drain(&store).await; + assert_eq!( + projected_message_content(registered, "message.batch.0").await, + "claimant source body" + ); + clear_temporal_reset(registered, &original[0]).await; + assert!(!temporal_reset_requested(registered, &original[0]).await); + + assert_eq!( + original_source.rewrite(2, 3, 0, &[&original[0]]).await, + 0, + "the re-offer settles as a live collision instead of rolling back" + ); + original_source.accept(3, &original[0]).await; + drain(&store).await; + assert_eq!( + projection_disposition(registered, &original[0]).await, + ProjectionSkipReason::OutputCollision.as_str() + ); + assert_eq!( + temporal_effect(registered, &original[0]).await, + original_effect, + "the original positive-output temporal receipt stays immutable" + ); + assert!( + temporal_reset_requested(registered, &original[0]).await, + "the retained positive receipt still requests a live-state reset" + ); + assert_eq!( + projected_message_content(registered, "message.batch.0").await, + "claimant source body", + "the collided re-offer must not reclaim the live output" + ); + original_source.offer(3, &[&original[0]]).await; + assert_eq!( + source_presence_generation(registered, &original[0]).await, + 3 + ); + } + + let runtime = runtime.reopen_profile_database_for_test().await.unwrap(); + let registered = runtime.profile_database(); + let original_source = Source::of(registered, &original); + original_source.offer(3, &[&original[0]]).await; + assert_eq!( + source_presence_generation(registered, &original[0]).await, + 3 + ); + assert_eq!( + projection_disposition(registered, &original[0]).await, + ProjectionSkipReason::OutputCollision.as_str() + ); + assert_eq!( + projected_message_content(registered, "message.batch.0").await, + "claimant source body" + ); +} + +#[asupersync::test] +async fn a_foreign_source_cannot_revive_a_retired_record() { + let tmp = TempDir::new().unwrap(); + let runtime = HostAdmissionTestRuntimeV1::profile(tmp.path()) + .await + .unwrap(); + let registered = runtime + .registered_database(HostAdmissionScope::Profile) + .unwrap(); + let store = registered.observation_store(); + let writes = sequential_writes(&SessionId::new("session.source-owner").unwrap(), 1); + let source = Source::of(registered, &writes); + source.offer(1, &[&writes[0]]).await; + persist(&store, &writes).await; + drain(&store).await; + assert_eq!(source.rewrite(1, 2, 0, &[]).await, 1); + + let foreign = + ObservationSourceIdentityV1::new(SessionId::new("session.source-foreign").unwrap()) + .unwrap(); + registered + .record_observation_source_presence( + &foreign, + &source.scope, + &[ObservationSourcePresenceV1 { + observation_id: writes[0].observation().observation_id().clone(), + generation: ObservationSourceGenerationV1::new(3).unwrap(), + start_offset: 0, + }], + ) + .await + .unwrap(); + store.persist_observation(writes[0].clone()).await.unwrap(); + drain(&store).await; + assert!(projected_messages(registered).await.is_empty()); + assert_eq!(source.rewrite(2, 3, 0, &[&writes[0]]).await, 0); + source.accept(3, &writes[0]).await; + drain(&store).await; + assert_eq!(projected_messages(registered).await, ["message.batch.0"]); +} + +#[asupersync::test] +async fn a_failed_presence_batch_does_not_authorize_revival() { + let tmp = TempDir::new().unwrap(); + let runtime = HostAdmissionTestRuntimeV1::profile(tmp.path()) + .await + .unwrap(); + let registered = runtime + .registered_database(HostAdmissionScope::Profile) + .unwrap(); + let store = registered.observation_store(); + let writes = sequential_writes(&SessionId::new("session.source-rollback").unwrap(), 2); + let source = Source::of(registered, &writes); + source.offer(1, &[&writes[0], &writes[1]]).await; + persist(&store, &writes).await; + drain(&store).await; + assert_eq!(source.rewrite(1, 2, 0, &[]).await, 2); + + let generation = ObservationSourceGenerationV1::new(3).unwrap(); + let error = registered + .record_observation_source_presence( + &source.identity, + &source.scope, + &[ + ObservationSourcePresenceV1 { + observation_id: writes[0].observation().observation_id().clone(), + generation, + start_offset: 0, + }, + ObservationSourcePresenceV1 { + observation_id: writes[1].observation().observation_id().clone(), + generation, + start_offset: u64::MAX, + }, + ], + ) + .await + .unwrap_err(); + assert!(matches!( + error, + tracedecay_store::ProjectionStoreError::SequenceOverflow(u64::MAX) + )); + assert!(projected_messages(registered).await.is_empty()); + rebuild_to_completion(registered).await; + assert!(projected_messages(registered).await.is_empty()); + assert_eq!(source.rewrite(2, 3, 0, &[&writes[0]]).await, 0); + source.accept(3, &writes[0]).await; + drain(&store).await; + assert_eq!(projected_messages(registered).await, ["message.batch.0"]); +} + +#[asupersync::test] +async fn a_record_retired_before_its_drain_projects_once_offered_again() { + let tmp = TempDir::new().unwrap(); + let runtime = HostAdmissionTestRuntimeV1::profile(tmp.path()) + .await + .unwrap(); + let store = runtime + .observation_store(HostAdmissionScope::Profile) + .unwrap(); + let registered = runtime + .registered_database(HostAdmissionScope::Profile) + .unwrap(); + let writes = sequential_writes(&SessionId::new("session.source-queued").unwrap(), 2); + let source = Source::of(registered, &writes); + source.offer(1, &[&writes[0], &writes[1]]).await; + persist(&store, &writes).await; + + assert_eq!(source.rewrite(1, 2, 0, &[]).await, 2); + assert_eq!(source.rewrite(2, 3, 0, &[&writes[0]]).await, 0); + source.accept(3, &writes[0]).await; + let outcomes = drain(&store).await; + assert!( + matches!( + outcomes.as_slice(), + [ + ProjectionPersistOutcome::Projected(_), + ProjectionPersistOutcome::Skipped { + reason: ProjectionSkipReason::SourceRecordRetired, + .. + }, + ] + ), + "{outcomes:?}" + ); + assert_eq!(projected_messages(registered).await, ["message.batch.0"]); +} + +#[asupersync::test] +async fn a_rebuild_that_staged_a_retirement_activates_the_revival() { + // Three alias pages and one build page fill the first rebuild call's + // step budget, so it stages the retirement and stops before activation. + const RECORDS: usize = 300; + + let tmp = TempDir::new().unwrap(); + let runtime = HostAdmissionTestRuntimeV1::profile(tmp.path()) + .await + .unwrap(); + let store = runtime + .observation_store(HostAdmissionScope::Profile) + .unwrap(); + let registered = runtime + .registered_database(HostAdmissionScope::Profile) + .unwrap(); + let writes = sequential_writes(&SessionId::new("session.source-rebuild").unwrap(), RECORDS); + let all: Vec<_> = writes.iter().collect(); + let source = Source::of(registered, &writes); + source.offer(1, &all).await; + persist(&store, &writes).await; + drain(&store).await; + assert_eq!(source.rewrite(1, 2, 0, &all[1..]).await, 1); + + let staged = rebuild(registered).await; + assert!(!staged.is_complete()); + assert_eq!(staged.skipped_observations(), 1, "the retirement is staged"); + + let retained_through = u64::try_from(RECORDS).unwrap(); + assert_eq!( + source.rewrite(2, 3, retained_through, &[&writes[0]]).await, + 0 + ); + source.accept(3, &writes[0]).await; + let rebuilt = rebuild_to_completion(registered).await; + assert_eq!(rebuilt.skipped_observations(), 0); + let messages = projected_messages(registered).await; + assert_eq!(messages.len(), RECORDS); + assert!(messages.iter().any(|message| message == "message.batch.0")); +} + +#[asupersync::test] +async fn a_rebuild_does_not_restore_a_revived_record_retired_again() { + let tmp = TempDir::new().unwrap(); + let runtime = RegisteredGlobalDbTestRuntime::profile(tmp.path()) + .await + .unwrap(); + let registered = runtime.profile_database(); + let store = registered.observation_store(); + let writes = sequential_writes( + &SessionId::new("session.source-retired-again").unwrap(), + 300, + ); + let all: Vec<_> = writes.iter().collect(); + let source = Source::of(registered, &writes); + source.offer(1, &all).await; + persist(&store, &writes).await; + drain(&store).await; + assert_eq!(source.rewrite(1, 2, 0, &all[1..]).await, 1); + let staged = rebuild(registered).await; + assert!(!staged.is_complete()); + assert_eq!(staged.skipped_observations(), 1); + source.rewrite(2, 3, 300, &[&writes[0]]).await; + source.accept(3, &writes[0]).await; + let first = store + .project_observation(writes[0].observation().observation_id()) + .await + .unwrap(); + assert!(matches!(first, ProjectionPersistOutcome::ExactDuplicate(_))); + assert_eq!(source.rewrite(3, 4, 0, &all[1..]).await, 1); + assert!( + !projected_messages(registered) + .await + .contains(&"message.batch.0".to_owned()) + ); + + let rebuilt = rebuild_to_completion(registered).await; + assert!( + !projected_messages(registered) + .await + .contains(&"message.batch.0".to_owned()), + "activation must preserve a retirement after staging its revival" + ); + assert_eq!(rebuilt.skipped_observations(), 1); +} + +#[asupersync::test] +async fn retirement_cancels_pending_revival_and_retained_prefix_preserves_acceptance() { + let tmp = TempDir::new().unwrap(); + let runtime = RegisteredGlobalDbTestRuntime::profile(tmp.path()) + .await + .unwrap(); + let registered = runtime.profile_database(); + let store = registered.observation_store(); + let writes = sequential_writes(&SessionId::new("session.source-pending").unwrap(), 1); + let source = Source::of(registered, &writes); + source.offer(1, &[&writes[0]]).await; + persist(&store, &writes).await; + drain(&store).await; + assert_eq!(source.rewrite(1, 2, 0, &[]).await, 1); + source.rewrite(2, 3, 0, &[&writes[0]]).await; + source.accept(3, &writes[0]).await; + source.rewrite(3, 4, 0, &[&writes[0]]).await; + drain(&store).await; + assert!( + projected_messages(registered).await.is_empty(), + "acceptance of an older layout does not accept the current offer" + ); + source.accept(4, &writes[0]).await; + assert_eq!(source.rewrite(4, 5, 0, &[]).await, 0); + drain(&store).await; + assert!(projected_messages(registered).await.is_empty()); + source.rewrite(5, 6, 0, &[&writes[0]]).await; + drain(&store).await; + assert!(projected_messages(registered).await.is_empty()); + + source.accept(6, &writes[0]).await; + source.rewrite(6, 7, 1, &[]).await; + drain(&store).await; + assert_eq!(projected_messages(registered).await, ["message.batch.0"]); +} + +#[asupersync::test] +async fn a_failed_accepted_batch_rolls_back_revival_and_coverage() { + let tmp = TempDir::new().unwrap(); + let runtime = RegisteredGlobalDbTestRuntime::profile(tmp.path()) + .await + .unwrap(); + let registered = runtime.profile_database(); + let store = registered.observation_store(); + let writes = sequential_writes( + &SessionId::new("session.source-accept-rollback").unwrap(), + 2, + ); + let source = Source::of(registered, &writes); + source.offer(1, &[&writes[0], &writes[1]]).await; + persist(&store, &writes).await; + drain(&store).await; + assert_eq!(source.rewrite(1, 2, 0, &[]).await, 2); + source.rewrite(2, 3, 0, &[&writes[0], &writes[1]]).await; + let previous = store + .get_source_cursor(&source.identity, &source.scope) + .await + .unwrap(); + let first = reoffer(&writes[0], 3, previous.clone()); + let second = reoffer(&writes[1], 3, Some(first.next_cursor().clone())); + registered + .writer_connection() + .unwrap() + .execute( + "CREATE TRIGGER fail_second_accepted_offer + BEFORE INSERT ON projection_queue + WHEN NEW.observation_id = (SELECT observation_id FROM observations WHERE sequence = 2) + BEGIN SELECT RAISE(ABORT, 'injected accepted-offer write failure'); END", + (), + ) + .await + .unwrap(); + let error = store + .persist_observations(vec![first.clone(), second.clone()]) + .await + .unwrap_err(); + let ObservationStoreError::Storage { + operation, + source: cause, + } = error + else { + panic!("expected the injected storage failure, got {error:?}"); + }; + assert_eq!(operation, "submit observation batch"); + assert!( + cause + .to_string() + .contains("injected accepted-offer write failure"), + "{cause}" + ); + registered + .writer_connection() + .unwrap() + .execute("DROP TRIGGER fail_second_accepted_offer", ()) + .await + .unwrap(); + assert_eq!( + store + .get_source_cursor(&source.identity, &source.scope) + .await + .unwrap(), + previous + ); + drain(&store).await; + assert!( + projected_messages(registered).await.is_empty(), + "the first offer rolled back with the second" + ); + + let outcomes = store + .persist_observations(vec![first, second]) + .await + .unwrap(); + assert!(outcomes.iter().all(|item| matches!( + item.outcome(), + ObservationPersistOutcome::CoveredDuplicate(_) + ))); + registered + .writer_connection() + .unwrap() + .execute( + "CREATE TRIGGER fail_revival_reset + BEFORE INSERT ON session_temporal_resets + BEGIN SELECT RAISE(ABORT, 'injected revival projection failure'); END", + (), + ) + .await + .unwrap(); + let error = store.project_queued_observations(128).await.unwrap_err(); + assert!( + error + .to_string() + .contains("injected revival projection failure"), + "{error:?}" + ); + assert!( + projected_messages(registered).await.is_empty(), + "the failed projector must roll back restored outputs" + ); + assert_eq!( + projection_disposition(registered, &writes[0]).await, + ProjectionSkipReason::SourceRecordRetired.as_str() + ); + registered + .writer_connection() + .unwrap() + .execute("DROP TRIGGER fail_revival_reset", ()) + .await + .unwrap(); + drain(&store).await; + assert_eq!( + projected_messages(registered).await, + ["message.batch.0", "message.batch.1"] + ); +} + +#[asupersync::test] +async fn an_exact_accepted_reoffer_revives_without_a_transport_receipt() { + let tmp = TempDir::new().unwrap(); + let runtime = RegisteredGlobalDbTestRuntime::profile(tmp.path()) + .await + .unwrap(); + let writes = sequential_writes(&SessionId::new("session.source-exact-reoffer").unwrap(), 1); + { + let registered = runtime.profile_database(); + let store = registered.observation_store(); + let source = Source::of(registered, &writes); + source.offer(1, &[&writes[0]]).await; + persist(&store, &writes).await; + drain(&store).await; + assert_eq!(source.rewrite(1, 2, 0, &[]).await, 1); + source.rewrite(2, 1, 0, &[&writes[0]]).await; + let outcome = store.persist_observation(writes[0].clone()).await.unwrap(); + assert!(matches!( + outcome, + ObservationPersistOutcome::ExactDuplicate(_) + )); + let snapshot = registered.read_snapshot().await.unwrap(); + let mut rows = snapshot + .query("SELECT COUNT(*) FROM td_runtime_writer_idempotency_v2", ()) + .await + .unwrap(); + assert_eq!( + rows.next().await.unwrap().unwrap().get::(0).unwrap(), + 0 + ); + } + let runtime = runtime.reopen_profile_database_for_test().await.unwrap(); + let registered = runtime.profile_database(); + drain(®istered.observation_store()).await; + assert_eq!(projected_messages(registered).await, ["message.batch.0"]); +} + +#[asupersync::test] +async fn a_released_queue_upgrades_without_accepting_its_old_work() { + let tmp = TempDir::new().unwrap(); + let runtime = RegisteredGlobalDbTestRuntime::profile(tmp.path()) + .await + .unwrap(); + let writes = sequential_writes(&SessionId::new("session.source-queue-upgrade").unwrap(), 1); + { + let registered = runtime.profile_database(); + let store = registered.observation_store(); + let source = Source::of(registered, &writes); + source.offer(1, &[&writes[0]]).await; + persist(&store, &writes).await; + assert_eq!(source.rewrite(1, 2, 0, &[]).await, 1); + let writer = registered.writer_connection().unwrap(); + writer + .execute( + "ALTER TABLE projection_queue DROP COLUMN revival_start_offset", + (), + ) + .await + .unwrap(); + writer + .execute( + "ALTER TABLE projection_queue DROP COLUMN revival_generation", + (), + ) + .await + .unwrap(); + } + let runtime = runtime.reopen_profile_database_for_test().await.unwrap(); + let registered = runtime.profile_database(); + let store = registered.observation_store(); + assert!( + registered + .writer_connection() + .unwrap() + .execute("UPDATE projection_queue SET revival_generation = '3'", (),) + .await + .is_err(), + "the upgraded queue rejects an incomplete fence" + ); + drain(&store).await; + assert!(projected_messages(registered).await.is_empty()); + let source = Source::of(registered, &writes); + source.rewrite(2, 3, 0, &[&writes[0]]).await; + source.accept(3, &writes[0]).await; + drain(&store).await; + assert_eq!(projected_messages(registered).await, ["message.batch.0"]); +} diff --git a/crates/tracedecay-rusqlite-runtime/src/repository/observation/mod.rs b/crates/tracedecay-rusqlite-runtime/src/repository/observation/mod.rs index a063f53808..9d006057f9 100644 --- a/crates/tracedecay-rusqlite-runtime/src/repository/observation/mod.rs +++ b/crates/tracedecay-rusqlite-runtime/src/repository/observation/mod.rs @@ -10,14 +10,15 @@ use tracedecay_domain::{ CanonicalObservationIdV1, DurableObservationV1, ObservationCollisionOutcomeV1, ObservationScopeV1, ObservationSourceCursorV1, ObservationSourceIdentityV1, ProjectionGenerationId, SanitizationReceiptV1, classify_observation_collision, + is_canonical_payload_revision_replay, }; use tracedecay_store::{ AnchoredObservationWrite, CanonicalBodyError, CursorAdvanceLedgerDisagreementV1, CursorAdvanceLedgerIdentityV1, LOAD_CANONICAL_BODY_SQL, ObservationCoverageReason, ObservationCursorAdvance, ObservationReadOperationV1, ObservationReadResultV1, PROJECTION_TERMINAL_RETRY_MICROS, ProjectionRebuildProgressV1, ProjectionRebuildStateV1, - SESSION_MESSAGE_PROJECTOR_VERSION, UPSERT_CANONICAL_BODY_SQL, parse_stored_observation, - slim_stored_json, + ProjectionSkipReason, SESSION_MESSAGE_PROJECTOR_VERSION, UPSERT_CANONICAL_BODY_SQL, + parse_stored_observation, slim_stored_json, }; use crate::operation::StorageOperationError; @@ -80,8 +81,11 @@ impl ObservationExecutor { let stored_observation = decode_stored_observation_on_savepoint(savepoint, stored_observation)?; let collision = classify_observation_collision(&stored_observation, observation); - if collision == ObservationCollisionOutcomeV1::ExactDuplicate - && stored_observation.identity() != observation.identity() + let canonical_revision = + is_canonical_payload_revision_replay(&stored_observation, observation); + if canonical_revision + || (collision == ObservationCollisionOutcomeV1::ExactDuplicate + && stored_observation.identity() != observation.identity()) { let identity = observation.identity(); let mut advance = ObservationCursorAdvance::for_ordering_with_sanitization_receipt( @@ -91,7 +95,11 @@ impl ObservationExecutor { identity.ordering_domain(), write.expected_cursor().cloned(), identity.position(), - ObservationCoverageReason::DuplicateObservation, + if canonical_revision { + ObservationCoverageReason::CanonicalPayloadRevision + } else { + ObservationCoverageReason::DuplicateObservation + }, observation.receipt().clone(), ) .map_err(invalid)?; @@ -105,7 +113,8 @@ impl ObservationExecutor { (None, None) => {} _ => return Err(invalid("cursor resume checkpoint is incomplete").into()), } - return self.execute_cursor_advance(savepoint, &advance); + self.execute_cursor_advance(savepoint, &advance)?; + return queue_accepted_source_revival(savepoint, observation, &stored_observation); } if collision != ObservationCollisionOutcomeV1::ExactDuplicate || stored_digest != payload_digest @@ -123,7 +132,7 @@ impl ObservationExecutor { return Err(invalid("sanitization receipt identity collision").into()); } verify_observation_authority(savepoint, write)?; - return Ok(()); + return queue_accepted_source_revival(savepoint, observation, &stored_observation); } let actual_cursor = read_cursor(savepoint, &source_json, &scope_json)?; @@ -420,6 +429,47 @@ impl ObservationExecutor { } } +fn queue_accepted_source_revival( + savepoint: &Savepoint<'_>, + candidate: &DurableObservationV1, + retained: &DurableObservationV1, +) -> Result<(), StorageOperationError> { + if candidate.source() != retained.source() || candidate.scope() != retained.scope() { + return Err(invalid("accepted source offer does not own the retained observation").into()); + } + let identity = candidate.identity(); + let start_offset = i64::try_from(identity.position().start()) + .map_err(|_| invalid("source offer offset exceeds SQLite integer"))?; + savepoint.execute_cached( + "INSERT INTO projection_queue ( + observation_id, observation_sequence, revival_generation, revival_start_offset + ) + SELECT observation.observation_id, observation.sequence, + presence.generation, presence.start_offset + FROM observations AS observation + JOIN observation_source_presence AS presence + ON presence.observation_id = observation.observation_id + JOIN observation_projection_dispositions AS disposition + ON disposition.observation_id = observation.observation_id + WHERE observation.observation_id = ?1 AND presence.source_key = ?2 + AND presence.generation = ?3 AND presence.start_offset = ?4 + AND disposition.projector_version = ?5 AND disposition.reason = ?6 + ON CONFLICT(observation_id) DO UPDATE SET + revival_generation = excluded.revival_generation, + revival_start_offset = excluded.revival_start_offset, + attempt_count = 0, next_retry_at_micros = 0, last_error = NULL", + params![ + candidate.observation_id().as_str(), + encode(&(candidate.source(), candidate.scope()))?, + identity.generation().generation_id().to_string(), + start_offset, + SESSION_MESSAGE_PROJECTOR_VERSION, + ProjectionSkipReason::SourceRecordRetired.as_str(), + ], + )?; + Ok(()) +} + /// Committed cursors of one source in commit order, read through the indexed /// session-id projection and narrowed to the exact source and scope. fn committed_source_cursors( diff --git a/crates/tracedecay-rusqlite-runtime/src/repository/observation/tests.rs b/crates/tracedecay-rusqlite-runtime/src/repository/observation/tests.rs index 741142120d..a0b582151b 100644 --- a/crates/tracedecay-rusqlite-runtime/src/repository/observation/tests.rs +++ b/crates/tracedecay-rusqlite-runtime/src/repository/observation/tests.rs @@ -394,7 +394,21 @@ fn connection() -> Connection { observation_sequence INTEGER NOT NULL UNIQUE, attempt_count INTEGER NOT NULL DEFAULT 0, next_retry_at_micros INTEGER NOT NULL DEFAULT 0, - last_error TEXT + last_error TEXT, + revival_generation TEXT, + revival_start_offset INTEGER + ); + CREATE TABLE observation_source_presence ( + source_key TEXT NOT NULL, + observation_id TEXT NOT NULL, + generation TEXT NOT NULL, + start_offset INTEGER NOT NULL, + PRIMARY KEY(source_key, observation_id) + ); + CREATE TABLE observation_projection_dispositions ( + projector_version TEXT NOT NULL, + observation_id TEXT NOT NULL, + reason TEXT NOT NULL ); CREATE TABLE retrieval_anchors ( anchor_id TEXT PRIMARY KEY, diff --git a/crates/tracedecay-session-temporal-store/src/lib.rs b/crates/tracedecay-session-temporal-store/src/lib.rs index bf57ec23e5..66539fc91b 100644 --- a/crates/tracedecay-session-temporal-store/src/lib.rs +++ b/crates/tracedecay-session-temporal-store/src/lib.rs @@ -111,7 +111,8 @@ pub use doctor_health::{ SessionTemporalHealthFindingKind, SessionTemporalHealthReport, SessionTemporalHealthStatus, }; pub use projection::{ - record_canonical_observation_effect, request_session_temporal_reset, + record_canonical_observation_effect, + record_live_collision_preserving_canonical_observation_effect, request_session_temporal_reset, request_session_temporal_reset_if_materialized, }; pub use refresh::{ diff --git a/crates/tracedecay-session-temporal-store/src/operations/message_anchor.rs b/crates/tracedecay-session-temporal-store/src/operations/message_anchor.rs index 96b3e837fa..89286346e4 100644 --- a/crates/tracedecay-session-temporal-store/src/operations/message_anchor.rs +++ b/crates/tracedecay-session-temporal-store/src/operations/message_anchor.rs @@ -32,6 +32,7 @@ use tracedecay_runtime_core::db::engine::params; use tracedecay_store::derive_canonical_projection; use super::sources::unavailable; +use crate::sql::live_effect_predicate; /// Resolved canonical source binding: anchor id, whether the publication still /// has to write an unobserved raw-message anchor row, and the source's @@ -126,7 +127,8 @@ async fn resolve_materialized_occurrences( serde_json::to_string(message_ids).map_err(|error| LcmError::Db(error.to_string()))?; let mut rows = conn .query( - "SELECT DISTINCT occurrence.message_id, occurrence.retrieval_anchor_id, + &format!( + "SELECT DISTINCT occurrence.message_id, occurrence.retrieval_anchor_id, anchor.anchor_json, anchor.owner_json, occurrence.knowledge_at, observation.observation_json, observation.receipt_id FROM session_occurrences occurrence @@ -134,10 +136,15 @@ async fn resolve_materialized_occurrences( ON anchor.anchor_id = occurrence.retrieval_anchor_id JOIN observations observation ON observation.observation_id = occurrence.source_observation_id + JOIN session_temporal_observation_effects AS effect + ON effect.observation_id = observation.observation_id WHERE occurrence.session_id = ?1 AND +occurrence.generation <= ?2 AND occurrence.message_id IN (SELECT value FROM json_each(?3)) + AND {} ORDER BY occurrence.message_id, occurrence.retrieval_anchor_id", + live_effect_predicate("(SELECT COALESCE(MAX(sequence), 0) FROM observations)") + ), params![session_id, generation, encoded_ids], ) .await?; @@ -218,7 +225,8 @@ async fn resolve_canonical_observations( let wanted = unresolved.iter().copied().collect::>(); let mut rows = conn .query( - "SELECT observation.observation_json, observation.receipt_id, + &format!( + "SELECT observation.observation_json, observation.receipt_id, effect.receipt_id, link.anchor_id, anchor.anchor_json, anchor.owner_json FROM session_temporal_observation_effects AS effect @@ -230,7 +238,10 @@ async fn resolve_canonical_observations( ON anchor.anchor_id = link.anchor_id WHERE effect.session_id = ?1 AND effect.output_count > 0 + AND {} ORDER BY effect.observation_sequence, link.anchor_id", + live_effect_predicate("(SELECT COALESCE(MAX(sequence), 0) FROM observations)") + ), params![session_id], ) .await?; @@ -707,6 +718,35 @@ mod tests { .expect("materialized occurrence"); } + async fn mark_output_collision( + conn: &impl crate::handle::SessionTemporalExec, + observation: &DurableObservationV1, + ) { + conn.execute( + "DELETE FROM observation_projection_provenance + WHERE projector_version = ?1 AND observation_id = ?2", + params![ + tracedecay_store::SESSION_MESSAGE_PROJECTOR_VERSION, + observation.observation_id().as_str(), + ], + ) + .await + .expect("collision releases live provenance"); + conn.execute( + "INSERT INTO observation_projection_dispositions ( + projector_version, observation_id, receipt_id, reason + ) VALUES (?1, ?2, ?3, ?4)", + params![ + tracedecay_store::SESSION_MESSAGE_PROJECTOR_VERSION, + observation.observation_id().as_str(), + observation.receipt().receipt().receipt_id().as_str(), + tracedecay_store::ProjectionSkipReason::OutputCollision.as_str(), + ], + ) + .await + .expect("live collision disposition"); + } + fn publication_over(store_ids: &[i64]) -> LcmImmutableSummaryPublication { let mut publication = publication(); publication.draft.source_refs = store_ids @@ -991,6 +1031,75 @@ mod tests { )); } + #[asupersync::test] + async fn collided_observation_is_not_a_canonical_unmaterialized_source() { + let directory = tempdir().expect("temporary directory"); + let runtime = HostAdmissionTestRuntimeV1::profile(directory.path()) + .await + .expect("registered profile runtime"); + let conn = runtime + .registered_database(HostAdmissionScope::Profile) + .expect("profile database") + .writer_connection() + .expect("profile writer"); + seed_raw_source(&conn, "1715000001").await; + let observation = + fixture_observation("codex", "session.message-anchor", "message.source", 1); + let anchor = fixture_anchor(&observation); + seed_canonical_binding( + &conn, + &serde_json::to_string(&observation).expect("observation json"), + &observation, + &anchor, + &anchor.owner_column_json().expect("owner json"), + ) + .await; + mark_output_collision(&conn, &observation).await; + + let sources = super::super::sources::prepare_sources(&conn, &publication()) + .await + .expect("collided canonical receipt is excluded from live ownership"); + + assert_eq!(sources.len(), 1); + assert!(sources[0].unobserved_raw_anchor); + assert_ne!(sources[0].canonical.id, anchor.anchor_id().as_str()); + } + + #[asupersync::test] + async fn collided_observation_is_not_a_canonical_materialized_source() { + let directory = tempdir().expect("temporary directory"); + let runtime = HostAdmissionTestRuntimeV1::profile(directory.path()) + .await + .expect("registered profile runtime"); + let conn = runtime + .registered_database(HostAdmissionScope::Profile) + .expect("profile database") + .writer_connection() + .expect("profile writer"); + seed_raw_source(&conn, "1715000001").await; + let observation = + fixture_observation("codex", "session.message-anchor", "message.source", 1); + let anchor = fixture_anchor(&observation); + seed_canonical_binding( + &conn, + &serde_json::to_string(&observation).expect("observation json"), + &observation, + &anchor, + &anchor.owner_column_json().expect("owner json"), + ) + .await; + materialize_occurrence(&conn, &observation, &anchor, "message.source").await; + mark_output_collision(&conn, &observation).await; + + let sources = super::super::sources::prepare_sources(&conn, &publication()) + .await + .expect("collided materialized receipt is excluded from live ownership"); + + assert_eq!(sources.len(), 1); + assert!(sources[0].unobserved_raw_anchor); + assert_ne!(sources[0].canonical.id, anchor.anchor_id().as_str()); + } + #[asupersync::test] async fn unavailable_session_owner_never_inserts_an_unobserved_anchor() { let directory = tempdir().expect("temporary directory"); diff --git a/crates/tracedecay-session-temporal-store/src/projection.rs b/crates/tracedecay-session-temporal-store/src/projection.rs index 53165af2aa..064f8aff06 100644 --- a/crates/tracedecay-session-temporal-store/src/projection.rs +++ b/crates/tracedecay-session-temporal-store/src/projection.rs @@ -28,7 +28,8 @@ use receipts::full_projection_coverage; pub(super) use receipts::validate_final_projection_receipt; pub(crate) use receipts::{base_source_frontier, digest_bytes}; pub use receipts::{ - record_canonical_observation_effect, request_session_temporal_reset, + record_canonical_observation_effect, + record_live_collision_preserving_canonical_observation_effect, request_session_temporal_reset, request_session_temporal_reset_if_materialized, }; diff --git a/crates/tracedecay-session-temporal-store/src/projection/receipts.rs b/crates/tracedecay-session-temporal-store/src/projection/receipts.rs index dc05ca15c8..99d2478555 100644 --- a/crates/tracedecay-session-temporal-store/src/projection/receipts.rs +++ b/crates/tracedecay-session-temporal-store/src/projection/receipts.rs @@ -6,8 +6,8 @@ use tracedecay_domain::RetrievalAnchorRecord; use tracedecay_domain::canonical_text::encode_tagged_lowercase_hex; use tracedecay_runtime_core::db::engine::{Executor, params}; use tracedecay_store::{ - ObservationProjection, ProjectionStoreError, ProjectionStoreResult, SessionStoreResult, - SessionTemporalDigestV1, SessionTemporalProjectionBatchReceiptV1, + ObservationProjection, ProjectionSkipReason, ProjectionStoreError, ProjectionStoreResult, + SessionStoreResult, SessionTemporalDigestV1, SessionTemporalProjectionBatchReceiptV1, SessionTemporalProjectionBatchV1, }; use tracedecay_temporal_query::execution::ExecutionControl; @@ -549,6 +549,82 @@ pub async fn record_canonical_observation_effect( } } +/// Records a live output-collision skip without rewriting an immutable +/// positive-output receipt from an earlier projection of the same observation. +/// The live disposition is the authority that the output is no longer +/// materialized; the retained receipt remains the authority for the original +/// projection and must still agree on observation sequence, session, and +/// sanitization receipt. Any disagreement remains a provenance collision. A +/// zero-output or absent row takes the ordinary full-digest path. +#[tracing::instrument( + name = "session_temporal.persist.live_collision_effect", + level = "trace", + skip_all +)] +pub async fn record_live_collision_preserving_canonical_observation_effect( + conn: &impl Executor, + sequence: u64, + observation: &tracedecay_domain::DurableObservationV1, + effect: &ObservationProjection, +) -> ProjectionStoreResult<()> { + if !matches!( + effect, + ObservationProjection::Skipped(ProjectionSkipReason::OutputCollision) + ) { + return record_canonical_observation_effect(conn, sequence, observation, effect).await; + } + let Ok(envelope) = observation_envelope_from_payload(observation.payload()) else { + return record_canonical_observation_effect(conn, sequence, observation, effect).await; + }; + let sequence_i64 = + i64::try_from(sequence).map_err(|_| ProjectionStoreError::SequenceOverflow(sequence))?; + let mut rows = conn + .query( + "SELECT observation_sequence, session_id, receipt_id, output_count + FROM session_temporal_observation_effects WHERE observation_id = ?1", + params![observation.observation_id().as_str()], + ) + .await + .map_err(|error| { + ProjectionStoreError::storage("read canonical temporal collision effect", error) + })?; + let Some(row) = rows.next().await.map_err(|error| { + ProjectionStoreError::storage("read canonical temporal collision effect", error) + })? + else { + drop(rows); + return record_canonical_observation_effect(conn, sequence, observation, effect).await; + }; + let actual = ( + row.get::(0).map_err(|error| { + ProjectionStoreError::storage("read canonical temporal collision effect", error) + })?, + row.get::(1).map_err(|error| { + ProjectionStoreError::storage("read canonical temporal collision effect", error) + })?, + row.get::(2).map_err(|error| { + ProjectionStoreError::storage("read canonical temporal collision effect", error) + })?, + row.get::(3).map_err(|error| { + ProjectionStoreError::storage("read canonical temporal collision effect", error) + })?, + ); + drop(rows); + if actual.3 <= 0 { + return record_canonical_observation_effect(conn, sequence, observation, effect).await; + } + let expected = ( + sequence_i64, + envelope.relations().session_id().as_str(), + observation.receipt().receipt().receipt_id().as_str(), + ); + if (actual.0, actual.1.as_str(), actual.2.as_str()) == expected { + Ok(()) + } else { + Err(ProjectionStoreError::ProvenanceCollision) + } +} + fn sorted_json(values: &[T]) -> SessionStoreResult> { let mut encoded = values .iter() diff --git a/crates/tracedecay-session-temporal-store/src/projection/tests.rs b/crates/tracedecay-session-temporal-store/src/projection/tests.rs index 3908b37ebb..1bc95d954c 100644 --- a/crates/tracedecay-session-temporal-store/src/projection/tests.rs +++ b/crates/tracedecay-session-temporal-store/src/projection/tests.rs @@ -27,7 +27,10 @@ use tracedecay_temporal_query::execution::ExecutionControl; use super::super::refresh::SessionRefreshRestartStateV1; use super::materialize::*; -use super::record_canonical_observation_effect; +use super::{ + record_canonical_observation_effect, + record_live_collision_preserving_canonical_observation_effect, +}; use crate::handle::SessionTemporalRegisteredDb; use crate::relations::{SessionRelationProjection, SessionRelationScope}; use crate::test_support::{FailingRegisteredDb, QueryCountingConnection}; @@ -2029,6 +2032,67 @@ async fn canonical_effect_replay_converges_on_an_identical_row() { ); } +#[asupersync::test] +async fn live_collision_preserves_an_immutable_positive_output_effect() { + let (_directory, connection) = open_effect_store("effect-live-collision").await; + let session_id = fixture_session("session.projector.effect-live-collision"); + let (observation, _) = fixture_observation(&session_id, 0, None, false); + let sequence = seed_effect_observation(&connection, &observation).await; + let original = tracedecay_store::derive_canonical_projection(&observation).unwrap(); + record_canonical_observation_effect(&connection, sequence, &observation, &original) + .await + .unwrap(); + let first = recorded_effect(&connection, &observation).await.unwrap(); + assert_eq!(first.3, 1); + + record_live_collision_preserving_canonical_observation_effect( + &connection, + sequence, + &observation, + &ObservationProjection::Skipped(ProjectionSkipReason::OutputCollision), + ) + .await + .unwrap(); + + assert_eq!( + recorded_effect(&connection, &observation).await, + Some(first) + ); +} + +#[asupersync::test] +async fn live_collision_rejects_a_mismatched_positive_output_authority() { + let (_directory, connection) = open_effect_store("effect-live-collision-mismatch").await; + let session_id = fixture_session("session.projector.effect-live-collision-mismatch"); + let (observation, _) = fixture_observation(&session_id, 0, None, false); + let sequence = seed_effect_observation(&connection, &observation).await; + connection + .execute( + "INSERT INTO session_temporal_observation_effects ( + observation_id, observation_sequence, session_id, receipt_id, + effect_digest, output_count, recorded_at + ) VALUES (?1, ?2, 'session.foreign', ?3, 'effect.foreign', 1, 1)", + params![ + observation.observation_id().as_str(), + i64::try_from(sequence).unwrap(), + observation.receipt().receipt().receipt_id().as_str(), + ], + ) + .await + .unwrap(); + + let error = record_live_collision_preserving_canonical_observation_effect( + &connection, + sequence, + &observation, + &ObservationProjection::Skipped(ProjectionSkipReason::OutputCollision), + ) + .await + .unwrap_err(); + + assert!(matches!(error, ProjectionStoreError::ProvenanceCollision)); +} + /// Case 3, conflict with a divergent payload. The durable row satisfies the /// insert guard (same observation, sequence, and receipt) yet disagrees on the /// projected effect, so the conflict-only read-back must still reject it. diff --git a/crates/tracedecay-sessions/src/runtime/source/jsonl.rs b/crates/tracedecay-sessions/src/runtime/source/jsonl.rs index 8fabfb74a5..e491142426 100644 --- a/crates/tracedecay-sessions/src/runtime/source/jsonl.rs +++ b/crates/tracedecay-sessions/src/runtime/source/jsonl.rs @@ -1,6 +1,6 @@ use std::cell::Cell; use std::collections::hash_map::DefaultHasher; -use std::collections::{BinaryHeap, HashMap, HashSet}; +use std::collections::{BinaryHeap, HashMap}; use std::hash::{Hash, Hasher}; use std::io::{BufRead, BufReader, Read, Seek, SeekFrom}; use std::path::Path; @@ -582,7 +582,8 @@ pub enum JsonlPrefixRecovery { /// can load the source's committed checkpoints. Report, /// Resume the rewritten generation after the longest prefix that still - /// hashes to one of these checkpoints; none matching rescans from zero. + /// hashes to one of these checkpoints the recorded generation committed; + /// none matching rescans from zero. Checkpoints(Arc<[JsonlPrefixCheckpoint]>), } @@ -705,16 +706,20 @@ pub(in crate::runtime) fn jsonl_prefix_digest( Ok((digest, hashed)) } -/// Longest checkpointed prefix of `file` whose digest still matches, or -/// `recorded` itself as soon as it matches, since resuming the recorded -/// generation needs no replacement. +/// Longest prefix of `file` that still hashes to a checkpoint the recorded +/// `generation` committed, or `recorded` itself as soon as it matches, since +/// resuming the recorded generation needs no replacement. /// -/// Checkpoints of one generation share one byte stream, so its first mismatch -/// rules out every later checkpoint of that generation; hashing stops once -/// every generation has diverged instead of walking the whole extent. +/// A rewrite keeps the recorded generation's records before the retained +/// prefix, so only that generation's checkpoints can prove one. An older +/// generation's checkpoint can match bytes the recorded generation had +/// replaced, and retaining them would keep the replacement's records and +/// never offer the restored ones again. The checkpoints share one byte +/// stream, so the first mismatch ends the walk. fn longest_retained_jsonl_prefix( file: &mut R, checkpoints: &[JsonlPrefixCheckpoint], + generation: u64, recorded: Option, extent: u64, ) -> std::io::Result<(Option<(JsonlPrefixCheckpoint, ResumeDigest)>, u64)> { @@ -722,33 +727,20 @@ fn longest_retained_jsonl_prefix( .iter() .copied() .chain(recorded) - .filter(|checkpoint| checkpoint.position > 0 && checkpoint.position <= extent) + .filter(|checkpoint| { + checkpoint.generation == generation + && checkpoint.position > 0 + && checkpoint.position <= extent + }) .collect::>(); - ordered.sort_unstable_by_key(|checkpoint| { - ( - checkpoint.position, - checkpoint.generation, - checkpoint.fingerprint, - ) - }); + ordered.sort_unstable_by_key(|checkpoint| (checkpoint.position, checkpoint.fingerprint)); ordered.dedup(); - let generations = ordered - .iter() - .map(|checkpoint| checkpoint.generation) - .collect::>(); - let mut diverged = HashSet::new(); let mut digest = ResumeDigest::new(); let mut hashed = 0_u64; let mut retained = None; let mut buffer = vec![0_u8; JSONL_HASH_CHUNK_BYTES]; file.seek(SeekFrom::Start(0))?; for checkpoint in ordered { - if diverged.len() == generations.len() { - break; - } - if diverged.contains(&checkpoint.generation) { - continue; - } while hashed < checkpoint.position { let requested = usize::try_from((checkpoint.position - hashed).min(buffer.len() as u64)) @@ -763,14 +755,13 @@ fn longest_retained_jsonl_prefix( digest.extend(&buffer[..read]); hashed = hashed.saturating_add(read as u64); } - if digest.fingerprint(checkpoint.position) == checkpoint.fingerprint { - if Some(checkpoint) == recorded { - return Ok((Some((checkpoint, digest)), hashed)); - } - retained = Some((checkpoint, digest.clone())); - } else { - diverged.insert(checkpoint.generation); + if digest.fingerprint(checkpoint.position) != checkpoint.fingerprint { + break; + } + if Some(checkpoint) == recorded { + return Ok((Some((checkpoint, digest)), hashed)); } + retained = Some((checkpoint, digest.clone())); } Ok((retained, hashed)) } @@ -787,6 +778,7 @@ enum RecordedPrefix { fn match_recorded_prefix( file: &mut MeasuredJsonlFile<'_>, path: &Path, + generation: u64, recorded: Option, prefix_recovery: &JsonlPrefixRecovery, extent: u64, @@ -809,7 +801,7 @@ fn match_recorded_prefix( } (_, JsonlPrefixRecovery::Checkpoints(checkpoints)) => { let (retained, hashed) = - longest_retained_jsonl_prefix(file, checkpoints, recorded, extent) + longest_retained_jsonl_prefix(file, checkpoints, generation, recorded, extent) .map_err(|error| TranscriptIngestError::scan_io("fingerprint", path, error))?; io.prefix_validation_bytes = io.prefix_validation_bytes.saturating_add(hashed); Ok(match retained { @@ -1326,6 +1318,7 @@ impl<'a> PreparedJsonlScan<'a> { match match_recorded_prefix( &mut file, path, + resume_state.generation, recorded, prefix_recovery, file_size, diff --git a/crates/tracedecay/tests/transcript_ingest_suite/claude.rs b/crates/tracedecay/tests/transcript_ingest_suite/claude.rs index 79e11aa960..c1b1b1b69c 100644 --- a/crates/tracedecay/tests/transcript_ingest_suite/claude.rs +++ b/crates/tracedecay/tests/transcript_ingest_suite/claude.rs @@ -1289,8 +1289,9 @@ async fn claude_observation_path_conflicting_redelivery_does_not_overwrite() { /// An atomically replaced transcript (temp file renamed over the original) /// is reconciled once: records the new file no longer holds stop being /// searchable, and an idle repoll of the replaced file reads no bytes. +/// Restoring the original makes its records searchable again. #[tokio::test] -async fn claude_atomic_replacement_retires_superseded_messages_and_settles() { +async fn claude_atomic_replacement_retires_superseded_messages_until_restored() { let tmp = TempDir::new().unwrap(); let (home, project) = setup(&tmp); init_git_repo(&project); @@ -1321,20 +1322,18 @@ async fn claude_atomic_replacement_retires_superseded_messages_and_settles() { let body: String = rows.iter().map(|row| format!("{row}\n")).collect(); std::fs::write(target, body).unwrap(); }; - write( - &path, - &[ - alpha.clone(), - row( - "c-a1", - Some("c-u1"), - "assistant", - 5, - "claudeprobe beta: square_area is called twice.", - ), - row("c-u2", Some("c-a1"), "user", 9, "claudeprobe gamma: thanks"), - ], - ); + let original = [ + alpha.clone(), + row( + "c-a1", + Some("c-u1"), + "assistant", + 5, + "claudeprobe beta: square_area is called twice.", + ), + row("c-u2", Some("c-a1"), "user", 9, "claudeprobe gamma: thanks"), + ]; + write(&path, &original); let db = open_project_session_db(&project).await.unwrap(); let source = ClaudeSource::with_home(&home); @@ -1401,6 +1400,44 @@ async fn claude_atomic_replacement_retires_superseded_messages_and_settles() { "idle repoll {repoll} of the replaced transcript must read nothing" ); } + + let mut conflicting = original.clone(); + conflicting[1]["message"]["content"][0]["text"] = + serde_json::json!("claudeprobe refused: a different answer under the retired UUID."); + write(&replacement, &conflicting); + std::fs::rename(&replacement, &path).unwrap(); + let refused = ingest().await; + assert_eq!(refused.records_rejected, 1); + let retired_hits = db.search_session_messages("claude", None, "beta", 10).await; + assert!( + retired_hits.is_empty(), + "a refused identity collision must not revive the retired record's old content: {retired_hits:?}" + ); + assert!( + db.search_session_messages("claude", None, "refused", 10) + .await + .is_empty() + ); + + write(&replacement, &original); + std::fs::rename(&replacement, &path).unwrap(); + ingest().await; + let mut texts: Vec = db + .search_session_messages("claude", None, "claudeprobe", 10) + .await + .into_iter() + .map(|hit| hit.message.text) + .collect(); + texts.sort(); + assert_eq!( + texts, + [ + "claudeprobe alpha: where is square_area called?", + "claudeprobe beta: square_area is called twice.", + "claudeprobe gamma: thanks", + ], + "restoring the original transcript must revive the messages the replacement retired" + ); } /// Production journal on #3393: Claude catch-up fails diff --git a/crates/tracedecay/tests/transcript_ingest_suite/codex_rollout_sources.rs b/crates/tracedecay/tests/transcript_ingest_suite/codex_rollout_sources.rs index 31932cc891..a4034b4c70 100644 --- a/crates/tracedecay/tests/transcript_ingest_suite/codex_rollout_sources.rs +++ b/crates/tracedecay/tests/transcript_ingest_suite/codex_rollout_sources.rs @@ -106,6 +106,41 @@ async fn revert_rollout_the_shipped_v2_binary_read( revert } +async fn seed_shipped_v2_rewrite_then_restore_thread( + home: &Path, + db: &ProjectSessionTestRuntime, + project: &Path, + thread_day: &str, + revert_day: &str, +) { + rollout( + home, + thread_day, + THREAD, + project, + &[LEDGER_MIGRATION, LEDGER_BACKFILL], + ); + ingest(home, db, project).await; + + let rewritten = rollout( + home, + thread_day, + THREAD, + project, + &[INVOICE_RETRY, INVOICE_ROLLBACK], + ); + ingest(home, db, project).await; + std::fs::rename(&rewritten, revert_rollout_path(home, revert_day)).unwrap(); + + rollout( + home, + thread_day, + THREAD, + project, + &[LEDGER_MIGRATION, LEDGER_BACKFILL], + ); +} + async fn assert_upgrade_leaves( home: &Path, db: &ProjectSessionTestRuntime, @@ -199,6 +234,36 @@ async fn rewriting_the_thread_rollout_supersedes_only_its_own_messages() { ); } +#[tokio::test] +async fn restoring_a_rewritten_rollout_revives_the_replies_the_rewrite_dropped() { + let tmp = TempDir::new().unwrap(); + let (home, project) = setup(&tmp); + let db = open_project_session_db(&project).await.unwrap(); + thread_rollout(&home, &project, &[LEDGER_MIGRATION, LEDGER_BACKFILL]); + ingest(&home, &db, &project).await; + + thread_rollout(&home, &project, &[LEDGER_MIGRATION, INVOICE_RETRY]); + ingest(&home, &db, &project).await; + assert_eq!( + replies(&db).await, + sorted(&[LEDGER_MIGRATION, INVOICE_RETRY]) + ); + + thread_rollout( + &home, + &project, + &[LEDGER_MIGRATION, LEDGER_BACKFILL, INVOICE_ROLLBACK], + ); + for pass in 1..=2 { + ingest(&home, &db, &project).await; + assert_eq!( + replies(&db).await, + sorted(&[LEDGER_MIGRATION, LEDGER_BACKFILL, INVOICE_ROLLBACK]), + "pass {pass}: the restored layout offers the dropped reply again and drops the rewrite's" + ); + } +} + async fn assert_a_shared_record_outlives_a_rewrite_of_either( thread_lines: &[serde_json::Value], revert_lines: &[serde_json::Value], @@ -298,28 +363,20 @@ async fn upgrade_keeps_both_rollouts_the_shipped_v2_binary_imported_together() { async fn upgrade_revives_a_revert_rollout_the_shipped_v2_binary_retired() { let tmp = TempDir::new().unwrap(); let (home, project) = setup(&tmp); - thread_rollout(&home, &project, &[LEDGER_MIGRATION, LEDGER_BACKFILL]); let db = open_project_session_db(&project).await.unwrap(); - ingest(&home, &db, &project).await; - assert_eq!( - replies(&db).await, - sorted(&[LEDGER_MIGRATION, LEDGER_BACKFILL]) - ); - revert_rollout_the_shipped_v2_binary_read( - &home, - &db, - &project, - "02", - &[INVOICE_RETRY, INVOICE_ROLLBACK], - ) - .await; + seed_shipped_v2_rewrite_then_restore_thread(&home, &db, &project, "01", "02").await; assert_upgrade_leaves( &home, &db, &project, - &[], &[INVOICE_RETRY, INVOICE_ROLLBACK], + &[ + LEDGER_MIGRATION, + LEDGER_BACKFILL, + INVOICE_RETRY, + INVOICE_ROLLBACK, + ], ) .await; } @@ -328,29 +385,20 @@ async fn upgrade_revives_a_revert_rollout_the_shipped_v2_binary_retired() { async fn upgrade_keeps_a_revert_rollout_the_shipped_v2_binary_read_last() { let tmp = TempDir::new().unwrap(); let (home, project) = setup(&tmp); - rollout( - &home, - "02", - THREAD, - &project, - &[LEDGER_MIGRATION, LEDGER_BACKFILL], - ); let db = open_project_session_db(&project).await.unwrap(); - revert_rollout_the_shipped_v2_binary_read( - &home, - &db, - &project, - "01", - &[INVOICE_RETRY, INVOICE_ROLLBACK], - ) - .await; + seed_shipped_v2_rewrite_then_restore_thread(&home, &db, &project, "02", "01").await; assert_upgrade_leaves( &home, &db, &project, &[INVOICE_RETRY, INVOICE_ROLLBACK], - &[INVOICE_RETRY, INVOICE_ROLLBACK], + &[ + LEDGER_MIGRATION, + LEDGER_BACKFILL, + INVOICE_RETRY, + INVOICE_ROLLBACK, + ], ) .await; }