From 34085b121ddc62f5d191aa3db64de40bcd4e3b2a Mon Sep 17 00:00:00 2001 From: Hare Date: Fri, 18 Sep 2026 06:08:35 +0900 Subject: [PATCH 01/13] worker: make segment activation atomic --- crates/worker/src/controller.rs | 1 + .../worker/src/runtime/worker_allocation.rs | 4 +- .../src/runtime/worker_allocation/error.rs | 8 + .../runtime/worker_allocation/lifecycle.rs | 255 +++++++++++++----- crates/worker/src/worker.rs | 171 +++++++++--- docs/design/compaction.md | 27 ++ 6 files changed, 361 insertions(+), 105 deletions(-) diff --git a/crates/worker/src/controller.rs b/crates/worker/src/controller.rs index 7fe3b02f4..2dda1ad49 100644 --- a/crates/worker/src/controller.rs +++ b/crates/worker/src/controller.rs @@ -2147,6 +2147,7 @@ async fn controller_loop( Err(WorkerError::Store(_)) | Err(WorkerError::WorkerStore(_)) | Err(WorkerError::InvalidState(_)) + | Err(WorkerError::SegmentActivationIncomplete { .. }) ) { set_controller_state( &shared_state, diff --git a/crates/worker/src/runtime/worker_allocation.rs b/crates/worker/src/runtime/worker_allocation.rs index 1eb3874bb..692d1ed82 100644 --- a/crates/worker/src/runtime/worker_allocation.rs +++ b/crates/worker/src/runtime/worker_allocation.rs @@ -19,8 +19,8 @@ pub use conflict::{ }; pub use error::ScopeLockError; pub use lifecycle::{ - ScopeAllocationGuard, SegmentLockInfo, adopt_allocation, install_top_level, - install_top_level_with_deny, lookup_segment, update_segment, + ScopeAllocationGuard, SegmentActivationGuard, SegmentLockInfo, adopt_allocation, + install_top_level, install_top_level_with_deny, lookup_segment, }; pub use mutate::{ delegate_scope, reclaim_delegated_scope, reclaim_stale, reclaim_stale_with, register_worker, diff --git a/crates/worker/src/runtime/worker_allocation/error.rs b/crates/worker/src/runtime/worker_allocation/error.rs index a0bcb78ef..601202693 100644 --- a/crates/worker/src/runtime/worker_allocation/error.rs +++ b/crates/worker/src/runtime/worker_allocation/error.rs @@ -37,4 +37,12 @@ pub enum ScopeLockError { worker_name: String, socket: PathBuf, }, + #[error( + "worker `{worker_name}` allocation moved from expected Segment {expected} to {actual:?}" + )] + SegmentChanged { + worker_name: String, + expected: SegmentId, + actual: Option, + }, } diff --git a/crates/worker/src/runtime/worker_allocation/lifecycle.rs b/crates/worker/src/runtime/worker_allocation/lifecycle.rs index 480d895af..15e02fd10 100644 --- a/crates/worker/src/runtime/worker_allocation/lifecycle.rs +++ b/crates/worker/src/runtime/worker_allocation/lifecycle.rs @@ -20,6 +20,18 @@ pub struct ScopeAllocationGuard { lock_path: PathBuf, } +/// Exclusive authority to activate a replacement Segment for one allocated Worker. +/// +/// The allocation lock remains held across the durable metadata CAS and the final +/// allocation-table update, so restore admission cannot observe an intermediate +/// ownership state. +pub struct SegmentActivationGuard { + guard: LockFileGuard, + worker_name: String, + expected: SegmentId, + replacement: SegmentId, +} + impl ScopeAllocationGuard { pub fn worker_name(&self) -> &str { &self.worker_name @@ -28,6 +40,79 @@ impl ScopeAllocationGuard { pub fn lock_path(&self) -> &Path { &self.lock_path } + + /// Lock the machine-wide table and validate a Segment replacement before + /// its durable metadata is changed. + pub fn begin_segment_activation( + &self, + expected: SegmentId, + replacement: SegmentId, + ) -> Result { + let guard = LockFileGuard::open(&self.lock_path)?; + let actual = guard + .data() + .allocations + .iter() + .find(|allocation| allocation.worker_name == self.worker_name) + .and_then(|allocation| allocation.segment_id); + + if actual != Some(expected) { + return Err(ScopeLockError::SegmentChanged { + worker_name: self.worker_name.clone(), + expected, + actual, + }); + } + + if let Some(existing) = guard.data().allocations.iter().find(|allocation| { + allocation.segment_id == Some(replacement) + && allocation.worker_name != self.worker_name + }) { + return Err(ScopeLockError::SegmentConflict { + segment_id: replacement, + worker_name: existing.worker_name.clone(), + socket: existing.socket.clone(), + }); + } + + Ok(SegmentActivationGuard { + guard, + worker_name: self.worker_name.clone(), + expected, + replacement, + }) + } +} + +impl SegmentActivationGuard { + /// Persist the replacement Segment while retaining exclusive allocation + /// authority. This is intentionally consumed so a successful metadata CAS + /// cannot accidentally be followed by an unlocked update. + pub fn commit(mut self) -> Result<(), ScopeLockError> { + let actual = self + .guard + .data() + .allocations + .iter() + .find(|allocation| allocation.worker_name == self.worker_name) + .and_then(|allocation| allocation.segment_id); + let Some(allocation) = self + .guard + .data_mut() + .allocations + .iter_mut() + .find(|allocation| allocation.worker_name == self.worker_name) + else { + return Err(ScopeLockError::SegmentChanged { + worker_name: self.worker_name, + expected: self.expected, + actual, + }); + }; + allocation.segment_id = Some(self.replacement); + self.guard.save()?; + Ok(()) + } } impl Drop for ScopeAllocationGuard { @@ -114,46 +199,6 @@ pub fn adopt_allocation( }) } -/// Rewrite the `segment_id` recorded for `worker_name` to -/// `new_segment_id`. -/// -/// The Worker's in-memory `segment_id` can change underneath the -/// allocation in two normal places: -/// -/// - `Worker::compact` mints a fresh Segment in the same Session. -/// - `session_store::ensure_head_or_fork` auto-forks within that Session when another -/// writer has advanced the store head behind our back. -/// -/// Both paths must call this so subsequent [`lookup_segment`] queries -/// find the live Segment id, not the old one. Without this update a -/// concurrent `restore_from_manifest(new_id)` would see "no live -/// writer" and proceed to register a competing allocation on the -/// Segment lineage this Worker just moved into. -/// -/// The lock is opened once and the allocation is rewritten inside the -/// guard, so the segment_id collision check is atomic with the -/// rewrite. -pub fn update_segment(worker_name: &str, new_segment_id: SegmentId) -> Result<(), ScopeLockError> { - let lock_path = default_allocation_path()?; - let mut guard = LockFileGuard::open(&lock_path)?; - if let Some(other) = guard.data().find_by_segment(new_segment_id) { - if other.worker_name != worker_name { - return Err(ScopeLockError::SegmentConflict { - segment_id: new_segment_id, - worker_name: other.worker_name.clone(), - socket: other.socket.clone(), - }); - } - } - let alloc = guard - .data_mut() - .find_mut(worker_name) - .ok_or_else(|| ScopeLockError::UnknownWorker(worker_name.into()))?; - alloc.segment_id = Some(new_segment_id); - guard.save()?; - Ok(()) -} - /// Information about a Worker that currently holds an allocation for a /// given session. #[derive(Debug, Clone)] @@ -283,32 +328,12 @@ mod tests { } #[test] - fn update_session_rewrites_allocation_session_id() { - let dir = TempDir::new().unwrap(); - let _sandbox = RuntimeDirSandbox::new(dir.path()); - let original = sid(); - let updated = sid(); - let _guard = install_top_level( - "p".into(), - std::process::id(), - sock("p"), - vec![write_rule("/work", true)], - original, - ) - .unwrap(); - update_segment("p", updated).unwrap(); - // lookup against the original is now empty, the updated id wins. - assert!(lookup_segment(original).unwrap().is_none()); - assert_eq!(lookup_segment(updated).unwrap().unwrap().worker_name, "p"); - } - - #[test] - fn update_session_rejects_when_target_already_held() { + fn segment_activation_rejects_when_target_already_held() { let dir = TempDir::new().unwrap(); let _sandbox = RuntimeDirSandbox::new(dir.path()); let s_a = sid(); let s_b = sid(); - let _g_a = install_top_level( + let g_a = install_top_level( "a".into(), std::process::id(), sock("a"), @@ -324,8 +349,11 @@ mod tests { s_b, ) .unwrap(); - // `a` cannot adopt b's live session id. - let err = update_segment("a", s_b).unwrap_err(); + // `a` cannot activate b's live Segment id. + let err = match g_a.begin_segment_activation(s_a, s_b) { + Ok(_) => panic!("expected SegmentConflict"), + Err(error) => error, + }; match err { ScopeLockError::SegmentConflict { worker_name, @@ -338,4 +366,103 @@ mod tests { other => panic!("expected SegmentConflict, got {other:?}"), } } + + #[test] + fn segment_activation_commits_verified_replacement() { + let dir = TempDir::new().unwrap(); + let _sandbox = RuntimeDirSandbox::new(dir.path()); + let lock_path = dir.path().join("workers.json"); + let old_segment = sid(); + let replacement = sid(); + let guard = install_top_level( + "activation".into(), + std::process::id(), + sock("activation"), + vec![write_rule("/work", true)], + old_segment, + ) + .unwrap(); + + guard + .begin_segment_activation(old_segment, replacement) + .unwrap() + .commit() + .unwrap(); + + let table = LockFileGuard::open(&lock_path).unwrap(); + assert_eq!( + table + .data() + .find("activation") + .and_then(|allocation| allocation.segment_id), + Some(replacement) + ); + } + + #[test] + fn dropping_segment_activation_preserves_source_allocation() { + let dir = TempDir::new().unwrap(); + let _sandbox = RuntimeDirSandbox::new(dir.path()); + let lock_path = dir.path().join("workers.json"); + let old_segment = sid(); + let replacement = sid(); + let guard = install_top_level( + "activation".into(), + std::process::id(), + sock("activation"), + vec![write_rule("/work", true)], + old_segment, + ) + .unwrap(); + + drop( + guard + .begin_segment_activation(old_segment, replacement) + .unwrap(), + ); + + let table = LockFileGuard::open(&lock_path).unwrap(); + assert_eq!( + table + .data() + .find("activation") + .and_then(|allocation| allocation.segment_id), + Some(old_segment) + ); + } + + #[test] + fn segment_activation_rejects_stale_expected_segment_without_mutation() { + let dir = TempDir::new().unwrap(); + let _sandbox = RuntimeDirSandbox::new(dir.path()); + let lock_path = dir.path().join("workers.json"); + let allocated = sid(); + let stale = sid(); + let replacement = sid(); + let guard = install_top_level( + "activation".into(), + std::process::id(), + sock("activation"), + vec![write_rule("/work", true)], + allocated, + ) + .unwrap(); + + assert!(matches!( + guard.begin_segment_activation(stale, replacement), + Err(ScopeLockError::SegmentChanged { + expected, + actual: Some(actual), + .. + }) if expected == stale && actual == allocated + )); + let table = LockFileGuard::open(&lock_path).unwrap(); + assert_eq!( + table + .data() + .find("activation") + .and_then(|allocation| allocation.segment_id), + Some(allocated) + ); + } } diff --git a/crates/worker/src/worker.rs b/crates/worker/src/worker.rs index af46ab494..459850acf 100644 --- a/crates/worker/src/worker.rs +++ b/crates/worker/src/worker.rs @@ -2,7 +2,7 @@ use std::collections::VecDeque; #[cfg(test)] use std::path::Path; use std::path::PathBuf; -use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use std::sync::{Arc, Mutex}; use std::time::{Duration, Instant}; @@ -1055,6 +1055,7 @@ pub struct SegmentState { location: ArcSwap, entries_written: AtomicUsize, append_lock: Mutex<()>, + activation_failed: AtomicBool, } impl SegmentState { @@ -1066,6 +1067,7 @@ impl SegmentState { }), entries_written: AtomicUsize::new(entries_written), append_lock: Mutex::new(()), + activation_failed: AtomicBool::new(false), }) } @@ -1093,6 +1095,19 @@ impl SegmentState { self.entries_written.store(n, Ordering::Release); } + fn ensure_append_allowed(&self) -> Result<(), StoreError> { + if self.activation_failed.load(Ordering::Acquire) { + return Err(StoreError::Io(std::io::Error::other( + "Segment activation is incomplete; restore the Worker before accepting more input", + ))); + } + Ok(()) + } + + fn fail_closed_activation(&self) { + self.activation_failed.store(true, Ordering::Release); + } + fn increment_entries(&self) { self.entries_written.fetch_add(1, Ordering::Release); } @@ -1223,6 +1238,7 @@ where } fn append_entry_locked(&self, entry: LogEntry) -> Result<(), StoreError> { + self.state.ensure_append_allowed()?; let loc = self.state.location(); self.store.append(loc.session_id, loc.segment_id, &entry)?; self.state.increment_entries(); @@ -1439,6 +1455,7 @@ where .append_lock .lock() .expect("segment append lock poisoned"); + self.writer.state.ensure_append_allowed()?; let mut current = self .state .lock() @@ -1580,6 +1597,7 @@ where .append_lock .lock() .expect("segment append lock poisoned"); + self.writer.state.ensure_append_allowed()?; let mut state = self .state .lock() @@ -4414,8 +4432,15 @@ impl Worker { let _rewrite_guard = self .prepare_session_rewrite(SessionRewriteKind::Fork) .await?; - let w = self.engine.as_ref().unwrap(); + let segment_state = Arc::clone(&self.segment_state); + let _append_guard = segment_state + .append_lock + .lock() + .expect("segment append lock poisoned"); + segment_state.ensure_append_allowed()?; + let loc = segment_state.location(); let fork_segment_id = session_store::new_segment_id(); + let w = self.engine.as_ref().unwrap(); let entry = LogEntry::AnnotatedSegmentStart { ts: segment_log::now_millis(), session_id: loc.session_id, @@ -4446,24 +4471,48 @@ impl Worker { { initial_entries.push(checkpoint); } + initial_entries.push(LogEntry::Extension { + ts: segment_log::now_millis(), + domain: SESSION_PENDING_ACTIVATIONS_EXTENSION_DOMAIN.to_owned(), + payload: serde_json::to_value( + &*self + .session + .pending_activations + .lock() + .expect("pending activation state poisoned"), + ) + .map_err(|error| { + WorkerError::InvalidState(format!( + "serialize pending submissions during automatic fork: {error}" + )) + })?, + }); self.store .create_segment(loc.session_id, fork_segment_id, &initial_entries) .map_err(WorkerError::from)?; - self.segment_state.set_location(SegmentLocation { + let new_location = SegmentLocation { session_id: loc.session_id, segment_id: fork_segment_id, - }); + }; + let allocation_activation = self + .scope_allocation + .as_ref() + .map(|allocation| { + allocation.begin_segment_activation(loc.segment_id, fork_segment_id) + }) + .transpose()?; + self.compare_and_swap_worker_metadata_segment(loc, new_location)?; + if let Some(allocation_activation) = allocation_activation { + if let Err(source) = allocation_activation.commit() { + segment_state.fail_closed_activation(); + return Err(WorkerError::SegmentActivationIncomplete { source }); + } + } + self.segment_state.set_location(new_location); self.segment_state .set_entries_written(initial_entries.len()); self.sink .reset_with_initial_entries(initial_entries.clone()); - if self.scope_allocation.is_some() { - worker_allocation::update_segment(&self.manifest.worker.name, fork_segment_id)?; - } - self.write_worker_metadata_active(SegmentLocation { - session_id: loc.session_id, - segment_id: fork_segment_id, - })?; Ok(()) } @@ -5058,12 +5107,12 @@ impl Worker { match outcome { Ok((new_segment_id, _summary, stats)) => { debug_assert_eq!(lifecycle.state, CompactionLifecycleState::Done); + self.set_compaction_progress(None); for metric in attempt.success_metrics(new_segment_id, started.elapsed(), &stats) { self.try_record_metric(&metric); } self.usage_tracker .note_compaction_correlation_id(attempt.correlation_id().to_string()); - self.release_compaction_service(&lifecycle).await; Ok(new_segment_id) } Err(error) => { @@ -5082,8 +5131,8 @@ impl Worker { CompactionLifecycleState::Failed }; lifecycle.ended_at_ms = Some(segment_log::now_millis()); - self.set_compaction_progress(None); self.release_compaction_service(&lifecycle).await; + self.set_compaction_progress(None); Err(error) } } @@ -5100,13 +5149,18 @@ impl Worker { let Some(registry) = &self.internal_worker_registry else { return; }; - if let Err(error) = registry.stop_service(session_id).await { - warn!( - compaction_id = %lifecycle.compaction_id, - internal_worker_session_id = %session_id, - error = %error, - "failed to release terminal compaction Internal Worker" - ); + loop { + match registry.stop_service(session_id).await { + Ok(_) => return, + Err(error) => { + warn!( + compaction_id = %lifecycle.compaction_id, + error = %error, + "compaction service cleanup failed; retaining cleanup authority and retrying" + ); + tokio::time::sleep(Duration::from_millis(50)).await; + } + } } } @@ -5643,8 +5697,31 @@ impl Worker { // rotate: create on disk, swap location, reset the broadcast // sink so existing subscribers see the new `SegmentStart // { compacted_from }` and reset their view. + // The Compactor has produced and validated its output. Retire it before + // activation so no internal process or registry authority survives into + // the replacement's live lifetime. Cleanup failures retain this future + // (and Busy state) until a retry succeeds. + self.release_compaction_service(lifecycle).await; + + self.set_compaction_progress(Some(InFlightCompaction { + phase: CompactionPhase::Committing, + started_at_ms: lifecycle.started_at_ms, + trigger, + })); + + // Activation lock order is append barrier -> allocation table -> metadata. + // From this point through live publication, every accepted Submit/Notify + // either completed on the source Segment or waits to observe the fully + // activated replacement. + let segment_state = Arc::clone(&self.segment_state); + let _append_guard = segment_state + .append_lock + .lock() + .expect("segment append lock poisoned"); + segment_state.ensure_append_allowed()?; + let new_segment_id = session_store::new_segment_id(); - let old_loc = self.segment_state.location(); + let old_loc = segment_state.location(); let source_turn_count = self.engine.as_ref().unwrap().turn_count(); let w = self.engine.as_ref().unwrap(); let entry = LogEntry::AnnotatedSegmentStart { @@ -5706,11 +5783,6 @@ impl Worker { })?, }); } - self.set_compaction_progress(Some(InFlightCompaction { - phase: CompactionPhase::Committing, - started_at_ms: lifecycle.started_at_ms, - trigger, - })); self.store .create_segment(old_loc.session_id, new_segment_id, &initial_entries)?; @@ -5724,21 +5796,32 @@ impl Worker { session_id: old_loc.session_id, segment_id: new_segment_id, }; - // The writer lease is stable for the Worker lifetime and is keyed by - // worker name. Compaction must not transfer or rewrite that lease: the - // active Segment is derived exclusively from the CAS-protected Worker - // metadata pointer below. A lost CAS therefore leaves every live and - // durable authority on the previous Segment. + // The machine-wide allocation lock is acquired only after staging, but + // before the durable commit point. Restore admission therefore cannot + // race the metadata CAS and allocation hand-off. + let allocation_activation = self + .scope_allocation + .as_ref() + .map(|allocation| { + allocation.begin_segment_activation(old_loc.segment_id, new_segment_id) + }) + .transpose()?; self.compare_and_swap_worker_metadata_segment(old_loc, new_location)?; + if let Some(allocation_activation) = allocation_activation { + if let Err(source) = allocation_activation.commit() { + // Metadata already names the complete replacement. Do not allow + // the old in-memory writer to accept another byte while durable + // authorities disagree; restore will converge from metadata. + segment_state.fail_closed_activation(); + return Err(WorkerError::SegmentActivationIncomplete { source }); + } + } // All live mutations after the durable commit are infallible and happen // before the replacement SegmentStart is broadcast. This keeps the // append destination, Session projection, Engine cache identity, and // reconnect snapshot on one side of the same activation boundary. - self.segment_state.set_location(SegmentLocation { - session_id: old_loc.session_id, - segment_id: new_segment_id, - }); + self.segment_state.set_location(new_location); self.segment_state .set_entries_written(initial_entries.len()); self.user_segments = retained_user_segments; @@ -5764,7 +5847,6 @@ impl Worker { lifecycle.revision = lifecycle.revision.saturating_add(1); lifecycle.state = CompactionLifecycleState::Done; lifecycle.ended_at_ms = Some(segment_log::now_millis()); - self.set_compaction_progress(None); Ok(( new_segment_id, summary_text, @@ -7169,7 +7251,9 @@ fn compact_failure_category(error: &WorkerError) -> CompactFailureCategory { WorkerError::CompactResultContextTooLarge { .. } => { CompactFailureCategory::ResultContextTooLarge } - WorkerError::WorkerStore(_) | WorkerError::CompactActiveSegmentChanged => { + WorkerError::WorkerStore(_) + | WorkerError::CompactActiveSegmentChanged + | WorkerError::SegmentActivationIncomplete { .. } => { CompactFailureCategory::ActiveSegmentCommit } WorkerError::Store(_) => CompactFailureCategory::Storage, @@ -7275,6 +7359,14 @@ pub enum WorkerError { #[error(transparent)] ScopeLock(#[from] ScopeLockError), + #[error( + "Segment activation could not be completed; the Worker must be restored before accepting more input" + )] + SegmentActivationIncomplete { + #[source] + source: ScopeLockError, + }, + #[error(transparent)] PromptCatalog(#[from] CatalogError), @@ -8373,8 +8465,9 @@ mod build_summary_prompt_tests { active_turn_count: 3, total_turn_count: 7, .. - } - ] + }, + LogEntry::Extension { domain, .. } + ] if domain == SESSION_PENDING_ACTIVATIONS_EXTENSION_DOMAIN )); let restored = session_store::collect_state(&fork_entries); assert!(restored.last_run_interrupted); diff --git a/docs/design/compaction.md b/docs/design/compaction.md index e6dd2b195..25a82c5cf 100644 --- a/docs/design/compaction.md +++ b/docs/design/compaction.md @@ -32,6 +32,33 @@ Compaction should preserve persisted reasoning history and avoid serializing unv The important property is explainability: after compaction, records should still show what summary replaced which older context and why future turns can rely on it. +## Activation boundary + +A compacted Segment is staged completely before it can become authoritative. The +activation boundary uses this lock order everywhere a live Worker changes Segment: + +1. stop the compaction service and remove its registry record; +2. acquire the Worker's append barrier; +3. write the final pending-submission checkpoint and stage the replacement Segment; +4. acquire and preflight the machine-wide Worker allocation lock; +5. compare-and-swap `metadata.json` from the source Segment to the replacement; +6. update the allocation table while still holding its lock; +7. publish the append destination, session projection, sink, and in-memory history; +8. release the append barrier. + +Submit and Notify acceptance take the same append barrier. Consequently an accepted +item is either durably included in the replacement checkpoint, or waits until every +live authority points at the replacement and is then appended there. Restore +admission takes the allocation lock, so it cannot observe the interval between the +metadata CAS and allocation hand-off. + +The metadata CAS is the commit point. Failures before it leave the source Segment +active. A failure updating allocation after it marks the old in-memory writer +unusable: further acceptance fails closed and process restart/restore converges from +the replacement named by metadata. The Worker must likewise remain non-idle while +terminal compaction-service cleanup is pending; cleanup authority is retained and +retried until the registry record is gone. + ## Metrics and comparison procedure Compaction measurements stay out of the ordinary transcript. They are appended as From fc24cb672eb931ffb9400dca7bdb599c805e39de Mon Sep 17 00:00:00 2001 From: Hare Date: Fri, 18 Sep 2026 06:27:11 +0900 Subject: [PATCH 02/13] worker: retain failed activation authority --- .../worker/src/runtime/worker_allocation.rs | 4 +- .../runtime/worker_allocation/lifecycle.rs | 93 ++++++++++++++- crates/worker/src/worker.rs | 112 ++++++++++++++---- crates/worker/tests/compact_events_test.rs | 4 +- docs/design/compaction.md | 11 +- 5 files changed, 187 insertions(+), 37 deletions(-) diff --git a/crates/worker/src/runtime/worker_allocation.rs b/crates/worker/src/runtime/worker_allocation.rs index 692d1ed82..726c294ad 100644 --- a/crates/worker/src/runtime/worker_allocation.rs +++ b/crates/worker/src/runtime/worker_allocation.rs @@ -19,8 +19,8 @@ pub use conflict::{ }; pub use error::ScopeLockError; pub use lifecycle::{ - ScopeAllocationGuard, SegmentActivationGuard, SegmentLockInfo, adopt_allocation, - install_top_level, install_top_level_with_deny, lookup_segment, + ScopeAllocationGuard, SegmentActivationCommitError, SegmentActivationGuard, SegmentLockInfo, + adopt_allocation, install_top_level, install_top_level_with_deny, lookup_segment, }; pub use mutate::{ delegate_scope, reclaim_delegated_scope, reclaim_stale, reclaim_stale_with, register_worker, diff --git a/crates/worker/src/runtime/worker_allocation/lifecycle.rs b/crates/worker/src/runtime/worker_allocation/lifecycle.rs index 15e02fd10..2f758480c 100644 --- a/crates/worker/src/runtime/worker_allocation/lifecycle.rs +++ b/crates/worker/src/runtime/worker_allocation/lifecycle.rs @@ -32,6 +32,29 @@ pub struct SegmentActivationGuard { replacement: SegmentId, } +/// A failed allocation-table publication together with the still-held machine-wide +/// activation authority. Callers must retain `authority` until recovery or process +/// teardown so restore admission cannot register a competing writer. +pub struct SegmentActivationCommitError { + source: ScopeLockError, + authority: SegmentActivationGuard, +} + +impl std::fmt::Debug for SegmentActivationCommitError { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter + .debug_struct("SegmentActivationCommitError") + .field("source", &self.source) + .finish_non_exhaustive() + } +} + +impl SegmentActivationCommitError { + pub fn into_parts(self) -> (ScopeLockError, SegmentActivationGuard) { + (self.source, self.authority) + } +} + impl ScopeAllocationGuard { pub fn worker_name(&self) -> &str { &self.worker_name @@ -88,7 +111,7 @@ impl SegmentActivationGuard { /// Persist the replacement Segment while retaining exclusive allocation /// authority. This is intentionally consumed so a successful metadata CAS /// cannot accidentally be followed by an unlocked update. - pub fn commit(mut self) -> Result<(), ScopeLockError> { + pub fn commit(mut self) -> Result<(), SegmentActivationCommitError> { let actual = self .guard .data() @@ -103,14 +126,22 @@ impl SegmentActivationGuard { .iter_mut() .find(|allocation| allocation.worker_name == self.worker_name) else { - return Err(ScopeLockError::SegmentChanged { - worker_name: self.worker_name, - expected: self.expected, - actual, + return Err(SegmentActivationCommitError { + source: ScopeLockError::SegmentChanged { + worker_name: self.worker_name.clone(), + expected: self.expected, + actual, + }, + authority: self, }); }; allocation.segment_id = Some(self.replacement); - self.guard.save()?; + if let Err(source) = self.guard.save() { + return Err(SegmentActivationCommitError { + source: ScopeLockError::Io(source), + authority: self, + }); + } Ok(()) } } @@ -399,6 +430,56 @@ mod tests { ); } + #[test] + fn failed_segment_activation_returns_locked_recovery_authority() { + let dir = TempDir::new().unwrap(); + let _sandbox = RuntimeDirSandbox::new(dir.path()); + let old_segment = sid(); + let replacement = sid(); + let guard = install_top_level( + "activation".into(), + std::process::id(), + sock("activation"), + vec![write_rule("/work", true)], + old_segment, + ) + .unwrap(); + let mut activation = guard + .begin_segment_activation(old_segment, replacement) + .unwrap(); + activation + .guard + .data_mut() + .allocations + .retain(|allocation| allocation.worker_name != "activation"); + + let error = activation.commit().unwrap_err(); + let (source, authority) = error.into_parts(); + assert!(matches!( + source, + ScopeLockError::SegmentChanged { + expected, + actual: None, + .. + } if expected == old_segment + )); + // The authority is returned rather than dropped on failure. Production + // stores it on SegmentState to keep restore admission blocked. + let (sent, received) = std::sync::mpsc::channel(); + let lookup = std::thread::spawn(move || { + sent.send(lookup_segment(replacement).unwrap()).unwrap(); + }); + assert!(received + .recv_timeout(std::time::Duration::from_millis(50)) + .is_err()); + drop(authority); + assert!(received + .recv_timeout(std::time::Duration::from_secs(1)) + .unwrap() + .is_none()); + lookup.join().unwrap(); + } + #[test] fn dropping_segment_activation_preserves_source_allocation() { let dir = TempDir::new().unwrap(); diff --git a/crates/worker/src/worker.rs b/crates/worker/src/worker.rs index 459850acf..7fc65102c 100644 --- a/crates/worker/src/worker.rs +++ b/crates/worker/src/worker.rs @@ -360,7 +360,9 @@ use crate::prompt::catalog::{CatalogError, PromptCatalog, WorkspacePromptProject use crate::prompt::source::PromptCatalogSource; use crate::prompt::system::{SystemPromptContext, SystemPromptError, SystemPromptTemplate}; use crate::runtime::dir; -use crate::runtime::worker_allocation::{self, ScopeAllocationGuard, ScopeLockError}; +use crate::runtime::worker_allocation::{ + self, ScopeAllocationGuard, ScopeLockError, SegmentActivationGuard, +}; use crate::skill::{SkillActivationResponse, SkillClientError}; #[cfg(test)] use async_trait::async_trait; @@ -1056,6 +1058,7 @@ pub struct SegmentState { entries_written: AtomicUsize, append_lock: Mutex<()>, activation_failed: AtomicBool, + activation_failure_authority: Mutex>, } impl SegmentState { @@ -1068,6 +1071,7 @@ impl SegmentState { entries_written: AtomicUsize::new(entries_written), append_lock: Mutex::new(()), activation_failed: AtomicBool::new(false), + activation_failure_authority: Mutex::new(None), }) } @@ -1104,7 +1108,11 @@ impl SegmentState { Ok(()) } - fn fail_closed_activation(&self) { + fn fail_closed_activation(&self, authority: SegmentActivationGuard) { + *self + .activation_failure_authority + .lock() + .expect("Segment activation failure authority mutex poisoned") = Some(authority); self.activation_failed.store(true, Ordering::Release); } @@ -2086,6 +2094,12 @@ pub(crate) trait SystemPromptContributionSource: Send + Sync { async fn load(&self) -> Option; } +#[derive(Debug, Clone)] +struct PendingCompactionCleanup { + session_id: String, + compaction_id: String, +} + /// An independent agent execution unit. /// /// Holds a [`Engine`] directly and persists session state via @@ -2184,6 +2198,8 @@ pub struct Worker { /// Parent-owned projection/control boundary for observable Internal service Workers. /// Service Workers are never exposed through the model-facing SubWorker control surface. internal_worker_registry: Option>, + /// Cleanup authority survives cancellation of an individual compaction future. + pending_compaction_cleanup: Mutex>, in_flight: Option, /// Monotonic counter incremented by worker event bridges when an /// assistant-side execution artifact becomes visible to clients before @@ -2461,6 +2477,7 @@ impl Worker { alerter: None, working_event_tx: None, internal_worker_registry: None, + pending_compaction_cleanup: Mutex::new(None), in_flight: None, ai_activity_counter: Arc::new(AtomicUsize::new(0)), pending_notifies: NotifyBuffer::new(), @@ -3629,7 +3646,7 @@ impl Worker { self.ensure_system_prompt_materialized().await?; self.ensure_segment_head().await?; if self.should_pre_run_compact() { - self.try_pre_run_compact().await; + self.try_pre_run_compact().await?; } Ok(()) } @@ -4503,8 +4520,9 @@ impl Worker { .transpose()?; self.compare_and_swap_worker_metadata_segment(loc, new_location)?; if let Some(allocation_activation) = allocation_activation { - if let Err(source) = allocation_activation.commit() { - segment_state.fail_closed_activation(); + if let Err(error) = allocation_activation.commit() { + let (source, authority) = error.into_parts(); + segment_state.fail_closed_activation(authority); return Err(WorkerError::SegmentActivationIncomplete { source }); } } @@ -4711,16 +4729,18 @@ impl Worker { /// This used to run in the controller's post-run path. Keeping it here /// preserves the ordering requirement that the next turn starts with a /// compacted history, without introducing a separate Busy controller state. - /// Best-effort: failures are logged and surfaced, but do not abort the - /// user turn that triggered the check. - pub async fn try_pre_run_compact(&mut self) { + /// Fatal activation failures are returned to the controller so it cannot + /// publish Idle or start the triggering turn with inconsistent authority. + pub async fn try_pre_run_compact(&mut self) -> Result<(), WorkerError> { let Some(state) = self.compact_state.clone() else { - return; + return Ok(()); }; let current_tokens = self.total_tokens().tokens; match state.evaluate_pre_run(current_tokens) { AutomaticCompactDecision::Start(AutomaticCompactTrigger::PreRun) => {} - AutomaticCompactDecision::Continue | AutomaticCompactDecision::Block(_) => return, + AutomaticCompactDecision::Continue | AutomaticCompactDecision::Block(_) => { + return Ok(()); + } AutomaticCompactDecision::Start(AutomaticCompactTrigger::RequestThreshold) => { unreachable!("pre-run evaluation returned request-threshold trigger") } @@ -4741,6 +4761,7 @@ impl Worker { completed, "automatic compaction must complete a claimed attempt" ); + Ok(()) } Err(e) => { warn!(error = %e, "Proactive pre-run compaction failed"); @@ -4759,6 +4780,11 @@ impl Worker { completed, "automatic compaction must complete a claimed attempt" ); + if matches!(e, WorkerError::SegmentActivationIncomplete { .. }) { + Err(e) + } else { + Ok(()) + } } } } @@ -5044,6 +5070,7 @@ impl Worker { mut cancel: Option>, trigger: CompactionTrigger, ) -> Result { + self.release_pending_compaction_service().await; let _rewrite_guard = self .prepare_session_rewrite(SessionRewriteKind::Compact) .await?; @@ -5131,30 +5158,43 @@ impl Worker { CompactionLifecycleState::Failed }; lifecycle.ended_at_ms = Some(segment_log::now_millis()); - self.release_compaction_service(&lifecycle).await; + self.release_compaction_service(&mut lifecycle).await; self.set_compaction_progress(None); Err(error) } } } - async fn release_compaction_service(&self, lifecycle: &CompactionLifecycle) { - let Some(session_id) = lifecycle - .internal_worker - .as_ref() - .map(|worker| worker.session_id.as_str()) - else { + async fn release_pending_compaction_service(&self) { + let cleanup = self + .pending_compaction_cleanup + .lock() + .expect("pending compaction cleanup mutex poisoned") + .clone(); + let Some(cleanup) = cleanup else { return; }; let Some(registry) = &self.internal_worker_registry else { return; }; loop { - match registry.stop_service(session_id).await { - Ok(_) => return, + match registry.stop_service(&cleanup.session_id).await { + Ok(_) => { + let mut pending = self + .pending_compaction_cleanup + .lock() + .expect("pending compaction cleanup mutex poisoned"); + if pending + .as_ref() + .is_some_and(|pending| pending.session_id == cleanup.session_id) + { + *pending = None; + } + return; + } Err(error) => { warn!( - compaction_id = %lifecycle.compaction_id, + compaction_id = %cleanup.compaction_id, error = %error, "compaction service cleanup failed; retaining cleanup authority and retrying" ); @@ -5164,6 +5204,18 @@ impl Worker { } } + async fn release_compaction_service(&self, lifecycle: &mut CompactionLifecycle) { + self.release_pending_compaction_service().await; + if self + .pending_compaction_cleanup + .lock() + .expect("pending compaction cleanup mutex poisoned") + .is_none() + { + lifecycle.internal_worker = None; + } + } + async fn compact_impl( &mut self, retained_tokens: u64, @@ -5399,6 +5451,14 @@ impl Worker { .map_err(|error| WorkerError::InvalidState(error.to_string()))?; lifecycle.revision = lifecycle.revision.saturating_add(1); lifecycle.internal_worker = Some(internal_ref); + let cleanup = PendingCompactionCleanup { + session_id: handle.session_id_string(), + compaction_id: lifecycle.compaction_id.clone(), + }; + *self + .pending_compaction_cleanup + .lock() + .expect("pending compaction cleanup mutex poisoned") = Some(cleanup); self.set_compaction_progress(Some(InFlightCompaction { phase: CompactionPhase::Summarizing, started_at_ms: lifecycle.started_at_ms, @@ -5808,11 +5868,13 @@ impl Worker { .transpose()?; self.compare_and_swap_worker_metadata_segment(old_loc, new_location)?; if let Some(allocation_activation) = allocation_activation { - if let Err(source) = allocation_activation.commit() { + if let Err(error) = allocation_activation.commit() { + let (source, authority) = error.into_parts(); // Metadata already names the complete replacement. Do not allow // the old in-memory writer to accept another byte while durable - // authorities disagree; restore will converge from metadata. - segment_state.fail_closed_activation(); + // authorities disagree; retain the allocation lock so restore + // admission cannot register a competing writer. + segment_state.fail_closed_activation(authority); return Err(WorkerError::SegmentActivationIncomplete { source }); } } @@ -6024,6 +6086,7 @@ where alerter: None, working_event_tx: None, internal_worker_registry: None, + pending_compaction_cleanup: Mutex::new(None), in_flight: None, ai_activity_counter: Arc::new(AtomicUsize::new(0)), pending_notifies: NotifyBuffer::new(), @@ -6110,6 +6173,7 @@ where alerter: None, working_event_tx: None, internal_worker_registry: None, + pending_compaction_cleanup: Mutex::new(None), in_flight: None, ai_activity_counter: Arc::new(AtomicUsize::new(0)), pending_notifies: NotifyBuffer::new(), @@ -6231,6 +6295,7 @@ where alerter: None, working_event_tx: None, internal_worker_registry: None, + pending_compaction_cleanup: Mutex::new(None), in_flight: None, ai_activity_counter: Arc::new(AtomicUsize::new(0)), pending_notifies: NotifyBuffer::new(), @@ -6609,6 +6674,7 @@ where alerter: None, working_event_tx: None, internal_worker_registry: None, + pending_compaction_cleanup: Mutex::new(None), in_flight: None, ai_activity_counter: Arc::new(AtomicUsize::new(0)), pending_notifies: NotifyBuffer::new(), diff --git a/crates/worker/tests/compact_events_test.rs b/crates/worker/tests/compact_events_test.rs index 416437931..e88fc223d 100644 --- a/crates/worker/tests/compact_events_test.rs +++ b/crates/worker/tests/compact_events_test.rs @@ -922,7 +922,7 @@ async fn pre_run_compact_publishes_runtime_progress_phases() { let session_before = worker.session_id(); let segment_before = worker.segment_id(); - worker.try_pre_run_compact().await; + worker.try_pre_run_compact().await.unwrap(); assert_eq!(worker.session_id(), session_before); assert_ne!(worker.segment_id(), segment_before); @@ -1083,7 +1083,7 @@ async fn pre_run_compact_failure_clears_runtime_progress() { let _ = drain(&mut rx); // Best-effort: returns Ok(()) even on failure and clears runtime progress. - worker.try_pre_run_compact().await; + worker.try_pre_run_compact().await.unwrap(); let events = drain(&mut rx); assert!(events.iter().any(|event| matches!( diff --git a/docs/design/compaction.md b/docs/design/compaction.md index 25a82c5cf..01abe4fd7 100644 --- a/docs/design/compaction.md +++ b/docs/design/compaction.md @@ -54,10 +54,13 @@ metadata CAS and allocation hand-off. The metadata CAS is the commit point. Failures before it leave the source Segment active. A failure updating allocation after it marks the old in-memory writer -unusable: further acceptance fails closed and process restart/restore converges from -the replacement named by metadata. The Worker must likewise remain non-idle while -terminal compaction-service cleanup is pending; cleanup authority is retained and -retried until the registry record is gone. +unusable and retains the machine-wide allocation lock: further acceptance fails +closed and restore admission cannot register a competing writer until process +teardown, after which restore converges from the replacement named by metadata. +The Worker must likewise remain non-idle while terminal compaction-service cleanup +is pending. Cleanup authority is stored on the Worker (not only on one compaction +future), retried until the registry record is gone, and cleared exactly once before +activation or a terminal result. ## Metrics and comparison procedure From e7c50dc5193a276df6ad6440595c9d0e64148787 Mon Sep 17 00:00:00 2001 From: Hare Date: Fri, 18 Sep 2026 06:40:46 +0900 Subject: [PATCH 03/13] worker: keep cleanup failures non-idle --- crates/worker/src/controller.rs | 84 +++++++++++++++++++++++++- crates/worker/src/worker.rs | 102 +++++++++++++++++++------------- docs/design/compaction.md | 6 +- 3 files changed, 148 insertions(+), 44 deletions(-) diff --git a/crates/worker/src/controller.rs b/crates/worker/src/controller.rs index 2dda1ad49..5848268bd 100644 --- a/crates/worker/src/controller.rs +++ b/crates/worker/src/controller.rs @@ -2148,6 +2148,7 @@ async fn controller_loop( | Err(WorkerError::WorkerStore(_)) | Err(WorkerError::InvalidState(_)) | Err(WorkerError::SegmentActivationIncomplete { .. }) + | Err(WorkerError::CompactionCleanupPending { .. }) ) { set_controller_state( &shared_state, @@ -2566,6 +2567,19 @@ where (WorkerStatus::Paused, shutdown_requested, false) } Err(e) => { + let status = if matches!( + &e, + WorkerError::SegmentActivationIncomplete { .. } + | WorkerError::CompactionCleanupPending { .. } + ) { + // Metadata has already committed the replacement, but + // one live authority failed to publish. Keep the + // controller visibly busy/fail-closed; publishing Idle + // would admit another turn against inconsistent state. + WorkerStatus::Running + } else { + WorkerStatus::Idle + }; let code = worker_error_code(&e); let message = e.to_string(); let _ = working_event_tx.send(Event::Error { @@ -2581,7 +2595,7 @@ where }, ); } - (WorkerStatus::Idle, shutdown_requested, false) + (status, shutdown_requested, false) } }; } @@ -3319,6 +3333,74 @@ mod tests { } } + #[tokio::test] + async fn segment_activation_failure_from_run_never_publishes_idle() { + let mut env = make_env().await; + let worker_future = async { + Err::(WorkerError::SegmentActivationIncomplete { + source: crate::runtime::worker_allocation::ScopeLockError::UnknownWorker( + "worker".into(), + ), + }) + }; + + let (status, shutdown, may_drain_pending) = drive_turn( + worker_future, + &mut env.method_rx, + &env.working_event_tx, + &env.cancel_tx, + &env.pause_tx, + &env.shared_state, + &env.runtime_dir, + None, + &env.notify_buffer, + &env.pending_submissions, + None, + "worker", + &env.spawned_registry, + false, + ) + .await; + + assert_eq!(status, WorkerStatus::Running); + assert!(!shutdown); + assert!(!may_drain_pending); + } + + #[tokio::test] + async fn pending_compaction_cleanup_from_run_never_publishes_idle() { + let mut env = make_env().await; + let worker_future = async { + Err::(WorkerError::CompactionCleanupPending { + source: crate::runtime::worker_allocation::ScopeLockError::UnknownWorker( + "worker".into(), + ), + }) + }; + + let (status, shutdown, may_drain_pending) = drive_turn( + worker_future, + &mut env.method_rx, + &env.working_event_tx, + &env.cancel_tx, + &env.pause_tx, + &env.shared_state, + &env.runtime_dir, + None, + &env.notify_buffer, + &env.pending_submissions, + None, + "worker", + &env.spawned_registry, + false, + ) + .await; + + assert_eq!(status, WorkerStatus::Running); + assert!(!shutdown); + assert!(!may_drain_pending); + } + #[tokio::test] async fn pause_waits_for_run_boundary_and_uses_safe_pause_channel() { let mut env = make_env().await; diff --git a/crates/worker/src/worker.rs b/crates/worker/src/worker.rs index 7fc65102c..eec9f1077 100644 --- a/crates/worker/src/worker.rs +++ b/crates/worker/src/worker.rs @@ -4780,7 +4780,11 @@ impl Worker { completed, "automatic compaction must complete a claimed attempt" ); - if matches!(e, WorkerError::SegmentActivationIncomplete { .. }) { + if matches!( + &e, + WorkerError::SegmentActivationIncomplete { .. } + | WorkerError::CompactionCleanupPending { .. } + ) { Err(e) } else { Ok(()) @@ -5070,7 +5074,7 @@ impl Worker { mut cancel: Option>, trigger: CompactionTrigger, ) -> Result { - self.release_pending_compaction_service().await; + self.release_pending_compaction_service().await?; let _rewrite_guard = self .prepare_session_rewrite(SessionRewriteKind::Compact) .await?; @@ -5143,6 +5147,14 @@ impl Worker { Ok(new_segment_id) } Err(error) => { + let error = if matches!(&error, WorkerError::CompactionCleanupPending { .. }) { + error + } else { + match self.release_compaction_service(&mut lifecycle).await { + Ok(()) => error, + Err(cleanup_error) => cleanup_error, + } + }; let observed_segment_id = self.segment_state.location().segment_id; for metric in attempt.failure_metrics( observed_segment_id, @@ -5152,68 +5164,70 @@ impl Worker { self.try_record_metric(&metric); } lifecycle.revision = lifecycle.revision.saturating_add(1); - lifecycle.state = if matches!(error, WorkerError::CompactCancelled) { + lifecycle.state = if matches!(&error, WorkerError::CompactCancelled) { CompactionLifecycleState::Interrupted } else { CompactionLifecycleState::Failed }; lifecycle.ended_at_ms = Some(segment_log::now_millis()); - self.release_compaction_service(&mut lifecycle).await; - self.set_compaction_progress(None); + if !matches!(&error, WorkerError::CompactionCleanupPending { .. }) { + self.set_compaction_progress(None); + } Err(error) } } } - async fn release_pending_compaction_service(&self) { + async fn release_pending_compaction_service(&self) -> Result<(), WorkerError> { let cleanup = self .pending_compaction_cleanup .lock() .expect("pending compaction cleanup mutex poisoned") .clone(); let Some(cleanup) = cleanup else { - return; + return Ok(()); }; let Some(registry) = &self.internal_worker_registry else { - return; + return Err(WorkerError::CompactionCleanupPending { + source: ScopeLockError::Io(std::io::Error::other( + "compaction registry unavailable during cleanup", + )), + }); }; - loop { - match registry.stop_service(&cleanup.session_id).await { - Ok(_) => { - let mut pending = self - .pending_compaction_cleanup - .lock() - .expect("pending compaction cleanup mutex poisoned"); - if pending - .as_ref() - .is_some_and(|pending| pending.session_id == cleanup.session_id) - { - *pending = None; - } - return; - } - Err(error) => { - warn!( - compaction_id = %cleanup.compaction_id, - error = %error, - "compaction service cleanup failed; retaining cleanup authority and retrying" - ); - tokio::time::sleep(Duration::from_millis(50)).await; + match registry.stop_service(&cleanup.session_id).await { + Ok(_) => { + let mut pending = self + .pending_compaction_cleanup + .lock() + .expect("pending compaction cleanup mutex poisoned"); + if pending + .as_ref() + .is_some_and(|pending| pending.session_id == cleanup.session_id) + { + *pending = None; } + Ok(()) + } + Err(source) => { + warn!( + compaction_id = %cleanup.compaction_id, + error = %source, + "compaction service cleanup failed; retaining cleanup authority" + ); + Err(WorkerError::CompactionCleanupPending { + source: ScopeLockError::Io(source), + }) } } } - async fn release_compaction_service(&self, lifecycle: &mut CompactionLifecycle) { - self.release_pending_compaction_service().await; - if self - .pending_compaction_cleanup - .lock() - .expect("pending compaction cleanup mutex poisoned") - .is_none() - { - lifecycle.internal_worker = None; - } + async fn release_compaction_service( + &self, + lifecycle: &mut CompactionLifecycle, + ) -> Result<(), WorkerError> { + self.release_pending_compaction_service().await?; + lifecycle.internal_worker = None; + Ok(()) } async fn compact_impl( @@ -5761,7 +5775,7 @@ impl Worker { // activation so no internal process or registry authority survives into // the replacement's live lifetime. Cleanup failures retain this future // (and Busy state) until a retry succeeds. - self.release_compaction_service(lifecycle).await; + self.release_compaction_service(lifecycle).await?; self.set_compaction_progress(Some(InFlightCompaction { phase: CompactionPhase::Committing, @@ -7433,6 +7447,12 @@ pub enum WorkerError { source: ScopeLockError, }, + #[error("compaction cleanup is pending; retry or shut down the Worker")] + CompactionCleanupPending { + #[source] + source: ScopeLockError, + }, + #[error(transparent)] PromptCatalog(#[from] CatalogError), diff --git a/docs/design/compaction.md b/docs/design/compaction.md index 01abe4fd7..3a46434a8 100644 --- a/docs/design/compaction.md +++ b/docs/design/compaction.md @@ -59,8 +59,10 @@ closed and restore admission cannot register a competing writer until process teardown, after which restore converges from the replacement named by metadata. The Worker must likewise remain non-idle while terminal compaction-service cleanup is pending. Cleanup authority is stored on the Worker (not only on one compaction -future), retried until the registry record is gone, and cleared exactly once before -activation or a terminal result. +future). A failed stop returns a typed cleanup-pending outcome without clearing that +authority; the next compaction boundary retries it. The authority and lifecycle +reference are cleared exactly once after the registry record is gone, before +activation or a terminal idle-capable result. ## Metrics and comparison procedure From 103ef40f57da6d0752f281a1f52e92c0b4cb6ed1 Mon Sep 17 00:00:00 2001 From: Hare Date: Fri, 18 Sep 2026 06:55:23 +0900 Subject: [PATCH 04/13] worker: fence controller during repair --- crates/worker/src/controller.rs | 138 +++++++++++++++++++++++++++++++- crates/worker/src/worker.rs | 18 ++++- docs/design/compaction.md | 5 +- 3 files changed, 157 insertions(+), 4 deletions(-) diff --git a/crates/worker/src/controller.rs b/crates/worker/src/controller.rs index 5848268bd..a137134b2 100644 --- a/crates/worker/src/controller.rs +++ b/crates/worker/src/controller.rs @@ -1489,6 +1489,44 @@ where Ok(workdir_for_view) } +fn reject_method_while_attention_locked( + method: &Method, + working_event_tx: &broadcast::Sender, + message: &str, +) -> bool { + match method { + Method::Shutdown { .. } | Method::ListPendingSubmissions => false, + Method::Submit { + submission_request_id, + .. + } + | Method::SubmitTracked { + submission_request_id, + .. + } => { + let _ = working_event_tx.send(Event::SubmissionRejected { + submission_request_id: submission_request_id.clone(), + message: message.to_owned(), + }); + true + } + Method::Notify { .. } | Method::NotifyTracked { .. } => { + let _ = working_event_tx.send(Event::Error { + code: ErrorCode::InvalidRequest, + message: message.to_owned(), + }); + true + } + _ => { + let _ = working_event_tx.send(Event::Error { + code: ErrorCode::InvalidRequest, + message: message.to_owned(), + }); + true + } + } +} + /// Idle/Paused event loop. Each iteration either fires a staged /// `PendingRun` (delegating to [`drive_turn`] for the Running phase) or /// waits for the next `Method`. Method handlers stop at "update state + @@ -1551,6 +1589,12 @@ async fn controller_loop( let mut deferred_methods = VecDeque::new(); 'controller: loop { + if worker.has_failed_segment_activation() || worker.has_pending_compaction_cleanup() { + // A queued activation may have been accepted while the failing turn + // was still running. Preserve its durable queue entry, but never + // start it while repair/cleanup attention is required. + pending = None; + } // here so the status flip → drive_turn → finish sequence lives // in one place, regardless of which Method caused it. if let Some(run) = pending.take() { @@ -1700,7 +1744,15 @@ async fn controller_loop( tokio::select! { method = method_rx.recv() => match method { Some(method) => method, - None => break, + None => { + while worker.has_pending_compaction_cleanup() { + if let Err(error) = worker.retry_pending_compaction_cleanup().await { + tracing::warn!(error = %error, "shutdown waiting for compaction cleanup"); + tokio::time::sleep(Duration::from_millis(50)).await; + } + } + break 'controller; + } }, _ = notify_buffer.notified() => continue, _ = tokio::time::sleep(remaining) => { @@ -1726,12 +1778,63 @@ async fn controller_loop( tokio::select! { method = method_rx.recv() => match method { Some(method) => method, - None => break, + None => { + while worker.has_pending_compaction_cleanup() { + if let Err(error) = worker.retry_pending_compaction_cleanup().await { + tracing::warn!(error = %error, "shutdown waiting for compaction cleanup"); + tokio::time::sleep(Duration::from_millis(50)).await; + } + } + break 'controller; + } }, _ = notify_buffer.notified() => continue, } }; + if worker.has_pending_compaction_cleanup() { + match worker.retry_pending_compaction_cleanup().await { + Ok(()) => { + set_controller_state( + &shared_state, + &runtime_dir, + &working_event_tx, + WorkerState::Idle, + ) + .await; + } + Err(error) => { + let message = error.to_string(); + let _ = working_event_tx.send(Event::Error { + code: worker_error_code(&error), + message: message.clone(), + }); + if reject_method_while_attention_locked( + &method, + &working_event_tx, + &message, + ) { + continue; + } + // Shutdown is an explicit cleanup barrier: it is not allowed + // to complete while the service remains unconfirmed. + if matches!(&method, Method::Shutdown { .. }) { + continue; + } + } + } + } + + if worker.has_failed_segment_activation() + && reject_method_while_attention_locked( + &method, + &working_event_tx, + "Segment activation is incomplete; restore or shut down the Worker", + ) + { + continue; + } + match method { Method::Submit { submission_request_id, @@ -2165,6 +2268,15 @@ async fn controller_loop( }); } if shutdown_after_compaction { + while worker.has_pending_compaction_cleanup() { + if let Err(error) = worker.retry_pending_compaction_cleanup().await { + let _ = working_event_tx.send(Event::Error { + code: worker_error_code(&error), + message: error.to_string(), + }); + tokio::time::sleep(Duration::from_millis(50)).await; + } + } let _ = working_event_tx.send(Event::Shutdown); break 'controller; } @@ -3333,6 +3445,28 @@ mod tests { } } + #[test] + fn attention_gate_rejects_submit_before_dispatch() { + let (event_tx, mut event_rx) = broadcast::channel(4); + let method = Method::Submit { + submission_request_id: "request-1".into(), + input: Vec::new(), + }; + + assert!(reject_method_while_attention_locked( + &method, + &event_tx, + "repair required", + )); + assert!(matches!( + event_rx.try_recv().unwrap(), + Event::SubmissionRejected { + submission_request_id, + message, + } if submission_request_id == "request-1" && message == "repair required" + )); + } + #[tokio::test] async fn segment_activation_failure_from_run_never_publishes_idle() { let mut env = make_env().await; diff --git a/crates/worker/src/worker.rs b/crates/worker/src/worker.rs index eec9f1077..f0b9dce9e 100644 --- a/crates/worker/src/worker.rs +++ b/crates/worker/src/worker.rs @@ -5178,6 +5178,21 @@ impl Worker { } } + pub(crate) fn has_pending_compaction_cleanup(&self) -> bool { + self.pending_compaction_cleanup + .lock() + .expect("pending compaction cleanup mutex poisoned") + .is_some() + } + + pub(crate) fn has_failed_segment_activation(&self) -> bool { + self.segment_state.activation_failed.load(Ordering::Acquire) + } + + pub(crate) async fn retry_pending_compaction_cleanup(&self) -> Result<(), WorkerError> { + self.release_pending_compaction_service().await + } + async fn release_pending_compaction_service(&self) -> Result<(), WorkerError> { let cleanup = self .pending_compaction_cleanup @@ -5480,7 +5495,8 @@ impl Worker { })); if let Err(error) = handle.send(summary_input.text).await { - let _ = registry.remove_service(&handle.session_id_string()); + // Keep the registry record and Worker-owned cleanup authority. The + // outer compaction boundary performs the single stop attempt. return Err(WorkerError::InvalidState(error.to_string())); } match handle.wait_until_idle().await { diff --git a/docs/design/compaction.md b/docs/design/compaction.md index 3a46434a8..4a9e26894 100644 --- a/docs/design/compaction.md +++ b/docs/design/compaction.md @@ -62,7 +62,10 @@ is pending. Cleanup authority is stored on the Worker (not only on one compactio future). A failed stop returns a typed cleanup-pending outcome without clearing that authority; the next compaction boundary retries it. The authority and lifecycle reference are cleared exactly once after the registry record is gone, before -activation or a terminal idle-capable result. +activation or a terminal idle-capable result. While either activation repair or +cleanup attention is outstanding, the controller fences Submit/Notify dispatch and +does not start queued work. Shutdown is itself a cleanup barrier: it is not emitted +until the retained service record has been stopped successfully. ## Metrics and comparison procedure From b62b7c8f2d70579fe677551695dea6a764d12e83 Mon Sep 17 00:00:00 2001 From: Hare Date: Fri, 18 Sep 2026 16:45:40 +0900 Subject: [PATCH 05/13] worker: await cleanup on automatic shutdown --- crates/worker/src/controller.rs | 28 ++++++++++++++++++++++++++-- 1 file changed, 26 insertions(+), 2 deletions(-) diff --git a/crates/worker/src/controller.rs b/crates/worker/src/controller.rs index a137134b2..2d3e5ad09 100644 --- a/crates/worker/src/controller.rs +++ b/crates/worker/src/controller.rs @@ -1719,6 +1719,15 @@ async fn controller_loop( ) .await; if shutdown { + while worker.has_pending_compaction_cleanup() { + if let Err(error) = worker.retry_pending_compaction_cleanup().await { + let _ = working_event_tx.send(Event::Error { + code: worker_error_code(&error), + message: error.to_string(), + }); + tokio::time::sleep(Duration::from_millis(50)).await; + } + } let _ = working_event_tx.send(Event::Shutdown); break; } @@ -3502,9 +3511,24 @@ mod tests { } #[tokio::test] - async fn pending_compaction_cleanup_from_run_never_publishes_idle() { + async fn pending_compaction_cleanup_preserves_shutdown_request_for_outer_barrier() { let mut env = make_env().await; + let method_tx = env._method_tx.clone(); + env.shared_state + .transition(WorkerState::Busy(WorkerBusyState::Run( + WorkerRunState::Running, + ))); + tokio::spawn(async move { + tokio::time::sleep(Duration::from_millis(10)).await; + method_tx + .send(Method::Shutdown { + command: WorkerCommandEnvelope::new(1), + }) + .await + .expect("send shutdown"); + }); let worker_future = async { + tokio::time::sleep(Duration::from_millis(50)).await; Err::(WorkerError::CompactionCleanupPending { source: crate::runtime::worker_allocation::ScopeLockError::UnknownWorker( "worker".into(), @@ -3531,7 +3555,7 @@ mod tests { .await; assert_eq!(status, WorkerStatus::Running); - assert!(!shutdown); + assert!(shutdown); assert!(!may_drain_pending); } From ff87ccf8558597d31a6b690c6bc85600af95d9b9 Mon Sep 17 00:00:00 2001 From: Hare Date: Fri, 18 Sep 2026 17:26:39 +0900 Subject: [PATCH 06/13] worker: harden activation recovery --- crates/worker/src/controller.rs | 88 ++++++++++--------- .../src/runtime/worker_allocation/table.rs | 80 ++++++++++++++--- crates/worker/src/spawn/registry.rs | 68 ++++++++++++++ crates/worker/src/worker.rs | 86 ++++++++++++++++-- docs/design/compaction.md | 8 +- 5 files changed, 263 insertions(+), 67 deletions(-) diff --git a/crates/worker/src/controller.rs b/crates/worker/src/controller.rs index 2d3e5ad09..6ab3ae64a 100644 --- a/crates/worker/src/controller.rs +++ b/crates/worker/src/controller.rs @@ -1719,14 +1719,13 @@ async fn controller_loop( ) .await; if shutdown { - while worker.has_pending_compaction_cleanup() { - if let Err(error) = worker.retry_pending_compaction_cleanup().await { - let _ = working_event_tx.send(Event::Error { - code: worker_error_code(&error), - message: error.to_string(), - }); - tokio::time::sleep(Duration::from_millis(50)).await; - } + while let Err(error) = worker.finish_pending_compaction_cleanup_for_shutdown().await + { + let _ = working_event_tx.send(Event::Error { + code: worker_error_code(&error), + message: error.to_string(), + }); + tokio::time::sleep(Duration::from_millis(50)).await; } let _ = working_event_tx.send(Event::Shutdown); break; @@ -1754,11 +1753,11 @@ async fn controller_loop( method = method_rx.recv() => match method { Some(method) => method, None => { - while worker.has_pending_compaction_cleanup() { - if let Err(error) = worker.retry_pending_compaction_cleanup().await { - tracing::warn!(error = %error, "shutdown waiting for compaction cleanup"); - tokio::time::sleep(Duration::from_millis(50)).await; - } + while let Err(error) = + worker.finish_pending_compaction_cleanup_for_shutdown().await + { + tracing::warn!(error = %error, "shutdown cleanup quarantine failed"); + tokio::time::sleep(Duration::from_millis(50)).await; } break 'controller; } @@ -1788,11 +1787,11 @@ async fn controller_loop( method = method_rx.recv() => match method { Some(method) => method, None => { - while worker.has_pending_compaction_cleanup() { - if let Err(error) = worker.retry_pending_compaction_cleanup().await { - tracing::warn!(error = %error, "shutdown waiting for compaction cleanup"); - tokio::time::sleep(Duration::from_millis(50)).await; - } + while let Err(error) = + worker.finish_pending_compaction_cleanup_for_shutdown().await + { + tracing::warn!(error = %error, "shutdown cleanup quarantine failed"); + tokio::time::sleep(Duration::from_millis(50)).await; } break 'controller; } @@ -1813,22 +1812,29 @@ async fn controller_loop( .await; } Err(error) => { - let message = error.to_string(); - let _ = working_event_tx.send(Event::Error { - code: worker_error_code(&error), - message: message.clone(), - }); - if reject_method_while_attention_locked( - &method, - &working_event_tx, - &message, - ) { - continue; - } - // Shutdown is an explicit cleanup barrier: it is not allowed - // to complete while the service remains unconfirmed. if matches!(&method, Method::Shutdown { .. }) { - continue; + if let Err(error) = + worker.finish_pending_compaction_cleanup_for_shutdown().await + { + let _ = working_event_tx.send(Event::Error { + code: worker_error_code(&error), + message: error.to_string(), + }); + continue; + } + } else { + let message = error.to_string(); + let _ = working_event_tx.send(Event::Error { + code: worker_error_code(&error), + message: message.clone(), + }); + if reject_method_while_attention_locked( + &method, + &working_event_tx, + &message, + ) { + continue; + } } } } @@ -2277,14 +2283,14 @@ async fn controller_loop( }); } if shutdown_after_compaction { - while worker.has_pending_compaction_cleanup() { - if let Err(error) = worker.retry_pending_compaction_cleanup().await { - let _ = working_event_tx.send(Event::Error { - code: worker_error_code(&error), - message: error.to_string(), - }); - tokio::time::sleep(Duration::from_millis(50)).await; - } + while let Err(error) = + worker.finish_pending_compaction_cleanup_for_shutdown().await + { + let _ = working_event_tx.send(Event::Error { + code: worker_error_code(&error), + message: error.to_string(), + }); + tokio::time::sleep(Duration::from_millis(50)).await; } let _ = working_event_tx.send(Event::Shutdown); break 'controller; diff --git a/crates/worker/src/runtime/worker_allocation/table.rs b/crates/worker/src/runtime/worker_allocation/table.rs index 3056074f1..b916cd178 100644 --- a/crates/worker/src/runtime/worker_allocation/table.rs +++ b/crates/worker/src/runtime/worker_allocation/table.rs @@ -1,6 +1,6 @@ //! On-disk allocation table and the `flock`-protected guard. -use std::fs::{DirBuilder, File, OpenOptions}; +use std::fs::{self, DirBuilder, File, OpenOptions}; use std::io::{self, Read, Seek, SeekFrom, Write}; use std::os::unix::fs::{DirBuilderExt, OpenOptionsExt}; use std::path::{Path, PathBuf}; @@ -101,7 +101,9 @@ pub fn default_allocation_path() -> io::Result { /// never call `save` leave the table unchanged, which is the right /// behaviour for error paths. pub struct LockFileGuard { - file: File, + data_file: File, + lock_file: File, + data_path: PathBuf, data: LockFile, } @@ -121,16 +123,19 @@ impl LockFileGuard { .mode(0o700) .create(parent)?; } - let file = OpenOptions::new() + let mut lock_os = path.as_os_str().to_owned(); + lock_os.push(".lock"); + let lock_path = PathBuf::from(lock_os); + let lock_file = OpenOptions::new() .read(true) .write(true) .create(true) .truncate(false) .mode(0o600) - .open(path)?; + .open(&lock_path)?; let started = Instant::now(); loop { - match FileExt::try_lock_exclusive(&file) { + match FileExt::try_lock_exclusive(&lock_file) { Ok(true) => break, Ok(false) => { if started.elapsed() >= LOCK_WAIT_TIMEOUT { @@ -159,8 +164,17 @@ impl LockFileGuard { Err(error) => return Err(error), } } + let data_file = OpenOptions::new() + .read(true) + .write(true) + .create(true) + .truncate(false) + .mode(0o600) + .open(path)?; let mut this = Self { - file, + data_file, + lock_file, + data_path: path.to_path_buf(), data: LockFile::default(), }; this.reload()?; @@ -168,9 +182,9 @@ impl LockFileGuard { } fn reload(&mut self) -> io::Result<()> { - self.file.seek(SeekFrom::Start(0))?; + self.data_file.seek(SeekFrom::Start(0))?; let mut buf = String::new(); - self.file.read_to_string(&mut buf)?; + self.data_file.read_to_string(&mut buf)?; self.data = if buf.trim().is_empty() { LockFile::default() } else { @@ -192,20 +206,37 @@ impl LockFileGuard { &mut self.data } - /// Serialise `self.data` back to the file (truncate + rewrite). + /// Persist with atomic replacement while the separate allocation lock stays + /// held. A crash exposes either the old complete table or the new one. pub fn save(&mut self) -> io::Result<()> { let json = serde_json::to_vec_pretty(&self.data).map_err(io::Error::other)?; - self.file.seek(SeekFrom::Start(0))?; - self.file.set_len(0)?; - self.file.write_all(&json)?; - self.file.sync_data()?; + let mut temp_os = self.data_path.as_os_str().to_owned(); + temp_os.push(".tmp"); + let temp_path = PathBuf::from(temp_os); + let mut temp_file = OpenOptions::new() + .write(true) + .create(true) + .truncate(true) + .mode(0o600) + .open(&temp_path)?; + temp_file.write_all(&json)?; + temp_file.sync_all()?; + fs::rename(&temp_path, &self.data_path)?; + if let Some(parent) = self.data_path.parent() { + File::open(parent)?.sync_all()?; + } + self.data_file = OpenOptions::new() + .read(true) + .write(true) + .mode(0o600) + .open(&self.data_path)?; Ok(()) } } impl Drop for LockFileGuard { fn drop(&mut self) { - let _ = FileExt::unlock(&self.file); + let _ = FileExt::unlock(&self.lock_file); } } @@ -259,6 +290,27 @@ mod tests { assert_eq!(guard.data().allocations[0].worker_name, "a"); } + #[test] + fn allocation_lock_remains_exclusive_across_atomic_save() { + let dir = TempDir::new().unwrap(); + let path = dir.path().join("workers.json"); + let mut guard = open_empty(&path); + guard.save().unwrap(); + + let (sent, received) = std::sync::mpsc::channel(); + let contender_path = path.clone(); + let contender = std::thread::spawn(move || { + let contender = LockFileGuard::open(&contender_path).unwrap(); + sent.send(contender.data().allocations.len()).unwrap(); + }); + assert!(received + .recv_timeout(Duration::from_millis(50)) + .is_err()); + drop(guard); + assert_eq!(received.recv_timeout(Duration::from_secs(1)).unwrap(), 0); + contender.join().unwrap(); + } + #[test] fn find_by_session_skips_none_placeholders() { let dir = TempDir::new().unwrap(); diff --git a/crates/worker/src/spawn/registry.rs b/crates/worker/src/spawn/registry.rs index f328838d0..19d44b99d 100644 --- a/crates/worker/src/spawn/registry.rs +++ b/crates/worker/src/spawn/registry.rs @@ -311,6 +311,7 @@ impl Drop for InternalSpawnReservation { pub struct SpawnedWorkerRegistry { internal_records: std::sync::Mutex>, service_records: std::sync::Mutex>, + quarantined_service_cleanup: std::sync::Mutex>, internal_names: std::sync::Mutex>, internal_shutting_down: AtomicBool, pending_internal_spawns: AtomicUsize, @@ -331,6 +332,7 @@ impl SpawnedWorkerRegistry { Arc::new(Self { internal_records: std::sync::Mutex::new(Vec::new()), service_records: std::sync::Mutex::new(Vec::new()), + quarantined_service_cleanup: std::sync::Mutex::new(HashSet::new()), internal_names: std::sync::Mutex::new(HashSet::new()), internal_shutting_down: AtomicBool::new(false), pending_internal_spawns: AtomicUsize::new(0), @@ -346,6 +348,7 @@ impl SpawnedWorkerRegistry { Arc::new(Self { internal_records: std::sync::Mutex::new(Vec::new()), service_records: std::sync::Mutex::new(Vec::new()), + quarantined_service_cleanup: std::sync::Mutex::new(HashSet::new()), internal_names: std::sync::Mutex::new(HashSet::new()), internal_shutting_down: AtomicBool::new(false), pending_internal_spawns: AtomicUsize::new(0), @@ -360,6 +363,7 @@ impl SpawnedWorkerRegistry { Arc::new(Self { internal_records: std::sync::Mutex::new(Vec::new()), service_records: std::sync::Mutex::new(Vec::new()), + quarantined_service_cleanup: std::sync::Mutex::new(HashSet::new()), internal_names: std::sync::Mutex::new(HashSet::new()), internal_shutting_down: AtomicBool::new(false), pending_internal_spawns: AtomicUsize::new(0), @@ -443,6 +447,7 @@ impl SpawnedWorkerRegistry { registry: Arc::new(Self { internal_records: std::sync::Mutex::new(Vec::new()), service_records: std::sync::Mutex::new(Vec::new()), + quarantined_service_cleanup: std::sync::Mutex::new(HashSet::new()), internal_names: std::sync::Mutex::new(HashSet::new()), internal_shutting_down: AtomicBool::new(false), pending_internal_spawns: AtomicUsize::new(0), @@ -556,6 +561,45 @@ impl SpawnedWorkerRegistry { self.remove_service(session_id) } + /// Transfer cleanup authority to a registry-owned quarantine task after the + /// controller has exhausted its bounded shutdown attempts. The service record + /// remains registered until a later stop succeeds. + pub(crate) fn quarantine_service_cleanup( + self: &Arc, + session_id: String, + ) -> io::Result<()> { + let mut quarantined = self + .quarantined_service_cleanup + .lock() + .map_err(|_| io::Error::other("service cleanup quarantine lock poisoned"))?; + if !quarantined.insert(session_id.clone()) { + return Ok(()); + } + drop(quarantined); + + let registry = Arc::clone(self); + tokio::spawn(async move { + loop { + match registry.stop_service(&session_id).await { + Ok(_) => { + if let Ok(mut quarantined) = registry.quarantined_service_cleanup.lock() { + quarantined.remove(&session_id); + } + return; + } + Err(error) => { + warn!( + error = %error, + "quarantined service cleanup retry failed" + ); + tokio::time::sleep(std::time::Duration::from_millis(250)).await; + } + } + } + }); + Ok(()) + } + /// Remove one service Worker and emit the terminal projection fence. pub(crate) fn remove_service(&self, session_id: &str) -> io::Result { let removed = { @@ -1535,6 +1579,30 @@ mod tests { )); } + #[tokio::test] + async fn cleanup_quarantine_owns_record_until_background_stop_finishes() { + let registry = registry(); + registry + .quarantine_service_cleanup("missing-service".into()) + .unwrap(); + + tokio::time::timeout(std::time::Duration::from_secs(1), async { + loop { + if !registry + .quarantined_service_cleanup + .lock() + .unwrap() + .contains("missing-service") + { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + } + #[tokio::test] async fn read_only_summary_omits_unavailable_change_stat() { let tracker = tools::Tracker::new(); diff --git a/crates/worker/src/worker.rs b/crates/worker/src/worker.rs index f0b9dce9e..bc8fc3afb 100644 --- a/crates/worker/src/worker.rs +++ b/crates/worker/src/worker.rs @@ -1058,7 +1058,6 @@ pub struct SegmentState { entries_written: AtomicUsize, append_lock: Mutex<()>, activation_failed: AtomicBool, - activation_failure_authority: Mutex>, } impl SegmentState { @@ -1071,7 +1070,6 @@ impl SegmentState { entries_written: AtomicUsize::new(entries_written), append_lock: Mutex::new(()), activation_failed: AtomicBool::new(false), - activation_failure_authority: Mutex::new(None), }) } @@ -1108,11 +1106,7 @@ impl SegmentState { Ok(()) } - fn fail_closed_activation(&self, authority: SegmentActivationGuard) { - *self - .activation_failure_authority - .lock() - .expect("Segment activation failure authority mutex poisoned") = Some(authority); + fn fail_closed_activation(&self) { self.activation_failed.store(true, Ordering::Release); } @@ -2224,6 +2218,10 @@ pub struct Worker { /// interceptor before Agen applies them to live typed history. pending_committed_history: Arc>>>, + /// Machine-wide lock retained after a post-CAS allocation publication + /// failure. Declared before `scope_allocation` so drop releases this nested + /// authority before the outer allocation guard reacquires the same lock. + failed_activation_authority: Mutex>, /// Scope allocation in the machine-wide lock file. `Some` for /// Workers built via `from_manifest` / `from_manifest_spawned` / /// `restore_from_manifest` (production paths); `None` for the @@ -2285,6 +2283,19 @@ pub struct Worker { log_writer: Option>, } +impl Drop for Worker { + fn drop(&mut self) { + // A failed activation authority nests the same machine-wide lock used by + // `ScopeAllocationGuard::drop`. Release the nested guard first so Worker + // teardown cannot self-deadlock while removing its outer allocation. + let authority = match self.failed_activation_authority.get_mut() { + Ok(authority) => authority.take(), + Err(poisoned) => poisoned.into_inner().take(), + }; + drop(authority); + } +} + impl Worker { pub async fn stop_feature_runtime(&mut self, reason: impl Into) { if let Some(hooks) = self.hook_registry.clone() { @@ -2484,6 +2495,7 @@ impl Worker { notification_coalesce_delay: DEFAULT_NOTIFICATION_COALESCE_DELAY, pending_attachments: Arc::new(Mutex::new(Vec::::new())), pending_committed_history: Arc::new(Mutex::new(std::collections::VecDeque::new())), + failed_activation_authority: Mutex::new(None), scope_allocation: None, callback_socket: None, runtime_ticket_role: None, @@ -4522,7 +4534,7 @@ impl Worker { if let Some(allocation_activation) = allocation_activation { if let Err(error) = allocation_activation.commit() { let (source, authority) = error.into_parts(); - segment_state.fail_closed_activation(authority); + self.fail_closed_segment_activation(authority); return Err(WorkerError::SegmentActivationIncomplete { source }); } } @@ -5185,6 +5197,14 @@ impl Worker { .is_some() } + fn fail_closed_segment_activation(&self, authority: SegmentActivationGuard) { + *self + .failed_activation_authority + .lock() + .expect("failed activation authority mutex poisoned") = Some(authority); + self.segment_state.fail_closed_activation(); + } + pub(crate) fn has_failed_segment_activation(&self) -> bool { self.segment_state.activation_failed.load(Ordering::Acquire) } @@ -5193,6 +5213,50 @@ impl Worker { self.release_pending_compaction_service().await } + pub(crate) async fn finish_pending_compaction_cleanup_for_shutdown( + &self, + ) -> Result<(), WorkerError> { + const SHUTDOWN_CLEANUP_ATTEMPTS: usize = 3; + let mut last_error = None; + for attempt in 0..SHUTDOWN_CLEANUP_ATTEMPTS { + match self.retry_pending_compaction_cleanup().await { + Ok(()) => return Ok(()), + Err(error) => { + last_error = Some(error); + if attempt + 1 < SHUTDOWN_CLEANUP_ATTEMPTS { + tokio::time::sleep(Duration::from_millis(50)).await; + } + } + } + } + + let cleanup = self + .pending_compaction_cleanup + .lock() + .expect("pending compaction cleanup mutex poisoned") + .clone(); + let Some(cleanup) = cleanup else { + return Ok(()); + }; + let Some(registry) = &self.internal_worker_registry else { + return Err(last_error.unwrap_or_else(|| WorkerError::CompactionCleanupPending { + source: ScopeLockError::Io(std::io::Error::other( + "compaction registry unavailable during shutdown quarantine", + )), + })); + }; + registry + .quarantine_service_cleanup(cleanup.session_id.clone()) + .map_err(|source| WorkerError::CompactionCleanupPending { + source: ScopeLockError::Io(source), + })?; + *self + .pending_compaction_cleanup + .lock() + .expect("pending compaction cleanup mutex poisoned") = None; + Ok(()) + } + async fn release_pending_compaction_service(&self) -> Result<(), WorkerError> { let cleanup = self .pending_compaction_cleanup @@ -5904,7 +5968,7 @@ impl Worker { // the old in-memory writer to accept another byte while durable // authorities disagree; retain the allocation lock so restore // admission cannot register a competing writer. - segment_state.fail_closed_activation(authority); + self.fail_closed_segment_activation(authority); return Err(WorkerError::SegmentActivationIncomplete { source }); } } @@ -6123,6 +6187,7 @@ where notification_coalesce_delay: DEFAULT_NOTIFICATION_COALESCE_DELAY, pending_attachments: Arc::new(Mutex::new(Vec::::new())), pending_committed_history: Arc::new(Mutex::new(std::collections::VecDeque::new())), + failed_activation_authority: Mutex::new(None), scope_allocation: Some(scope_allocation), callback_socket: None, runtime_ticket_role: None, @@ -6210,6 +6275,7 @@ where notification_coalesce_delay: DEFAULT_NOTIFICATION_COALESCE_DELAY, pending_attachments: Arc::new(Mutex::new(Vec::::new())), pending_committed_history: Arc::new(Mutex::new(std::collections::VecDeque::new())), + failed_activation_authority: Mutex::new(None), scope_allocation: None, callback_socket: None, runtime_ticket_role: None, @@ -6332,6 +6398,7 @@ where notification_coalesce_delay: DEFAULT_NOTIFICATION_COALESCE_DELAY, pending_attachments: Arc::new(Mutex::new(Vec::::new())), pending_committed_history: Arc::new(Mutex::new(std::collections::VecDeque::new())), + failed_activation_authority: Mutex::new(None), scope_allocation: Some(scope_allocation), callback_socket: Some(callback_socket), runtime_ticket_role: None, @@ -6711,6 +6778,7 @@ where notification_coalesce_delay: DEFAULT_NOTIFICATION_COALESCE_DELAY, pending_attachments: Arc::new(Mutex::new(Vec::::new())), pending_committed_history: Arc::new(Mutex::new(std::collections::VecDeque::new())), + failed_activation_authority: Mutex::new(None), scope_allocation: Some(scope_allocation), callback_socket: None, runtime_ticket_role: None, diff --git a/docs/design/compaction.md b/docs/design/compaction.md index 4a9e26894..76ca2c9bf 100644 --- a/docs/design/compaction.md +++ b/docs/design/compaction.md @@ -42,7 +42,7 @@ activation boundary uses this lock order everywhere a live Worker changes Segmen 3. write the final pending-submission checkpoint and stage the replacement Segment; 4. acquire and preflight the machine-wide Worker allocation lock; 5. compare-and-swap `metadata.json` from the source Segment to the replacement; -6. update the allocation table while still holding its lock; +6. atomically replace the allocation table while still holding its separate lock; 7. publish the append destination, session projection, sink, and in-memory history; 8. release the append barrier. @@ -64,8 +64,10 @@ authority; the next compaction boundary retries it. The authority and lifecycle reference are cleared exactly once after the registry record is gone, before activation or a terminal idle-capable result. While either activation repair or cleanup attention is outstanding, the controller fences Submit/Notify dispatch and -does not start queued work. Shutdown is itself a cleanup barrier: it is not emitted -until the retained service record has been stopped successfully. +does not start queued work. Shutdown is itself a cleanup barrier: it performs a +bounded set of direct stop attempts and does not emit its terminal event until either +the registry record is gone or cleanup authority has moved to a runtime-owned +quarantine task. ## Metrics and comparison procedure From 6214e88fac3435d595845d143b2a20f84c392cbc Mon Sep 17 00:00:00 2001 From: Hare Date: Fri, 18 Sep 2026 17:52:06 +0900 Subject: [PATCH 07/13] worker: test activation fault recovery --- .../worker/src/runtime/worker_allocation.rs | 4 + .../runtime/worker_allocation/lifecycle.rs | 62 +++++++++++- .../src/runtime/worker_allocation/table.rs | 16 ++++ crates/worker/src/spawn/registry.rs | 38 ++++++++ crates/worker/src/worker.rs | 94 +++++++++++++++++++ 5 files changed, 212 insertions(+), 2 deletions(-) diff --git a/crates/worker/src/runtime/worker_allocation.rs b/crates/worker/src/runtime/worker_allocation.rs index 726c294ad..cb8b474b1 100644 --- a/crates/worker/src/runtime/worker_allocation.rs +++ b/crates/worker/src/runtime/worker_allocation.rs @@ -22,6 +22,10 @@ pub use lifecycle::{ ScopeAllocationGuard, SegmentActivationCommitError, SegmentActivationGuard, SegmentLockInfo, adopt_allocation, install_top_level, install_top_level_with_deny, lookup_segment, }; +#[cfg(test)] +pub(crate) use lifecycle::install_top_level_at_for_test; +#[cfg(test)] +pub(crate) use table::fail_next_save_for_test; pub use mutate::{ delegate_scope, reclaim_delegated_scope, reclaim_stale, reclaim_stale_with, register_worker, register_worker_with_deny, release_worker, diff --git a/crates/worker/src/runtime/worker_allocation/lifecycle.rs b/crates/worker/src/runtime/worker_allocation/lifecycle.rs index 2f758480c..6dd9c6537 100644 --- a/crates/worker/src/runtime/worker_allocation/lifecycle.rs +++ b/crates/worker/src/runtime/worker_allocation/lifecycle.rs @@ -201,6 +201,31 @@ pub fn install_top_level_with_deny( }) } +#[cfg(test)] +pub(crate) fn install_top_level_at_for_test( + lock_path: PathBuf, + worker_name: String, + pid: u32, + socket: PathBuf, + scope_allow: Vec, + segment_id: SegmentId, +) -> Result { + let mut guard = LockFileGuard::open(&lock_path)?; + super::mutate::register_worker_with_deny( + &mut guard, + worker_name.clone(), + pid, + socket, + scope_allow, + Vec::new(), + segment_id, + )?; + Ok(ScopeAllocationGuard { + worker_name, + lock_path, + }) +} + /// Take ownership of an existing allocation that was pre-registered by /// a spawning Worker. /// @@ -260,7 +285,7 @@ pub fn lookup_segment(segment_id: SegmentId) -> Result, #[cfg(test)] mod tests { - use super::super::table::Allocation; + use super::super::table::{Allocation, LockFile}; use super::super::test_util::*; use super::*; use tempfile::TempDir; @@ -464,7 +489,7 @@ mod tests { } if expected == old_segment )); // The authority is returned rather than dropped on failure. Production - // stores it on SegmentState to keep restore admission blocked. + // stores it on Worker to keep restore admission blocked. let (sent, received) = std::sync::mpsc::channel(); let lookup = std::thread::spawn(move || { sent.send(lookup_segment(replacement).unwrap()).unwrap(); @@ -480,6 +505,39 @@ mod tests { lookup.join().unwrap(); } + #[test] + fn allocation_save_failure_preserves_source_table_and_returns_authority() { + let dir = TempDir::new().unwrap(); + let _sandbox = RuntimeDirSandbox::new(dir.path()); + let lock_path = dir.path().join("workers.json"); + let old_segment = sid(); + let replacement = sid(); + let guard = install_top_level( + "activation".into(), + std::process::id(), + sock("activation"), + vec![write_rule("/work", true)], + old_segment, + ) + .unwrap(); + let activation = guard + .begin_segment_activation(old_segment, replacement) + .unwrap(); + crate::runtime::worker_allocation::table::fail_next_save_for_test(&lock_path); + + let error = activation.commit().unwrap_err(); + let (source, authority) = error.into_parts(); + assert!(matches!(source, ScopeLockError::Io(_))); + let persisted: LockFile = serde_json::from_slice(&std::fs::read(&lock_path).unwrap()).unwrap(); + assert_eq!( + persisted + .find("activation") + .and_then(|allocation| allocation.segment_id), + Some(old_segment) + ); + drop(authority); + } + #[test] fn dropping_segment_activation_preserves_source_allocation() { let dir = TempDir::new().unwrap(); diff --git a/crates/worker/src/runtime/worker_allocation/table.rs b/crates/worker/src/runtime/worker_allocation/table.rs index b916cd178..d06dad5cd 100644 --- a/crates/worker/src/runtime/worker_allocation/table.rs +++ b/crates/worker/src/runtime/worker_allocation/table.rs @@ -7,6 +7,14 @@ use std::path::{Path, PathBuf}; use std::thread; use std::time::{Duration, Instant}; +#[cfg(test)] +static FAIL_NEXT_SAVE_PATH: std::sync::Mutex> = std::sync::Mutex::new(None); + +#[cfg(test)] +pub(crate) fn fail_next_save_for_test(path: &Path) { + *FAIL_NEXT_SAVE_PATH.lock().unwrap() = Some(path.to_path_buf()); +} + use fs4::fs_std::FileExt; use manifest::{ScopeRule, paths}; use serde::{Deserialize, Serialize}; @@ -209,6 +217,14 @@ impl LockFileGuard { /// Persist with atomic replacement while the separate allocation lock stays /// held. A crash exposes either the old complete table or the new one. pub fn save(&mut self) -> io::Result<()> { + #[cfg(test)] + { + let mut fail_path = FAIL_NEXT_SAVE_PATH.lock().unwrap(); + if fail_path.as_ref() == Some(&self.data_path) { + *fail_path = None; + return Err(io::Error::other("injected allocation save failure")); + } + } let json = serde_json::to_vec_pretty(&self.data).map_err(io::Error::other)?; let mut temp_os = self.data_path.as_os_str().to_owned(); temp_os.push(".tmp"); diff --git a/crates/worker/src/spawn/registry.rs b/crates/worker/src/spawn/registry.rs index 19d44b99d..fc4a21dcb 100644 --- a/crates/worker/src/spawn/registry.rs +++ b/crates/worker/src/spawn/registry.rs @@ -9,6 +9,8 @@ //! rules to the parent scope. use std::collections::{BTreeMap, HashSet}; +#[cfg(test)] +use std::collections::HashMap; use std::io; use std::sync::{ Arc, Mutex, @@ -312,6 +314,8 @@ pub struct SpawnedWorkerRegistry { internal_records: std::sync::Mutex>, service_records: std::sync::Mutex>, quarantined_service_cleanup: std::sync::Mutex>, + #[cfg(test)] + fail_service_stops: std::sync::Mutex>, internal_names: std::sync::Mutex>, internal_shutting_down: AtomicBool, pending_internal_spawns: AtomicUsize, @@ -333,6 +337,8 @@ impl SpawnedWorkerRegistry { internal_records: std::sync::Mutex::new(Vec::new()), service_records: std::sync::Mutex::new(Vec::new()), quarantined_service_cleanup: std::sync::Mutex::new(HashSet::new()), + #[cfg(test)] + fail_service_stops: std::sync::Mutex::new(HashMap::new()), internal_names: std::sync::Mutex::new(HashSet::new()), internal_shutting_down: AtomicBool::new(false), pending_internal_spawns: AtomicUsize::new(0), @@ -349,6 +355,8 @@ impl SpawnedWorkerRegistry { internal_records: std::sync::Mutex::new(Vec::new()), service_records: std::sync::Mutex::new(Vec::new()), quarantined_service_cleanup: std::sync::Mutex::new(HashSet::new()), + #[cfg(test)] + fail_service_stops: std::sync::Mutex::new(HashMap::new()), internal_names: std::sync::Mutex::new(HashSet::new()), internal_shutting_down: AtomicBool::new(false), pending_internal_spawns: AtomicUsize::new(0), @@ -364,6 +372,8 @@ impl SpawnedWorkerRegistry { internal_records: std::sync::Mutex::new(Vec::new()), service_records: std::sync::Mutex::new(Vec::new()), quarantined_service_cleanup: std::sync::Mutex::new(HashSet::new()), + #[cfg(test)] + fail_service_stops: std::sync::Mutex::new(HashMap::new()), internal_names: std::sync::Mutex::new(HashSet::new()), internal_shutting_down: AtomicBool::new(false), pending_internal_spawns: AtomicUsize::new(0), @@ -448,6 +458,8 @@ impl SpawnedWorkerRegistry { internal_records: std::sync::Mutex::new(Vec::new()), service_records: std::sync::Mutex::new(Vec::new()), quarantined_service_cleanup: std::sync::Mutex::new(HashSet::new()), + #[cfg(test)] + fail_service_stops: std::sync::Mutex::new(HashMap::new()), internal_names: std::sync::Mutex::new(HashSet::new()), internal_shutting_down: AtomicBool::new(false), pending_internal_spawns: AtomicUsize::new(0), @@ -540,9 +552,35 @@ impl SpawnedWorkerRegistry { Ok(worker_ref) } + #[cfg(test)] + pub(crate) fn fail_service_stops_for_test(&self, session_id: &str, count: usize) { + self.fail_service_stops + .lock() + .unwrap() + .insert(session_id.to_owned(), count); + } + /// Stop and remove one parent-owned service Worker. This is host-only and is /// intentionally separate from the SubWorker control surface. + #[cfg(test)] + pub(crate) fn is_service_cleanup_quarantined_for_test(&self, session_id: &str) -> bool { + self.quarantined_service_cleanup + .lock() + .unwrap() + .contains(session_id) + } + pub(crate) async fn stop_service(&self, session_id: &str) -> io::Result { + #[cfg(test)] + { + let mut failures = self.fail_service_stops.lock().unwrap(); + if let Some(remaining) = failures.get_mut(session_id) { + if *remaining > 0 { + *remaining -= 1; + return Err(io::Error::other("injected service stop failure")); + } + } + } let record = self .service_records .lock() diff --git a/crates/worker/src/worker.rs b/crates/worker/src/worker.rs index bc8fc3afb..0c3d62aa3 100644 --- a/crates/worker/src/worker.rs +++ b/crates/worker/src/worker.rs @@ -8584,6 +8584,100 @@ mod build_summary_prompt_tests { assert_eq!(restored.active_run_turn_count, None); } + #[tokio::test] + async fn failed_activation_worker_drop_releases_nested_authority_before_allocation() { + let dir = tempfile::tempdir().unwrap(); + let store = session_store::FsStore::new(dir.path().join("sessions")).unwrap(); + let mut worker = Worker::new( + minimal_manifest(), + Engine::<_, Mutable, SessionHistoryMetadata>::new_annotated(NoopClient), + store, + WorkerWorkspaceContext::no_workspace(), + WorkerFilesystemAuthority::None, + Scope::empty(), + ) + .await + .unwrap(); + worker.ensure_segment_head().await.unwrap(); + let old_segment = worker.segment_id(); + let replacement = session_store::new_segment_id(); + let lock_path = dir.path().join("workers.json"); + let allocation = worker_allocation::install_top_level_at_for_test( + lock_path.clone(), + "test-worker".into(), + std::process::id(), + dir.path().join("worker.sock"), + Vec::new(), + old_segment, + ) + .unwrap(); + let activation = allocation + .begin_segment_activation(old_segment, replacement) + .unwrap(); + worker.scope_allocation = Some(allocation); + worker_allocation::fail_next_save_for_test(&lock_path); + let (source, authority) = activation.commit().unwrap_err().into_parts(); + assert!(matches!(source, ScopeLockError::Io(_))); + worker.fail_closed_segment_activation(authority); + assert!(worker.segment_state.ensure_append_allowed().is_err()); + + let (sent, received) = std::sync::mpsc::channel(); + let teardown = std::thread::spawn(move || { + drop(worker); + sent.send(()).unwrap(); + }); + received + .recv_timeout(Duration::from_secs(1)) + .expect("Worker teardown must not deadlock on nested allocation locks"); + teardown.join().unwrap(); + } + + #[tokio::test] + async fn shutdown_cleanup_failures_transfer_service_authority_to_quarantine() { + let dir = tempfile::tempdir().unwrap(); + let store = session_store::FsStore::new(dir.path().join("sessions")).unwrap(); + let mut worker = Worker::new( + minimal_manifest(), + Engine::<_, Mutable, SessionHistoryMetadata>::new_annotated(NoopClient), + store, + WorkerWorkspaceContext::no_workspace(), + WorkerFilesystemAuthority::None, + Scope::empty(), + ) + .await + .unwrap(); + let registry = + crate::spawn::registry::SpawnedWorkerRegistry::new_for_internal_services(); + registry.fail_service_stops_for_test("cleanup-session", 3); + worker.internal_worker_registry = Some(Arc::clone(®istry)); + *worker + .pending_compaction_cleanup + .lock() + .expect("pending cleanup lock poisoned") = Some(PendingCompactionCleanup { + session_id: "cleanup-session".into(), + compaction_id: "compaction".into(), + }); + + tokio::time::timeout( + Duration::from_secs(1), + worker.finish_pending_compaction_cleanup_for_shutdown(), + ) + .await + .expect("shutdown cleanup must be bounded") + .unwrap(); + assert!(!worker.has_pending_compaction_cleanup()); + tokio::time::timeout(Duration::from_secs(1), async { + loop { + if !registry.is_service_cleanup_quarantined_for_test("cleanup-session") { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + } + #[tokio::test] async fn auto_fork_checkpoints_interrupted_run_budget_for_restore() { let dir = tempfile::tempdir().unwrap(); From d1d391e51169a06280ae4e422e1a20c6fe49aacb Mon Sep 17 00:00:00 2001 From: Hare Date: Fri, 18 Sep 2026 18:15:33 +0900 Subject: [PATCH 08/13] worker: exercise activation recovery faults --- .../src/runtime/worker_allocation/table.rs | 17 +++-- crates/worker/src/spawn/registry.rs | 26 ++++++++ crates/worker/src/worker.rs | 62 +++++++++++++++++-- 3 files changed, 96 insertions(+), 9 deletions(-) diff --git a/crates/worker/src/runtime/worker_allocation/table.rs b/crates/worker/src/runtime/worker_allocation/table.rs index d06dad5cd..6bef5b1c7 100644 --- a/crates/worker/src/runtime/worker_allocation/table.rs +++ b/crates/worker/src/runtime/worker_allocation/table.rs @@ -8,11 +8,16 @@ use std::thread; use std::time::{Duration, Instant}; #[cfg(test)] -static FAIL_NEXT_SAVE_PATH: std::sync::Mutex> = std::sync::Mutex::new(None); +static FAIL_NEXT_SAVE_PATHS: std::sync::LazyLock< + std::sync::Mutex>, +> = std::sync::LazyLock::new(|| std::sync::Mutex::new(std::collections::HashSet::new())); #[cfg(test)] pub(crate) fn fail_next_save_for_test(path: &Path) { - *FAIL_NEXT_SAVE_PATH.lock().unwrap() = Some(path.to_path_buf()); + FAIL_NEXT_SAVE_PATHS + .lock() + .unwrap() + .insert(path.to_path_buf()); } use fs4::fs_std::FileExt; @@ -219,9 +224,11 @@ impl LockFileGuard { pub fn save(&mut self) -> io::Result<()> { #[cfg(test)] { - let mut fail_path = FAIL_NEXT_SAVE_PATH.lock().unwrap(); - if fail_path.as_ref() == Some(&self.data_path) { - *fail_path = None; + if FAIL_NEXT_SAVE_PATHS + .lock() + .unwrap() + .remove(&self.data_path) + { return Err(io::Error::other("injected allocation save failure")); } } diff --git a/crates/worker/src/spawn/registry.rs b/crates/worker/src/spawn/registry.rs index fc4a21dcb..b234da4a5 100644 --- a/crates/worker/src/spawn/registry.rs +++ b/crates/worker/src/spawn/registry.rs @@ -552,6 +552,32 @@ impl SpawnedWorkerRegistry { Ok(worker_ref) } + #[cfg(test)] + pub(crate) fn install_service_for_test( + &self, + ) -> ( + String, + tokio::sync::broadcast::Sender, + ) { + let (session, status_tx) = crate::internal_worker::test_internal_worker_session( + crate::internal_worker::InternalWorkerVisibility::ServicePrivate, + ); + let session_id = session.session_id_string(); + self.service_records.lock().unwrap().push( + InternalServiceWorkerRecord::new("test-service", "test-service", session), + ); + (session_id, status_tx) + } + + #[cfg(test)] + pub(crate) fn has_service_for_test(&self, session_id: &str) -> bool { + self.service_records + .lock() + .unwrap() + .iter() + .any(|record| record.session.session_id_string() == session_id) + } + #[cfg(test)] pub(crate) fn fail_service_stops_for_test(&self, session_id: &str, count: usize) { self.fail_service_stops diff --git a/crates/worker/src/worker.rs b/crates/worker/src/worker.rs index 0c3d62aa3..6044d4b75 100644 --- a/crates/worker/src/worker.rs +++ b/crates/worker/src/worker.rs @@ -8585,7 +8585,7 @@ mod build_summary_prompt_tests { } #[tokio::test] - async fn failed_activation_worker_drop_releases_nested_authority_before_allocation() { + async fn post_metadata_allocation_failure_fences_appends_and_restart_converges() { let dir = tempfile::tempdir().unwrap(); let store = session_store::FsStore::new(dir.path().join("sessions")).unwrap(); let mut worker = Worker::new( @@ -8601,6 +8601,27 @@ mod build_summary_prompt_tests { worker.ensure_segment_head().await.unwrap(); let old_segment = worker.segment_id(); let replacement = session_store::new_segment_id(); + let old_location = SegmentLocation { + session_id: worker.session_id(), + segment_id: old_segment, + }; + let replacement_location = SegmentLocation { + session_id: worker.session_id(), + segment_id: replacement, + }; + let persisted_metadata = Arc::new(Mutex::new(WorkerActiveSegmentRef::active_segment( + old_location.session_id, + old_location.segment_id, + ))); + let metadata_for_cas = Arc::clone(&persisted_metadata); + worker.worker_metadata_segment_cas = Some(Arc::new(move |_worker_name, expected, replacement| { + let mut persisted = metadata_for_cas.lock().unwrap(); + if &*persisted != expected { + return Ok(false); + } + *persisted = replacement; + Ok(true) + })); let lock_path = dir.path().join("workers.json"); let allocation = worker_allocation::install_top_level_at_for_test( lock_path.clone(), @@ -8615,6 +8636,16 @@ mod build_summary_prompt_tests { .begin_segment_activation(old_segment, replacement) .unwrap(); worker.scope_allocation = Some(allocation); + worker + .compare_and_swap_worker_metadata_segment(old_location, replacement_location) + .unwrap(); + assert_eq!( + *persisted_metadata.lock().unwrap(), + WorkerActiveSegmentRef::active_segment( + replacement_location.session_id, + replacement_location.segment_id, + ) + ); worker_allocation::fail_next_save_for_test(&lock_path); let (source, authority) = activation.commit().unwrap_err().into_parts(); assert!(matches!(source, ScopeLockError::Io(_))); @@ -8630,6 +8661,24 @@ mod build_summary_prompt_tests { .recv_timeout(Duration::from_secs(1)) .expect("Worker teardown must not deadlock on nested allocation locks"); teardown.join().unwrap(); + + // Process-stop releases the retained authority and the old allocation; + // restart follows the replacement named by committed metadata. + let recovered = worker_allocation::install_top_level_at_for_test( + lock_path, + "test-worker".into(), + std::process::id(), + dir.path().join("worker-restored.sock"), + Vec::new(), + replacement, + ) + .unwrap(); + drop( + recovered + .begin_segment_activation(replacement, session_store::new_segment_id()) + .unwrap(), + ); + drop(recovered); } #[tokio::test] @@ -8648,13 +8697,14 @@ mod build_summary_prompt_tests { .unwrap(); let registry = crate::spawn::registry::SpawnedWorkerRegistry::new_for_internal_services(); - registry.fail_service_stops_for_test("cleanup-session", 3); + let (cleanup_session_id, _events) = registry.install_service_for_test(); + registry.fail_service_stops_for_test(&cleanup_session_id, 4); worker.internal_worker_registry = Some(Arc::clone(®istry)); *worker .pending_compaction_cleanup .lock() .expect("pending cleanup lock poisoned") = Some(PendingCompactionCleanup { - session_id: "cleanup-session".into(), + session_id: cleanup_session_id.clone(), compaction_id: "compaction".into(), }); @@ -8666,9 +8716,13 @@ mod build_summary_prompt_tests { .expect("shutdown cleanup must be bounded") .unwrap(); assert!(!worker.has_pending_compaction_cleanup()); + assert!(registry.has_service_for_test(&cleanup_session_id)); + assert!(registry.is_service_cleanup_quarantined_for_test(&cleanup_session_id)); tokio::time::timeout(Duration::from_secs(1), async { loop { - if !registry.is_service_cleanup_quarantined_for_test("cleanup-session") { + if !registry.is_service_cleanup_quarantined_for_test(&cleanup_session_id) + && !registry.has_service_for_test(&cleanup_session_id) + { break; } tokio::task::yield_now().await; From d59879cc4b45b6b01e6102be744b5fd84934104f Mon Sep 17 00:00:00 2001 From: Hare Date: Fri, 18 Sep 2026 18:35:38 +0900 Subject: [PATCH 09/13] worker: cover activation race boundaries --- crates/worker/src/controller.rs | 4 +- .../src/runtime/worker_allocation/table.rs | 100 ++++++++++-- crates/worker/src/spawn/registry.rs | 78 ---------- crates/worker/src/worker.rs | 147 +++++++++++------- docs/design/compaction.md | 7 +- 5 files changed, 182 insertions(+), 154 deletions(-) diff --git a/crates/worker/src/controller.rs b/crates/worker/src/controller.rs index 6ab3ae64a..a6717f442 100644 --- a/crates/worker/src/controller.rs +++ b/crates/worker/src/controller.rs @@ -1756,7 +1756,7 @@ async fn controller_loop( while let Err(error) = worker.finish_pending_compaction_cleanup_for_shutdown().await { - tracing::warn!(error = %error, "shutdown cleanup quarantine failed"); + tracing::warn!(error = %error, "shutdown cleanup retry failed"); tokio::time::sleep(Duration::from_millis(50)).await; } break 'controller; @@ -1790,7 +1790,7 @@ async fn controller_loop( while let Err(error) = worker.finish_pending_compaction_cleanup_for_shutdown().await { - tracing::warn!(error = %error, "shutdown cleanup quarantine failed"); + tracing::warn!(error = %error, "shutdown cleanup retry failed"); tokio::time::sleep(Duration::from_millis(50)).await; } break 'controller; diff --git a/crates/worker/src/runtime/worker_allocation/table.rs b/crates/worker/src/runtime/worker_allocation/table.rs index 6bef5b1c7..0df4c0a80 100644 --- a/crates/worker/src/runtime/worker_allocation/table.rs +++ b/crates/worker/src/runtime/worker_allocation/table.rs @@ -8,16 +8,39 @@ use std::thread; use std::time::{Duration, Instant}; #[cfg(test)] -static FAIL_NEXT_SAVE_PATHS: std::sync::LazyLock< - std::sync::Mutex>, -> = std::sync::LazyLock::new(|| std::sync::Mutex::new(std::collections::HashSet::new())); +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum SaveFaultPoint { + BeforeRename, + AfterRename, +} + +#[cfg(test)] +static SAVE_FAULTS: std::sync::LazyLock< + std::sync::Mutex>, +> = std::sync::LazyLock::new(|| std::sync::Mutex::new(std::collections::HashMap::new())); #[cfg(test)] pub(crate) fn fail_next_save_for_test(path: &Path) { - FAIL_NEXT_SAVE_PATHS + fail_next_save_at_for_test(path, SaveFaultPoint::BeforeRename); +} + +#[cfg(test)] +pub(crate) fn fail_next_save_at_for_test(path: &Path, point: SaveFaultPoint) { + SAVE_FAULTS .lock() .unwrap() - .insert(path.to_path_buf()); + .insert(path.to_path_buf(), point); +} + +#[cfg(test)] +fn take_save_fault(path: &Path, point: SaveFaultPoint) -> bool { + let mut faults = SAVE_FAULTS.lock().unwrap(); + if faults.get(path) == Some(&point) { + faults.remove(path); + true + } else { + false + } } use fs4::fs_std::FileExt; @@ -222,16 +245,6 @@ impl LockFileGuard { /// Persist with atomic replacement while the separate allocation lock stays /// held. A crash exposes either the old complete table or the new one. pub fn save(&mut self) -> io::Result<()> { - #[cfg(test)] - { - if FAIL_NEXT_SAVE_PATHS - .lock() - .unwrap() - .remove(&self.data_path) - { - return Err(io::Error::other("injected allocation save failure")); - } - } let json = serde_json::to_vec_pretty(&self.data).map_err(io::Error::other)?; let mut temp_os = self.data_path.as_os_str().to_owned(); temp_os.push(".tmp"); @@ -244,7 +257,19 @@ impl LockFileGuard { .open(&temp_path)?; temp_file.write_all(&json)?; temp_file.sync_all()?; + #[cfg(test)] + if take_save_fault(&self.data_path, SaveFaultPoint::BeforeRename) { + return Err(io::Error::other( + "injected allocation save failure before rename", + )); + } fs::rename(&temp_path, &self.data_path)?; + #[cfg(test)] + if take_save_fault(&self.data_path, SaveFaultPoint::AfterRename) { + return Err(io::Error::other( + "injected allocation save failure after rename", + )); + } if let Some(parent) = self.data_path.parent() { File::open(parent)?.sync_all()?; } @@ -313,6 +338,51 @@ mod tests { assert_eq!(guard.data().allocations[0].worker_name, "a"); } + #[test] + fn atomic_save_reopens_complete_table_at_both_rename_boundaries() { + let dir = TempDir::new().unwrap(); + let path = dir.path().join("workers.json"); + let old_segment = sid(); + let replacement = sid(); + { + let mut guard = open_empty(&path); + register_worker( + &mut guard, + "a".into(), + std::process::id(), + sock("a"), + vec![write_rule("/src", true)], + old_segment, + ) + .unwrap(); + } + + { + let mut guard = LockFileGuard::open(&path).unwrap(); + guard.data_mut().find_mut("a").unwrap().segment_id = Some(replacement); + fail_next_save_at_for_test(&path, SaveFaultPoint::BeforeRename); + assert!(guard.save().is_err()); + } + let guard = LockFileGuard::open(&path).unwrap(); + assert_eq!( + guard.data().find("a").and_then(|allocation| allocation.segment_id), + Some(old_segment) + ); + drop(guard); + + { + let mut guard = LockFileGuard::open(&path).unwrap(); + guard.data_mut().find_mut("a").unwrap().segment_id = Some(replacement); + fail_next_save_at_for_test(&path, SaveFaultPoint::AfterRename); + assert!(guard.save().is_err()); + } + let guard = LockFileGuard::open(&path).unwrap(); + assert_eq!( + guard.data().find("a").and_then(|allocation| allocation.segment_id), + Some(replacement) + ); + } + #[test] fn allocation_lock_remains_exclusive_across_atomic_save() { let dir = TempDir::new().unwrap(); diff --git a/crates/worker/src/spawn/registry.rs b/crates/worker/src/spawn/registry.rs index b234da4a5..8592c7d66 100644 --- a/crates/worker/src/spawn/registry.rs +++ b/crates/worker/src/spawn/registry.rs @@ -313,7 +313,6 @@ impl Drop for InternalSpawnReservation { pub struct SpawnedWorkerRegistry { internal_records: std::sync::Mutex>, service_records: std::sync::Mutex>, - quarantined_service_cleanup: std::sync::Mutex>, #[cfg(test)] fail_service_stops: std::sync::Mutex>, internal_names: std::sync::Mutex>, @@ -336,7 +335,6 @@ impl SpawnedWorkerRegistry { Arc::new(Self { internal_records: std::sync::Mutex::new(Vec::new()), service_records: std::sync::Mutex::new(Vec::new()), - quarantined_service_cleanup: std::sync::Mutex::new(HashSet::new()), #[cfg(test)] fail_service_stops: std::sync::Mutex::new(HashMap::new()), internal_names: std::sync::Mutex::new(HashSet::new()), @@ -354,7 +352,6 @@ impl SpawnedWorkerRegistry { Arc::new(Self { internal_records: std::sync::Mutex::new(Vec::new()), service_records: std::sync::Mutex::new(Vec::new()), - quarantined_service_cleanup: std::sync::Mutex::new(HashSet::new()), #[cfg(test)] fail_service_stops: std::sync::Mutex::new(HashMap::new()), internal_names: std::sync::Mutex::new(HashSet::new()), @@ -371,7 +368,6 @@ impl SpawnedWorkerRegistry { Arc::new(Self { internal_records: std::sync::Mutex::new(Vec::new()), service_records: std::sync::Mutex::new(Vec::new()), - quarantined_service_cleanup: std::sync::Mutex::new(HashSet::new()), #[cfg(test)] fail_service_stops: std::sync::Mutex::new(HashMap::new()), internal_names: std::sync::Mutex::new(HashSet::new()), @@ -457,7 +453,6 @@ impl SpawnedWorkerRegistry { registry: Arc::new(Self { internal_records: std::sync::Mutex::new(Vec::new()), service_records: std::sync::Mutex::new(Vec::new()), - quarantined_service_cleanup: std::sync::Mutex::new(HashSet::new()), #[cfg(test)] fail_service_stops: std::sync::Mutex::new(HashMap::new()), internal_names: std::sync::Mutex::new(HashSet::new()), @@ -586,16 +581,6 @@ impl SpawnedWorkerRegistry { .insert(session_id.to_owned(), count); } - /// Stop and remove one parent-owned service Worker. This is host-only and is - /// intentionally separate from the SubWorker control surface. - #[cfg(test)] - pub(crate) fn is_service_cleanup_quarantined_for_test(&self, session_id: &str) -> bool { - self.quarantined_service_cleanup - .lock() - .unwrap() - .contains(session_id) - } - pub(crate) async fn stop_service(&self, session_id: &str) -> io::Result { #[cfg(test)] { @@ -625,45 +610,6 @@ impl SpawnedWorkerRegistry { self.remove_service(session_id) } - /// Transfer cleanup authority to a registry-owned quarantine task after the - /// controller has exhausted its bounded shutdown attempts. The service record - /// remains registered until a later stop succeeds. - pub(crate) fn quarantine_service_cleanup( - self: &Arc, - session_id: String, - ) -> io::Result<()> { - let mut quarantined = self - .quarantined_service_cleanup - .lock() - .map_err(|_| io::Error::other("service cleanup quarantine lock poisoned"))?; - if !quarantined.insert(session_id.clone()) { - return Ok(()); - } - drop(quarantined); - - let registry = Arc::clone(self); - tokio::spawn(async move { - loop { - match registry.stop_service(&session_id).await { - Ok(_) => { - if let Ok(mut quarantined) = registry.quarantined_service_cleanup.lock() { - quarantined.remove(&session_id); - } - return; - } - Err(error) => { - warn!( - error = %error, - "quarantined service cleanup retry failed" - ); - tokio::time::sleep(std::time::Duration::from_millis(250)).await; - } - } - } - }); - Ok(()) - } - /// Remove one service Worker and emit the terminal projection fence. pub(crate) fn remove_service(&self, session_id: &str) -> io::Result { let removed = { @@ -1643,30 +1589,6 @@ mod tests { )); } - #[tokio::test] - async fn cleanup_quarantine_owns_record_until_background_stop_finishes() { - let registry = registry(); - registry - .quarantine_service_cleanup("missing-service".into()) - .unwrap(); - - tokio::time::timeout(std::time::Duration::from_secs(1), async { - loop { - if !registry - .quarantined_service_cleanup - .lock() - .unwrap() - .contains("missing-service") - { - break; - } - tokio::task::yield_now().await; - } - }) - .await - .unwrap(); - } - #[tokio::test] async fn read_only_summary_omits_unavailable_change_stat() { let tracker = tools::Tracker::new(); diff --git a/crates/worker/src/worker.rs b/crates/worker/src/worker.rs index 6044d4b75..27f7cd354 100644 --- a/crates/worker/src/worker.rs +++ b/crates/worker/src/worker.rs @@ -5216,45 +5216,18 @@ impl Worker { pub(crate) async fn finish_pending_compaction_cleanup_for_shutdown( &self, ) -> Result<(), WorkerError> { - const SHUTDOWN_CLEANUP_ATTEMPTS: usize = 3; - let mut last_error = None; - for attempt in 0..SHUTDOWN_CLEANUP_ATTEMPTS { + loop { match self.retry_pending_compaction_cleanup().await { Ok(()) => return Ok(()), Err(error) => { - last_error = Some(error); - if attempt + 1 < SHUTDOWN_CLEANUP_ATTEMPTS { - tokio::time::sleep(Duration::from_millis(50)).await; - } + warn!( + error = %error, + "shutdown remains blocked on compaction service cleanup" + ); + tokio::time::sleep(Duration::from_millis(50)).await; } } } - - let cleanup = self - .pending_compaction_cleanup - .lock() - .expect("pending compaction cleanup mutex poisoned") - .clone(); - let Some(cleanup) = cleanup else { - return Ok(()); - }; - let Some(registry) = &self.internal_worker_registry else { - return Err(last_error.unwrap_or_else(|| WorkerError::CompactionCleanupPending { - source: ScopeLockError::Io(std::io::Error::other( - "compaction registry unavailable during shutdown quarantine", - )), - })); - }; - registry - .quarantine_service_cleanup(cleanup.session_id.clone()) - .map_err(|source| WorkerError::CompactionCleanupPending { - source: ScopeLockError::Io(source), - })?; - *self - .pending_compaction_cleanup - .lock() - .expect("pending compaction cleanup mutex poisoned") = None; - Ok(()) } async fn release_pending_compaction_service(&self) -> Result<(), WorkerError> { @@ -8682,7 +8655,7 @@ mod build_summary_prompt_tests { } #[tokio::test] - async fn shutdown_cleanup_failures_transfer_service_authority_to_quarantine() { + async fn shutdown_cleanup_stays_blocked_until_registered_service_stops() { let dir = tempfile::tempdir().unwrap(); let store = session_store::FsStore::new(dir.path().join("sessions")).unwrap(); let mut worker = Worker::new( @@ -8708,28 +8681,20 @@ mod build_summary_prompt_tests { compaction_id: "compaction".into(), }); - tokio::time::timeout( - Duration::from_secs(1), - worker.finish_pending_compaction_cleanup_for_shutdown(), - ) - .await - .expect("shutdown cleanup must be bounded") - .unwrap(); - assert!(!worker.has_pending_compaction_cleanup()); + let cleanup = worker.finish_pending_compaction_cleanup_for_shutdown(); + tokio::pin!(cleanup); + tokio::select! { + result = &mut cleanup => panic!("shutdown cleanup completed before injected failures: {result:?}"), + _ = tokio::time::sleep(Duration::from_millis(75)) => {} + } + assert!(worker.has_pending_compaction_cleanup()); assert!(registry.has_service_for_test(&cleanup_session_id)); - assert!(registry.is_service_cleanup_quarantined_for_test(&cleanup_session_id)); - tokio::time::timeout(Duration::from_secs(1), async { - loop { - if !registry.is_service_cleanup_quarantined_for_test(&cleanup_session_id) - && !registry.has_service_for_test(&cleanup_session_id) - { - break; - } - tokio::task::yield_now().await; - } - }) - .await - .unwrap(); + tokio::time::timeout(Duration::from_secs(1), cleanup) + .await + .expect("shutdown cleanup must converge after retry") + .unwrap(); + assert!(!worker.has_pending_compaction_cleanup()); + assert!(!registry.has_service_for_test(&cleanup_session_id)); } #[tokio::test] @@ -10479,6 +10444,78 @@ mod build_summary_prompt_tests { ); } + #[test] + fn submit_and_notification_wait_for_activation_barrier_then_append_to_replacement() { + let temp = tempfile::tempdir().unwrap(); + let handle = PendingSubmissionHandle::for_test(temp.path()); + let old_location = handle.writer.state.location(); + let replacement = session_store::new_segment_id(); + handle + .writer + .store + .create_segment(old_location.session_id, replacement, &[]) + .unwrap(); + let old_entries = handle + .writer + .store + .read_all(old_location.session_id, old_location.segment_id) + .unwrap() + .len(); + let append_guard = handle.writer.state.append_lock.lock().unwrap(); + let submit_handle = handle.clone(); + let notify_handle = handle.clone(); + let (started, starts) = std::sync::mpsc::channel(); + let (finished, finishes) = std::sync::mpsc::channel(); + let submit_started = started.clone(); + let submit_finished = finished.clone(); + let submit = std::thread::spawn(move || { + submit_started.send(()).unwrap(); + submit_handle + .accept("barrier-submit".into(), vec![Segment::text("queued")], false) + .unwrap(); + submit_finished.send(()).unwrap(); + }); + let notify = std::thread::spawn(move || { + started.send(()).unwrap(); + notify_handle + .accept_notification("barrier-notify".into(), "completed".into()) + .unwrap(); + finished.send(()).unwrap(); + }); + starts.recv_timeout(Duration::from_secs(1)).unwrap(); + starts.recv_timeout(Duration::from_secs(1)).unwrap(); + assert!(finishes.recv_timeout(Duration::from_millis(50)).is_err()); + + handle.writer.state.set_location(SegmentLocation { + session_id: old_location.session_id, + segment_id: replacement, + }); + drop(append_guard); + finishes.recv_timeout(Duration::from_secs(1)).unwrap(); + finishes.recv_timeout(Duration::from_secs(1)).unwrap(); + submit.join().unwrap(); + notify.join().unwrap(); + + assert_eq!( + handle + .writer + .store + .read_all(old_location.session_id, old_location.segment_id) + .unwrap() + .len(), + old_entries + ); + assert_eq!( + handle + .writer + .store + .read_all(old_location.session_id, replacement) + .unwrap() + .len(), + 2 + ); + } + #[test] fn pending_submission_queue_is_durable_idempotent_and_bounded() { let temp = tempfile::tempdir().unwrap(); diff --git a/docs/design/compaction.md b/docs/design/compaction.md index 76ca2c9bf..f12740c07 100644 --- a/docs/design/compaction.md +++ b/docs/design/compaction.md @@ -64,10 +64,9 @@ authority; the next compaction boundary retries it. The authority and lifecycle reference are cleared exactly once after the registry record is gone, before activation or a terminal idle-capable result. While either activation repair or cleanup attention is outstanding, the controller fences Submit/Notify dispatch and -does not start queued work. Shutdown is itself a cleanup barrier: it performs a -bounded set of direct stop attempts and does not emit its terminal event until either -the registry record is gone or cleanup authority has moved to a runtime-owned -quarantine task. +does not start queued work. Shutdown is itself a cleanup barrier: it retries the +Worker-owned stop authority with backoff and does not emit its terminal event until +the registry record is gone. ## Metrics and comparison procedure From 72a61e4952c5032d7811182bc795ea3db2b7e39a Mon Sep 17 00:00:00 2001 From: Hare Date: Fri, 18 Sep 2026 18:53:47 +0900 Subject: [PATCH 10/13] worker: persist compaction-time notifications --- crates/worker/src/controller.rs | 147 +++++++++++++++++++++++++++++++- crates/worker/src/worker.rs | 96 +++++++++++++++++++++ docs/design/compaction.md | 12 +-- 3 files changed, 248 insertions(+), 7 deletions(-) diff --git a/crates/worker/src/controller.rs b/crates/worker/src/controller.rs index a6717f442..842b9a4a6 100644 --- a/crates/worker/src/controller.rs +++ b/crates/worker/src/controller.rs @@ -25,7 +25,9 @@ use crate::shutdown_after_idle::{ }; use crate::spawn::registry::SpawnedWorkerRegistry; use crate::spawn::tool::sub_worker_spawn_tool; -use crate::worker::{SystemItemCommitter, Worker, WorkerError, WorkerRunResult}; +use crate::worker::{ + PendingSubmissionHandle, SystemItemCommitter, Worker, WorkerError, WorkerRunResult, +}; use protocol::{ AlertLevel, AlertSource, CommandEvent as ProtocolCommandEvent, CommandSnapshot as ProtocolCommandSnapshot, CommandStatus as ProtocolCommandStatus, @@ -1489,6 +1491,115 @@ where Ok(workdir_for_view) } +fn durably_accept_method_while_busy( + method: Method, + pending_submissions: &PendingSubmissionHandle, + working_event_tx: &broadcast::Sender, +) -> Option +where + St: session_store::Store + Clone, +{ + match method { + Method::Submit { + submission_request_id, + input, + } => { + let request_id = submission_request_id.clone(); + match pending_submissions.accept_from_source( + submission_request_id, + input, + pending_submissions.direct_client_namespace(), + session_store::LoggedSessionHistoryOrigin::LegacyUnknown, + false, + ) { + Ok(acceptance) => { + let _ = working_event_tx.send(Event::SubmissionAccepted { + submission_request_id: acceptance.submission_request_id, + submission_id: acceptance.submission_id, + disposition: acceptance.disposition, + }); + } + Err(error) => { + let _ = working_event_tx.send(Event::SubmissionRejected { + submission_request_id: request_id, + message: error.to_string(), + }); + } + } + None + } + Method::SubmitTracked { + submission_request_id, + input, + source, + } => { + let request_id = submission_request_id.clone(); + let (source_namespace, provenance) = + resolved_input_source(pending_submissions, &source); + match pending_submissions.accept_from_source( + submission_request_id, + input, + source_namespace, + provenance, + false, + ) { + Ok(acceptance) => { + let _ = working_event_tx.send(Event::SubmissionAccepted { + submission_request_id: acceptance.submission_request_id, + submission_id: acceptance.submission_id, + disposition: acceptance.disposition, + }); + } + Err(error) => { + let _ = working_event_tx.send(Event::SubmissionRejected { + submission_request_id: request_id, + message: error.to_string(), + }); + } + } + None + } + Method::Notify { + notification_request_id, + message, + } => { + if let Err(error) = pending_submissions.accept_notification_from_source( + notification_request_id, + message, + pending_submissions.direct_client_namespace(), + session_store::LoggedSessionHistoryOrigin::LegacyUnknown, + ) { + let _ = working_event_tx.send(Event::Error { + code: ErrorCode::InvalidRequest, + message: error.to_string(), + }); + } + None + } + Method::NotifyTracked { + notification_request_id, + message, + source, + } => { + let (source_namespace, provenance) = + resolved_input_source(pending_submissions, &source); + if let Err(error) = pending_submissions.accept_notification_from_source( + notification_request_id, + message, + source_namespace, + provenance, + ) { + let _ = working_event_tx.send(Event::Error { + code: ErrorCode::InvalidRequest, + message: error.to_string(), + }); + } + None + } + method => Some(method), + } +} + fn reject_method_while_attention_locked( method: &Method, working_event_tx: &broadcast::Sender, @@ -2250,7 +2361,15 @@ async fn controller_loop( ); let _ = cancel_tx.send(true); } - Some(method) => deferred_methods.push_back(method), + Some(method) => { + if let Some(method) = durably_accept_method_while_busy( + method, + &pending_submissions, + &working_event_tx, + ) { + deferred_methods.push_back(method); + } + } None => { shutdown_after_compaction = true; let _ = cancel_tx.send(true); @@ -3460,6 +3579,30 @@ mod tests { } } + #[test] + fn busy_notification_is_durable_before_shutdown_can_observe_it() { + let dir = tempfile::tempdir().unwrap(); + let pending = PendingSubmissionHandle::for_test(dir.path()); + let (event_tx, _event_rx) = broadcast::channel(4); + + assert!(durably_accept_method_while_busy( + Method::NotifyTracked { + notification_request_id: "completion-1".into(), + message: "subworker completed".into(), + source: protocol::AuthenticatedInputSource::UntrustedWire, + }, + &pending, + &event_tx, + ) + .is_none()); + let entries = pending.persisted_entries_for_test(); + assert!(entries.iter().any(|entry| matches!( + entry, + LogEntry::Extension { domain, .. } + if domain == "worker.pending_activations.v1" + ))); + } + #[test] fn attention_gate_rejects_submit_before_dispatch() { let (event_tx, mut event_rx) = broadcast::channel(4); diff --git a/crates/worker/src/worker.rs b/crates/worker/src/worker.rs index 27f7cd354..adf4de779 100644 --- a/crates/worker/src/worker.rs +++ b/crates/worker/src/worker.rs @@ -1930,6 +1930,15 @@ impl PendingSubmissionHandle { }, } } + + #[cfg(test)] + pub(crate) fn persisted_entries_for_test(&self) -> Vec { + let location = self.writer.state.location(); + self.writer + .store + .read_all(location.session_id, location.segment_id) + .expect("read test pending entries") + } } /// Type-erased commit handle for the interceptor. Lets the interceptor commit `SystemItem`s without being generic over the @@ -8654,6 +8663,93 @@ mod build_summary_prompt_tests { drop(recovered); } + #[tokio::test] + async fn crash_after_allocation_before_live_switch_recovers_committed_replacement() { + let dir = tempfile::tempdir().unwrap(); + let store = session_store::FsStore::new(dir.path().join("sessions")).unwrap(); + let mut worker = Worker::new( + minimal_manifest(), + Engine::<_, Mutable, SessionHistoryMetadata>::new_annotated(NoopClient), + store, + WorkerWorkspaceContext::no_workspace(), + WorkerFilesystemAuthority::None, + Scope::empty(), + ) + .await + .unwrap(); + worker.ensure_segment_head().await.unwrap(); + let old_location = worker.segment_state.location(); + let replacement_location = SegmentLocation { + session_id: old_location.session_id, + segment_id: session_store::new_segment_id(), + }; + let persisted_metadata = Arc::new(Mutex::new(WorkerActiveSegmentRef::active_segment( + old_location.session_id, + old_location.segment_id, + ))); + let metadata_for_cas = Arc::clone(&persisted_metadata); + worker.worker_metadata_segment_cas = Some(Arc::new(move |_worker_name, expected, replacement| { + let mut persisted = metadata_for_cas.lock().unwrap(); + if &*persisted != expected { + return Ok(false); + } + *persisted = replacement; + Ok(true) + })); + let lock_path = dir.path().join("workers.json"); + let allocation = worker_allocation::install_top_level_at_for_test( + lock_path.clone(), + "test-worker".into(), + std::process::id(), + dir.path().join("worker.sock"), + Vec::new(), + old_location.segment_id, + ) + .unwrap(); + let activation = allocation + .begin_segment_activation( + old_location.segment_id, + replacement_location.segment_id, + ) + .unwrap(); + worker.scope_allocation = Some(allocation); + worker + .compare_and_swap_worker_metadata_segment(old_location, replacement_location) + .unwrap(); + activation.commit().unwrap(); + + // Simulated crash point: durable metadata and allocation moved, while the + // in-memory append destination still names the source Segment. + assert_eq!(worker.segment_id(), old_location.segment_id); + assert_eq!( + *persisted_metadata.lock().unwrap(), + WorkerActiveSegmentRef::active_segment( + replacement_location.session_id, + replacement_location.segment_id, + ) + ); + drop(worker); + + let recovered = worker_allocation::install_top_level_at_for_test( + lock_path, + "test-worker".into(), + std::process::id(), + dir.path().join("worker-restored.sock"), + Vec::new(), + replacement_location.segment_id, + ) + .unwrap(); + drop( + recovered + .begin_segment_activation( + replacement_location.segment_id, + session_store::new_segment_id(), + ) + .unwrap(), + ); + drop(recovered); + } + #[tokio::test] async fn shutdown_cleanup_stays_blocked_until_registered_service_stops() { let dir = tempfile::tempdir().unwrap(); diff --git a/docs/design/compaction.md b/docs/design/compaction.md index f12740c07..6d33041b5 100644 --- a/docs/design/compaction.md +++ b/docs/design/compaction.md @@ -46,11 +46,13 @@ activation boundary uses this lock order everywhere a live Worker changes Segmen 7. publish the append destination, session projection, sink, and in-memory history; 8. release the append barrier. -Submit and Notify acceptance take the same append barrier. Consequently an accepted -item is either durably included in the replacement checkpoint, or waits until every -live authority points at the replacement and is then appended there. Restore -admission takes the allocation lock, so it cannot observe the interval between the -metadata CAS and allocation hand-off. +Submit and Notify acceptance take the same append barrier. The controller persists +requests received while manual compaction is running immediately rather than holding +volatile deferred methods, so an overlapping Shutdown cannot discard a SubWorker +completion. Consequently an accepted item is either durably included in the +replacement checkpoint, or waits until every live authority points at the replacement +and is then appended there. Restore admission takes the allocation lock, so it cannot +observe the interval between the metadata CAS and allocation hand-off. The metadata CAS is the commit point. Failures before it leave the source Segment active. A failure updating allocation after it marks the old in-memory writer From 19b04a8d8f000c36e46e5e03733cfa462e400a13 Mon Sep 17 00:00:00 2001 From: Hare Date: Fri, 18 Sep 2026 18:58:05 +0900 Subject: [PATCH 11/13] worker: verify completion durability on shutdown --- crates/worker/tests/compact_events_test.rs | 47 ++++++++++++++++++++-- 1 file changed, 43 insertions(+), 4 deletions(-) diff --git a/crates/worker/tests/compact_events_test.rs b/crates/worker/tests/compact_events_test.rs index e88fc223d..8e2a91af6 100644 --- a/crates/worker/tests/compact_events_test.rs +++ b/crates/worker/tests/compact_events_test.rs @@ -344,7 +344,10 @@ target = "./" permission = "write" "#; -async fn make_worker_with_manifest(manifest_toml: &str, client: C) -> Worker +async fn make_worker_with_manifest_and_store( + manifest_toml: &str, + client: C, +) -> (Worker, TestStore) where C: LlmClient + Clone + Send + Sync + 'static, { @@ -364,6 +367,7 @@ where let worker = Engine::<_, agen::state::Mutable, worker::SessionHistoryMetadata>::new_annotated(client); + let observed_store = store.clone(); let mut worker = Worker::new( manifest, worker, @@ -375,7 +379,16 @@ where .await .unwrap(); worker.enable_worker_metadata_write_through().unwrap(); - worker + (worker, observed_store) +} + +async fn make_worker_with_manifest(manifest_toml: &str, client: C) -> Worker +where + C: LlmClient + Clone + Send + Sync + 'static, +{ + make_worker_with_manifest_and_store(manifest_toml, client) + .await + .0 } async fn make_worker(client: MockClient) -> Worker { @@ -1104,8 +1117,13 @@ async fn pre_run_compact_failure_clears_runtime_progress() { #[tokio::test] async fn manual_compact_cancel_clears_progress_before_returning_idle() { - let worker = - make_worker_with_manifest(POST_RUN_MANIFEST_TOML, BlockingCompactClient::new()).await; + let (worker, observed_store) = make_worker_with_manifest_and_store( + POST_RUN_MANIFEST_TOML, + BlockingCompactClient::new(), + ) + .await; + let source_session = worker.session_id(); + let source_segment = worker.segment_id(); let runtime_tmp = tempfile::tempdir().unwrap(); let bash_output_dir = runtime_tmp.path().join("bash-output"); let (handle, shutdown_receiver) = @@ -1201,6 +1219,15 @@ async fn manual_compact_cancel_clears_progress_before_returning_idle() { break; } } + let completion_request_id = "subworker-completion-during-compact"; + handle + .send(Method::NotifyTracked { + notification_request_id: completion_request_id.into(), + message: "subworker completed".into(), + source: protocol::AuthenticatedInputSource::UntrustedWire, + }) + .await + .expect("send completion notification during compact"); let shutdown = protocol::WorkerCommandEnvelope::new(4); handle .send(Method::Shutdown { command: shutdown }) @@ -1230,6 +1257,18 @@ async fn manual_compact_cancel_clears_progress_before_returning_idle() { .await .expect("controller shutdown timeout") .expect("shutdown confirmation"); + + let persisted = observed_store + .read_all(source_session, source_segment) + .expect("read source Segment after shutdown"); + assert!(persisted.iter().any(|entry| { + matches!( + entry, + LogEntry::Extension { domain, payload, .. } + if domain == "worker.pending_activations.v1" + && payload.to_string().contains(completion_request_id) + ) + })); } #[tokio::test] From 72f0b82ae51615217e5ea410645116cef2b106e6 Mon Sep 17 00:00:00 2001 From: Hare Date: Fri, 18 Sep 2026 19:12:55 +0900 Subject: [PATCH 12/13] worker: persist child completion before wakeup --- crates/worker/src/spawn/tool.rs | 35 ++++++++++++++++++++++++++++----- 1 file changed, 30 insertions(+), 5 deletions(-) diff --git a/crates/worker/src/spawn/tool.rs b/crates/worker/src/spawn/tool.rs index 9acbdded3..a8e9722bd 100644 --- a/crates/worker/src/spawn/tool.rs +++ b/crates/worker/src/spawn/tool.rs @@ -272,17 +272,18 @@ impl ParentNotificationTarget { }; match self { Self::Controller { sender, fallback } => { + // Persist through the Worker-owned pending authority before the + // child completion path returns. The Controller delivery is only + // a wake-up hint; replay is idempotent by notification_request_id. + fallback(method.clone()); let Some(parent_method_tx) = sender.upgrade() else { - fallback(method); return; }; - let fallback = fallback.clone(); tokio::spawn(async move { - if let Err(error) = parent_method_tx.send(method).await { + if parent_method_tx.send(method).await.is_err() { tracing::warn!( - "failed to notify parent Controller; using durable pending authority" + "failed to wake parent Controller after durable notification acceptance" ); - fallback(error.0); } }); } @@ -1201,6 +1202,30 @@ enabled = false assert!(*captured.lock().unwrap()); } + #[tokio::test] + async fn parent_controller_notification_is_durable_before_wakeup_delivery() { + let (parent_method_tx, mut parent_method_rx) = mpsc::channel(1); + let durable = Arc::new(AtomicBool::new(false)); + let durable_for_fallback = Arc::clone(&durable); + let target = ParentNotificationTarget::with_controller_fallback( + parent_method_tx.downgrade(), + ParentNotificationTarget::Durable(Arc::new(move |_| { + durable_for_fallback.store(true, Ordering::Release); + })), + ); + + target.notify("child-session".into(), "completed".into()); + let wakeup = parent_method_rx.recv().await.unwrap(); + assert!(durable.load(Ordering::Acquire)); + assert!(matches!( + wakeup, + Method::NotifyTracked { + source: protocol::AuthenticatedInputSource::SubWorker { session_id }, + .. + } if session_id == "child-session" + )); + } + #[test] fn durable_parent_notification_target_preserves_child_source() { let captured = Arc::new(std::sync::Mutex::new(None)); From 57fa8f6c761d54826e5f05c2985e88b093095481 Mon Sep 17 00:00:00 2001 From: Hare Date: Fri, 18 Sep 2026 19:35:04 +0900 Subject: [PATCH 13/13] fix: apply rustfmt to T-620 changes --- crates/worker/src/controller.rs | 36 +++++++------ .../worker/src/runtime/worker_allocation.rs | 8 +-- .../runtime/worker_allocation/lifecycle.rs | 24 +++++---- .../src/runtime/worker_allocation/table.rs | 14 +++-- crates/worker/src/spawn/registry.rs | 22 ++++---- crates/worker/src/worker.rs | 52 +++++++++---------- crates/worker/tests/compact_events_test.rs | 8 ++- 7 files changed, 89 insertions(+), 75 deletions(-) diff --git a/crates/worker/src/controller.rs b/crates/worker/src/controller.rs index 842b9a4a6..2ea198113 100644 --- a/crates/worker/src/controller.rs +++ b/crates/worker/src/controller.rs @@ -1830,7 +1830,9 @@ async fn controller_loop( ) .await; if shutdown { - while let Err(error) = worker.finish_pending_compaction_cleanup_for_shutdown().await + while let Err(error) = worker + .finish_pending_compaction_cleanup_for_shutdown() + .await { let _ = working_event_tx.send(Event::Error { code: worker_error_code(&error), @@ -1924,8 +1926,9 @@ async fn controller_loop( } Err(error) => { if matches!(&method, Method::Shutdown { .. }) { - if let Err(error) = - worker.finish_pending_compaction_cleanup_for_shutdown().await + if let Err(error) = worker + .finish_pending_compaction_cleanup_for_shutdown() + .await { let _ = working_event_tx.send(Event::Error { code: worker_error_code(&error), @@ -2402,8 +2405,9 @@ async fn controller_loop( }); } if shutdown_after_compaction { - while let Err(error) = - worker.finish_pending_compaction_cleanup_for_shutdown().await + while let Err(error) = worker + .finish_pending_compaction_cleanup_for_shutdown() + .await { let _ = working_event_tx.send(Event::Error { code: worker_error_code(&error), @@ -3585,16 +3589,18 @@ mod tests { let pending = PendingSubmissionHandle::for_test(dir.path()); let (event_tx, _event_rx) = broadcast::channel(4); - assert!(durably_accept_method_while_busy( - Method::NotifyTracked { - notification_request_id: "completion-1".into(), - message: "subworker completed".into(), - source: protocol::AuthenticatedInputSource::UntrustedWire, - }, - &pending, - &event_tx, - ) - .is_none()); + assert!( + durably_accept_method_while_busy( + Method::NotifyTracked { + notification_request_id: "completion-1".into(), + message: "subworker completed".into(), + source: protocol::AuthenticatedInputSource::UntrustedWire, + }, + &pending, + &event_tx, + ) + .is_none() + ); let entries = pending.persisted_entries_for_test(); assert!(entries.iter().any(|entry| matches!( entry, diff --git a/crates/worker/src/runtime/worker_allocation.rs b/crates/worker/src/runtime/worker_allocation.rs index cb8b474b1..c7445cdaa 100644 --- a/crates/worker/src/runtime/worker_allocation.rs +++ b/crates/worker/src/runtime/worker_allocation.rs @@ -18,16 +18,16 @@ pub use conflict::{ ConflictOwner, find_conflict_owner, find_conflict_owners, is_within_effective_write, }; pub use error::ScopeLockError; +#[cfg(test)] +pub(crate) use lifecycle::install_top_level_at_for_test; pub use lifecycle::{ ScopeAllocationGuard, SegmentActivationCommitError, SegmentActivationGuard, SegmentLockInfo, adopt_allocation, install_top_level, install_top_level_with_deny, lookup_segment, }; -#[cfg(test)] -pub(crate) use lifecycle::install_top_level_at_for_test; -#[cfg(test)] -pub(crate) use table::fail_next_save_for_test; pub use mutate::{ delegate_scope, reclaim_delegated_scope, reclaim_stale, reclaim_stale_with, register_worker, register_worker_with_deny, release_worker, }; +#[cfg(test)] +pub(crate) use table::fail_next_save_for_test; pub use table::{Allocation, LockFile, LockFileGuard, default_allocation_path}; diff --git a/crates/worker/src/runtime/worker_allocation/lifecycle.rs b/crates/worker/src/runtime/worker_allocation/lifecycle.rs index 6dd9c6537..a596c1aa9 100644 --- a/crates/worker/src/runtime/worker_allocation/lifecycle.rs +++ b/crates/worker/src/runtime/worker_allocation/lifecycle.rs @@ -88,8 +88,7 @@ impl ScopeAllocationGuard { } if let Some(existing) = guard.data().allocations.iter().find(|allocation| { - allocation.segment_id == Some(replacement) - && allocation.worker_name != self.worker_name + allocation.segment_id == Some(replacement) && allocation.worker_name != self.worker_name }) { return Err(ScopeLockError::SegmentConflict { segment_id: replacement, @@ -494,14 +493,18 @@ mod tests { let lookup = std::thread::spawn(move || { sent.send(lookup_segment(replacement).unwrap()).unwrap(); }); - assert!(received - .recv_timeout(std::time::Duration::from_millis(50)) - .is_err()); + assert!( + received + .recv_timeout(std::time::Duration::from_millis(50)) + .is_err() + ); drop(authority); - assert!(received - .recv_timeout(std::time::Duration::from_secs(1)) - .unwrap() - .is_none()); + assert!( + received + .recv_timeout(std::time::Duration::from_secs(1)) + .unwrap() + .is_none() + ); lookup.join().unwrap(); } @@ -528,7 +531,8 @@ mod tests { let error = activation.commit().unwrap_err(); let (source, authority) = error.into_parts(); assert!(matches!(source, ScopeLockError::Io(_))); - let persisted: LockFile = serde_json::from_slice(&std::fs::read(&lock_path).unwrap()).unwrap(); + let persisted: LockFile = + serde_json::from_slice(&std::fs::read(&lock_path).unwrap()).unwrap(); assert_eq!( persisted .find("activation") diff --git a/crates/worker/src/runtime/worker_allocation/table.rs b/crates/worker/src/runtime/worker_allocation/table.rs index 0df4c0a80..af7c6159a 100644 --- a/crates/worker/src/runtime/worker_allocation/table.rs +++ b/crates/worker/src/runtime/worker_allocation/table.rs @@ -365,7 +365,10 @@ mod tests { } let guard = LockFileGuard::open(&path).unwrap(); assert_eq!( - guard.data().find("a").and_then(|allocation| allocation.segment_id), + guard + .data() + .find("a") + .and_then(|allocation| allocation.segment_id), Some(old_segment) ); drop(guard); @@ -378,7 +381,10 @@ mod tests { } let guard = LockFileGuard::open(&path).unwrap(); assert_eq!( - guard.data().find("a").and_then(|allocation| allocation.segment_id), + guard + .data() + .find("a") + .and_then(|allocation| allocation.segment_id), Some(replacement) ); } @@ -396,9 +402,7 @@ mod tests { let contender = LockFileGuard::open(&contender_path).unwrap(); sent.send(contender.data().allocations.len()).unwrap(); }); - assert!(received - .recv_timeout(Duration::from_millis(50)) - .is_err()); + assert!(received.recv_timeout(Duration::from_millis(50)).is_err()); drop(guard); assert_eq!(received.recv_timeout(Duration::from_secs(1)).unwrap(), 0); contender.join().unwrap(); diff --git a/crates/worker/src/spawn/registry.rs b/crates/worker/src/spawn/registry.rs index 8592c7d66..7c2606f74 100644 --- a/crates/worker/src/spawn/registry.rs +++ b/crates/worker/src/spawn/registry.rs @@ -8,9 +8,9 @@ //! Parent registry drop closes all session handles and synchronously returns delegated Write deny //! rules to the parent scope. -use std::collections::{BTreeMap, HashSet}; #[cfg(test)] use std::collections::HashMap; +use std::collections::{BTreeMap, HashSet}; use std::io; use std::sync::{ Arc, Mutex, @@ -453,8 +453,8 @@ impl SpawnedWorkerRegistry { registry: Arc::new(Self { internal_records: std::sync::Mutex::new(Vec::new()), service_records: std::sync::Mutex::new(Vec::new()), - #[cfg(test)] - fail_service_stops: std::sync::Mutex::new(HashMap::new()), + #[cfg(test)] + fail_service_stops: std::sync::Mutex::new(HashMap::new()), internal_names: std::sync::Mutex::new(HashSet::new()), internal_shutting_down: AtomicBool::new(false), pending_internal_spawns: AtomicUsize::new(0), @@ -550,17 +550,19 @@ impl SpawnedWorkerRegistry { #[cfg(test)] pub(crate) fn install_service_for_test( &self, - ) -> ( - String, - tokio::sync::broadcast::Sender, - ) { + ) -> (String, tokio::sync::broadcast::Sender) { let (session, status_tx) = crate::internal_worker::test_internal_worker_session( crate::internal_worker::InternalWorkerVisibility::ServicePrivate, ); let session_id = session.session_id_string(); - self.service_records.lock().unwrap().push( - InternalServiceWorkerRecord::new("test-service", "test-service", session), - ); + self.service_records + .lock() + .unwrap() + .push(InternalServiceWorkerRecord::new( + "test-service", + "test-service", + session, + )); (session_id, status_tx) } diff --git a/crates/worker/src/worker.rs b/crates/worker/src/worker.rs index adf4de779..b631ba6c2 100644 --- a/crates/worker/src/worker.rs +++ b/crates/worker/src/worker.rs @@ -4535,9 +4535,7 @@ impl Worker { let allocation_activation = self .scope_allocation .as_ref() - .map(|allocation| { - allocation.begin_segment_activation(loc.segment_id, fork_segment_id) - }) + .map(|allocation| allocation.begin_segment_activation(loc.segment_id, fork_segment_id)) .transpose()?; self.compare_and_swap_worker_metadata_segment(loc, new_location)?; if let Some(allocation_activation) = allocation_activation { @@ -8596,14 +8594,15 @@ mod build_summary_prompt_tests { old_location.segment_id, ))); let metadata_for_cas = Arc::clone(&persisted_metadata); - worker.worker_metadata_segment_cas = Some(Arc::new(move |_worker_name, expected, replacement| { - let mut persisted = metadata_for_cas.lock().unwrap(); - if &*persisted != expected { - return Ok(false); - } - *persisted = replacement; - Ok(true) - })); + worker.worker_metadata_segment_cas = + Some(Arc::new(move |_worker_name, expected, replacement| { + let mut persisted = metadata_for_cas.lock().unwrap(); + if &*persisted != expected { + return Ok(false); + } + *persisted = replacement; + Ok(true) + })); let lock_path = dir.path().join("workers.json"); let allocation = worker_allocation::install_top_level_at_for_test( lock_path.clone(), @@ -8688,14 +8687,15 @@ mod build_summary_prompt_tests { old_location.segment_id, ))); let metadata_for_cas = Arc::clone(&persisted_metadata); - worker.worker_metadata_segment_cas = Some(Arc::new(move |_worker_name, expected, replacement| { - let mut persisted = metadata_for_cas.lock().unwrap(); - if &*persisted != expected { - return Ok(false); - } - *persisted = replacement; - Ok(true) - })); + worker.worker_metadata_segment_cas = + Some(Arc::new(move |_worker_name, expected, replacement| { + let mut persisted = metadata_for_cas.lock().unwrap(); + if &*persisted != expected { + return Ok(false); + } + *persisted = replacement; + Ok(true) + })); let lock_path = dir.path().join("workers.json"); let allocation = worker_allocation::install_top_level_at_for_test( lock_path.clone(), @@ -8707,10 +8707,7 @@ mod build_summary_prompt_tests { ) .unwrap(); let activation = allocation - .begin_segment_activation( - old_location.segment_id, - replacement_location.segment_id, - ) + .begin_segment_activation(old_location.segment_id, replacement_location.segment_id) .unwrap(); worker.scope_allocation = Some(allocation); worker @@ -8764,8 +8761,7 @@ mod build_summary_prompt_tests { ) .await .unwrap(); - let registry = - crate::spawn::registry::SpawnedWorkerRegistry::new_for_internal_services(); + let registry = crate::spawn::registry::SpawnedWorkerRegistry::new_for_internal_services(); let (cleanup_session_id, _events) = registry.install_service_for_test(); registry.fail_service_stops_for_test(&cleanup_session_id, 4); worker.internal_worker_registry = Some(Arc::clone(®istry)); @@ -10567,7 +10563,11 @@ mod build_summary_prompt_tests { let submit = std::thread::spawn(move || { submit_started.send(()).unwrap(); submit_handle - .accept("barrier-submit".into(), vec![Segment::text("queued")], false) + .accept( + "barrier-submit".into(), + vec![Segment::text("queued")], + false, + ) .unwrap(); submit_finished.send(()).unwrap(); }); diff --git a/crates/worker/tests/compact_events_test.rs b/crates/worker/tests/compact_events_test.rs index 8e2a91af6..c6c37367c 100644 --- a/crates/worker/tests/compact_events_test.rs +++ b/crates/worker/tests/compact_events_test.rs @@ -1117,11 +1117,9 @@ async fn pre_run_compact_failure_clears_runtime_progress() { #[tokio::test] async fn manual_compact_cancel_clears_progress_before_returning_idle() { - let (worker, observed_store) = make_worker_with_manifest_and_store( - POST_RUN_MANIFEST_TOML, - BlockingCompactClient::new(), - ) - .await; + let (worker, observed_store) = + make_worker_with_manifest_and_store(POST_RUN_MANIFEST_TOML, BlockingCompactClient::new()) + .await; let source_session = worker.session_id(); let source_segment = worker.segment_id(); let runtime_tmp = tempfile::tempdir().unwrap();