distributed.rs 25.7 KB
Newer Older
1
2
// SPDX-FileCopyrightText: Copyright (c) 2024-2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
// SPDX-License-Identifier: Apache-2.0
Ryan Olson's avatar
Ryan Olson committed
3

4
use crate::component::{Component, Instance};
5
use crate::pipeline::PipelineError;
6
use crate::pipeline::network::manager::NetworkManager;
7
use crate::service::{ServiceClient, ServiceSet};
8
use crate::storage::kv::{self, Store as _};
Ryan Olson's avatar
Ryan Olson committed
9
use crate::{
10
    component::{self, ComponentBuilder, Endpoint, Namespace},
11
    discovery::Discovery,
12
13
    metrics::PrometheusUpdateCallback,
    metrics::{MetricsHierarchy, MetricsRegistry},
Ryan Olson's avatar
Ryan Olson committed
14
15
    transports::{etcd, nats, tcp},
};
16
use crate::{discovery, system_status_server, transports};
Ryan Olson's avatar
Ryan Olson committed
17

18
use super::utils::GracefulShutdownTracker;
19
20
use crate::SystemHealth;
use crate::runtime::Runtime;
Ryan Olson's avatar
Ryan Olson committed
21

22
// Used instead of std::cell::OnceCell because get_or_try_init there is nightly
23
use async_once_cell::OnceCell;
24

25
use std::fmt;
26
use std::sync::{Arc, OnceLock, Weak};
27
use std::time::Duration;
28
use tokio::sync::watch::Receiver;
29
30

use anyhow::Result;
Ryan Olson's avatar
Ryan Olson committed
31
32
use derive_getters::Dissolve;
use figment::error;
33
34
use std::collections::HashMap;
use tokio::sync::Mutex;
35
use tokio_util::sync::CancellationToken;
Ryan Olson's avatar
Ryan Olson committed
36

37
38
type InstanceMap = HashMap<Endpoint, Weak<Receiver<Vec<Instance>>>>;

39
40
41
42
43
44
45
46
/// Distributed [Runtime] which provides access to shared resources across the cluster, this includes
/// communication protocols and transports.
#[derive(Clone)]
pub struct DistributedRuntime {
    // local runtime
    runtime: Runtime,

    nats_client: Option<transports::nats::Client>,
47
    store: kv::Manager,
48
    network_manager: Arc<NetworkManager>,
49
50
    tcp_server: Arc<OnceCell<Arc<transports::tcp::server::TcpStreamServer>>>,
    system_status_server: Arc<OnceLock<Arc<system_status_server::SystemStatusServerInfo>>>,
51
    request_plane: RequestPlaneMode,
52
53
54
55

    // Service discovery client
    discovery_client: Arc<dyn discovery::Discovery>,

56
57
58
59
    // Discovery metadata (only used for Kubernetes backend)
    // Shared with system status server to expose via /metadata endpoint
    discovery_metadata: Option<Arc<tokio::sync::RwLock<discovery::DiscoveryMetadata>>>,

60
61
62
63
64
65
66
    // local registry for components
    // the registry allows us to use share runtime resources across instances of the same component object.
    // take for example two instances of a client to the same remote component. The registry allows us to use
    // a single endpoint watcher for both clients, this keeps the number background tasking watching specific
    // paths in etcd to a minimum.
    component_registry: component::Registry,

67
    instance_sources: Arc<tokio::sync::Mutex<InstanceMap>>,
68
69
70
71

    // Health Status
    system_health: Arc<parking_lot::Mutex<SystemHealth>>,

72
73
74
    // Local endpoint registry for in-process calls
    local_endpoint_registry: crate::local_endpoint_registry::LocalEndpointRegistry,

75
76
    // This hierarchy's own metrics registry
    metrics_registry: MetricsRegistry,
77
78
79

    // Registry for /engine/* route callbacks
    engine_routes: crate::engine_routes::EngineRouteRegistry,
80
81
}

82
impl MetricsHierarchy for DistributedRuntime {
83
84
85
86
    fn basename(&self) -> String {
        "".to_string() // drt has no basename. Basename only begins with the Namespace.
    }

87
88
89
90
91
92
    fn parent_hierarchies(&self) -> Vec<&dyn MetricsHierarchy> {
        vec![] // drt is the root, so no parent hierarchies
    }

    fn get_metrics_registry(&self) -> &MetricsRegistry {
        &self.metrics_registry
93
94
95
    }
}

Ryan Olson's avatar
Ryan Olson committed
96
97
98
99
100
101
impl std::fmt::Debug for DistributedRuntime {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "DistributedRuntime")
    }
}

Ryan Olson's avatar
Ryan Olson committed
102
103
impl DistributedRuntime {
    pub async fn new(runtime: Runtime, config: DistributedConfig) -> Result<Self> {
104
        let (selected_kv_store, nats_config, request_plane) = config.dissolve();
Ryan Olson's avatar
Ryan Olson committed
105
106
107

        let runtime_clone = runtime.clone();

108
        let store = match selected_kv_store {
109
            kv::Selector::Etcd(etcd_config) => {
110
111
112
113
                let etcd_client = etcd::Client::new(*etcd_config, runtime_clone).await.inspect_err(|err|
                    // The returned error doesn't show because of a dropped runtime error, so
                    // log it first.
                    tracing::error!(%err, "Could not connect to etcd. Pass `--store-kv ..` to use a different backend or start etcd."))?;
114
                kv::Manager::etcd(etcd_client)
115
            }
116
            kv::Selector::File(root) => kv::Manager::file(runtime.primary_token(), root),
117
            kv::Selector::Memory => kv::Manager::memory(),
118
        };
Ryan Olson's avatar
Ryan Olson committed
119

120
121
122
123
        let nats_client = match nats_config {
            Some(nc) => Some(nc.connect().await?),
            None => None,
        };
Ryan Olson's avatar
Ryan Olson committed
124

125
        // Start system status server for health and metrics if enabled in configuration
126
127
128
129
130
131
132
133
        let config = crate::config::RuntimeConfig::from_settings().unwrap_or_default();
        // IMPORTANT: We must extract cancel_token from runtime BEFORE moving runtime into the struct below.
        // This is because after moving, runtime is no longer accessible in this scope (ownership rules).
        let cancel_token = if config.system_server_enabled() {
            Some(runtime.clone().child_token())
        } else {
            None
        };
134
135
        let starting_health_status = config.starting_health_status.clone();
        let use_endpoint_health_status = config.use_endpoint_health_status.clone();
136
137
        let health_endpoint_path = config.system_health_path.clone();
        let live_endpoint_path = config.system_live_path.clone();
138
        let system_health = Arc::new(parking_lot::Mutex::new(SystemHealth::new(
139
140
            starting_health_status,
            use_endpoint_health_status,
141
142
            health_endpoint_path,
            live_endpoint_path,
143
        )));
144

145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
        // Initialize discovery client based on backend configuration
        let discovery_backend =
            std::env::var("DYN_DISCOVERY_BACKEND").unwrap_or_else(|_| "kv_store".to_string());

        let (discovery_client, discovery_metadata) = match discovery_backend.as_str() {
            "kubernetes" => {
                tracing::info!("Initializing Kubernetes discovery backend");
                let metadata = Arc::new(tokio::sync::RwLock::new(
                    crate::discovery::DiscoveryMetadata::new(),
                ));
                let client = crate::discovery::KubeDiscoveryClient::new(
                    metadata.clone(),
                    runtime.primary_token(),
                )
                .await
                .inspect_err(
                    |err| tracing::error!(%err, "Failed to initialize Kubernetes discovery client"),
                )?;
                (Arc::new(client) as Arc<dyn Discovery>, Some(metadata))
            }
            _ => {
                tracing::info!("Initializing KV store discovery backend");
                use crate::discovery::KVStoreDiscovery;
                (
                    Arc::new(KVStoreDiscovery::new(
                        store.clone(),
                        runtime.primary_token(),
                    )) as Arc<dyn Discovery>,
                    None,
                )
            }
176
177
        };

178
        let component_registry = component::Registry::new();
179

180
181
182
183
184
185
186
187
        // NetworkManager for request plane
        let network_manager = NetworkManager::new(
            runtime.child_token(),
            nats_client.clone().map(|c| c.client().clone()),
            component_registry.clone(),
            request_plane,
        );

188
        let distributed_runtime = Self {
Ryan Olson's avatar
Ryan Olson committed
189
            runtime,
190
            store,
191
            network_manager: Arc::new(network_manager),
Ryan Olson's avatar
Ryan Olson committed
192
193
            nats_client,
            tcp_server: Arc::new(OnceCell::new()),
194
            system_status_server: Arc::new(OnceLock::new()),
195
            discovery_client,
196
            discovery_metadata,
197
            component_registry,
198
            instance_sources: Arc::new(Mutex::new(HashMap::new())),
199
            metrics_registry: crate::MetricsRegistry::new(),
200
            system_health,
201
            request_plane,
202
            local_endpoint_registry: crate::local_endpoint_registry::LocalEndpointRegistry::new(),
203
            engine_routes: crate::engine_routes::EngineRouteRegistry::new(),
204
205
        };

206
207
208
209
210
211
        // Initialize the uptime gauge in SystemHealth
        distributed_runtime
            .system_health
            .lock()
            .initialize_uptime_gauge(&distributed_runtime)?;

212
        // Handle system status server initialization
213
        if let Some(cancel_token) = cancel_token {
214
            // System server is enabled - start both the state and HTTP server
215
            let host = config.system_host.clone();
216
            let port = config.system_port as u16;
217

218
            // Start system status server (it creates SystemStatusState internally)
219
            match crate::system_status_server::spawn_system_status_server(
220
221
222
223
                &host,
                port,
                cancel_token,
                Arc::new(distributed_runtime.clone()),
224
                distributed_runtime.discovery_metadata.clone(),
225
226
227
            )
            .await
            {
228
                Ok((addr, handle)) => {
229
                    tracing::info!("System status server started successfully on {}", addr);
230

231
232
233
234
235
236
                    // Store system status server information
                    let system_status_server_info =
                        crate::system_status_server::SystemStatusServerInfo::new(
                            addr,
                            Some(handle),
                        );
237

238
                    // Initialize the system_status_server field
239
                    distributed_runtime
240
241
242
                        .system_status_server
                        .set(Arc::new(system_status_server_info))
                        .expect("System status server info should only be set once");
243
244
                }
                Err(e) => {
245
                    tracing::error!("System status server startup failed: {}", e);
246
                }
247
            }
248
        } else {
249
            // System server HTTP is disabled, but uptime metrics are still being tracked via SystemHealth
250
251
252
            tracing::debug!(
                "System status server HTTP endpoints disabled, but uptime metrics are being tracked"
            );
253
254
        }

255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
        // Start health check manager if enabled
        if config.health_check_enabled {
            let health_check_config = crate::health_check::HealthCheckConfig {
                canary_wait_time: std::time::Duration::from_secs(config.canary_wait_time_secs),
                request_timeout: std::time::Duration::from_secs(
                    config.health_check_request_timeout_secs,
                ),
            };

            // Start the health check manager (spawns per-endpoint monitoring tasks)
            match crate::health_check::start_health_check_manager(
                distributed_runtime.clone(),
                Some(health_check_config),
            )
            .await
            {
                Ok(()) => tracing::info!(
                    "Health check manager started (canary_wait_time: {}s, request_timeout: {}s)",
                    config.canary_wait_time_secs,
                    config.health_check_request_timeout_secs
                ),
                Err(e) => tracing::error!("Health check manager failed to start: {}", e),
            }
        }

280
        Ok(distributed_runtime)
Ryan Olson's avatar
Ryan Olson committed
281
282
283
    }

    pub async fn from_settings(runtime: Runtime) -> Result<Self> {
284
        let config = DistributedConfig::from_settings();
Ryan Olson's avatar
Ryan Olson committed
285
286
287
288
289
290
291
        Self::new(runtime, config).await
    }

    pub fn runtime(&self) -> &Runtime {
        &self.runtime
    }

292
293
294
295
    pub fn primary_token(&self) -> CancellationToken {
        self.runtime.primary_token()
    }

296
297
298
299
300
301
302
303
304
305
306
    // TODO: Don't hand out pointers, instead have methods to use the registry in friendly ways
    // (without being aware of async locks and so on)
    pub fn component_registry(&self) -> &component::Registry {
        &self.component_registry
    }

    // TODO: Don't hand out pointers, instead provide system health related services.
    pub fn system_health(&self) -> Arc<parking_lot::Mutex<SystemHealth>> {
        self.system_health.clone()
    }

307
308
309
310
311
312
313
    /// Get the local endpoint registry for in-process endpoint calls
    pub fn local_endpoint_registry(
        &self,
    ) -> &crate::local_endpoint_registry::LocalEndpointRegistry {
        &self.local_endpoint_registry
    }

314
315
316
317
318
    /// Get the engine route registry for registering custom /engine/* routes
    pub fn engine_routes(&self) -> &crate::engine_routes::EngineRouteRegistry {
        &self.engine_routes
    }

319
    pub fn connection_id(&self) -> u64 {
320
        self.discovery_client.instance_id()
Ryan Olson's avatar
Ryan Olson committed
321
322
323
324
    }

    pub fn shutdown(&self) {
        self.runtime.shutdown();
325
        self.store.shutdown();
Ryan Olson's avatar
Ryan Olson committed
326
327
328
329
    }

    /// Create a [`Namespace`]
    pub fn namespace(&self, name: impl Into<String>) -> Result<Namespace> {
330
        Namespace::new(self.clone(), name.into())
Ryan Olson's avatar
Ryan Olson committed
331
332
    }

333
334
335
    /// Returns the discovery interface for service registration and discovery
    pub fn discovery(&self) -> Arc<dyn Discovery> {
        self.discovery_client.clone()
336
337
    }

338
    pub async fn tcp_server(&self) -> Result<Arc<tcp::server::TcpStreamServer>> {
Ryan Olson's avatar
Ryan Olson committed
339
340
341
342
343
        Ok(self
            .tcp_server
            .get_or_try_init(async move {
                let options = tcp::server::ServerOptions::default();
                let server = tcp::server::TcpStreamServer::new(options).await?;
344
                Ok::<_, PipelineError>(server)
Ryan Olson's avatar
Ryan Olson committed
345
346
347
348
349
            })
            .await?
            .clone())
    }

350
    /// Get the network manager
351
352
353
    ///
    /// The network manager consolidates all network configuration and provides
    /// unified access to request plane servers and clients.
354
355
    pub fn network_manager(&self) -> Arc<NetworkManager> {
        self.network_manager.clone()
356
357
358
359
360
361
362
363
364
    }

    /// Get the request plane server (convenience method)
    ///
    /// This is a shortcut for `network_manager().await?.server().await`.
    pub async fn request_plane_server(
        &self,
    ) -> Result<Arc<dyn crate::pipeline::network::ingress::unified_server::RequestPlaneServer>>
    {
365
        self.network_manager().server().await
Ryan Olson's avatar
Ryan Olson committed
366
367
    }

368
369
370
371
372
    /// Get system status server information if available
    pub fn system_status_server_info(
        &self,
    ) -> Option<Arc<crate::system_status_server::SystemStatusServerInfo>> {
        self.system_status_server.get().cloned()
373
374
    }

375
    /// An interface to store things outside of the process. Usually backed by something like etcd.
376
    /// Currently does key-value, but will grow to include whatever we need to store.
377
    pub fn store(&self) -> &kv::Manager {
378
        &self.store
379
380
    }

381
382
383
384
385
    /// How the frontend should talk to the backend.
    pub fn request_plane(&self) -> RequestPlaneMode {
        self.request_plane
    }

386
387
388
    pub fn child_token(&self) -> CancellationToken {
        self.runtime.child_token()
    }
389

390
391
392
393
    pub(crate) fn graceful_shutdown_tracker(&self) -> Arc<GracefulShutdownTracker> {
        self.runtime.graceful_shutdown_tracker()
    }

394
    pub fn instance_sources(&self) -> Arc<Mutex<InstanceMap>> {
395
396
        self.instance_sources.clone()
    }
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422

    /// TODO: This is a temporary KV router measure for component/component.rs EventPublisher impl for
    /// Component, to allow it to publish to NATS. KV Router is the only user.
    pub(crate) async fn kv_router_nats_publish(
        &self,
        subject: String,
        payload: bytes::Bytes,
    ) -> anyhow::Result<()> {
        let Some(nats_client) = self.nats_client.as_ref() else {
            anyhow::bail!("KV router's EventPublisher requires NATS");
        };
        Ok(nats_client.client().publish(subject, payload).await?)
    }

    /// TODO: This is a temporary KV router measure for component/component.rs EventSubscriber impl for
    /// Component, to allow it to subscribe to NATS. KV Router is the only user.
    pub(crate) async fn kv_router_nats_subscribe(
        &self,
        subject: String,
    ) -> Result<async_nats::Subscriber> {
        let Some(nats_client) = self.nats_client.as_ref() else {
            anyhow::bail!("KV router's EventSubscriber requires NATS");
        };
        Ok(nats_client.client().subscribe(subject).await?)
    }

423
424
425
426
    /// DEPRECATED: This method exists only for NATS request plane support.
    /// Once everything uses the TCP request plane, this can be removed along with
    /// the NATS service registration infrastructure.
    pub fn register_nats_service(&self, component: Component) {
427
428
429
430
        let drt = self.clone();
        self.runtime().secondary().spawn(async move {
            let service_name = component.service_name();

431
432
433
434
435
436
437
438
439
440
441
442
443
444
            // Pre-check to save cost of creating the service, but don't hold the lock
            if drt
                .component_registry()
                .inner
                .lock()
                .await
                .services
                .contains_key(&service_name)
            {
                // The NATS service is per component, but it is called from `serve_endpoint`, and there
                // are often multiple endpoints for a component (e.g. `clear_kv_blocks` and `generate`).
                tracing::trace!("Service {service_name} already exists");
                return;
            }
445

446
447
448
449
450
451
452
453
454
455
            let Some(nats_client) = drt.nats_client.as_ref() else {
                tracing::error!("Cannot create NATS service without NATS.");
                return;
            };
            let description = None;
            let nats_service = match crate::component::service::build_nats_service(
                nats_client,
                &component,
                description,
            )
456
            .await
457
458
459
460
461
462
463
            {
                Ok(service) => service,
                Err(err) => {
                    tracing::error!(error = %err, component = service_name, "Failed to build NATS service");
                    return;
                }
            };
464

465
466
467
468
            let mut guard = drt.component_registry().inner.lock().await;
            if !guard.services.contains_key(&service_name) {
                // Normal case
                guard.services.insert(service_name.clone(), nats_service);
469

470
                tracing::info!("Added NATS service {service_name}");
471

472
473
474
475
476
477
478
                drop(guard);
            } else {
                drop(guard);
                let _ = nats_service.stop().await;
                // The NATS service is per component, but it is called from `serve_endpoint`, and there
                // are often multiple endpoints for a component (e.g. `clear_kv_blocks` and `generate`).
                // TODO: Is this still true?
479
            }
480
        });
481
    }
Ryan Olson's avatar
Ryan Olson committed
482
483
484
485
}

#[derive(Dissolve)]
pub struct DistributedConfig {
486
    pub store_backend: kv::Selector,
487
    pub nats_config: Option<nats::ClientOptions>,
488
    pub request_plane: RequestPlaneMode,
Ryan Olson's avatar
Ryan Olson committed
489
490
491
}

impl DistributedConfig {
492
    pub fn from_settings() -> DistributedConfig {
493
        let request_plane = RequestPlaneMode::from_env();
Ryan Olson's avatar
Ryan Olson committed
494
        DistributedConfig {
495
            store_backend: kv::Selector::Etcd(Box::default()),
496
497
498
499
500
501
            nats_config: if request_plane.is_nats() {
                Some(nats::ClientOptions::default())
            } else {
                None
            },
            request_plane,
Ryan Olson's avatar
Ryan Olson committed
502
503
        }
    }
Ryan Olson's avatar
Ryan Olson committed
504
505

    pub fn for_cli() -> DistributedConfig {
506
507
508
509
        let etcd_config = etcd::ClientOptions {
            attach_lease: false,
            ..Default::default()
        };
510
        let request_plane = RequestPlaneMode::from_env();
511
        DistributedConfig {
512
            store_backend: kv::Selector::Etcd(Box::new(etcd_config)),
513
514
515
516
517
518
            nats_config: if request_plane.is_nats() {
                Some(nats::ClientOptions::default())
            } else {
                None
            },
            request_plane,
519
520
        }
    }
521
522
523
524
525

    /// A DistributedConfig that isn't distributed, for when the frontend and backend are in the
    /// same process.
    pub fn process_local() -> DistributedConfig {
        DistributedConfig {
526
            store_backend: kv::Selector::Memory,
527
528
529
530
531
532
            nats_config: None,
            // This won't be used in process local, so we likely need a "none" option to
            // communicate that and avoid opening the ports.
            request_plane: RequestPlaneMode::Tcp,
        }
    }
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
}

/// Request plane transport mode configuration
///
/// This determines how requests are distributed from routers to workers:
/// - `Nats`: Use NATS for request distribution (default, legacy)
/// - `Http`: Use HTTP/2 for request distribution
/// - `Tcp`: Use raw TCP for request distribution with msgpack support
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RequestPlaneMode {
    /// Use NATS for request plane (default for backward compatibility)
    Nats,
    /// Use HTTP/2 for request plane
    Http,
    /// Use raw TCP for request plane with msgpack support
    Tcp,
}

impl Default for RequestPlaneMode {
    fn default() -> Self {
        Self::Nats
    }
}

impl fmt::Display for RequestPlaneMode {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Nats => write!(f, "nats"),
            Self::Http => write!(f, "http"),
            Self::Tcp => write!(f, "tcp"),
        }
    }
}

impl std::str::FromStr for RequestPlaneMode {
    type Err = anyhow::Error;

    fn from_str(s: &str) -> std::result::Result<Self, Self::Err> {
        match s.to_lowercase().as_str() {
            "nats" => Ok(Self::Nats),
            "http" => Ok(Self::Http),
            "tcp" => Ok(Self::Tcp),
            _ => Err(anyhow::anyhow!(
                "Invalid request plane mode: '{}'. Valid options are: 'nats', 'http', 'tcp'",
                s
            )),
579
        }
Ryan Olson's avatar
Ryan Olson committed
580
    }
Ryan Olson's avatar
Ryan Olson committed
581
}
582

583
584
585
586
587
588
589
590
591
impl RequestPlaneMode {
    /// Get the request plane mode from environment variable (uncached)
    /// Reads from `DYN_REQUEST_PLANE` environment variable.
    fn from_env() -> Self {
        std::env::var("DYN_REQUEST_PLANE")
            .ok()
            .and_then(|s| s.parse().ok())
            .unwrap_or_default()
    }
592
593
594
595

    pub fn is_nats(&self) -> bool {
        matches!(self, RequestPlaneMode::Nats)
    }
596
597
}

598
pub mod distributed_test_utils {
599
600
    //! Common test helper functions for DistributedRuntime tests

601
    /// Helper function to create a DRT instance for integration-only tests.
602
    /// Uses from_current to leverage existing tokio runtime
603
    /// Note: Settings are read from environment variables inside DistributedRuntime::from_settings
604
    #[cfg(feature = "integration")]
605
    pub async fn create_test_drt_async() -> super::DistributedRuntime {
606
        use crate::{storage::kv, transports::nats};
607

608
        let rt = crate::Runtime::from_current().unwrap();
609
        let config = super::DistributedConfig {
610
            store_backend: kv::Selector::Memory,
611
            nats_config: Some(nats::ClientOptions::default()),
612
            request_plane: crate::distributed::RequestPlaneMode::default(),
613
614
        };
        super::DistributedRuntime::new(rt, config).await.unwrap()
615
616
    }
}
617

618
#[cfg(all(test, feature = "integration"))]
619
mod tests {
620
    use super::RequestPlaneMode;
621
622
623
624
    use super::distributed_test_utils::create_test_drt_async;

    #[tokio::test]
    async fn test_drt_uptime_after_delay_system_disabled() {
625
        use crate::config::environment_names::runtime::system as env_system;
626
        // Test uptime with system status server disabled
627
        temp_env::async_with_vars([(env_system::DYN_SYSTEM_PORT, None::<&str>)], async {
628
629
630
631
632
633
634
            // Start a DRT
            let drt = create_test_drt_async().await;

            // Wait 50ms
            tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;

            // Check that uptime is 50+ ms
635
            let uptime = drt.system_health.lock().uptime();
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
            assert!(
                uptime >= std::time::Duration::from_millis(50),
                "Expected uptime to be at least 50ms, but got {:?}",
                uptime
            );

            println!(
                "✓ DRT uptime test passed (system disabled): uptime = {:?}",
                uptime
            );
        })
        .await;
    }

    #[tokio::test]
    async fn test_drt_uptime_after_delay_system_enabled() {
652
        use crate::config::environment_names::runtime::system as env_system;
653
        // Test uptime with system status server enabled
654
        temp_env::async_with_vars([(env_system::DYN_SYSTEM_PORT, Some("8081"))], async {
655
656
657
658
659
660
661
            // Start a DRT
            let drt = create_test_drt_async().await;

            // Wait 50ms
            tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;

            // Check that uptime is 50+ ms
662
            let uptime = drt.system_health.lock().uptime();
663
664
665
666
667
668
669
670
671
672
673
674
675
            assert!(
                uptime >= std::time::Duration::from_millis(50),
                "Expected uptime to be at least 50ms, but got {:?}",
                uptime
            );

            println!(
                "✓ DRT uptime test passed (system enabled): uptime = {:?}",
                uptime
            );
        })
        .await;
    }
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711

    #[test]
    fn test_request_plane_mode_from_str() {
        assert_eq!(
            "nats".parse::<RequestPlaneMode>().unwrap(),
            RequestPlaneMode::Nats
        );
        assert_eq!(
            "http".parse::<RequestPlaneMode>().unwrap(),
            RequestPlaneMode::Http
        );
        assert_eq!(
            "tcp".parse::<RequestPlaneMode>().unwrap(),
            RequestPlaneMode::Tcp
        );
        assert_eq!(
            "NATS".parse::<RequestPlaneMode>().unwrap(),
            RequestPlaneMode::Nats
        );
        assert_eq!(
            "HTTP".parse::<RequestPlaneMode>().unwrap(),
            RequestPlaneMode::Http
        );
        assert_eq!(
            "TCP".parse::<RequestPlaneMode>().unwrap(),
            RequestPlaneMode::Tcp
        );
        assert!("invalid".parse::<RequestPlaneMode>().is_err());
    }

    #[test]
    fn test_request_plane_mode_display() {
        assert_eq!(RequestPlaneMode::Nats.to_string(), "nats");
        assert_eq!(RequestPlaneMode::Http.to_string(), "http");
        assert_eq!(RequestPlaneMode::Tcp.to_string(), "tcp");
    }
712
}