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 6e41dcd610..f9cb1159b0 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 @@ -104,6 +104,28 @@ fn history_continuation(outcome: Option) -> Hist } } +/// Only an actually executed, unblocked history pass clears terminal backoff. +/// A suppressed wake has no evidence of recovery; a blocked pass may have +/// advanced healthy providers while the corrupt one remains blocked. +/// Conversely, a completed, pending, or retryable pass may clear the terminal +/// blocker without committing a source row. +fn clears_blocked_history_backoff( + outcome: Option<&SessionHistoricalIngestOutcome>, + release: Option<&HistoryRelease>, +) -> bool { + // Admission refusal returns Retryable without ever running the ingestor. + // Only a held history capacity subscription proves that a pass executed. + matches!(release, Some(HistoryRelease::Capacity(_))) + && matches!( + outcome, + Some( + SessionHistoricalIngestOutcome::Complete + | SessionHistoricalIngestOutcome::Pending { .. } + | SessionHistoricalIngestOutcome::Retryable { .. } + ) + ) +} + /// What a history pass that committed nothing waits on before it runs again. enum HistoryRelease { /// The daemon-wide historical admission had no free permit. @@ -333,17 +355,10 @@ pub(super) async fn run_session_temporal_refresh_scheduler( ) || history_outcome .as_ref() .is_some_and(SessionHistoricalIngestOutcome::made_progress); - // A pass that made progress does NOT prove the blocked cause - // healed: one healthy provider advancing while another stays - // blocked reports Blocked { made_progress: true }, and the - // blocked provider would be re-read at full cadence again. The - // backoff resets only on an outcome with no blocker at all. - if history_requires_projection - && !matches!( - history_outcome, - Some(SessionHistoricalIngestOutcome::Blocked { .. }) - ) - { + // A nonblocked attempt clears old terminal backoff even when it + // committed no new rows. A suppressed wake or progress by healthy + // providers while another remains Blocked cannot clear it. + if clears_blocked_history_backoff(history_outcome.as_ref(), history_release.as_ref()) { blocked_history_attempts = 0; blocked_history_until = None; } @@ -2604,6 +2619,60 @@ mod tests { assert!(!state.has_pending_work()); } + #[test] + fn terminal_history_backoff_clears_only_after_an_admitted_unblocked_pass() { + let admitted = HistoryRelease::Capacity(SessionHistoricalCapacityRelease::new(Vec::new())); + let refused = HistoryRelease::Admission; + assert!(!clears_blocked_history_backoff(None, Some(&admitted))); + for made_progress in [false, true] { + let blocked = SessionHistoricalIngestOutcome::Blocked { + reason_code: "projection_store_corrupt".to_owned(), + made_progress, + }; + assert!( + !clears_blocked_history_backoff(Some(&blocked), Some(&admitted)), + "a healthy sibling's progress cannot reset a corrupt provider's backoff" + ); + let pending = SessionHistoricalIngestOutcome::Pending { made_progress }; + assert!(clears_blocked_history_backoff( + Some(&pending), + Some(&admitted) + )); + assert!(!clears_blocked_history_backoff( + Some(&pending), + Some(&refused) + )); + let retryable = SessionHistoricalIngestOutcome::Retryable { + reason_code: "provider_temporarily_unavailable".to_owned(), + made_progress, + }; + assert!(clears_blocked_history_backoff( + Some(&retryable), + Some(&admitted) + )); + assert!(!clears_blocked_history_backoff( + Some(&retryable), + Some(&refused) + )); + } + let admission_saturated = SessionHistoricalIngestOutcome::Retryable { + reason_code: HISTORY_ADMISSION_SATURATED_REASON.to_owned(), + made_progress: false, + }; + assert!( + !clears_blocked_history_backoff(Some(&admission_saturated), Some(&refused)), + "blocked, admission-refused, then blocked again must retain escalating backoff" + ); + assert!(clears_blocked_history_backoff( + Some(&SessionHistoricalIngestOutcome::Complete), + Some(&admitted) + )); + assert!(!clears_blocked_history_backoff( + Some(&SessionHistoricalIngestOutcome::Cancelled), + Some(&admitted) + )); + } + #[test] fn a_window_that_committed_coverage_continues_and_one_that_did_not_awaits_release() { assert_eq!(