distributed.rs 23.9 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::storage::key_value_store::{
7
8
    EtcdStore, KeyValueStore, KeyValueStoreEnum, KeyValueStoreManager, KeyValueStoreSelect,
    MemoryStore,
9
};
10
use crate::transports::nats::DRTNatsClientPrometheusMetrics;
Ryan Olson's avatar
Ryan Olson committed
11
use crate::{
12
    component::{self, ComponentBuilder, Endpoint, Namespace},
13
    discovery::Discovery,
14
15
    metrics::PrometheusUpdateCallback,
    metrics::{MetricsHierarchy, MetricsRegistry},
Ryan Olson's avatar
Ryan Olson committed
16
17
18
    service::ServiceClient,
    transports::{etcd, nats, tcp},
};
19
use crate::{discovery, system_status_server, transports};
Ryan Olson's avatar
Ryan Olson committed
20

21
use super::utils::GracefulShutdownTracker;
22
23
use crate::SystemHealth;
use crate::runtime::Runtime;
Ryan Olson's avatar
Ryan Olson committed
24

25
use async_once_cell::OnceCell;
26
use std::fmt;
27
use std::sync::{Arc, OnceLock, Weak};
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
47
48
/// 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>,
    store: KeyValueStoreManager,
    tcp_server: Arc<OnceCell<Arc<transports::tcp::server::TcpStreamServer>>>,
49
    network_manager: Arc<OnceCell<Arc<crate::pipeline::network::manager::NetworkManager>>>,
50
    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
72
73
74
75

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

    // This hierarchy's own metrics registry
    metrics_registry: MetricsRegistry,
}

76
impl MetricsHierarchy for DistributedRuntime {
77
78
79
80
    fn basename(&self) -> String {
        "".to_string() // drt has no basename. Basename only begins with the Namespace.
    }

81
82
83
84
85
86
    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
87
88
89
    }
}

Ryan Olson's avatar
Ryan Olson committed
90
91
92
93
94
95
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
96
97
impl DistributedRuntime {
    pub async fn new(runtime: Runtime, config: DistributedConfig) -> Result<Self> {
98
        let (selected_kv_store, nats_config, request_plane) = config.dissolve();
Ryan Olson's avatar
Ryan Olson committed
99
100
101

        let runtime_clone = runtime.clone();

102
        let store = match selected_kv_store {
103
            KeyValueStoreSelect::Etcd(etcd_config) => {
104
105
106
107
                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."))?;
108
                KeyValueStoreManager::etcd(etcd_client)
109
            }
110
111
            KeyValueStoreSelect::File(root) => KeyValueStoreManager::file(root),
            KeyValueStoreSelect::Memory => KeyValueStoreManager::memory(),
112
        };
Ryan Olson's avatar
Ryan Olson committed
113

114
115
116
117
        let nats_client = match nats_config {
            Some(nc) => Some(nc.connect().await?),
            None => None,
        };
Ryan Olson's avatar
Ryan Olson committed
118

119
        // Start system status server for health and metrics if enabled in configuration
120
121
122
123
124
125
126
127
        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
        };
128
129
        let starting_health_status = config.starting_health_status.clone();
        let use_endpoint_health_status = config.use_endpoint_health_status.clone();
130
131
        let health_endpoint_path = config.system_health_path.clone();
        let live_endpoint_path = config.system_live_path.clone();
132
        let system_health = Arc::new(parking_lot::Mutex::new(SystemHealth::new(
133
134
            starting_health_status,
            use_endpoint_health_status,
135
136
            health_endpoint_path,
            live_endpoint_path,
137
        )));
138

139
140
        let nats_client_for_metrics = nats_client.clone();

141
142
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
        // 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,
                )
            }
172
173
        };

174
        let distributed_runtime = Self {
Ryan Olson's avatar
Ryan Olson committed
175
            runtime,
176
            store,
Ryan Olson's avatar
Ryan Olson committed
177
178
            nats_client,
            tcp_server: Arc::new(OnceCell::new()),
179
            network_manager: Arc::new(OnceCell::new()),
180
            system_status_server: Arc::new(OnceLock::new()),
181
            discovery_client,
182
            discovery_metadata,
Ryan Olson's avatar
Ryan Olson committed
183
            component_registry: component::Registry::new(),
184
            instance_sources: Arc::new(Mutex::new(HashMap::new())),
185
            metrics_registry: crate::MetricsRegistry::new(),
186
            system_health,
187
            request_plane,
188
189
        };

190
191
192
193
194
        if let Some(nats_client_for_metrics) = nats_client_for_metrics {
            let nats_client_metrics = DRTNatsClientPrometheusMetrics::new(
                &distributed_runtime,
                nats_client_for_metrics.client().clone(),
            )?;
195
            // Register a callback to update NATS client metrics on the DRT's metrics registry
196
197
198
199
200
201
202
203
            let nats_client_callback = Arc::new({
                let nats_client_clone = nats_client_metrics.clone();
                move || {
                    nats_client_clone.set_from_client_stats();
                    Ok(())
                }
            });
            distributed_runtime
204
205
                .metrics_registry
                .add_update_callback(nats_client_callback);
206
        }
207

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

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

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

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

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

282
        Ok(distributed_runtime)
Ryan Olson's avatar
Ryan Olson committed
283
284
285
    }

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

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

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

298
299
300
301
302
303
304
305
306
307
308
    // 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()
    }

309
    pub fn connection_id(&self) -> u64 {
310
        self.discovery_client.instance_id()
Ryan Olson's avatar
Ryan Olson committed
311
312
313
314
    }

    pub fn shutdown(&self) {
        self.runtime.shutdown();
315
        self.store.shutdown();
Ryan Olson's avatar
Ryan Olson committed
316
317
318
319
    }

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

323
324
325
    /// Returns the discovery interface for service registration and discovery
    pub fn discovery(&self) -> Arc<dyn Discovery> {
        self.discovery_client.clone()
326
327
    }

328
329
    pub(crate) fn service_client(&self) -> Option<ServiceClient> {
        self.nats_client().map(|nc| ServiceClient::new(nc.clone()))
Ryan Olson's avatar
Ryan Olson committed
330
331
    }

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

344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
    /// Get the network manager (lazy initialization)
    ///
    /// The network manager consolidates all network configuration and provides
    /// unified access to request plane servers and clients.
    pub async fn network_manager(
        &self,
    ) -> Result<Arc<crate::pipeline::network::manager::NetworkManager>> {
        use crate::pipeline::network::manager::NetworkManager;

        let manager = self
            .network_manager
            .get_or_try_init(async {
                // Get NATS client if available
                let nats_client = self.nats_client().map(|c| c.client().clone());

                // NetworkManager handles all config reading and mode selection
                anyhow::Ok(NetworkManager::new(
                    self.child_token(),
                    nats_client,
                    self.component_registry.clone(),
364
                    self.request_plane,
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
                ))
            })
            .await?;

        Ok(manager.clone())
    }

    /// 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>>
    {
        let manager = self.network_manager().await?;
        manager.server().await
    }

    /// DEPRECATED: Use network_manager().server() instead
    #[deprecated(note = "Use request_plane_server() or network_manager().server() instead")]
    pub async fn http_server(
        &self,
    ) -> Result<Arc<crate::pipeline::network::ingress::http_endpoint::SharedHttpServer>> {
        // For backward compatibility, try to downcast
        let _server = self.request_plane_server().await?;
        // This will only work if we're actually in HTTP mode
        // For now, just return an error suggesting the new API
        anyhow::bail!(
            "http_server() is deprecated. Use request_plane_server() instead, which returns a trait object that works with all transport types."
        )
    }

    /// DEPRECATED: Use network_manager().server() instead
    #[deprecated(note = "Use request_plane_server() or network_manager().server() instead")]
    pub async fn shared_tcp_server(
        &self,
    ) -> Result<Arc<crate::pipeline::network::ingress::shared_tcp_endpoint::SharedTcpServer>> {
        // For backward compatibility, try to downcast
        let _server = self.request_plane_server().await?;
        // This will only work if we're actually in TCP mode
        // For now, just return an error suggesting the new API
        anyhow::bail!(
            "shared_tcp_server() is deprecated. Use request_plane_server() instead, which returns a trait object that works with all transport types."
        )
    }

411
412
    pub fn nats_client(&self) -> Option<&nats::Client> {
        self.nats_client.as_ref()
Ryan Olson's avatar
Ryan Olson committed
413
414
    }

415
416
417
418
419
    /// 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()
420
421
    }

422
    /// An interface to store things outside of the process. Usually backed by something like etcd.
423
    /// Currently does key-value, but will grow to include whatever we need to store.
424
425
    pub fn store(&self) -> &KeyValueStoreManager {
        &self.store
426
427
    }

428
429
430
431
432
    /// How the frontend should talk to the backend.
    pub fn request_plane(&self) -> RequestPlaneMode {
        self.request_plane
    }

433
434
435
    pub fn child_token(&self) -> CancellationToken {
        self.runtime.child_token()
    }
436

437
438
439
440
    pub(crate) fn graceful_shutdown_tracker(&self) -> Arc<GracefulShutdownTracker> {
        self.runtime.graceful_shutdown_tracker()
    }

441
    pub fn instance_sources(&self) -> Arc<Mutex<InstanceMap>> {
442
443
        self.instance_sources.clone()
    }
Ryan Olson's avatar
Ryan Olson committed
444
445
446
447
}

#[derive(Dissolve)]
pub struct DistributedConfig {
448
    pub store_backend: KeyValueStoreSelect,
449
    pub nats_config: Option<nats::ClientOptions>,
450
    pub request_plane: RequestPlaneMode,
Ryan Olson's avatar
Ryan Olson committed
451
452
453
}

impl DistributedConfig {
454
    pub fn from_settings() -> DistributedConfig {
455
        let request_plane = RequestPlaneMode::from_env();
Ryan Olson's avatar
Ryan Olson committed
456
        DistributedConfig {
457
            store_backend: KeyValueStoreSelect::Etcd(Box::default()),
458
459
460
461
462
463
            nats_config: if request_plane.is_nats() {
                Some(nats::ClientOptions::default())
            } else {
                None
            },
            request_plane,
Ryan Olson's avatar
Ryan Olson committed
464
465
        }
    }
Ryan Olson's avatar
Ryan Olson committed
466
467

    pub fn for_cli() -> DistributedConfig {
468
469
470
471
        let etcd_config = etcd::ClientOptions {
            attach_lease: false,
            ..Default::default()
        };
472
        let request_plane = RequestPlaneMode::from_env();
473
474
        DistributedConfig {
            store_backend: KeyValueStoreSelect::Etcd(Box::new(etcd_config)),
475
476
477
478
479
480
            nats_config: if request_plane.is_nats() {
                Some(nats::ClientOptions::default())
            } else {
                None
            },
            request_plane,
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
        }
    }
}

/// 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
            )),
529
        }
Ryan Olson's avatar
Ryan Olson committed
530
    }
Ryan Olson's avatar
Ryan Olson committed
531
}
532

533
534
535
536
537
538
539
540
541
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()
    }
542
543
544
545

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

548
pub mod distributed_test_utils {
549
550
    //! Common test helper functions for DistributedRuntime tests

551
    /// Helper function to create a DRT instance for integration-only tests.
552
    /// Uses from_current to leverage existing tokio runtime
553
    /// Note: Settings are read from environment variables inside DistributedRuntime::from_settings
554
    #[cfg(feature = "integration")]
555
556
557
    pub async fn create_test_drt_async() -> super::DistributedRuntime {
        use crate::{storage::key_value_store::KeyValueStoreSelect, transports::nats};

558
        let rt = crate::Runtime::from_current().unwrap();
559
560
        let config = super::DistributedConfig {
            store_backend: KeyValueStoreSelect::Memory,
561
            nats_config: Some(nats::ClientOptions::default()),
562
            request_plane: crate::distributed::RequestPlaneMode::default(),
563
564
        };
        super::DistributedRuntime::new(rt, config).await.unwrap()
565
566
    }
}
567

568
#[cfg(all(test, feature = "integration"))]
569
mod tests {
570
    use super::RequestPlaneMode;
571
572
573
574
    use super::distributed_test_utils::create_test_drt_async;

    #[tokio::test]
    async fn test_drt_uptime_after_delay_system_disabled() {
575
        use crate::config::environment_names::runtime::system as env_system;
576
        // Test uptime with system status server disabled
577
        temp_env::async_with_vars([(env_system::DYN_SYSTEM_PORT, None::<&str>)], async {
578
579
580
581
582
583
584
            // 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
585
            let uptime = drt.system_health.lock().uptime();
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
            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() {
602
        use crate::config::environment_names::runtime::system as env_system;
603
        // Test uptime with system status server enabled
604
        temp_env::async_with_vars([(env_system::DYN_SYSTEM_PORT, Some("8081"))], async {
605
606
607
608
609
610
611
            // 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
612
            let uptime = drt.system_health.lock().uptime();
613
614
615
616
617
618
619
620
621
622
623
624
625
            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;
    }
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661

    #[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");
    }
662
}