From d1d391e51169a06280ae4e422e1a20c6fe49aacb Mon Sep 17 00:00:00 2001 From: Hare Date: Fri, 18 Sep 2026 18:15:33 +0900 Subject: [PATCH] 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;