From 5d986e052575009919352288dfcdfa37f4f3b1e9 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Tue, 3 Nov 2026 04:17:51 -0800 Subject: [PATCH 1/3] fix(sessions): reset blocked history backoff after recovery Clear the previous terminal retry escalation when an admitted history pass is no longer Blocked, even if it commits no new source rows. A suppressed wake and a Blocked result with healthy-sibling progress still cannot clear a corrupt provider's backoff. Add a regression covering pending/retryable zero-progress recovery and the blocked, cancelled and suppressed cases. Refs #4569. --- .../worker.rs | 59 +++++++++++++++---- 1 file changed, 48 insertions(+), 11 deletions(-) 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..eb5835eb12 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,22 @@ 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>) -> bool { + 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 +349,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()) { blocked_history_attempts = 0; blocked_history_until = None; } @@ -2604,6 +2613,34 @@ mod tests { assert!(!state.has_pending_work()); } + #[test] + fn terminal_history_backoff_clears_on_recovery_without_source_progress() { + assert!(!clears_blocked_history_backoff(None)); + 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)), + "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))); + let retryable = SessionHistoricalIngestOutcome::Retryable { + reason_code: "provider_temporarily_unavailable".to_owned(), + made_progress, + }; + assert!(clears_blocked_history_backoff(Some(&retryable))); + } + assert!(clears_blocked_history_backoff(Some( + &SessionHistoricalIngestOutcome::Complete + ))); + assert!(!clears_blocked_history_backoff(Some( + &SessionHistoricalIngestOutcome::Cancelled + ))); + } + #[test] fn a_window_that_committed_coverage_continues_and_one_that_did_not_awaits_release() { assert_eq!( From 075beb95a4459491438906b3963167e9ea549eb6 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Tue, 3 Nov 2026 04:28:04 -0800 Subject: [PATCH 2/3] fix(sessions): retain corruption backoff on denied history admission Use the existing typed HistoryRelease authority to distinguish an admitted history pass from synthetic retryable admission saturation. A denied permit cannot prove a blocked provider has recovered. Extend the existing outcome regressions with actual admission and no-progress distinctions. --- .../worker.rs | 60 ++++++++++++------- 1 file changed, 40 insertions(+), 20 deletions(-) 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 eb5835eb12..82a2d3f030 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 @@ -109,15 +109,21 @@ fn history_continuation(outcome: Option) -> Hist /// 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>) -> bool { - matches!( - outcome, - Some( - SessionHistoricalIngestOutcome::Complete - | SessionHistoricalIngestOutcome::Pending { .. } - | SessionHistoricalIngestOutcome::Retryable { .. } +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. @@ -352,7 +358,7 @@ pub(super) async fn run_session_temporal_refresh_scheduler( // 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()) { + if clears_blocked_history_backoff(history_outcome.as_ref(), history_release.as_ref()) { blocked_history_attempts = 0; blocked_history_until = None; } @@ -2614,31 +2620,45 @@ mod tests { } #[test] - fn terminal_history_backoff_clears_on_recovery_without_source_progress() { - assert!(!clears_blocked_history_backoff(None)); + 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)), + !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))); + 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))); + assert!(clears_blocked_history_backoff(Some(&retryable), Some(&admitted))); + assert!(!clears_blocked_history_backoff(Some(&retryable), Some(&refused))); } - assert!(clears_blocked_history_backoff(Some( - &SessionHistoricalIngestOutcome::Complete - ))); - assert!(!clears_blocked_history_backoff(Some( - &SessionHistoricalIngestOutcome::Cancelled - ))); + 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] From 319166d8b4cfd8d152da1667505f0bacc47b3586 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Tue, 3 Nov 2026 06:21:03 -0800 Subject: [PATCH 3/3] style(sessions): format history recovery regressions Apply the exact rustfmt_check_2024 output to the existing assertions. Preserve behavior and keep the history-backoff regression intact. --- .../worker.rs | 20 +++++++++++++++---- 1 file changed, 16 insertions(+), 4 deletions(-) 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 82a2d3f030..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 @@ -2634,14 +2634,26 @@ mod tests { "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))); + 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))); + 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(),