formating

Author: yuandrew

core-c-bridge/src/runtime.rs

@@ -16,7 +16,8 @@ use std::{
     time::{Duration, UNIX_EPOCH},
 };
 use temporal_sdk_core::{
-    CoreRuntime, RuntimeOptions as CoreRuntimeOptions, RuntimeOptionsBuilder as CoreRuntimeOptionsBuilder, TokioRuntimeBuilder,
+    CoreRuntime, RuntimeOptions as CoreRuntimeOptions,
+    RuntimeOptionsBuilder as CoreRuntimeOptionsBuilder, TokioRuntimeBuilder,
     telemetry::{build_otlp_metric_exporter, start_prometheus_metric_exporter},
 };
 use temporal_sdk_core_api::telemetry::{

core/src/lib.rs

@@ -30,7 +30,6 @@ mod core_tests;
 #[macro_use]
 pub mod test_help;
 
-use std::collections::HashMap;
 pub(crate) use temporal_sdk_core_api::errors;
 
 pub use pollers::{
@@ -119,7 +118,7 @@ where
         client_ident.clone(),
         worker_config.versioning_strategy.clone(),
     ));
-    
+
     let worker = Worker::new(
         worker_config.clone(),
         sticky_q,
@@ -370,10 +369,6 @@ impl CoreRuntime {
     pub fn telemetry_mut(&mut self) -> &mut TelemetryInstance {
         &mut self.telemetry
     }
-
-    fn task_queue_key(&self) -> Uuid {
-        self.process_key
-    }
 }
 
 impl Drop for CoreRuntime {

core/src/telemetry/metrics.rs

@@ -1,4 +1,4 @@
-use crate::{abstractions::dbg_panic, telemetry::TelemetryInstance};
+use crate::abstractions::dbg_panic;
 
 use std::{
     fmt::{Debug, Display},

core/src/worker/client.rs

@@ -1,14 +1,9 @@
 //! Worker-specific client needs
 
 pub(crate) mod mocks;
-use crate::{
-    abstractions::dbg_panic, protosext::legacy_query_failure, worker::heartbeat::HeartbeatFn,
-};
+use crate::protosext::legacy_query_failure;
 use parking_lot::RwLock;
-use std::{
-    sync::{Arc, OnceLock},
-    time::Duration,
-};
+use std::{sync::Arc, time::Duration};
 use temporal_client::{
     Client, ClientWorkerSet, IsWorkerTaskLongPoll, Namespace, NamespacedClient, NoRetryOnMatching,
     RetryClient, WorkflowService,

core/src/worker/client/mocks.rs

@@ -35,8 +35,7 @@ pub fn mock_worker_client() -> MockWorkerClient {
         .returning(|| ("test-core".to_string(), "0.0.0".to_string()));
     r.expect_get_identity()
         .returning(|| "test-identity".to_string());
-    r.expect_get_process_key()
-        .returning(|| Uuid::new_v4());
+    r.expect_get_process_key().returning(Uuid::new_v4);
     r
 }
 

core/src/worker/heartbeat.rs

@@ -1,15 +1,17 @@
-use crate::{WorkerClient, abstractions::dbg_panic};
-use gethostname::gethostname;
+use crate::WorkerClient;
+use crate::worker::{TaskPollers, WorkerTelemetry};
 use parking_lot::Mutex;
 use prost_types::Duration as PbDuration;
-use std::{fmt, sync::{Arc, OnceLock}, time::{Duration, SystemTime}};
 use std::collections::HashMap;
-use temporal_sdk_core_api::worker::{WorkerConfig, WorkerConfigBuilder, WorkerVersioningStrategy};
-use temporal_sdk_core_protos::temporal::api::worker::v1::{WorkerHeartbeat, WorkerHostInfo};
-use tokio::{sync::Notify, task::JoinHandle, time::MissedTickBehavior};
-use uuid::Uuid;
+use std::{
+    fmt,
+    sync::Arc,
+    time::{Duration, SystemTime},
+};
 use temporal_client::SharedNamespaceWorkerTrait;
-use crate::worker::{TaskPollers, WorkerTelemetry};
+use temporal_sdk_core_api::worker::{WorkerConfigBuilder, WorkerVersioningStrategy};
+use temporal_sdk_core_protos::temporal::api::worker::v1::WorkerHeartbeat;
+use tokio::sync::Notify;
 
 /// Callback used to collect heartbeat data from each worker at the time of heartbeat
 pub(crate) type HeartbeatFn = Box<dyn Fn() -> WorkerHeartbeat + Send + Sync>;
@@ -29,12 +31,16 @@ impl SharedNamespaceWorker {
         heartbeat_interval: Duration,
         telemetry: Option<WorkerTelemetry>,
     ) -> Self {
-        println!("SharedNamespaceWorker::new() {:?}\n\tsdk_name_and_version key{:?}", client.get_identity(), client.sdk_name_and_version());
+        println!(
+            "SharedNamespaceWorker::new() {:?}\n\tsdk_name_and_version key{:?}",
+            client.get_identity(),
+            client.sdk_name_and_version()
+        );
         let config = WorkerConfigBuilder::default()
             .namespace(namespace.clone())
             .task_queue(format!(
                 "temporal-sys/worker-commands/{namespace}/{}",
-                client.get_process_key().to_string()
+                client.get_process_key()
             ))
             .no_remote_activities(true)
             .max_outstanding_nexus_tasks(5_usize)
@@ -68,7 +74,9 @@ impl SharedNamespaceWorker {
             loop {
                 // TODO: Race condition here, can technically shut down before anything is ever initialized
                 if heartbeat_map_clone.lock().is_empty() {
-                    println!("// TODO: Race condition here, can technically shut down before anything is ever initialized");
+                    println!(
+                        "// TODO: Race condition here, can technically shut down before anything is ever initialized"
+                    );
                     worker.shutdown().await;
                     return;
                 }
@@ -174,11 +182,15 @@ mod tests {
         worker,
         worker::client::mocks::mock_worker_client,
     };
-    use std::{sync::{Arc, atomic::{AtomicUsize, Ordering}}, time::Duration};
-    use mockall::mock;
+    use std::{
+        sync::{
+            Arc,
+            atomic::{AtomicUsize, Ordering},
+        },
+        time::Duration,
+    };
     use temporal_sdk_core_api::worker::PollerBehavior;
     use temporal_sdk_core_protos::temporal::api::workflowservice::v1::RecordWorkerHeartbeatResponse;
-    use uuid::Uuid;
 
     #[tokio::test]
     async fn worker_heartbeat_basic() {
@@ -217,7 +229,13 @@ mod tests {
             .into();
 
         let client = Arc::new(mock);
-        let worker = worker::Worker::new(config, None, client.clone(), None, Some(Duration::from_millis(100)));
+        let worker = worker::Worker::new(
+            config,
+            None,
+            client.clone(),
+            None,
+            Some(Duration::from_millis(100)),
+        );
 
         tokio::time::sleep(Duration::from_millis(250)).await;
         worker.drain_activity_poller_and_shutdown().await;

core/src/worker/mod.rs

@@ -53,7 +53,7 @@ use std::{
     convert::TryInto,
     future,
     sync::{
-        Arc, OnceLock,
+        Arc,
         atomic::{AtomicBool, Ordering},
     },
     time::Duration,
@@ -67,7 +67,6 @@ use temporal_sdk_core_api::{
     errors::{CompleteNexusError, WorkerValidationError},
     worker::PollerBehavior,
 };
-use temporal_sdk_core_protos::temporal::api::cloud::account::v1::Metrics;
 use temporal_sdk_core_protos::temporal::api::worker::v1::WorkerHostInfo;
 use temporal_sdk_core_protos::{
     TaskToken,
@@ -84,7 +83,7 @@ use temporal_sdk_core_protos::{
         taskqueue::v1::{StickyExecutionAttributes, TaskQueue},
     },
 };
-use tokio::sync::{OnceCell, mpsc::unbounded_channel, watch};
+use tokio::sync::{mpsc::unbounded_channel, watch};
 use tokio_stream::wrappers::UnboundedReceiverStream;
 use tokio_util::sync::CancellationToken;
 use tracing::Subscriber;
@@ -167,9 +166,6 @@ impl WorkerTelemetry {
 struct WorkerHeartbeat {
     /// Instance key used to identify this worker in worker heartbeating
     worker_instance_key: Uuid,
-    /// Used to remove this worker from the parent map used to track this worker for
-    /// worker heartbeat
-    shutdown_callback: OnceCell<Arc<dyn Fn() + Send + Sync>>,
     /// Heartbeat interval, defaults to 60s
     heartbeat_interval: Duration,
     /// Telemetry instance, needed to initialize [SharedNamespaceWorker] when replacing client
@@ -388,15 +384,11 @@ impl Worker {
         telem_instance: Option<&TelemetryInstance>,
         worker_heartbeat_interval: Option<Duration>,
     ) -> Self {
-        let worker_telemetry = if let Some(telem) = telem_instance {
-            Some(WorkerTelemetry {
-                metric_meter: telem.get_metric_meter(),
-                temporal_metric_meter: telem.get_temporal_metric_meter(),
-                trace_subscriber: telem.trace_subscriber(),
-            })
-        } else {
-            None
-        };
+        let worker_telemetry = telem_instance.map(|telem| WorkerTelemetry {
+            metric_meter: telem.get_metric_meter(),
+            temporal_metric_meter: telem.get_temporal_metric_meter(),
+            trace_subscriber: telem.trace_subscriber(),
+        });
 
         Worker::new_with_pollers_inner(
             config,
@@ -661,8 +653,7 @@ impl Worker {
 
             Some(WorkerHeartbeat {
                 worker_instance_key,
-                shutdown_callback: OnceCell::new(),
-                heartbeat_interval: heartbeat_interval,
+                heartbeat_interval,
                 telemetry: worker_telemetry.clone(),
             })
         } else {

core/src/worker/workflow/mod.rs

@@ -23,7 +23,7 @@ use crate::{
     internal_flags::InternalFlags,
     pollers::TrackedPermittedTqResp,
     protosext::{ValidPollWFTQResponse, protocol_messages::IncomingProtocolMessage},
-    telemetry::{TelemetryInstance, VecDisplayer, set_trace_subscriber_for_current_thread},
+    telemetry::{VecDisplayer, set_trace_subscriber_for_current_thread},
     worker::{
         LocalActRequest, LocalActivityExecutionResult, LocalActivityResolution,
         PostActivateHookData,

tests/common/mod.rs

@@ -38,7 +38,12 @@ use temporal_sdk::{
         WorkerInterceptor,
     },
 };
-use temporal_sdk_core::{ClientOptions, ClientOptionsBuilder, CoreRuntime, WorkerConfigBuilder, init_replay_worker, init_worker, replay::{HistoryForReplay, ReplayWorkerInput}, telemetry::{build_otlp_metric_exporter, start_prometheus_metric_exporter}, RuntimeOptionsBuilder, RuntimeOptions};
+use temporal_sdk_core::{
+    ClientOptions, ClientOptionsBuilder, CoreRuntime, RuntimeOptionsBuilder, WorkerConfigBuilder,
+    init_replay_worker, init_worker,
+    replay::{HistoryForReplay, ReplayWorkerInput},
+    telemetry::{build_otlp_metric_exporter, start_prometheus_metric_exporter},
+};
 use temporal_sdk_core_api::{
     Worker as CoreWorker,
     telemetry::{
@@ -809,10 +814,6 @@ pub(crate) fn get_integ_telem_options() -> TelemetryOptions {
     .unwrap()
 }
 
-pub fn get_integ_runtime_options(heartbeat_interval: Option<Duration>) -> RuntimeOptions {
-    RuntimeOptions::new(get_integ_telem_options(), heartbeat_interval)
-}
-
 #[async_trait::async_trait(?Send)]
 pub(crate) trait WorkflowHandleExt {
     async fn fetch_history_and_replay(

tests/integ_tests/client_tests.rs

@@ -446,7 +446,10 @@ async fn http_proxy() {
     opts.target_url = format!("http://127.0.0.1:{}", server.addr.port())
         .parse()
         .unwrap();
-    let client = opts.connect("my-namespace", None, Uuid::new_v4()).await.unwrap();
+    let client = opts
+        .connect("my-namespace", None, Uuid::new_v4())
+        .await
+        .unwrap();
     let _ = client.list_namespaces().await;
     assert!(call_count.load(Ordering::SeqCst) == 1);
     assert!(tcp_proxy.hit_count() == 0);
@@ -456,7 +459,10 @@ async fn http_proxy() {
         target_addr: tcp_proxy_addr.to_string(),
         basic_auth: None,
     });
-    let proxied_client = opts.connect("my-namespace", None, Uuid::new_v4()).await.unwrap();
+    let proxied_client = opts
+        .connect("my-namespace", None, Uuid::new_v4())
+        .await
+        .unwrap();
     let _ = proxied_client.list_namespaces().await;
     assert!(call_count.load(Ordering::SeqCst) == 2);
     assert!(tcp_proxy.hit_count() == 1);
@@ -478,7 +484,10 @@ async fn http_proxy() {
             target_addr: format!("unix:{}", sock_path.to_str().unwrap()),
             basic_auth: None,
         });
-        let proxied_client = opts.connect("my-namespace", None, Uuid::new_v4()).await.unwrap();
+        let proxied_client = opts
+            .connect("my-namespace", None, Uuid::new_v4())
+            .await
+            .unwrap();
         let _ = proxied_client.list_namespaces().await;
         assert!(call_count.load(Ordering::SeqCst) == 3);
         assert!(unix_proxy.hit_count() == 1);