cargo fmt

Author: yuandrew

client/src/worker_registry/mod.rs

@@ -2,10 +2,13 @@
 //! This is needed to implement Eager Workflow Start, a latency optimization in which the client,
 //!  after reserving a slot, directly forwards a WFT to a local worker.
 
+use anyhow::bail;
 use parking_lot::RwLock;
-use std::collections::{HashMap, hash_map::Entry::{ Occupied, Vacant}};
+use std::collections::{
+    HashMap,
+    hash_map::Entry::{Occupied, Vacant},
+};
 use std::sync::Arc;
-use anyhow::bail;
 use temporal_sdk_core_protos::temporal::api::worker::v1::WorkerHeartbeat;
 use temporal_sdk_core_protos::temporal::api::workflowservice::v1::PollWorkflowTaskQueueResponse;
 use uuid::Uuid;
@@ -70,15 +73,21 @@ impl ClientWorkerSetImpl {
         None
     }
 
-    fn register(&mut self, worker: Arc<dyn ClientWorker + Send + Sync>) -> Result<(), anyhow::Error> {
+    fn register(
+        &mut self,
+        worker: Arc<dyn ClientWorker + Send + Sync>,
+    ) -> Result<(), anyhow::Error> {
         let slot_key = SlotKey::new(
             worker.namespace().to_string(),
             worker.task_queue().to_string(),
         );
         if let Vacant(p) = self.slot_providers.entry(slot_key.clone()) {
             p.insert(worker.worker_instance_key());
         } else {
-            bail!("Registration of multiple workers on the same namespace and task queue for the same client not allowed: {slot_key:?}, worker_instance_key: {:?}.", worker.worker_instance_key());
+            bail!(
+                "Registration of multiple workers on the same namespace and task queue for the same client not allowed: {slot_key:?}, worker_instance_key: {:?}.",
+                worker.worker_instance_key()
+            );
         }
 
         if worker.heartbeat_enabled()
@@ -107,12 +116,15 @@ impl ClientWorkerSetImpl {
         &mut self,
         worker_instance_key: Uuid,
     ) -> Result<Arc<dyn ClientWorker + Send + Sync>, anyhow::Error> {
-        let worker = self.all_workers.remove(&worker_instance_key).ok_or_else(|| {
-            anyhow::anyhow!(
-                "Worker with worker_instance_key {} not found",
-                worker_instance_key
-            )
-        })?;
+        let worker = self
+            .all_workers
+            .remove(&worker_instance_key)
+            .ok_or_else(|| {
+                anyhow::anyhow!(
+                    "Worker with worker_instance_key {} not found",
+                    worker_instance_key
+                )
+            })?;
 
         let slot_key = SlotKey::new(
             worker.namespace().to_string(),
@@ -211,7 +223,10 @@ impl ClientWorkerSet {
     }
 
     /// Register a local worker that can provide WFT processing slots and potentially worker heartbeating.
-    pub fn register_worker(&self, worker: Arc<dyn ClientWorker + Send + Sync>) -> Result<(), anyhow::Error> {
+    pub fn register_worker(
+        &self,
+        worker: Arc<dyn ClientWorker + Send + Sync>,
+    ) -> Result<(), anyhow::Error> {
         self.worker_manager.write().register(worker)
     }
 
@@ -343,7 +358,9 @@ mod tests {
                 worker_keys.push(worker_instance_key);
             } else {
                 // Should get error for duplicate namespace+task_queue combinations
-                assert!(result.unwrap_err().to_string().contains("Registration of multiple workers on the same namespace and task queue"));
+                assert!(result.unwrap_err().to_string().contains(
+                    "Registration of multiple workers on the same namespace and task queue"
+                ));
             }
         }
 
@@ -479,15 +496,20 @@ mod tests {
         // second worker register should fail due to duplicate namespace+task_queue
         let result = manager.register_worker(Arc::new(worker2));
         assert!(result.is_err());
-        assert!(result.unwrap_err().to_string().contains("Registration of multiple workers on the same namespace and task queue"));
+        assert!(
+            result
+                .unwrap_err()
+                .to_string()
+                .contains("Registration of multiple workers on the same namespace and task queue")
+        );
 
         assert_eq!((1, 1), manager.num_providers());
         assert_eq!(manager.num_heartbeat_workers(), 1);
 
         let impl_ref = manager.worker_manager.read();
         assert_eq!(impl_ref.shared_worker.len(), 1);
         assert!(impl_ref.shared_worker.contains_key("test_namespace"));
-}
+    }
 
     #[test]
     fn multiple_workers_same_namespace_share_heartbeat_manager() {

core/src/core_tests/workflow_tasks.rs

@@ -2997,7 +2997,8 @@ async fn both_normal_and_sticky_pollers_poll_concurrently() {
         None,
         None,
         false,
-    ).unwrap();
+    )
+    .unwrap();
 
     for _ in 1..50 {
         let activation = worker.poll_workflow_activation().await.unwrap();

core/src/test_help/integ_helpers.rs

@@ -206,7 +206,8 @@ pub fn mock_worker(mocks: MocksHolder) -> Worker {
         None,
         None,
         false,
-    ).unwrap()
+    )
+    .unwrap()
 }
 
 pub struct FakeWfResponses {

core/src/worker/heartbeat.rs

@@ -227,7 +227,8 @@ mod tests {
             None,
             Some(Duration::from_millis(100)),
             false,
-        ).unwrap();
+        )
+        .unwrap();
 
         tokio::time::sleep(Duration::from_millis(250)).await;
         worker.drain_activity_poller_and_shutdown().await;

core/src/worker/mod.rs

@@ -48,6 +48,7 @@ use crate::{
     worker::workflow::wft_poller,
 };
 use activities::WorkerActivityTasks;
+use anyhow::bail;
 use futures_util::{StreamExt, stream};
 use gethostname::gethostname;
 use parking_lot::{Mutex, RwLock};
@@ -61,7 +62,6 @@ use std::{
     },
     time::Duration,
 };
-use anyhow::bail;
 use temporal_client::{ClientWorker, HeartbeatCallback, Slot as SlotTrait};
 use temporal_client::{
     ConfiguredClient, SharedNamespaceWorkerTrait, TemporalServiceClientWithMetrics,
@@ -311,9 +311,11 @@ impl WorkerTrait for Worker {
         }
         self.shutdown_token.cancel();
         // First, unregister worker from the client
-        if let Err(e) = self.client
+        if let Err(e) = self
+            .client
             .workers()
-            .unregister_worker(self.worker_instance_key) {
+            .unregister_worker(self.worker_instance_key)
+        {
             error!(
                 task_queue=%self.config.task_queue,
                 namespace=%self.config.namespace,
@@ -384,7 +386,10 @@ impl Worker {
     /// For worker heartbeat, this will remove an existing shared worker if it is the last worker of
     /// the old client and create a new nexus worker if it's the first client of the namespace on
     /// the new client.
-    pub fn replace_client(&self, new_client: ConfiguredClient<TemporalServiceClientWithMetrics>) -> Result<(), anyhow::Error> {
+    pub fn replace_client(
+        &self,
+        new_client: ConfiguredClient<TemporalServiceClientWithMetrics>,
+    ) -> Result<(), anyhow::Error> {
         // Unregister worker from current client, register in new client at the end
         let client_worker = self
             .client

tests/integ_tests/polling_tests.rs

@@ -198,7 +198,9 @@ async fn switching_worker_client_changes_poll() {
 
     // Swap client, poll for next task, confirm it's second wf, and respond w/ empty
     info!("Replacing client and polling again");
-    worker.replace_client(client2.get_client().inner().clone());
+    worker
+        .replace_client(client2.get_client().inner().clone())
+        .unwrap();
     let act2 = worker.poll_workflow_activation().await.unwrap();
     assert_eq!(wf2.run_id, act2.run_id);
     worker.complete_execution(&act2.run_id).await;