diff --git a/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/worker.rs b/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/worker.rs index 8c775cb9a8..e94b21337e 100644 --- a/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/worker.rs +++ b/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/worker.rs @@ -1057,7 +1057,15 @@ async fn reject_staged_completion<'a>( &recovery, durable_projector_failure_code(REFRESH_COMPLETION_REFUSED), ) { - Some(request) => apply_fail_effect(store, state, &recovery, request, report).await, + Some(request) => { + let disposition = match error { + SessionStoreError::RefreshSupersededByReset => { + RefreshFailureDisposition::Rediscover + } + _ => RefreshFailureDisposition::SuppressDiscovery, + }; + apply_fail_effect(store, state, &recovery, request, disposition, report).await; + } None => report.terminal_errors += 1, } return; @@ -1431,7 +1439,15 @@ pub async fn apply_refresh_effect( durable_projector_failure_code(REFRESH_PROGRESS_REFUSED), ) { Some(request) => { - apply_fail_effect(store, state, recovery, request, report).await; + apply_fail_effect( + store, + state, + recovery, + request, + RefreshFailureDisposition::SuppressDiscovery, + report, + ) + .await; } None => report.terminal_errors += 1, } @@ -1446,7 +1462,15 @@ pub async fn apply_refresh_effect( } } SessionTemporalRefreshEffect::Fail(request) => { - apply_fail_effect(store, state, recovery, request, report).await; + apply_fail_effect( + store, + state, + recovery, + request, + RefreshFailureDisposition::SuppressDiscovery, + report, + ) + .await; } SessionTemporalRefreshEffect::Deferred => report.deferred += 1, } @@ -1530,7 +1554,15 @@ async fn apply_prepared_refresh_effect( SessionRefreshDispositionV1::Joined => report.joined += 1, } if receipt.operation_id() == prepared.recovery.operation_id() { - apply_fail_effect(store, state, &prepared.recovery, request, report).await; + apply_fail_effect( + store, + state, + &prepared.recovery, + request, + RefreshFailureDisposition::SuppressDiscovery, + report, + ) + .await; } else { report.terminal_errors += 1; } @@ -1550,11 +1582,17 @@ async fn apply_prepared_refresh_effect( } } +enum RefreshFailureDisposition { + SuppressDiscovery, + Rediscover, +} + async fn apply_fail_effect( store: &SessionTemporalStore<'_, tracedecay_global_db::RegisteredGlobalDb>, state: &SessionTemporalRefreshWakeState, recovery: &SessionRefreshRecoveryV1, request: SessionRefreshFailureRequestV1, + disposition: RefreshFailureDisposition, report: &mut SessionTemporalRefreshPassReport, ) { if !state.claim_terminal_attempt(recovery) { @@ -1571,7 +1609,9 @@ async fn apply_fail_effect( "session temporal refresh failed durably" ); report.failed += 1; - state.record_terminal_discovery_failure(recovery); + if matches!(disposition, RefreshFailureDisposition::SuppressDiscovery) { + state.record_terminal_discovery_failure(recovery); + } } Err(error) if is_retryable_storage(&error) => { report.last_error = Some(format!("{error:?}")); diff --git a/crates/tracedecay-session-temporal-store/src/refresh.rs b/crates/tracedecay-session-temporal-store/src/refresh.rs index 3b6ae3fad2..3cd1217f78 100644 --- a/crates/tracedecay-session-temporal-store/src/refresh.rs +++ b/crates/tracedecay-session-temporal-store/src/refresh.rs @@ -3225,14 +3225,8 @@ async fn complete_session_refresh_in_transaction( COMPLETE_REFRESH, ) .await?; - // A begin applies any pending reset, so one pending now arrived while - // this refresh ran: its candidate covers effects the reset retired and can - // never match the live frontier. Refusing it terminally ends the operation - // so the next begin can apply the reset. if session_reset_is_pending(transaction, request.session_id()).await? { - return Err(SessionStoreError::InvalidStateTransition { - context: "refresh candidate was superseded by a requested session reset", - }); + return Err(SessionStoreError::RefreshSupersededByReset); } let progress = require_exact_terminal_progress( transaction, diff --git a/crates/tracedecay-store/src/session/common.rs b/crates/tracedecay-store/src/session/common.rs index 1412e43f50..ecb3f1d603 100644 --- a/crates/tracedecay-store/src/session/common.rs +++ b/crates/tracedecay-store/src/session/common.rs @@ -274,6 +274,8 @@ pub enum SessionStoreError { }, #[error("session temporal idempotency conflict in {context}")] IdempotencyConflict { context: &'static str }, + #[error("refresh candidate was superseded by a requested session reset")] + RefreshSupersededByReset, #[error("invalid session temporal state transition in {context}")] InvalidStateTransition { context: &'static str }, #[error("session temporal capability {capability:?} is unsupported")] diff --git a/crates/tracedecay/tests/session_suite/temporal_projection/activation.rs b/crates/tracedecay/tests/session_suite/temporal_projection/activation.rs index 4f81293b0a..4dd2d7b146 100644 --- a/crates/tracedecay/tests/session_suite/temporal_projection/activation.rs +++ b/crates/tracedecay/tests/session_suite/temporal_projection/activation.rs @@ -1,3 +1,5 @@ +use tracedecay_session_runtime::session_temporal_refresh_scheduler::SessionTemporalRefreshSchedulerRegistry; + use super::*; const OMITTED_LINEAGE_REFUSAL: &str = "session-temporal storage operation activate session temporal generation failed: candidate omits canonical typed assertion lineage through the frozen frontier"; @@ -315,11 +317,6 @@ async fn activation_is_pinned_to_the_generation_the_refresh_extends() { ); } -/// A source rewrite that retires an observation a running refresh already -/// projected asks for a session reset. That refresh can never prove its -/// frontier again, so completion refuses it as a terminal state transition -/// (never a retryable storage fault that would block the reset forever), and -/// the next refresh applies the reset and activates the rewritten history. #[tokio::test] async fn reset_requested_during_a_refresh_supersedes_its_candidate() { let tmp = TempDir::new().unwrap(); @@ -343,8 +340,6 @@ async fn reset_requested_during_a_refresh_supersedes_its_candidate() { ], ) .unwrap(); - // The reset request `request_session_temporal_reset` records for a - // retired observation that produced session output. let requested = rusqlite::Connection::open(&path) .unwrap() .execute( @@ -359,36 +354,42 @@ async fn reset_requested_during_a_refresh_supersedes_its_candidate() { .unwrap(); assert_eq!(requested, 1, "the retired observation requests one reset"); - let store = runtime - .session_temporal_store(HostAdmissionScope::Profile) - .unwrap(); - let refused = complete_candidate(&store, &stale).await; - assert!( - matches!( - refused, - Err(SessionStoreError::InvalidStateTransition { - context: "refresh candidate was superseded by a requested session reset" - }) - ), - "{refused:?}" - ); - assert_eq!(active_generations(&path).await, vec!["1:active"]); + let registry = SessionTemporalRefreshSchedulerRegistry::default(); + registry + .ensure_profile( + path.clone(), + runtime + .registered_database_arc(HostAdmissionScope::Profile) + .unwrap(), + ) + .await; + registry.wait_profile_idle(&path).await; + registry.shutdown().await; - fail_candidate(&store, &stale).await; - let rebuilt = begin_candidate(&store, &session_id, 2).await; - assert_ne!(rebuilt.operation_id(), stale.operation_id()); - let first = occurrence(&session_id, &observation(&session_id, 0, "one")); - persist_batch( - &store, - &rebuilt, - batch(&rebuilt, vec![first], vec![], vec![]), + assert_eq!( + rows( + &path, + "SELECT state FROM session_refresh_operations ORDER BY created_at, operation_id" + ) + .await, + vec!["failed", "complete"] + ); + let replacement_generation = rows( + &path, + "SELECT binding.generation || ':active' + FROM session_refresh_bindings AS binding + JOIN session_refresh_operations AS operation + ON operation.session_id = binding.session_id + AND operation.operation_id = binding.operation_id + WHERE operation.state = 'complete'", ) - .await - .unwrap(); - complete_candidate(&store, &rebuilt).await.unwrap(); + .await; + assert_eq!(replacement_generation.len(), 1); + assert_ne!(replacement_generation, vec!["1:active"]); + assert_eq!(active_generations(&path).await, replacement_generation); assert_eq!( - active_generations(&path).await, - vec![format!("{}:active", rebuilt.candidate_generation().value())] + rows(&path, "SELECT CAST(MAX(observation_sequence) AS TEXT) FROM session_temporal_observation_effects").await, + vec!["2"] ); assert_eq!( rows( @@ -399,6 +400,80 @@ async fn reset_requested_during_a_refresh_supersedes_its_candidate() { .await, vec!["1"] ); + assert_eq!( + rows( + &path, + "SELECT occurrence.message_id FROM session_occurrences AS occurrence + JOIN session_temporal_generations AS generation + ON generation.session_id = occurrence.session_id + AND generation.generation = occurrence.generation + WHERE generation.state = 'active' ORDER BY occurrence.message_id" + ) + .await, + vec![format!("message.temporal.{}.0", session_id.as_str())] + ); +} + +#[tokio::test] +async fn ambiguous_candidate_failure_does_not_repeat_for_an_unchanged_frontier() { + let tmp = TempDir::new().unwrap(); + let runtime = profile_runtime(&tmp).await; + let path = runtime + .database_path(HostAdmissionScope::Profile) + .unwrap() + .to_path_buf(); + let session_id = session("session.temporal.ambiguous-discovery"); + let observation_store = runtime + .observation_store(HostAdmissionScope::Profile) + .unwrap(); + let store = runtime + .session_temporal_store(HostAdmissionScope::Profile) + .unwrap(); + let duplicate = "message.temporal.duplicate"; + let mut occurrences = Vec::new(); + for ordinal in 0..2 { + let observation = persist_custom_observation( + &observation_store, + observation_with_message_ids(&session_id, ordinal, "duplicate", duplicate, None), + ) + .await; + occurrences.push(occurrence_with_message_id( + &session_id, + &observation, + duplicate, + )); + } + let candidate = begin_candidate(&store, &session_id, 2).await; + persist_batch( + &store, + &candidate, + batch(&candidate, occurrences, vec![], vec![]), + ) + .await + .unwrap(); + let registry = SessionTemporalRefreshSchedulerRegistry::default(); + let wake = registry + .ensure_profile( + path.clone(), + runtime + .registered_database_arc(HostAdmissionScope::Profile) + .unwrap(), + ) + .await; + registry.wait_profile_idle(&path).await; + assert_eq!( + rows(&path, "SELECT state FROM session_refresh_operations").await, + vec!["failed"] + ); + + wake.wake(); + registry.wait_profile_idle(&path).await; + registry.shutdown().await; + assert_eq!( + rows(&path, "SELECT state FROM session_refresh_operations").await, + vec!["failed"] + ); + assert_eq!(active_generations(&path).await, vec!["1:active"]); } /// Persists observations `one` and `two` in a fresh profile and projects the