Skip to main content

dynamo_runtime/
config.rs

1// SPDX-FileCopyrightText: Copyright (c) 2024-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
2// SPDX-License-Identifier: Apache-2.0
3
4use anyhow::Result;
5use derive_builder::Builder;
6use figment::{
7    Figment,
8    providers::{Env, Format, Serialized, Toml},
9};
10use serde::{Deserialize, Serialize};
11use std::fmt;
12use std::sync::OnceLock;
13use validator::Validate;
14
15pub mod environment_names;
16
17/// Default system host for health and metrics endpoints
18const DEFAULT_SYSTEM_HOST: &str = "0.0.0.0";
19
20/// Default system port for health and metrics endpoints (-1 = disabled)
21const DEFAULT_SYSTEM_PORT: i16 = -1;
22
23/// Default health endpoint paths
24const DEFAULT_SYSTEM_HEALTH_PATH: &str = "/health";
25const DEFAULT_SYSTEM_LIVE_PATH: &str = "/live";
26
27/// Default health check configuration
28/// This is the wait time before sending canary health checks when no activity is detected
29pub const DEFAULT_CANARY_WAIT_TIME_SECS: u64 = 10;
30/// Default timeout for individual health check requests
31pub const DEFAULT_HEALTH_CHECK_REQUEST_TIMEOUT_SECS: u64 = 3;
32
33#[derive(Debug, Clone, Serialize, Deserialize)]
34pub struct WorkerConfig {
35    /// Grace shutdown period for the system server.
36    pub graceful_shutdown_timeout: u64,
37}
38
39impl WorkerConfig {
40    /// Instantiates and reads server configurations from appropriate sources.
41    /// Panics on invalid configuration.
42    pub fn from_settings() -> Self {
43        // All calls should be global and thread safe.
44        Figment::new()
45            .merge(Serialized::defaults(Self::default()))
46            .merge(Env::prefixed("DYN_WORKER_"))
47            .extract()
48            .unwrap() // safety: Called on startup, so panic is reasonable
49    }
50}
51
52impl Default for WorkerConfig {
53    fn default() -> Self {
54        WorkerConfig {
55            graceful_shutdown_timeout: if cfg!(debug_assertions) {
56                1 // Debug build: 1 second
57            } else {
58                30 // Release build: 30 seconds
59            },
60        }
61    }
62}
63
64#[derive(Debug, Deserialize, Serialize, PartialEq, Clone)]
65#[serde(rename_all = "lowercase")]
66pub enum HealthStatus {
67    Ready,
68    NotReady,
69}
70
71/// Runtime configuration
72/// Defines the configuration for Tokio runtimes
73#[derive(Serialize, Deserialize, Validate, Debug, Builder, Clone)]
74#[builder(build_fn(private, name = "build_internal"), derive(Debug, Serialize))]
75pub struct RuntimeConfig {
76    /// Number of async worker threads
77    /// If set to 1, the runtime will run in single-threaded mode
78    /// Set this at runtime with environment variable DYN_RUNTIME_NUM_WORKER_THREADS. Defaults to
79    /// number of cores.
80    #[validate(range(min = 1))]
81    #[builder_field_attr(serde(skip_serializing_if = "Option::is_none"))]
82    pub num_worker_threads: Option<usize>,
83
84    /// Maximum number of blocking threads
85    /// Blocking threads are used for blocking operations, this value must be greater than 0.
86    /// Set this at runtime with environment variable DYN_RUNTIME_MAX_BLOCKING_THREADS.
87    ///
88    /// Defaults to the core count (`impl Default`). The `#[builder(default = "512")]` below
89    /// applies only when building through `RuntimeConfigBuilder` without setting this field.
90    ///
91    /// This is a ceiling, not a preallocation: Tokio spawns blocking threads on demand and reaps
92    /// them when idle, so measure at steady state under load.
93    #[validate(range(min = 1))]
94    #[builder(default = "512")]
95    #[builder_field_attr(serde(skip_serializing_if = "Option::is_none"))]
96    pub max_blocking_threads: usize,
97
98    /// System status server host for health and metrics endpoints
99    /// Set this at runtime with environment variable DYN_SYSTEM_HOST
100    #[builder(default = "DEFAULT_SYSTEM_HOST.to_string()")]
101    #[builder_field_attr(serde(skip_serializing_if = "Option::is_none"))]
102    pub system_host: String,
103
104    /// System status server port for health and metrics endpoints
105    /// Set to -1 to disable the system status server (default)
106    /// Set to 0 to bind to a random available port
107    /// Set to a positive port number (e.g. 8081) to bind to a specific port
108    /// Set this at runtime with environment variable DYN_SYSTEM_PORT
109    #[builder(default = "DEFAULT_SYSTEM_PORT")]
110    #[builder_field_attr(serde(skip_serializing_if = "Option::is_none"))]
111    pub system_port: i16,
112
113    /// Health and metrics System status server enabled (DEPRECATED)
114    /// This field is deprecated. Use system_port instead (set to positive value to enable)
115    /// Environment variable DYN_SYSTEM_ENABLED is deprecated
116    #[deprecated(
117        note = "Use system_port instead. Set DYN_SYSTEM_PORT to enable the system metrics server."
118    )]
119    #[builder(default = "false")]
120    #[builder_field_attr(serde(skip_serializing_if = "Option::is_none"))]
121    pub system_enabled: bool,
122
123    /// Starting Health Status
124    /// Set this at runtime with environment variable DYN_SYSTEM_STARTING_HEALTH_STATUS
125    #[builder(default = "HealthStatus::NotReady")]
126    #[builder_field_attr(serde(skip_serializing_if = "Option::is_none"))]
127    pub starting_health_status: HealthStatus,
128
129    /// Use Endpoint Health Status
130    /// When using endpoint health status, health status
131    /// is the AND of individual endpoint health
132    /// Set this at runtime with environment variable DYN_SYSTEM_USE_ENDPOINT_HEALTH_STATUS
133    /// with the list of endpoints to consider for system health
134    #[builder(default = "vec![]")]
135    #[builder_field_attr(serde(skip_serializing_if = "Option::is_none"))]
136    pub use_endpoint_health_status: Vec<String>,
137
138    /// Health endpoint paths
139    /// Set this at runtime with environment variable DYN_SYSTEM_HEALTH_PATH
140    #[builder(default = "DEFAULT_SYSTEM_HEALTH_PATH.to_string()")]
141    #[builder_field_attr(serde(skip_serializing_if = "Option::is_none"))]
142    pub system_health_path: String,
143    /// Set this at runtime with environment variable DYN_SYSTEM_LIVE_PATH
144    #[builder(default = "DEFAULT_SYSTEM_LIVE_PATH.to_string()")]
145    #[builder_field_attr(serde(skip_serializing_if = "Option::is_none"))]
146    pub system_live_path: String,
147
148    /// Number of threads for the Rayon compute pool
149    /// If not set, defaults to num_cpus / 2
150    /// Set this at runtime with environment variable DYN_COMPUTE_THREADS
151    #[builder(default = "None")]
152    #[builder_field_attr(serde(skip_serializing_if = "Option::is_none"))]
153    pub compute_threads: Option<usize>,
154
155    /// Stack size for compute threads in bytes
156    /// Defaults to 2MB (2097152 bytes)
157    /// Set this at runtime with environment variable DYN_COMPUTE_STACK_SIZE
158    #[builder(default = "Some(2 * 1024 * 1024)")]
159    #[builder_field_attr(serde(skip_serializing_if = "Option::is_none"))]
160    pub compute_stack_size: Option<usize>,
161
162    /// Thread name prefix for compute pool threads
163    /// Set this at runtime with environment variable DYN_COMPUTE_THREAD_PREFIX
164    #[builder(default = "\"compute\".to_string()")]
165    #[builder_field_attr(serde(skip_serializing_if = "Option::is_none"))]
166    pub compute_thread_prefix: String,
167
168    /// Enable active health checking with payloads
169    /// Set this at runtime with environment variable DYN_HEALTH_CHECK_ENABLED
170    #[builder(default = "false")]
171    #[builder_field_attr(serde(skip_serializing_if = "Option::is_none"))]
172    pub health_check_enabled: bool,
173
174    /// Canary wait time in seconds (time to wait before sending health check when no activity)
175    /// Set this at runtime with environment variable DYN_CANARY_WAIT_TIME
176    #[builder(default = "DEFAULT_CANARY_WAIT_TIME_SECS")]
177    #[builder_field_attr(serde(skip_serializing_if = "Option::is_none"))]
178    pub canary_wait_time_secs: u64,
179
180    /// Health check request timeout in seconds
181    /// Set this at runtime with environment variable DYN_HEALTH_CHECK_REQUEST_TIMEOUT
182    #[builder(default = "DEFAULT_HEALTH_CHECK_REQUEST_TIMEOUT_SECS")]
183    #[builder_field_attr(serde(skip_serializing_if = "Option::is_none"))]
184    pub health_check_request_timeout_secs: u64,
185}
186
187impl fmt::Display for RuntimeConfig {
188    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
189        // If None, it defaults to "number of cores", so we indicate that.
190        match self.num_worker_threads {
191            Some(val) => write!(f, "num_worker_threads={val}, ")?,
192            None => write!(f, "num_worker_threads=default (num_cores), ")?,
193        }
194
195        write!(f, "max_blocking_threads={}, ", self.max_blocking_threads)?;
196        write!(f, "system_host={}, ", self.system_host)?;
197        write!(f, "system_port={}, ", self.system_port)?;
198        write!(
199            f,
200            "use_endpoint_health_status={:?}",
201            self.use_endpoint_health_status
202        )?;
203        write!(
204            f,
205            "starting_health_status={:?}",
206            self.starting_health_status
207        )?;
208        write!(f, ", system_health_path={}", self.system_health_path)?;
209        write!(f, ", system_live_path={}", self.system_live_path)?;
210        write!(f, ", health_check_enabled={}", self.health_check_enabled)?;
211        write!(f, ", canary_wait_time_secs={}", self.canary_wait_time_secs)?;
212        write!(
213            f,
214            ", health_check_request_timeout_secs={}",
215            self.health_check_request_timeout_secs
216        )?;
217
218        Ok(())
219    }
220}
221
222impl RuntimeConfig {
223    pub fn builder() -> RuntimeConfigBuilder {
224        RuntimeConfigBuilder::default()
225    }
226
227    pub(crate) fn figment() -> Figment {
228        Figment::new()
229            .merge(Serialized::defaults(RuntimeConfig::default()))
230            .merge(Toml::file("/opt/dynamo/defaults/runtime.toml"))
231            .merge(Toml::file("/opt/dynamo/etc/runtime.toml"))
232            .merge(Env::prefixed("DYN_RUNTIME_").filter_map(|k| {
233                let full_key = format!("DYN_RUNTIME_{}", k.as_str());
234                // filters out empty environment variables
235                match std::env::var(&full_key) {
236                    Ok(v) if !v.is_empty() => Some(k.into()),
237                    _ => None,
238                }
239            }))
240            .merge(Env::prefixed("DYN_SYSTEM_").filter_map(|k| {
241                let full_key = format!("DYN_SYSTEM_{}", k.as_str());
242                // filters out empty environment variables
243                match std::env::var(&full_key) {
244                    Ok(v) if !v.is_empty() => {
245                        // Map DYN_SYSTEM_* to the correct field names
246                        let mapped_key = match k.as_str() {
247                            "HOST" => "system_host",
248                            "PORT" => "system_port",
249                            "ENABLED" => "system_enabled",
250                            "USE_ENDPOINT_HEALTH_STATUS" => "use_endpoint_health_status",
251                            "STARTING_HEALTH_STATUS" => "starting_health_status",
252                            "HEALTH_PATH" => "system_health_path",
253                            "LIVE_PATH" => "system_live_path",
254                            _ => k.as_str(),
255                        };
256                        Some(mapped_key.into())
257                    }
258                    _ => None,
259                }
260            }))
261            .merge(Env::prefixed("DYN_COMPUTE_").filter_map(|k| {
262                let full_key = format!("DYN_COMPUTE_{}", k.as_str());
263                // filters out empty environment variables
264                match std::env::var(&full_key) {
265                    Ok(v) if !v.is_empty() => {
266                        // Map DYN_COMPUTE_* to the correct field names
267                        let mapped_key = match k.as_str() {
268                            "THREADS" => "compute_threads",
269                            "STACK_SIZE" => "compute_stack_size",
270                            "THREAD_PREFIX" => "compute_thread_prefix",
271                            _ => k.as_str(),
272                        };
273                        Some(mapped_key.into())
274                    }
275                    _ => None,
276                }
277            }))
278            .merge(Env::prefixed("DYN_HEALTH_CHECK_").filter_map(|k| {
279                let full_key = format!("DYN_HEALTH_CHECK_{}", k.as_str());
280                // filters out empty environment variables
281                match std::env::var(&full_key) {
282                    Ok(v) if !v.is_empty() => {
283                        // Map DYN_HEALTH_CHECK_* to the correct field names
284                        let mapped_key = match k.as_str() {
285                            "ENABLED" => "health_check_enabled",
286                            "REQUEST_TIMEOUT" => "health_check_request_timeout_secs",
287                            _ => k.as_str(),
288                        };
289                        Some(mapped_key.into())
290                    }
291                    _ => None,
292                }
293            }))
294            .merge(Env::prefixed("DYN_CANARY_").filter_map(|k| {
295                let full_key = format!("DYN_CANARY_{}", k.as_str());
296                // filters out empty environment variables
297                match std::env::var(&full_key) {
298                    Ok(v) if !v.is_empty() => {
299                        // Map DYN_CANARY_* to the correct field names
300                        let mapped_key = match k.as_str() {
301                            "WAIT_TIME" => "canary_wait_time_secs",
302                            _ => k.as_str(),
303                        };
304                        Some(mapped_key.into())
305                    }
306                    _ => None,
307                }
308            }))
309    }
310
311    /// Load the runtime configuration from the environment and configuration files
312    /// Configuration is priorities in the following order, where the last has the lowest priority:
313    /// 1. Environment variables (top priority)
314    ///    TO DO: Add documentation for configuration files. Paths should be configurable.
315    /// 2. /opt/dynamo/etc/runtime.toml
316    /// 3. /opt/dynamo/defaults/runtime.toml (lowest priority)
317    ///
318    /// Environment variables are prefixed with `DYN_RUNTIME_` and `DYN_SYSTEM`
319    pub fn from_settings() -> Result<RuntimeConfig> {
320        use environment_names::runtime::system as env_system;
321        // Check for deprecated environment variables
322        if std::env::var(env_system::DYN_SYSTEM_USE_ENDPOINT_HEALTH_STATUS).is_ok() {
323            tracing::warn!(
324                "DYN_SYSTEM_USE_ENDPOINT_HEALTH_STATUS is deprecated and no longer used. \
325                System health is now determined by endpoints that register with health check payloads. \
326                Please update your configuration to register health check payloads directly on endpoints."
327            );
328        }
329
330        if std::env::var(env_system::DYN_SYSTEM_ENABLED).is_ok() {
331            tracing::warn!(
332                "DYN_SYSTEM_ENABLED is deprecated. \
333                System metrics server is now controlled solely by DYN_SYSTEM_PORT. \
334                Set DYN_SYSTEM_PORT to a positive value to enable the server, or set to -1 to disable (default)."
335            );
336        }
337
338        let config: RuntimeConfig = Self::figment().extract()?;
339        config.validate()?;
340        Ok(config)
341    }
342
343    /// Check if System server should be enabled
344    /// System server is enabled when DYN_SYSTEM_PORT is set to 0 or a positive value
345    /// Port 0 binds to a random available port
346    /// Negative values disable the server
347    pub fn system_server_enabled(&self) -> bool {
348        self.system_port >= 0
349    }
350
351    pub fn single_threaded() -> Self {
352        RuntimeConfig {
353            num_worker_threads: Some(1),
354            max_blocking_threads: 1,
355            system_host: DEFAULT_SYSTEM_HOST.to_string(),
356            system_port: DEFAULT_SYSTEM_PORT,
357            #[allow(deprecated)]
358            system_enabled: false,
359            starting_health_status: HealthStatus::NotReady,
360            use_endpoint_health_status: vec![],
361            system_health_path: DEFAULT_SYSTEM_HEALTH_PATH.to_string(),
362            system_live_path: DEFAULT_SYSTEM_LIVE_PATH.to_string(),
363            compute_threads: Some(1),
364            compute_stack_size: Some(2 * 1024 * 1024),
365            compute_thread_prefix: "compute".to_string(),
366            health_check_enabled: false,
367            canary_wait_time_secs: DEFAULT_CANARY_WAIT_TIME_SECS,
368            health_check_request_timeout_secs: DEFAULT_HEALTH_CHECK_REQUEST_TIMEOUT_SECS,
369        }
370    }
371
372    /// The Tokio builder for this config, not yet built.
373    ///
374    /// Separate from [`Self::create_runtime`] because the pyo3 bridge builds its own runtime:
375    /// `pyo3_async_runtimes::tokio::init` takes a builder and calls `build()` later. Handing it
376    /// this builder is the only way to bound that runtime's size. Both paths go through here so
377    /// they cannot drift apart.
378    pub fn tokio_builder(&self) -> tokio::runtime::Builder {
379        let mut builder = tokio::runtime::Builder::new_multi_thread();
380        builder
381            .worker_threads(
382                self.num_worker_threads
383                    .unwrap_or_else(|| std::thread::available_parallelism().unwrap().get()),
384            )
385            .max_blocking_threads(self.max_blocking_threads)
386            .enable_all();
387        if env_is_truthy(environment_names::runtime::DYN_ENABLE_POLL_HISTOGRAM) {
388            tracing::info!(
389                "Tokio poll-time histogram enabled (DYN_ENABLE_POLL_HISTOGRAM); \
390                 expect ~2× Instant::now() overhead per task poll"
391            );
392            builder.enable_metrics_poll_time_histogram();
393        }
394        builder
395    }
396
397    /// Create a new default runtime configuration
398    pub(crate) fn create_runtime(&self) -> std::io::Result<tokio::runtime::Runtime> {
399        self.tokio_builder().build()
400    }
401}
402
403impl Default for RuntimeConfig {
404    fn default() -> Self {
405        let num_cores = std::thread::available_parallelism().unwrap().get();
406        Self {
407            num_worker_threads: Some(num_cores),
408            max_blocking_threads: num_cores,
409            system_host: DEFAULT_SYSTEM_HOST.to_string(),
410            system_port: DEFAULT_SYSTEM_PORT,
411            #[allow(deprecated)]
412            system_enabled: false,
413            starting_health_status: HealthStatus::NotReady,
414            use_endpoint_health_status: vec![],
415            system_health_path: DEFAULT_SYSTEM_HEALTH_PATH.to_string(),
416            system_live_path: DEFAULT_SYSTEM_LIVE_PATH.to_string(),
417            compute_threads: None,
418            compute_stack_size: Some(2 * 1024 * 1024),
419            compute_thread_prefix: "compute".to_string(),
420            health_check_enabled: false,
421            canary_wait_time_secs: DEFAULT_CANARY_WAIT_TIME_SECS,
422            health_check_request_timeout_secs: DEFAULT_HEALTH_CHECK_REQUEST_TIMEOUT_SECS,
423        }
424    }
425}
426
427impl RuntimeConfigBuilder {
428    /// Build and validate the runtime configuration
429    pub fn build(&self) -> Result<RuntimeConfig> {
430        let config = self.build_internal()?;
431        config.validate()?;
432        Ok(config)
433    }
434}
435
436// Canonical truthy/falsy/bool parsing for user-supplied configuration
437// (environment variables, headers, config values). The single implementation
438// lives in the zero-dependency `dynamo-truthy` crate so that crates which
439// cannot depend on `dynamo-runtime` share it too; this re-export is the
440// canonical import path for everything that can.
441pub use dynamo_truthy::{
442    env_is_falsey, env_is_truthy, is_falsey, is_truthy, parse_bool, parse_bool_opt,
443};
444
445#[derive(Debug, Clone, Copy, PartialEq, Eq)]
446pub enum ConsoleLogFormat {
447    Readable,
448    Jsonl,
449}
450
451impl ConsoleLogFormat {
452    fn from_env_value(value: &str) -> Option<Self> {
453        match value.to_ascii_lowercase().as_str() {
454            "readable" => Some(Self::Readable),
455            "jsonl" => Some(Self::Jsonl),
456            _ => None,
457        }
458    }
459
460    pub fn as_str(self) -> &'static str {
461        match self {
462            Self::Readable => "readable",
463            Self::Jsonl => "jsonl",
464        }
465    }
466}
467
468/// Return whether the legacy `DYN_LOGGING_JSONL` switch is enabled.
469///
470/// This remains a separate compatibility signal because older deployments
471/// also use it to enable local trace-context propagation.
472pub(crate) fn legacy_jsonl_logging_enabled() -> bool {
473    env_is_truthy(environment_names::logging::DYN_LOGGING_JSONL)
474}
475
476/// Return the console log format.
477///
478/// `DYN_LOGGING_CONSOLE_FORMAT` takes precedence. `DYN_LOGGING_JSONL` remains
479/// supported as a legacy fallback when the new setting is unset or blank.
480pub fn console_log_format() -> ConsoleLogFormat {
481    let legacy_format = || {
482        if legacy_jsonl_logging_enabled() {
483            ConsoleLogFormat::Jsonl
484        } else {
485            ConsoleLogFormat::Readable
486        }
487    };
488
489    match std::env::var(environment_names::logging::DYN_LOGGING_CONSOLE_FORMAT) {
490        Ok(value) if value.trim().is_empty() => legacy_format(),
491        Ok(value) => match ConsoleLogFormat::from_env_value(value.trim()) {
492            Some(format) => format,
493            None => {
494                eprintln!(
495                    "Invalid {} value '{}'; using readable console logs",
496                    environment_names::logging::DYN_LOGGING_CONSOLE_FORMAT,
497                    value
498                );
499                ConsoleLogFormat::Readable
500            }
501        },
502        Err(_) => legacy_format(),
503    }
504}
505
506/// Return whether the effective console log format is JSONL.
507pub fn jsonl_logging_enabled() -> bool {
508    console_log_format() == ConsoleLogFormat::Jsonl
509}
510
511/// Check whether logging with ANSI terminal escape codes and colors is disabled.
512/// Set the `DYN_SDK_DISABLE_ANSI_LOGGING` environment variable a [`is_truthy`] value
513pub fn disable_ansi_logging() -> bool {
514    env_is_truthy(environment_names::logging::DYN_SDK_DISABLE_ANSI_LOGGING)
515}
516
517/// Check whether to use local timezone for logging timestamps (default is UTC)
518/// Set the `DYN_LOG_USE_LOCAL_TZ` environment variable to a [`is_truthy`] value
519pub fn use_local_timezone() -> bool {
520    env_is_truthy(environment_names::logging::DYN_LOG_USE_LOCAL_TZ)
521}
522
523/// Returns true if `DYN_LOGGING_SPAN_EVENTS` is set to a truthy value.
524pub fn span_events_enabled() -> bool {
525    env_is_truthy(environment_names::logging::DYN_LOGGING_SPAN_EVENTS)
526}
527
528#[cfg(test)]
529mod tests {
530    use super::*;
531
532    #[test]
533    fn test_runtime_config_builder_overrides_related_fields() -> Result<()> {
534        let config = RuntimeConfig::builder()
535            .num_worker_threads(Some(24))
536            .max_blocking_threads(32)
537            .system_host("127.0.0.1".to_string())
538            .system_port(9090)
539            .build()?;
540
541        assert_eq!(config.num_worker_threads, Some(24));
542        assert_eq!(config.max_blocking_threads, 32);
543        assert_eq!(config.system_host, "127.0.0.1");
544        assert_eq!(config.system_port, 9090);
545        Ok(())
546    }
547
548    /// Both thread-pool variables must survive `from_settings()`.
549    ///
550    /// Covers parsing on its own, so if a frontend's thread count ignores
551    /// `DYN_RUNTIME_MAX_BLOCKING_THREADS` the cause is wiring rather than parsing.
552    ///
553    /// `temp_env::with_vars` restores the old values on the way out, including on panic.
554    #[test]
555    fn test_from_settings_reads_both_thread_env_vars() {
556        const WORKERS: &str = "DYN_RUNTIME_NUM_WORKER_THREADS";
557        const BLOCKING: &str = "DYN_RUNTIME_MAX_BLOCKING_THREADS";
558
559        temp_env::with_vars([(WORKERS, Some("7")), (BLOCKING, Some("11"))], || {
560            let config = RuntimeConfig::from_settings().expect("from_settings failed");
561            assert_eq!(config.num_worker_threads, Some(7), "{WORKERS} was not read");
562            assert_eq!(config.max_blocking_threads, 11, "{BLOCKING} was not read");
563        });
564    }
565
566    /// The builder given to the pyo3 bridge must carry the configured worker count.
567    ///
568    /// The bridge calls `build()` itself, so nothing on our side sees the resulting runtime. If
569    /// this stopped applying the config, a bridge-built runtime would quietly go back to one
570    /// worker per CPU — the original bug, in a place no other test looks.
571    #[test]
572    fn test_tokio_builder_applies_configured_worker_threads() -> Result<()> {
573        let config = RuntimeConfig::builder()
574            .num_worker_threads(Some(3))
575            .max_blocking_threads(5)
576            .build()?;
577
578        let runtime = config.tokio_builder().build()?;
579        assert_eq!(runtime.metrics().num_workers(), 3);
580        Ok(())
581    }
582
583    /// With `num_worker_threads` unset, the builder falls back to the core count.
584    #[test]
585    fn test_tokio_builder_defaults_worker_threads_to_core_count() -> Result<()> {
586        let config = RuntimeConfig {
587            num_worker_threads: None,
588            ..RuntimeConfig::default()
589        };
590
591        let runtime = config.tokio_builder().build()?;
592        assert_eq!(
593            runtime.metrics().num_workers(),
594            std::thread::available_parallelism()?.get()
595        );
596        Ok(())
597    }
598
599    /// `max_blocking_threads` must actually cap concurrent blocking work.
600    ///
601    /// This is the setting whose effect on a frontend's thread count could not be observed, and
602    /// `num_workers()` cannot show it — Tokio counts blocking threads separately and only
603    /// exposes that count under `tokio_unstable`. Measuring concurrency works on stable instead:
604    /// blocking threads are spawned on demand up to the cap, so queueing more tasks than the cap
605    /// must serialize them.
606    ///
607    /// Only the upper bound is asserted. A missing cap shows up as a peak near the task count,
608    /// while asserting a lower bound would make the test depend on the scheduler overlapping
609    /// tasks, which a loaded CI machine need not do.
610    #[test]
611    fn test_tokio_builder_applies_max_blocking_threads() -> Result<()> {
612        use std::sync::Arc;
613        use std::sync::atomic::{AtomicUsize, Ordering};
614
615        const CAP: usize = 2;
616
617        let config = RuntimeConfig::builder()
618            .num_worker_threads(Some(2))
619            .max_blocking_threads(CAP)
620            .build()?;
621        let runtime = config.tokio_builder().build()?;
622
623        let in_flight = Arc::new(AtomicUsize::new(0));
624        let peak = Arc::new(AtomicUsize::new(0));
625
626        runtime.block_on(async {
627            let tasks: Vec<_> = (0..CAP * 4)
628                .map(|_| {
629                    let in_flight = Arc::clone(&in_flight);
630                    let peak = Arc::clone(&peak);
631                    tokio::task::spawn_blocking(move || {
632                        let now = in_flight.fetch_add(1, Ordering::SeqCst) + 1;
633                        peak.fetch_max(now, Ordering::SeqCst);
634                        // Long enough that tasks overlap if the cap allows it.
635                        std::thread::sleep(std::time::Duration::from_millis(20));
636                        in_flight.fetch_sub(1, Ordering::SeqCst);
637                    })
638                })
639                .collect();
640
641            for task in tasks {
642                task.await.expect("blocking task panicked");
643            }
644        });
645
646        let observed = peak.load(Ordering::SeqCst);
647        assert!(
648            observed <= CAP,
649            "{observed} blocking tasks ran at once, but the cap was {CAP}"
650        );
651        Ok(())
652    }
653
654    /// `Default` sets `max_blocking_threads` to the core count, not the `#[builder(default)]`
655    /// of 512 — that applies only when building through `RuntimeConfigBuilder`.
656    #[test]
657    fn test_default_max_blocking_threads_is_core_count() {
658        let cores = std::thread::available_parallelism().unwrap().get();
659        let config = RuntimeConfig::default();
660        assert_eq!(config.max_blocking_threads, cores);
661        assert_eq!(config.num_worker_threads, Some(cores));
662    }
663
664    #[test]
665    fn test_runtime_config_rejects_invalid_thread_count() -> Result<()> {
666        let result = RuntimeConfig::builder()
667            .num_worker_threads(Some(0))
668            .max_blocking_threads(0)
669            .build();
670
671        let error = result.unwrap_err().to_string();
672        assert!(error.contains("num_worker_threads: Validation error"));
673        assert!(error.contains("max_blocking_threads: Validation error"));
674        Ok(())
675    }
676
677    #[test]
678    fn test_system_server_enabled_by_nonnegative_port() {
679        let mut config = RuntimeConfig::default();
680        for (port, enabled) in [(-1, false), (0, true), (9527, true)] {
681            config.system_port = port;
682            assert_eq!(config.system_server_enabled(), enabled);
683        }
684    }
685
686    #[test]
687    fn test_is_truthy_and_falsey() {
688        // Test truthy values
689        assert!(is_truthy("1"));
690        assert!(is_truthy("true"));
691        assert!(is_truthy("TRUE"));
692        assert!(is_truthy("on"));
693        assert!(is_truthy("yes"));
694
695        // Test falsey values
696        assert!(is_falsey("0"));
697        assert!(is_falsey("false"));
698        assert!(is_falsey("FALSE"));
699        assert!(is_falsey("off"));
700        assert!(is_falsey("no"));
701
702        // Test opposite behavior
703        assert!(!is_truthy("0"));
704        assert!(!is_falsey("1"));
705    }
706
707    #[test]
708    fn test_console_log_format() {
709        use environment_names::logging;
710
711        for (console_format, legacy_jsonl, expected) in [
712            (None, None, ConsoleLogFormat::Readable),
713            (None, Some("true"), ConsoleLogFormat::Jsonl),
714            (Some(""), Some("true"), ConsoleLogFormat::Jsonl),
715            (Some("   "), Some("true"), ConsoleLogFormat::Jsonl),
716            (Some(" jsonl "), Some("false"), ConsoleLogFormat::Jsonl),
717            (Some("readable"), Some("true"), ConsoleLogFormat::Readable),
718            (Some("jsonl"), Some("false"), ConsoleLogFormat::Jsonl),
719            (
720                Some("unsupported"),
721                Some("true"),
722                ConsoleLogFormat::Readable,
723            ),
724        ] {
725            temp_env::with_vars(
726                [
727                    (logging::DYN_LOGGING_CONSOLE_FORMAT, console_format),
728                    (logging::DYN_LOGGING_JSONL, legacy_jsonl),
729                ],
730                || {
731                    assert_eq!(console_log_format(), expected);
732                    assert_eq!(jsonl_logging_enabled(), expected == ConsoleLogFormat::Jsonl);
733                },
734            );
735        }
736    }
737}