From fbbea1bf912233c1a06dd88ee35bd7fe9f7311c3 Mon Sep 17 00:00:00 2001 From: Hare Date: Wed, 16 Sep 2026 07:27:21 +0900 Subject: [PATCH] fix: scope compaction suppression to one run --- crates/worker/src/compact/state.rs | 288 ++++++++++++--------------- crates/worker/src/ipc/interceptor.rs | 10 +- crates/worker/src/worker.rs | 18 +- 3 files changed, 140 insertions(+), 176 deletions(-) diff --git a/crates/worker/src/compact/state.rs b/crates/worker/src/compact/state.rs index cebab0f2..2c49e68f 100644 --- a/crates/worker/src/compact/state.rs +++ b/crates/worker/src/compact/state.rs @@ -2,16 +2,17 @@ use std::sync::Mutex; use super::telemetry::CompactFailureCategory; -/// The process-local state of automatic compaction for one logical run. +/// Process-local automatic compaction guard for the current logical run. /// -/// This is deliberately not persisted: restoring a worker starts from -/// [`AutomaticCompactState::NotAttempted`]. The failure counter is likewise a -/// guard for one live worker process, not session authority. +/// This guard is deliberately not persisted or reconstructed from session +/// history, compaction metrics, or replacement-segment state. #[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub(crate) enum AutomaticCompactState { - NotAttempted, - Attempted, - Completed(CompactionOutcome), +pub(crate) enum AutomaticCompactGuard { + Ready, + SuppressedForCurrentRun { + failure_category: CompactFailureCategory, + }, + AwaitingPostCompactRequest, } #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -38,40 +39,35 @@ pub(crate) enum AutomaticCompactDecision { /// Typed reason why a provider request may not proceed. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub(crate) enum AutomaticCompactBlock { - /// Compaction succeeded but the replacement still exceeds the request - /// safety threshold before a post-compaction request was durably recorded. + /// An automatic attempt was already claimed for this logical run and has + /// not yet produced an outcome. + Attempted, + /// Compaction succeeded, but no post-compaction provider request has yet + /// committed a new occupancy UsageRecord. Thrash, /// This logical run already used its automatic attempt and it failed. Failed(CompactFailureCategory), - /// This logical run already used its automatic attempt and it was - /// cancelled. Cancellation is not counted as a compaction failure. + /// This logical run's automatic attempt was cancelled. Cancellation is not + /// classified or counted as a compaction failure. Cancelled, - /// Consecutive automatic failures disabled further attempts until a - /// successful manual compaction or explicit re-enable. - Disabled(CompactFailureCategory), } #[derive(Debug)] struct AutomaticCompactRuntimeState { - attempt: AutomaticCompactState, - consecutive_failures: u32, - automatic_disabled: bool, - last_failure: Option, + guard: AutomaticCompactGuard, + attempt_claimed: bool, + cancelled_attempt: bool, pending_request_block: Option, } -const MAX_CONSECUTIVE_AUTOMATIC_FAILURES: u32 = 2; - -/// Tracks automatic compaction thresholds and process-local loop guards. +/// Tracks automatic compaction thresholds and the current logical-run guard. #[derive(Debug)] pub(crate) struct CompactState { - /// Post-run threshold. Checked before a new user run starts. + /// Proactive threshold checked before a fresh user run starts. compact_threshold: Option, - /// In-request safety threshold. Checked immediately before each provider - /// request, both before and after pre-request hooks. + /// Safety threshold checked immediately before every provider request. request_threshold: Option, retained_tokens: u64, - max_consecutive_failures: u32, runtime: Mutex, } @@ -85,12 +81,10 @@ impl CompactState { compact_threshold, request_threshold, retained_tokens, - max_consecutive_failures: MAX_CONSECUTIVE_AUTOMATIC_FAILURES, runtime: Mutex::new(AutomaticCompactRuntimeState { - attempt: AutomaticCompactState::NotAttempted, - consecutive_failures: 0, - automatic_disabled: false, - last_failure: None, + guard: AutomaticCompactGuard::Ready, + attempt_claimed: false, + cancelled_attempt: false, pending_request_block: None, }), } @@ -108,26 +102,21 @@ impl CompactState { return false; } let runtime = self.lock_runtime(); - !runtime.automatic_disabled && runtime.attempt == AutomaticCompactState::NotAttempted + runtime.guard == AutomaticCompactGuard::Ready && !runtime.attempt_claimed } /// Starts a fresh logical run. Pause/resume paths must not call this. pub(crate) fn begin_logical_run(&self) { - let mut runtime = self.lock_runtime(); - runtime.attempt = AutomaticCompactState::NotAttempted; - runtime.pending_request_block = None; + self.clear_logical_run(); } - /// Clears per-run state after a terminal run outcome. The consecutive - /// failure circuit breaker intentionally survives logical-run boundaries. + /// Clears per-run state after a terminal run outcome. pub(crate) fn finish_logical_run(&self) { - let mut runtime = self.lock_runtime(); - runtime.attempt = AutomaticCompactState::NotAttempted; - runtime.pending_request_block = None; + self.clear_logical_run(); } - /// Atomically evaluates the pre-run threshold and claims this logical - /// run's one automatic compaction attempt when eligible. + /// Atomically evaluates the proactive threshold and claims this logical + /// run's automatic attempt when eligible. pub(crate) fn evaluate_pre_run(&self, total_tokens: u64) -> AutomaticCompactDecision { if !self .compact_threshold @@ -138,8 +127,8 @@ impl CompactState { self.claim_attempt(AutomaticCompactTrigger::PreRun) } - /// Atomically evaluates the request safety threshold and either claims the - /// run's one automatic attempt or records a typed request-block reason. + /// Atomically evaluates the request safety threshold and either claims an + /// automatic attempt or returns the typed reason the request must stop. pub(crate) fn evaluate_request(&self, total_tokens: u64) -> AutomaticCompactDecision { if !self .request_threshold @@ -147,45 +136,43 @@ impl CompactState { { return AutomaticCompactDecision::Continue; } - self.claim_attempt(AutomaticCompactTrigger::RequestThreshold) } + /// Claims a hook-originated compaction yield under the same guard used by + /// threshold evaluation. This exists even in manual-only configurations. + pub(crate) fn claim_hook_yield(&self) -> AutomaticCompactDecision { + self.claim_attempt(AutomaticCompactTrigger::RequestThreshold) + } + + pub(crate) fn has_claimed_attempt(&self) -> bool { + self.lock_runtime().attempt_claimed + } + pub(crate) fn record_request_block(&self, block: AutomaticCompactBlock) { self.lock_runtime().pending_request_block = Some(block); } - /// Claims a hook-originated compaction yield under the same one-attempt - /// guard used by threshold evaluation. - pub(crate) fn claim_hook_yield(&self) -> AutomaticCompactDecision { - self.claim_attempt(AutomaticCompactTrigger::RequestThreshold) - } - - pub(crate) fn has_claimed_attempt(&self) -> bool { - self.lock_runtime().attempt == AutomaticCompactState::Attempted - } - /// Completes the currently claimed automatic attempt exactly once. pub(crate) fn complete_automatic(&self, outcome: CompactionOutcome) -> bool { let mut runtime = self.lock_runtime(); - if runtime.attempt != AutomaticCompactState::Attempted { + if !runtime.attempt_claimed + || runtime.guard != AutomaticCompactGuard::Ready + || runtime.cancelled_attempt + { return false; } - runtime.attempt = AutomaticCompactState::Completed(outcome); match outcome { CompactionOutcome::Succeeded => { - runtime.consecutive_failures = 0; - runtime.last_failure = None; + runtime.guard = AutomaticCompactGuard::AwaitingPostCompactRequest; } - CompactionOutcome::Failed(category) => { - runtime.consecutive_failures = runtime.consecutive_failures.saturating_add(1); - runtime.last_failure = Some(category); - if runtime.consecutive_failures >= self.max_consecutive_failures { - runtime.automatic_disabled = true; - } + CompactionOutcome::Failed(failure_category) => { + runtime.guard = AutomaticCompactGuard::SuppressedForCurrentRun { failure_category }; + } + CompactionOutcome::Cancelled => { + runtime.cancelled_attempt = true; } - CompactionOutcome::Cancelled => {} } true } @@ -194,53 +181,47 @@ impl CompactState { /// following successful compaction has a durably committed UsageRecord. pub(crate) fn post_compact_request_committed(&self) { let mut runtime = self.lock_runtime(); - if runtime.attempt == AutomaticCompactState::Completed(CompactionOutcome::Succeeded) { - runtime.attempt = AutomaticCompactState::NotAttempted; + if runtime.guard == AutomaticCompactGuard::AwaitingPostCompactRequest { + runtime.guard = AutomaticCompactGuard::Ready; + runtime.attempt_claimed = false; + runtime.cancelled_attempt = false; } } - /// A successful manual compaction is the recovery path for a disabled - /// automatic circuit breaker and does not count as an automatic attempt. - pub(crate) fn reenable_automatic(&self) { - let mut runtime = self.lock_runtime(); - runtime.attempt = AutomaticCompactState::NotAttempted; - runtime.consecutive_failures = 0; - runtime.automatic_disabled = false; - runtime.last_failure = None; - runtime.pending_request_block = None; - } - pub(crate) fn take_pending_request_block(&self) -> Option { self.lock_runtime().pending_request_block.take() } fn claim_attempt(&self, trigger: AutomaticCompactTrigger) -> AutomaticCompactDecision { let mut runtime = self.lock_runtime(); - if runtime.automatic_disabled { - let category = runtime - .last_failure - .expect("disabled automatic compaction must retain its failure category"); - return AutomaticCompactDecision::Block(AutomaticCompactBlock::Disabled(category)); - } - - match runtime.attempt { - AutomaticCompactState::NotAttempted => { - runtime.attempt = AutomaticCompactState::Attempted; + match runtime.guard { + AutomaticCompactGuard::Ready if !runtime.attempt_claimed => { + runtime.attempt_claimed = true; AutomaticCompactDecision::Start(trigger) } - AutomaticCompactState::Attempted => AutomaticCompactDecision::Continue, - AutomaticCompactState::Completed(CompactionOutcome::Succeeded) => { - AutomaticCompactDecision::Block(AutomaticCompactBlock::Thrash) - } - AutomaticCompactState::Completed(CompactionOutcome::Failed(category)) => { - AutomaticCompactDecision::Block(AutomaticCompactBlock::Failed(category)) - } - AutomaticCompactState::Completed(CompactionOutcome::Cancelled) => { + AutomaticCompactGuard::Ready if runtime.cancelled_attempt => { AutomaticCompactDecision::Block(AutomaticCompactBlock::Cancelled) } + AutomaticCompactGuard::Ready => { + AutomaticCompactDecision::Block(AutomaticCompactBlock::Attempted) + } + AutomaticCompactGuard::SuppressedForCurrentRun { failure_category } => { + AutomaticCompactDecision::Block(AutomaticCompactBlock::Failed(failure_category)) + } + AutomaticCompactGuard::AwaitingPostCompactRequest => { + AutomaticCompactDecision::Block(AutomaticCompactBlock::Thrash) + } } } + fn clear_logical_run(&self) { + let mut runtime = self.lock_runtime(); + runtime.guard = AutomaticCompactGuard::Ready; + runtime.attempt_claimed = false; + runtime.cancelled_attempt = false; + runtime.pending_request_block = None; + } + fn lock_runtime(&self) -> std::sync::MutexGuard<'_, AutomaticCompactRuntimeState> { self.runtime .lock() @@ -248,18 +229,8 @@ impl CompactState { } #[cfg(test)] - pub(crate) fn attempt_state(&self) -> AutomaticCompactState { - self.lock_runtime().attempt - } - - #[cfg(test)] - pub(crate) fn consecutive_failures(&self) -> u32 { - self.lock_runtime().consecutive_failures - } - - #[cfg(test)] - pub(crate) fn automatic_disabled(&self) -> bool { - self.lock_runtime().automatic_disabled + pub(crate) fn guard(&self) -> AutomaticCompactGuard { + self.lock_runtime().guard } } @@ -270,70 +241,52 @@ mod tests { const FAILURE: CompactFailureCategory = CompactFailureCategory::Storage; #[test] - fn same_logical_run_claims_only_one_automatic_attempt() { + fn automatic_failure_suppresses_only_current_logical_run() { let state = CompactState::new(Some(10), Some(10), 2); - assert_eq!( state.evaluate_pre_run(11), AutomaticCompactDecision::Start(AutomaticCompactTrigger::PreRun) ); - assert_eq!( - state.evaluate_request(11), - AutomaticCompactDecision::Continue - ); - assert_eq!(state.attempt_state(), AutomaticCompactState::Attempted); - } - - #[test] - fn completed_attempt_blocks_retry_and_preserves_failure_category() { - let state = CompactState::new(None, Some(10), 2); - assert!(matches!( - state.evaluate_request(11), - AutomaticCompactDecision::Start(_) - )); assert!(state.complete_automatic(CompactionOutcome::Failed(FAILURE))); - - let block = AutomaticCompactBlock::Failed(FAILURE); + assert_eq!( + state.guard(), + AutomaticCompactGuard::SuppressedForCurrentRun { + failure_category: FAILURE + } + ); + assert_eq!( + state.evaluate_request(10), + AutomaticCompactDecision::Continue, + "a failed proactive compact still permits a request below the safety threshold" + ); assert_eq!( state.evaluate_request(11), - AutomaticCompactDecision::Block(block) + AutomaticCompactDecision::Block(AutomaticCompactBlock::Failed(FAILURE)) ); - state.record_request_block(block); - assert_eq!(state.take_pending_request_block(), Some(block)); - assert_eq!(state.consecutive_failures(), 1); - } - - #[test] - fn two_failed_automatic_attempts_disable_until_manual_success() { - let state = CompactState::new(Some(10), Some(10), 2); - for expected in 1..=2 { - state.begin_logical_run(); - assert!(matches!( - state.evaluate_pre_run(11), - AutomaticCompactDecision::Start(_) - )); - assert!(state.complete_automatic(CompactionOutcome::Failed(FAILURE))); - assert_eq!(state.consecutive_failures(), expected); - } - assert!(state.automatic_disabled()); state.begin_logical_run(); + assert_eq!(state.guard(), AutomaticCompactGuard::Ready); assert_eq!( state.evaluate_pre_run(11), - AutomaticCompactDecision::Block(AutomaticCompactBlock::Disabled(FAILURE)) + AutomaticCompactDecision::Start(AutomaticCompactTrigger::PreRun) ); + } - state.reenable_automatic(); - assert!(!state.automatic_disabled()); - assert_eq!(state.consecutive_failures(), 0); + #[test] + fn claimed_attempt_cannot_be_started_twice() { + let state = CompactState::new(Some(10), Some(10), 2); assert!(matches!( state.evaluate_pre_run(11), AutomaticCompactDecision::Start(_) )); + assert_eq!( + state.evaluate_request(11), + AutomaticCompactDecision::Block(AutomaticCompactBlock::Attempted) + ); } #[test] - fn cancellation_consumes_run_attempt_without_counting_as_failure() { + fn cancellation_consumes_run_attempt_without_becoming_failure() { let state = CompactState::new(None, Some(10), 2); assert!(matches!( state.evaluate_request(11), @@ -341,12 +294,16 @@ mod tests { )); assert!(state.complete_automatic(CompactionOutcome::Cancelled)); - assert_eq!(state.consecutive_failures(), 0); - assert!(!state.automatic_disabled()); + assert_eq!(state.guard(), AutomaticCompactGuard::Ready); assert_eq!( state.evaluate_request(11), AutomaticCompactDecision::Block(AutomaticCompactBlock::Cancelled) ); + state.begin_logical_run(); + assert!(matches!( + state.evaluate_request(11), + AutomaticCompactDecision::Start(_) + )); } #[test] @@ -363,7 +320,7 @@ mod tests { AutomaticCompactDecision::Block(AutomaticCompactBlock::Thrash) ); state.post_compact_request_committed(); - assert_eq!(state.attempt_state(), AutomaticCompactState::NotAttempted); + assert_eq!(state.guard(), AutomaticCompactGuard::Ready); assert!(matches!( state.evaluate_request(11), AutomaticCompactDecision::Start(_) @@ -371,16 +328,33 @@ mod tests { } #[test] - fn pause_resume_preserves_attempt_while_terminal_finish_clears_it() { + fn pause_resume_preserves_guard_while_terminal_finish_clears_it() { let state = CompactState::new(None, Some(10), 2); assert!(matches!( state.evaluate_request(11), AutomaticCompactDecision::Start(_) )); + assert!(state.complete_automatic(CompactionOutcome::Failed(FAILURE))); // Pause/resume deliberately performs no state transition. - assert_eq!(state.attempt_state(), AutomaticCompactState::Attempted); + assert!(matches!( + state.guard(), + AutomaticCompactGuard::SuppressedForCurrentRun { .. } + )); state.finish_logical_run(); - assert_eq!(state.attempt_state(), AutomaticCompactState::NotAttempted); + assert_eq!(state.guard(), AutomaticCompactGuard::Ready); + } + + #[test] + fn hook_yield_is_guarded_without_threshold_configuration() { + let state = CompactState::new(None, None, 2); + assert!(matches!( + state.claim_hook_yield(), + AutomaticCompactDecision::Start(_) + )); + assert_eq!( + state.claim_hook_yield(), + AutomaticCompactDecision::Block(AutomaticCompactBlock::Attempted) + ); } } diff --git a/crates/worker/src/ipc/interceptor.rs b/crates/worker/src/ipc/interceptor.rs index 79bc9593..01498e4d 100644 --- a/crates/worker/src/ipc/interceptor.rs +++ b/crates/worker/src/ipc/interceptor.rs @@ -870,10 +870,7 @@ mod tests { .unwrap(); assert!(matches!(action, PreRequestAction::Yield)); - assert_eq!( - state.attempt_state(), - crate::compact::state::AutomaticCompactState::Attempted - ); + assert!(state.has_claimed_attempt()); // Hook must not run when an internal mechanism short-circuits first. assert_eq!(count.load(Ordering::Relaxed), 0); } @@ -916,10 +913,7 @@ mod tests { PreRequestAction::YieldWith(items) => assert_eq!(items.len(), 1), other => panic!("expected YieldWith queued system item, got {other:?}"), } - assert_eq!( - state.attempt_state(), - crate::compact::state::AutomaticCompactState::Attempted - ); + assert!(state.has_claimed_attempt()); assert!(saw_handle.load(Ordering::Relaxed)); assert_eq!(committed.lock().expect("committed system items").len(), 1); } diff --git a/crates/worker/src/worker.rs b/crates/worker/src/worker.rs index 7385f514..f708abe5 100644 --- a/crates/worker/src/worker.rs +++ b/crates/worker/src/worker.rs @@ -3476,7 +3476,7 @@ impl Worker { tracker_for_usage.record_usage(event); }); - let compact_state = if post_run_threshold.is_some() || request_threshold.is_some() { + let compact_state = { if let (Some(post), Some(req)) = (post_run_threshold, request_threshold) { if post > req { warn!( @@ -3494,8 +3494,6 @@ impl Worker { )); self.compact_state = Some(state.clone()); Some(state) - } else { - None }; let usage_history_handle = compact_state.as_ref().map(|_| self.usage_history.clone()); @@ -4837,9 +4835,6 @@ impl Worker { { Ok(new_segment_id) => { info!(new_segment_id = %new_segment_id, "Manual compaction succeeded"); - if let Some(ref state) = state { - state.reenable_automatic(); - } Ok(ManualCompactResult::Compacted { new_segment_id }) } Err(e) => { @@ -7172,11 +7167,12 @@ fn restored_flow_runtime_state( fn automatic_compact_block_error(block: AutomaticCompactBlock) -> WorkerError { match block { AutomaticCompactBlock::Thrash => WorkerError::CompactThrash, - AutomaticCompactBlock::Failed(category) | AutomaticCompactBlock::Disabled(category) => { - WorkerError::AutomaticCompactFailed { - category: category.as_str(), - } - } + AutomaticCompactBlock::Failed(category) => WorkerError::AutomaticCompactFailed { + category: category.as_str(), + }, + AutomaticCompactBlock::Attempted => WorkerError::AutomaticCompactState( + "automatic compaction attempt already claimed for this logical run".to_string(), + ), AutomaticCompactBlock::Cancelled => WorkerError::CompactCancelled, } }