Autoformat

Sushisource · Sushisource · commit 54df6c4182a6 · 2025-12-05T10:49:29.000-08:00
diff --git a/crates/common/src/worker.rs b/crates/common/src/worker.rs
@@ -240,6 +240,40 @@ pub struct WorkerConfig {
     pub skip_client_worker_set_check: bool,
 }
 
+impl WorkerConfig {
+    /// Returns true if the configuration specifies we should fail a workflow on a certain error
+    /// type rather than failing the workflow task.
+    pub fn should_fail_workflow(
+        &self,
+        workflow_type: &str,
+        error_type: &WorkflowErrorType,
+    ) -> bool {
+        self.workflow_failure_errors.contains(error_type)
+            || self
+                .workflow_types_to_failure_errors
+                .get(workflow_type)
+                .map(|s| s.contains(error_type))
+                .unwrap_or(false)
+    }
+
+    pub fn computed_deployment_version(&self) -> Option<WorkerDeploymentVersion> {
+        let wdv = match self.versioning_strategy {
+            WorkerVersioningStrategy::None { ref build_id } => WorkerDeploymentVersion {
+                deployment_name: "".to_owned(),
+                build_id: build_id.clone(),
+            },
+            WorkerVersioningStrategy::WorkerDeploymentBased(ref opts) => opts.version.clone(),
+            WorkerVersioningStrategy::LegacyBuildIdBased { ref build_id } => {
+                WorkerDeploymentVersion {
+                    deployment_name: "".to_owned(),
+                    build_id: build_id.clone(),
+                }
+            }
+        };
+        if wdv.is_empty() { None } else { Some(wdv) }
+    }
+}
+
 impl<S: worker_config_builder::IsComplete> WorkerConfigBuilder<S> {
     pub fn build(self) -> Result<WorkerConfig, String> {
         let config = self.build_internal();
@@ -260,12 +294,12 @@ impl<S: worker_config_builder::IsComplete> WorkerConfigBuilder<S> {
             config.nexus_task_poller_behavior.validate()?;
 
             if let Some(ref x) = config.max_worker_activities_per_second
-                && (!x.is_normal() || x.is_sign_negative()) {
-                    return Err(
-                        "`max_worker_activities_per_second` must be positive and nonzero"
-                            .to_string(),
-                    );
-                }
+                && (!x.is_normal() || x.is_sign_negative())
+            {
+                return Err(
+                    "`max_worker_activities_per_second` must be positive and nonzero".to_string(),
+                );
+            }
 
             if matches!(config.max_outstanding_workflow_tasks, Some(v) if v == 0) {
                 return Err("`max_outstanding_workflow_tasks` must be > 0".to_string());
@@ -282,9 +316,13 @@ impl<S: worker_config_builder::IsComplete> WorkerConfigBuilder<S> {
 
             if config.max_cached_workflows > 0 {
                 if let Some(max_wft) = config.max_outstanding_workflow_tasks
-                    && max_wft < 2 {
-                        return Err("`max_cached_workflows` > 0 requires `max_outstanding_workflow_tasks` >= 2".to_string());
-                    }
+                    && max_wft < 2
+                {
+                    return Err(
+                        "`max_cached_workflows` > 0 requires `max_outstanding_workflow_tasks` >= 2"
+                            .to_string(),
+                    );
+                }
                 if matches!(config.workflow_task_poller_behavior, PollerBehavior::SimpleMaximum(u) if u < 2)
                 {
                     return Err("`max_cached_workflows` > 0 requires `workflow_task_poller_behavior` to be at least 2".to_string());
@@ -325,44 +363,6 @@ impl<S: worker_config_builder::IsComplete> WorkerConfigBuilder<S> {
     }
 }
 
-impl WorkerConfig {
-    /// Returns true if the configuration specifies we should fail a workflow on a certain error
-    /// type rather than failing the workflow task.
-    pub fn should_fail_workflow(
-        &self,
-        workflow_type: &str,
-        error_type: &WorkflowErrorType,
-    ) -> bool {
-        self.workflow_failure_errors.contains(error_type)
-            || self
-                .workflow_types_to_failure_errors
-                .get(workflow_type)
-                .map(|s| s.contains(error_type))
-                .unwrap_or(false)
-    }
-
-    pub fn computed_deployment_version(&self) -> Option<WorkerDeploymentVersion> {
-        let wdv = match self.versioning_strategy {
-            WorkerVersioningStrategy::None { ref build_id } => WorkerDeploymentVersion {
-                deployment_name: "".to_owned(),
-                build_id: build_id.clone(),
-            },
-            WorkerVersioningStrategy::WorkerDeploymentBased(ref opts) => opts.version.clone(),
-            WorkerVersioningStrategy::LegacyBuildIdBased { ref build_id } => {
-                WorkerDeploymentVersion {
-                    deployment_name: "".to_owned(),
-                    build_id: build_id.clone(),
-                }
-            }
-        };
-        if wdv.is_empty() { None } else { Some(wdv) }
-    }
-}
-
-// Note: WorkerConfigBuilder is now generated by bon.
-// The custom clear_max_outstanding_opts method is removed because bon's builder
-// doesn't support custom methods the same way. Users should manually unset these fields if needed.
-
 /// This trait allows users to customize the performance characteristics of workers dynamically.
 /// For more, see the docstrings of the traits in the return types of its functions.
 pub trait WorkerTuner {
diff --git a/crates/sdk-core-c-bridge/src/runtime.rs b/crates/sdk-core-c-bridge/src/runtime.rs
@@ -17,12 +17,11 @@ use std::{
 };
 use temporalio_common::telemetry::{
     CoreLog, CoreLogConsumer, HistogramBucketOverrides, Logger, MetricTemporality,
-    OtelCollectorOptions, PrometheusExporterOptions,
-    TelemetryOptions as CoreTelemetryOptions, metrics::CoreMeter,
+    OtelCollectorOptions, PrometheusExporterOptions, TelemetryOptions as CoreTelemetryOptions,
+    metrics::CoreMeter,
 };
 use temporalio_sdk_core::{
-    CoreRuntime, RuntimeOptions as CoreRuntimeOptions,
-    TokioRuntimeBuilder as TokioRuntime,
+    CoreRuntime, RuntimeOptions as CoreRuntimeOptions, TokioRuntimeBuilder as TokioRuntime,
     telemetry::{build_otlp_metric_exporter, start_prometheus_metric_exporter},
 };
 use tracing::Level;
@@ -143,11 +142,8 @@ pub extern "C" fn temporal_core_runtime_new(options: *const RuntimeOptions) -> R
             // freeable
             let mut runtime = Runtime {
                 core: Arc::new(
-                    CoreRuntime::new(
-                        CoreRuntimeOptions::default(),
-                        TokioRuntime::default(),
-                    )
-                    .unwrap(),
+                    CoreRuntime::new(CoreRuntimeOptions::default(), TokioRuntime::default())
+                        .unwrap(),
                 ),
                 log_forwarder: None,
             };
@@ -208,13 +204,15 @@ impl Runtime {
         let mut log_forwarder = None;
         let telemetry_options = if let Some(v) = unsafe { options.telemetry.as_ref() } {
             // Prepare values
-            let (attach_service_name, metric_prefix) = if let Some(v) = unsafe { v.metrics.as_ref() } {
-                (v.attach_service_name, v.metric_prefix.to_option_string())
-            } else {
-                (true, None)
-            };
+            let (attach_service_name, metric_prefix) =
+                if let Some(v) = unsafe { v.metrics.as_ref() } {
+                    (v.attach_service_name, v.metric_prefix.to_option_string())
+                } else {
+                    (true, None)
+                };
 
-            let logging = unsafe { v.logging.as_ref() }.map(|v| if let Some(callback) = v.forward_to {
+            let logging = unsafe { v.logging.as_ref() }.map(|v| {
+                if let Some(callback) = v.forward_to {
                     let consumer = Arc::new(LogForwarder {
                         callback,
                         active: AtomicBool::new(false),
@@ -228,7 +226,8 @@ impl Runtime {
                     Logger::Console {
                         filter: v.filter.to_string(),
                     }
-                });
+                }
+            });
 
             // Build with all values
             CoreTelemetryOptions::builder()
diff --git a/crates/sdk-core-c-bridge/src/tests/utils.rs b/crates/sdk-core-c-bridge/src/tests/utils.rs
@@ -5,9 +5,7 @@ use crate::{
 };
 use std::{collections::HashMap, ops::Deref};
 use temporalio_client::ClientOptions;
-use temporalio_sdk_core::ephemeral_server::{
-    TemporalDevServerConfig, default_cached_download,
-};
+use temporalio_sdk_core::ephemeral_server::{TemporalDevServerConfig, default_cached_download};
 use url::Url;
 
 pub fn byte_array_to_vec(runtime: *mut Runtime, byte_array: *const ByteArray) -> Option<Vec<u8>> {
diff --git a/crates/sdk-core/benches/workflow_replay_bench.rs b/crates/sdk-core/benches/workflow_replay_bench.rs
@@ -86,21 +86,9 @@ pub fn bench_metrics(c: &mut Criterion) {
     c.bench_function("Record with new attributes on each call", move |b| {
         b.iter_batched(
             || {
-                let c = meter.counter(
-                    MetricParameters::builder()
-                        .name("c")
-                        .build(),
-                );
-                let h = meter.histogram(
-                    MetricParameters::builder()
-                        .name("h")
-                        .build(),
-                );
-                let g = meter.gauge(
-                    MetricParameters::builder()
-                        .name("g")
-                        .build(),
-                );
+                let c = meter.counter(MetricParameters::builder().name("c").build());
+                let h = meter.histogram(MetricParameters::builder().name("h").build());
+                let g = meter.gauge(MetricParameters::builder().name("g").build());
 
                 let vals = [1, 2, 3, 4, 5];
                 let labels = ["l1", "l2"];
diff --git a/crates/sdk-core/src/worker/mod.rs b/crates/sdk-core/src/worker/mod.rs
@@ -1458,7 +1458,7 @@ mod tests {
         // The validation would need to be added manually via a builder validator
         // For now, this test just documents that a config with 0 pollers can be created
         // but would fail at runtime when attempting to create the worker
-        use temporalio_common::worker::{WorkerConfig, WorkerVersioningStrategy, WorkerTaskTypes};
+        use temporalio_common::worker::{WorkerConfig, WorkerTaskTypes, WorkerVersioningStrategy};
         let _cfg = WorkerConfig::builder()
             .namespace("test")
             .task_queue("test")
diff --git a/crates/sdk-core/src/worker/workflow/mod.rs b/crates/sdk-core/src/worker/workflow/mod.rs
@@ -98,8 +98,7 @@ use tokio::{
     task::{LocalSet, spawn_blocking},
 };
 use tokio_stream::wrappers::UnboundedReceiverStream;
-use tokio_util::either::Either;
-use tokio_util::sync::CancellationToken;
+use tokio_util::{either::Either, sync::CancellationToken};
 use tracing::{Span, Subscriber};
 
 /// Id used by server for "legacy" queries. IE: Queries that come in the `query` rather than
diff --git a/crates/sdk-core/tests/common/mod.rs b/crates/sdk-core/tests/common/mod.rs
@@ -46,8 +46,7 @@ use temporalio_common::{
         },
     },
     telemetry::{
-        Logger, OtelCollectorOptions, PrometheusExporterOptions,
-        TelemetryOptions,
+        Logger, OtelCollectorOptions, PrometheusExporterOptions, TelemetryOptions,
         metrics::CoreMeter,
     },
     worker::{WorkerTaskTypes, WorkerVersioningStrategy},
@@ -62,8 +61,7 @@ use temporalio_sdk::{
 #[cfg(any(feature = "test-utilities", test))]
 pub(crate) use temporalio_sdk_core::test_help::NAMESPACE;
 use temporalio_sdk_core::{
-    ClientOptions, CoreRuntime, RuntimeOptions, WorkerConfig,
-    init_replay_worker, init_worker,
+    ClientOptions, CoreRuntime, RuntimeOptions, WorkerConfig, init_replay_worker, init_worker,
     replay::{HistoryForReplay, ReplayWorkerInput},
     telemetry::{build_otlp_metric_exporter, start_prometheus_metric_exporter},
     test_help::{MockPollCfg, build_mock_pollers, mock_worker},
@@ -807,9 +805,7 @@ pub(crate) fn get_integ_telem_options() -> TelemetryOptions {
         .ok()
         .map(|x| x.parse::<Url>().unwrap())
     {
-        let opts = OtelCollectorOptions::builder()
-            .url(url)
-            .build();
+        let opts = OtelCollectorOptions::builder().url(url).build();
         TelemetryOptions::builder()
             .metrics(Arc::new(build_otlp_metric_exporter(opts).unwrap()) as Arc<dyn CoreMeter>)
             .logging(Logger::Console {
diff --git a/crates/sdk-core/tests/integ_tests/ephemeral_server_tests.rs b/crates/sdk-core/tests/integ_tests/ephemeral_server_tests.rs
@@ -63,9 +63,11 @@ async fn temporal_cli_shutdown_port_reuse() {
 #[ignore]
 async fn temporal_cli_concurrent_starts() -> Result<(), Box<dyn std::error::Error>> {
     stream::iter((0..80).map(|_| {
-        Ok::<TemporalDevServerConfig, Box<dyn std::error::Error>>(TemporalDevServerConfig::builder()
-            .exe(default_cached_download())
-            .build())
+        Ok::<TemporalDevServerConfig, Box<dyn std::error::Error>>(
+            TemporalDevServerConfig::builder()
+                .exe(default_cached_download())
+                .build(),
+        )
     }))
     .try_for_each_concurrent(8, |config| async move {
         let mut server = config.start_server().await?;
diff --git a/crates/sdk-core/tests/integ_tests/metrics_tests.rs b/crates/sdk-core/tests/integ_tests/metrics_tests.rs
@@ -48,17 +48,17 @@ use temporalio_common::{
         },
     },
     telemetry::{
-        HistogramBucketOverrides, OtelCollectorOptions, OtlpProtocol,
-        PrometheusExporterOptions, TaskQueueLabelStrategy, TelemetryOptions,
+        HistogramBucketOverrides, OtelCollectorOptions, OtlpProtocol, PrometheusExporterOptions,
+        TaskQueueLabelStrategy, TelemetryOptions,
         metrics::{
             CoreMeter, CounterBase, Gauge, GaugeBase, HistogramBase, MetricKeyValue,
             MetricParameters, NewAttributes,
         },
     },
     worker::{
         PollerBehavior, SlotKind, SlotMarkUsedContext, SlotReleaseContext, SlotReservationContext,
-        SlotSupplier, SlotSupplierPermit, WorkerConfig, WorkerTaskTypes,
-        WorkerVersioningStrategy, WorkflowSlotKind,
+        SlotSupplier, SlotSupplierPermit, WorkerConfig, WorkerTaskTypes, WorkerVersioningStrategy,
+        WorkflowSlotKind,
     },
 };
 use temporalio_sdk::{
@@ -668,19 +668,15 @@ async fn request_fail_codes_otel() {
         .ok()
         .map(|x| x.parse::<Url>().unwrap())
     {
-        let opts = OtelCollectorOptions::builder()
-            .url(url)
-            .build();
+        let opts = OtelCollectorOptions::builder().url(url).build();
         build_otlp_metric_exporter(opts).unwrap()
     } else {
         // skip
         return;
     };
     let exporter = Arc::new(exporter);
-    let telemopts = TelemetryOptions::builder()
-        .metrics(exporter as Arc<dyn CoreMeter>);
-    let rt = CoreRuntime::new_assume_tokio(get_integ_runtime_options(telemopts.build()))
-        .unwrap();
+    let telemopts = TelemetryOptions::builder().metrics(exporter as Arc<dyn CoreMeter>);
+    let rt = CoreRuntime::new_assume_tokio(get_integ_runtime_options(telemopts.build())).unwrap();
     let opts = get_integ_server_options();
     let mut client = opts
         .connect(NAMESPACE, rt.telemetry().get_temporal_metric_meter())
@@ -1313,16 +1309,11 @@ async fn test_prometheus_metric_format_consistency() {
 
 #[tokio::test]
 async fn prometheus_label_nonsense() {
-    let opts_builder = PrometheusExporterOptions::builder()
-        .socket_addr(ANY_PORT.parse().unwrap());
+    let opts_builder = PrometheusExporterOptions::builder().socket_addr(ANY_PORT.parse().unwrap());
     let (telemopts, addr, _aborter) = prom_metrics(Some(opts_builder.build()));
     let meter = telemopts.metrics.clone().unwrap();
 
-    let ctr = meter.counter(
-        MetricParameters::builder()
-            .name("some_counter")
-            .build(),
-    );
+    let ctr = meter.counter(MetricParameters::builder().name("some_counter").build());
     let a1 = meter.new_attributes(NewAttributes::from([MetricKeyValue::new("thing", "foo")]));
     let a2 = meter.new_attributes(NewAttributes::from([MetricKeyValue::new("blerp", "baz")]));
     ctr.add(1, &a1);
diff --git a/crates/sdk-core/tests/integ_tests/polling_tests.rs b/crates/sdk-core/tests/integ_tests/polling_tests.rs
@@ -193,12 +193,7 @@ async fn switching_worker_client_changes_poll() {
         let mut config = integ_worker_config("my-task-queue");
         // We want a cache so we don't get extra remove-job activations
         config.max_cached_workflows = 100_usize;
-        let worker = init_worker(
-            init_integ_telem().unwrap(),
-            config,
-            client1.clone(),
-        )
-        .unwrap();
+        let worker = init_worker(init_integ_telem().unwrap(), config, client1.clone()).unwrap();
 
         // Poll for first task, confirm it's first wf, complete, and wait for complete
         info!("Doing initial poll");
@@ -398,14 +393,8 @@ async fn replace_client_works_after_polling_failure() {
 
             let mut config = integ_worker_config(&task_queue);
             config.max_cached_workflows = 100_usize;
-            let worker = Arc::new(
-                init_worker(
-                    &rt,
-                    config,
-                    client_for_initial_server.clone(),
-                )
-                .unwrap(),
-            );
+            let worker =
+                Arc::new(init_worker(&rt, config, client_for_initial_server.clone()).unwrap());
 
             // Polling the initial server the first time is successful.
             let wf_1 = client_for_initial_server
diff --git a/crates/sdk-core/tests/integ_tests/worker_heartbeat_tests.rs b/crates/sdk-core/tests/integ_tests/worker_heartbeat_tests.rs
diff --git a/crates/sdk-core/tests/integ_tests/worker_versioning_tests.rs b/crates/sdk-core/tests/integ_tests/worker_versioning_tests.rs
