diff --git a/CHANGELOG.md b/CHANGELOG.md index cc4448d9bf..7c42d0b1c9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1710,6 +1710,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- *(admission)* `NotApplicable` is a terminal no-op in the shared replay-pass + decision. A closed status that leaves the spool unchanged now stops until + the next kick instead of entering the retryable backoff arm. + - *(code-index)* the background worker consults the typed publication-authority park instead of a loop-local bool, so a park it has not yet observed still stops reconcile. Branch publication handles `NotApplicable` as a closed diff --git a/crates/tracedecay-daemon-service/src/profile_host_admission_replay.rs b/crates/tracedecay-daemon-service/src/profile_host_admission_replay.rs index f8d79e2cd4..255e0622ef 100644 --- a/crates/tracedecay-daemon-service/src/profile_host_admission_replay.rs +++ b/crates/tracedecay-daemon-service/src/profile_host_admission_replay.rs @@ -841,6 +841,10 @@ impl ProfileHostAdmissionReplayWorker { // Non-retryable failure: stop until the next explicit kick. break; } + ReplayPassDecision::TerminalNoop => { + consecutive_retryable = 0; + break; + } ReplayPassDecision::Requeue => { consecutive_retryable = 0; } @@ -1391,6 +1395,43 @@ mod tests { registry.shutdown().await; } + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn not_applicable_pending_record_does_not_back_off() { + let temp = tempfile::TempDir::new().unwrap(); + let profile_root = temp.path().join("profile"); + std::fs::create_dir_all(&profile_root).unwrap(); + let db_path = tracedecay_sessions::runtime::user_sessions_db_path(&profile_root); + let (runtime, _) = + tracedecay_host_admission::HostAdmissionRuntime::open_for_database(&db_path).unwrap(); + let broker = Arc::new(tracedecay_host_admission::HostAdmissionBroker::new(runtime)); + broker.admit("test:pending", b"pending").await.unwrap(); + let registry = ProfileHostAdmissionReplayRegistry::default(); + let pass_override = Arc::new(|| { + Box::pin(async { HostAdmissionOutcome::not_applicable("code_index_not_applicable") }) + as std::pin::Pin + Send>> + }); + + registry + .ensure_with_pass_override(&db_path, &profile_root, &broker, pass_override) + .await; + tokio::time::timeout(Duration::from_secs(1), async { + while registry.pass_count(&db_path).await == 0 { + tokio::task::yield_now().await; + } + }) + .await + .expect("profile replay must attempt the not-applicable record"); + tokio::time::sleep(Duration::from_millis(150)).await; + let passes = registry.pass_count(&db_path).await; + assert!(passes >= 1); + assert_eq!(registry.backoff_count(&db_path).await, 0); + tokio::time::sleep(Duration::from_millis(150)).await; + + assert_eq!(registry.pass_count(&db_path).await, passes); + assert_eq!(registry.backoff_count(&db_path).await, 0); + registry.shutdown().await; + } + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn shutdown_cancels_and_joins_an_in_flight_pass() { let temp = tempfile::TempDir::new().unwrap(); diff --git a/crates/tracedecay-host-admission/src/replay.rs b/crates/tracedecay-host-admission/src/replay.rs index ceffe51442..b9b1588c39 100644 --- a/crates/tracedecay-host-admission/src/replay.rs +++ b/crates/tracedecay-host-admission/src/replay.rs @@ -7,7 +7,7 @@ use std::time::Duration; -use tracedecay_sessions::admission::HostAdmissionOutcome; +use tracedecay_sessions::admission::{HostAdmissionOutcome, HostAdmissionStatus}; const MAX_BACKOFF: Duration = Duration::from_secs(2); const INITIAL_BACKOFF: Duration = Duration::from_millis(25); @@ -28,6 +28,7 @@ pub fn replay_backoff(attempt: u32, shift_cap: u32) -> Duration { } /// How a worker should proceed after one replay pass. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] pub enum ReplayPassDecision { /// The spool shrank and more work remains. Yield and re-run immediately. ProgressPending, @@ -37,15 +38,33 @@ pub enum ReplayPassDecision { Stop, /// Re-evaluate the work condition without backoff. Requeue, + /// Closed `NotApplicable` left the spool unchanged. Stop until the next + /// kick without backoff and without a failure log. + TerminalNoop, } /// Classify one replay pass from its pending-count delta and outcome. +/// +/// `NotApplicable` already closes the replay record. It must not enter the +/// retryable backoff arm, even when `is_replay_progress` is true and the +/// spool did not shrink: that arm is for work that may succeed later. A shrink +/// with records still pending continues immediately; a drained spool requeues; +/// an unchanged spool stops until the next external kick. pub fn classify_replay_pass( pending_before: usize, pending_after: usize, outcome: &HostAdmissionOutcome, ) -> ReplayPassDecision { let made_progress = pending_after < pending_before; + if outcome.status == HostAdmissionStatus::NotApplicable { + if made_progress && pending_after > 0 { + return ReplayPassDecision::ProgressPending; + } + if pending_after == 0 { + return ReplayPassDecision::Requeue; + } + return ReplayPassDecision::TerminalNoop; + } if made_progress && pending_after > 0 { ReplayPassDecision::ProgressPending } else if !made_progress @@ -58,3 +77,42 @@ pub fn classify_replay_pass( ReplayPassDecision::Requeue } } + +#[cfg(test)] +mod tests { + use super::{ReplayPassDecision, classify_replay_pass}; + use tracedecay_sessions::admission::HostAdmissionOutcome; + + #[test] + fn not_applicable_is_a_terminal_noop_unless_the_spool_actually_moves() { + let closed = HostAdmissionOutcome::not_applicable("code_index_not_applicable"); + let mut flagged_retryable = closed.clone(); + flagged_retryable.retryable = true; + + assert_eq!( + classify_replay_pass(2, 2, &closed), + ReplayPassDecision::TerminalNoop + ); + assert_eq!( + classify_replay_pass(2, 3, &flagged_retryable), + ReplayPassDecision::TerminalNoop + ); + assert_eq!( + classify_replay_pass(2, 1, &closed), + ReplayPassDecision::ProgressPending + ); + assert_eq!( + classify_replay_pass(1, 0, &closed), + ReplayPassDecision::Requeue + ); + + assert_eq!( + classify_replay_pass(2, 2, &HostAdmissionOutcome::accepted_for_replay()), + ReplayPassDecision::Backoff + ); + assert_eq!( + classify_replay_pass(2, 2, &HostAdmissionOutcome::spool_corrupted()), + ReplayPassDecision::Stop + ); + } +} diff --git a/crates/tracedecay-mcp/src/server/project_host_admission_replay.rs b/crates/tracedecay-mcp/src/server/project_host_admission_replay.rs index 6d4f8c5a1d..851e75b05a 100644 --- a/crates/tracedecay-mcp/src/server/project_host_admission_replay.rs +++ b/crates/tracedecay-mcp/src/server/project_host_admission_replay.rs @@ -203,6 +203,10 @@ impl ProjectHostAdmissionReplayWorker { ); break; } + ReplayPassDecision::TerminalNoop => { + consecutive_retryable = 0; + break; + } ReplayPassDecision::Requeue => { consecutive_retryable = 0; if self.dirty.load(Ordering::Acquire) || pending_after > 0 { @@ -275,6 +279,42 @@ mod tests { task.shutdown().await; } + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn not_applicable_pending_record_stops_without_backoff() { + let temp = tempfile::TempDir::new().unwrap(); + let (runtime, _) = tracedecay_host_admission::HostAdmissionRuntime::open( + temp.path(), + tracedecay_host_admission::SpoolBounds::default(), + ) + .unwrap(); + let broker = Arc::new(tracedecay_host_admission::HostAdmissionBroker::new(runtime)); + broker.admit("test:pending", b"pending").await.unwrap(); + let passes = Arc::new(AtomicUsize::new(0)); + let passes_for_run = Arc::clone(&passes); + let pass: PassFn = Arc::new(move || { + let passes = Arc::clone(&passes_for_run); + Box::pin(async move { + passes.fetch_add(1, Ordering::AcqRel); + HostAdmissionOutcome::not_applicable("code_index_not_applicable") + }) + }); + let task = ProjectHostAdmissionReplayTask::start(broker, pass); + + tokio::time::timeout(Duration::from_secs(1), async { + while task.pass_count() == 0 { + tokio::task::yield_now().await; + } + }) + .await + .expect("project replay must attempt the not-applicable record"); + tokio::time::sleep(Duration::from_millis(100)).await; + + assert_eq!(passes.load(Ordering::Acquire), 1); + assert_eq!(task.pass_count(), 1); + assert_eq!(task.backoff_count(), 0); + task.shutdown().await; + } + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn dropping_task_aborts_an_in_flight_pass_without_an_arc_cycle() { let temp = tempfile::TempDir::new().unwrap(); diff --git a/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/projector.rs b/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/projector.rs index 10818733c1..632898ca24 100644 --- a/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/projector.rs +++ b/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/projector.rs @@ -119,11 +119,11 @@ impl SessionTemporalRefreshProjector for CanonicalSessionTemporalProjector { // Empty remaining range is a durable no-op: terminalize with an // empty complete progress batch instead of deferring forever. Ok(None) => canonical_noop_complete_effect(&recovery), - Err(error) if error.is_storage() => Err( - SessionTemporalRefreshProjectorError::retryable(format!( + Err(error) if error.is_storage() => { + Err(SessionTemporalRefreshProjectorError::retryable(format!( "source_busy: {error}" - )), - ), + ))) + } Err(_) => Err(SessionTemporalRefreshProjectorError::terminal( "projector_failed", )), diff --git a/crates/tracedecay-session-temporal-store/src/query.rs b/crates/tracedecay-session-temporal-store/src/query.rs index 367b14bb79..ae0ab6b4de 100644 --- a/crates/tracedecay-session-temporal-store/src/query.rs +++ b/crates/tracedecay-session-temporal-store/src/query.rs @@ -246,8 +246,7 @@ pub(super) async fn read_observations( // ceiling. Split until a single observation remains; that // observation is then a typed storage failure, not a retry // that looks like a busy source. - if observation_prefetch_exceeded_materialization_limit(&error) && chunk.len() > 1 - { + if observation_prefetch_exceeded_materialization_limit(&error) && chunk.len() > 1 { let mid = start + chunk.len() / 2; pending.push((mid, end)); pending.push((start, mid)); @@ -289,11 +288,18 @@ fn observation_prefetch_exceeded_materialization_limit(error: &SessionStoreError } } +/// The error `read_observation` raises for an id the store does not hold, reused +/// by callers that resolve prefetched observations out of a batch map. +pub(super) fn missing_observation(observation_id: &CanonicalObservationIdV1) -> SessionStoreError { + storage_message( + PERSIST_OPERATION, + format!("source observation {} is missing", observation_id.as_str()), + ) +} + #[cfg(test)] mod tests { - use super::{ - PERSIST_OPERATION, observation_prefetch_exceeded_materialization_limit, storage, - }; + use super::{PERSIST_OPERATION, observation_prefetch_exceeded_materialization_limit, storage}; #[test] fn materialization_limit_is_the_prefetch_split_signal() { @@ -308,15 +314,8 @@ mod tests { assert!(observation_prefetch_exceeded_materialization_limit( &exceeded )); - assert!(!observation_prefetch_exceeded_materialization_limit(&locked)); + assert!(!observation_prefetch_exceeded_materialization_limit( + &locked + )); } } - -/// The error `read_observation` raises for an id the store does not hold, reused -/// by callers that resolve prefetched observations out of a batch map. -pub(super) fn missing_observation(observation_id: &CanonicalObservationIdV1) -> SessionStoreError { - storage_message( - PERSIST_OPERATION, - format!("source observation {} is missing", observation_id.as_str()), - ) -}