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

fix: apply rustfmt to T-620 changes

Browse Source
This commit is contained in:
Hare
2026-09-18 19:35:04 +09:00
parent 72f0b82ae5
commit 57fa8f6c76
7 changed files with 89 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/worker/src/controller.rs
+21 -15
View File
@@ -1830,7 +1830,9 @@ async fn controller_loop<C, St>(
) )
.await; .await;
if shutdown { 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 { let _ = working_event_tx.send(Event::Error {
code: worker_error_code(&error), code: worker_error_code(&error),
@@ -1924,8 +1926,9 @@ async fn controller_loop<C, St>(
} }
Err(error) => { Err(error) => {
if matches!(&method, Method::Shutdown { .. }) { if matches!(&method, Method::Shutdown { .. }) {
if let Err(error) = if let Err(error) = worker
worker.finish_pending_compaction_cleanup_for_shutdown().await .finish_pending_compaction_cleanup_for_shutdown()
.await
{ {
let _ = working_event_tx.send(Event::Error { let _ = working_event_tx.send(Event::Error {
code: worker_error_code(&error), code: worker_error_code(&error),
@@ -2402,8 +2405,9 @@ async fn controller_loop<C, St>(
}); });
} }
if shutdown_after_compaction { if shutdown_after_compaction {
while let Err(error) = while let Err(error) = worker
worker.finish_pending_compaction_cleanup_for_shutdown().await .finish_pending_compaction_cleanup_for_shutdown()
.await
{ {
let _ = working_event_tx.send(Event::Error { let _ = working_event_tx.send(Event::Error {
code: worker_error_code(&error), code: worker_error_code(&error),
@@ -3585,16 +3589,18 @@ mod tests {
let pending = PendingSubmissionHandle::for_test(dir.path()); let pending = PendingSubmissionHandle::for_test(dir.path());
let (event_tx, _event_rx) = broadcast::channel(4); let (event_tx, _event_rx) = broadcast::channel(4);
assert!(durably_accept_method_while_busy( assert!(
Method::NotifyTracked { durably_accept_method_while_busy(
notification_request_id: "completion-1".into(), Method::NotifyTracked {
message: "subworker completed".into(), notification_request_id: "completion-1".into(),
source: protocol::AuthenticatedInputSource::UntrustedWire, message: "subworker completed".into(),
}, source: protocol::AuthenticatedInputSource::UntrustedWire,
&pending, },
&event_tx, &pending,
) &event_tx,
.is_none()); )
.is_none()
);
let entries = pending.persisted_entries_for_test(); let entries = pending.persisted_entries_for_test();
assert!(entries.iter().any(|entry| matches!( assert!(entries.iter().any(|entry| matches!(
entry, entry,
crates/worker/src/runtime/worker_allocation.rs
+4 -4
View File
@@ -18,16 +18,16 @@ pub use conflict::{
ConflictOwner, find_conflict_owner, find_conflict_owners, is_within_effective_write, ConflictOwner, find_conflict_owner, find_conflict_owners, is_within_effective_write,
}; };
pub use error::ScopeLockError; pub use error::ScopeLockError;
#[cfg(test)]
pub(crate) use lifecycle::install_top_level_at_for_test;
pub use lifecycle::{ pub use lifecycle::{
ScopeAllocationGuard, SegmentActivationCommitError, SegmentActivationGuard, SegmentLockInfo, ScopeAllocationGuard, SegmentActivationCommitError, SegmentActivationGuard, SegmentLockInfo,
adopt_allocation, install_top_level, install_top_level_with_deny, lookup_segment, 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::{ pub use mutate::{
delegate_scope, reclaim_delegated_scope, reclaim_stale, reclaim_stale_with, register_worker, delegate_scope, reclaim_delegated_scope, reclaim_stale, reclaim_stale_with, register_worker,
register_worker_with_deny, release_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}; pub use table::{Allocation, LockFile, LockFileGuard, default_allocation_path};
crates/worker/src/runtime/worker_allocation/lifecycle.rs
+14 -10
View File
@@ -88,8 +88,7 @@ impl ScopeAllocationGuard {
} }
if let Some(existing) = guard.data().allocations.iter().find(|allocation| { if let Some(existing) = guard.data().allocations.iter().find(|allocation| {
allocation.segment_id == Some(replacement) allocation.segment_id == Some(replacement) && allocation.worker_name != self.worker_name
&& allocation.worker_name != self.worker_name
}) { }) {
return Err(ScopeLockError::SegmentConflict { return Err(ScopeLockError::SegmentConflict {
segment_id: replacement, segment_id: replacement,
@@ -494,14 +493,18 @@ mod tests {
let lookup = std::thread::spawn(move || { let lookup = std::thread::spawn(move || {
sent.send(lookup_segment(replacement).unwrap()).unwrap(); sent.send(lookup_segment(replacement).unwrap()).unwrap();
}); });
assert!(received assert!(
.recv_timeout(std::time::Duration::from_millis(50)) received
.is_err()); .recv_timeout(std::time::Duration::from_millis(50))
.is_err()
);
drop(authority); drop(authority);
assert!(received assert!(
.recv_timeout(std::time::Duration::from_secs(1)) received
.unwrap() .recv_timeout(std::time::Duration::from_secs(1))
.is_none()); .unwrap()
.is_none()
);
lookup.join().unwrap(); lookup.join().unwrap();
} }
@@ -528,7 +531,8 @@ mod tests {
let error = activation.commit().unwrap_err(); let error = activation.commit().unwrap_err();
let (source, authority) = error.into_parts(); let (source, authority) = error.into_parts();
assert!(matches!(source, ScopeLockError::Io(_))); 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!( assert_eq!(
persisted persisted
.find("activation") .find("activation")
crates/worker/src/runtime/worker_allocation/table.rs
+9 -5
View File
@@ -365,7 +365,10 @@ mod tests {
} }
let guard = LockFileGuard::open(&path).unwrap(); let guard = LockFileGuard::open(&path).unwrap();
assert_eq!( 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) Some(old_segment)
); );
drop(guard); drop(guard);
@@ -378,7 +381,10 @@ mod tests {
} }
let guard = LockFileGuard::open(&path).unwrap(); let guard = LockFileGuard::open(&path).unwrap();
assert_eq!( assert_eq!(
guard.data().find("a").and_then(|allocation| allocation.segment_id), guard
.data()
.find("a")
.and_then(|allocation| allocation.segment_id),
Some(replacement) Some(replacement)
); );
} }
@@ -396,9 +402,7 @@ mod tests {
let contender = LockFileGuard::open(&contender_path).unwrap(); let contender = LockFileGuard::open(&contender_path).unwrap();
sent.send(contender.data().allocations.len()).unwrap(); sent.send(contender.data().allocations.len()).unwrap();
}); });
assert!(received assert!(received.recv_timeout(Duration::from_millis(50)).is_err());
.recv_timeout(Duration::from_millis(50))
.is_err());
drop(guard); drop(guard);
assert_eq!(received.recv_timeout(Duration::from_secs(1)).unwrap(), 0); assert_eq!(received.recv_timeout(Duration::from_secs(1)).unwrap(), 0);
contender.join().unwrap(); contender.join().unwrap();
crates/worker/src/spawn/registry.rs
+12 -10
View File
@@ -8,9 +8,9 @@
//! Parent registry drop closes all session handles and synchronously returns delegated Write deny //! Parent registry drop closes all session handles and synchronously returns delegated Write deny
//! rules to the parent scope. //! rules to the parent scope.
use std::collections::{BTreeMap, HashSet};
#[cfg(test)] #[cfg(test)]
use std::collections::HashMap; use std::collections::HashMap;
use std::collections::{BTreeMap, HashSet};
use std::io; use std::io;
use std::sync::{ use std::sync::{
Arc, Mutex, Arc, Mutex,
@@ -453,8 +453,8 @@ impl SpawnedWorkerRegistry {
registry: Arc::new(Self { registry: Arc::new(Self {
internal_records: std::sync::Mutex::new(Vec::new()), internal_records: std::sync::Mutex::new(Vec::new()),
service_records: std::sync::Mutex::new(Vec::new()), service_records: std::sync::Mutex::new(Vec::new()),
#[cfg(test)] #[cfg(test)]
fail_service_stops: std::sync::Mutex::new(HashMap::new()), fail_service_stops: std::sync::Mutex::new(HashMap::new()),
internal_names: std::sync::Mutex::new(HashSet::new()), internal_names: std::sync::Mutex::new(HashSet::new()),
internal_shutting_down: AtomicBool::new(false), internal_shutting_down: AtomicBool::new(false),
pending_internal_spawns: AtomicUsize::new(0), pending_internal_spawns: AtomicUsize::new(0),
@@ -550,17 +550,19 @@ impl SpawnedWorkerRegistry {
#[cfg(test)] #[cfg(test)]
pub(crate) fn install_service_for_test( pub(crate) fn install_service_for_test(
&self, &self,
) -> ( ) -> (String, tokio::sync::broadcast::Sender<protocol::Event>) {
String,
tokio::sync::broadcast::Sender<protocol::Event>,
) {
let (session, status_tx) = crate::internal_worker::test_internal_worker_session( let (session, status_tx) = crate::internal_worker::test_internal_worker_session(
crate::internal_worker::InternalWorkerVisibility::ServicePrivate, crate::internal_worker::InternalWorkerVisibility::ServicePrivate,
); );
let session_id = session.session_id_string(); let session_id = session.session_id_string();
self.service_records.lock().unwrap().push( self.service_records
InternalServiceWorkerRecord::new("test-service", "test-service", session), .lock()
); .unwrap()
.push(InternalServiceWorkerRecord::new(
"test-service",
"test-service",
session,
));
(session_id, status_tx) (session_id, status_tx)
} }
crates/worker/src/worker.rs
+26 -26
View File
@@ -4535,9 +4535,7 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
let allocation_activation = self let allocation_activation = self
.scope_allocation .scope_allocation
.as_ref() .as_ref()
.map(|allocation| { .map(|allocation| allocation.begin_segment_activation(loc.segment_id, fork_segment_id))
allocation.begin_segment_activation(loc.segment_id, fork_segment_id)
})
.transpose()?; .transpose()?;
self.compare_and_swap_worker_metadata_segment(loc, new_location)?; self.compare_and_swap_worker_metadata_segment(loc, new_location)?;
if let Some(allocation_activation) = allocation_activation { if let Some(allocation_activation) = allocation_activation {
@@ -8596,14 +8594,15 @@ mod build_summary_prompt_tests {
old_location.segment_id, old_location.segment_id,
))); )));
let metadata_for_cas = Arc::clone(&persisted_metadata); let metadata_for_cas = Arc::clone(&persisted_metadata);
worker.worker_metadata_segment_cas = Some(Arc::new(move |_worker_name, expected, replacement| { worker.worker_metadata_segment_cas =
let mut persisted = metadata_for_cas.lock().unwrap(); Some(Arc::new(move |_worker_name, expected, replacement| {
if &*persisted != expected { let mut persisted = metadata_for_cas.lock().unwrap();
return Ok(false); if &*persisted != expected {
} return Ok(false);
*persisted = replacement; }
Ok(true) *persisted = replacement;
})); Ok(true)
}));
let lock_path = dir.path().join("workers.json"); let lock_path = dir.path().join("workers.json");
let allocation = worker_allocation::install_top_level_at_for_test( let allocation = worker_allocation::install_top_level_at_for_test(
lock_path.clone(), lock_path.clone(),
@@ -8688,14 +8687,15 @@ mod build_summary_prompt_tests {
old_location.segment_id, old_location.segment_id,
))); )));
let metadata_for_cas = Arc::clone(&persisted_metadata); let metadata_for_cas = Arc::clone(&persisted_metadata);
worker.worker_metadata_segment_cas = Some(Arc::new(move |_worker_name, expected, replacement| { worker.worker_metadata_segment_cas =
let mut persisted = metadata_for_cas.lock().unwrap(); Some(Arc::new(move |_worker_name, expected, replacement| {
if &*persisted != expected { let mut persisted = metadata_for_cas.lock().unwrap();
return Ok(false); if &*persisted != expected {
} return Ok(false);
*persisted = replacement; }
Ok(true) *persisted = replacement;
})); Ok(true)
}));
let lock_path = dir.path().join("workers.json"); let lock_path = dir.path().join("workers.json");
let allocation = worker_allocation::install_top_level_at_for_test( let allocation = worker_allocation::install_top_level_at_for_test(
lock_path.clone(), lock_path.clone(),
@@ -8707,10 +8707,7 @@ mod build_summary_prompt_tests {
) )
.unwrap(); .unwrap();
let activation = allocation let activation = allocation
.begin_segment_activation( .begin_segment_activation(old_location.segment_id, replacement_location.segment_id)
old_location.segment_id,
replacement_location.segment_id,
)
.unwrap(); .unwrap();
worker.scope_allocation = Some(allocation); worker.scope_allocation = Some(allocation);
worker worker
@@ -8764,8 +8761,7 @@ mod build_summary_prompt_tests {
) )
.await .await
.unwrap(); .unwrap();
let registry = let registry = crate::spawn::registry::SpawnedWorkerRegistry::new_for_internal_services();
crate::spawn::registry::SpawnedWorkerRegistry::new_for_internal_services();
let (cleanup_session_id, _events) = registry.install_service_for_test(); let (cleanup_session_id, _events) = registry.install_service_for_test();
registry.fail_service_stops_for_test(&cleanup_session_id, 4); registry.fail_service_stops_for_test(&cleanup_session_id, 4);
worker.internal_worker_registry = Some(Arc::clone(&registry)); worker.internal_worker_registry = Some(Arc::clone(&registry));
@@ -10567,7 +10563,11 @@ mod build_summary_prompt_tests {
let submit = std::thread::spawn(move || { let submit = std::thread::spawn(move || {
submit_started.send(()).unwrap(); submit_started.send(()).unwrap();
submit_handle submit_handle
.accept("barrier-submit".into(), vec![Segment::text("queued")], false) .accept(
"barrier-submit".into(),
vec![Segment::text("queued")],
false,
)
.unwrap(); .unwrap();
submit_finished.send(()).unwrap(); submit_finished.send(()).unwrap();
}); });
crates/worker/tests/compact_events_test.rs
+3 -5
View File
@@ -1117,11 +1117,9 @@ async fn pre_run_compact_failure_clears_runtime_progress() {
#[tokio::test] #[tokio::test]
async fn manual_compact_cancel_clears_progress_before_returning_idle() { async fn manual_compact_cancel_clears_progress_before_returning_idle() {
let (worker, observed_store) = make_worker_with_manifest_and_store( let (worker, observed_store) =
POST_RUN_MANIFEST_TOML, make_worker_with_manifest_and_store(POST_RUN_MANIFEST_TOML, BlockingCompactClient::new())
BlockingCompactClient::new(), .await;
)
.await;
let source_session = worker.session_id(); let source_session = worker.session_id();
let source_segment = worker.segment_id(); let source_segment = worker.segment_id();
let runtime_tmp = tempfile::tempdir().unwrap(); let runtime_tmp = tempfile::tempdir().unwrap();