Hare/yoi
1
1
Fork 0
Code Actions Packages Releases Activity

fix: scope compaction suppression to one run

Browse Source
This commit is contained in:
Hare
2026-09-16 07:27:21 +09:00
parent a0f8ee51a4
commit fbbea1bf91
3 changed files with 140 additions and 176 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/worker/src/compact/state.rs
+130 -156
View File
@@ -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<CompactFailureCategory>,
guard: AutomaticCompactGuard,
attempt_claimed: bool,
cancelled_attempt: bool,
pending_request_block: Option<AutomaticCompactBlock>,
}
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<u64>,
/// 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<u64>,
retained_tokens: u64,
max_consecutive_failures: u32,
runtime: Mutex<AutomaticCompactRuntimeState>,
}
@@ -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,51 +181,45 @@ 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<AutomaticCompactBlock> {
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> {
@@ -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.begin_logical_run();
assert_eq!(state.guard(), AutomaticCompactGuard::Ready);
assert_eq!(
state.evaluate_pre_run(11),
AutomaticCompactDecision::Start(AutomaticCompactTrigger::PreRun)
);
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() {
fn claimed_attempt_cannot_be_started_twice() {
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.evaluate_pre_run(11),
AutomaticCompactDecision::Block(AutomaticCompactBlock::Disabled(FAILURE))
state.evaluate_request(11),
AutomaticCompactDecision::Block(AutomaticCompactBlock::Attempted)
);
state.reenable_automatic();
assert!(!state.automatic_disabled());
assert_eq!(state.consecutive_failures(), 0);
assert!(matches!(
state.evaluate_pre_run(11),
AutomaticCompactDecision::Start(_)
));
}
#[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)
);
}
}
crates/worker/src/ipc/interceptor.rs
+2 -8
View File
@@ -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);
}
crates/worker/src/worker.rs
+6 -10
View File
@@ -3476,7 +3476,7 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
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<C: LlmClient + 'static, St: Store> Worker<C, St> {
));
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<C: LlmClient + 'static, St: Store> Worker<C, St> {
{
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 {
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,
}
}