From 57fa8f6c761d54826e5f05c2985e88b093095481 Mon Sep 17 00:00:00 2001 From: Hare Date: Fri, 18 Sep 2026 19:35:04 +0900 Subject: [PATCH] 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();