diff --git a/docs/fern/pages/developer-guide/knowledge-base/concepts/system-architecture/architecture.md b/docs/fern/pages/developer-guide/knowledge-base/concepts/system-architecture/architecture.md index c762354a17c5..85a532593294 100644 --- a/docs/fern/pages/developer-guide/knowledge-base/concepts/system-architecture/architecture.md +++ b/docs/fern/pages/developer-guide/knowledge-base/concepts/system-architecture/architecture.md @@ -69,6 +69,11 @@ The request plane carries RPC traffic between Dynamo components. `DYN_REQUEST_PL - `tcp` is the default and uses direct pooled connections. - `nats` uses brokered request transport. +When the runtime reads this setting from its environment, an absent value selects +`tcp`. Every present value, including an empty value, must be `tcp` or `nats` +(case-insensitive); any other value stops startup instead of silently selecting +`tcp`. + `DYN_REQUEST_PLANE_CODEC` selects `msgpack` or `json`. The destination endpoint advertises its codec, so one client can communicate with endpoints that use different codecs. ### Event Plane diff --git a/lib/runtime/src/distributed.rs b/lib/runtime/src/distributed.rs index 45e9bed1d07a..be0a139e3642 100644 --- a/lib/runtime/src/distributed.rs +++ b/lib/runtime/src/distributed.rs @@ -413,7 +413,7 @@ impl DistributedRuntime { } pub async fn from_settings(runtime: Runtime) -> Result { - let config = DistributedConfig::from_settings(); + let config = DistributedConfig::try_from_settings()?; Self::new(runtime, config).await } @@ -839,8 +839,11 @@ impl DistributedConfig { /// # Panics /// Panics if a discovery or transport setting is invalid. pub fn from_settings() -> DistributedConfig { + Self::try_from_settings().unwrap_or_else(|err| panic!("{err}")) + } + + pub fn try_from_settings() -> Result { Self::from_settings_with_overrides(None, None, None) - .unwrap_or_else(|error| panic!("{error}")) } /// Resolve per-worker options before environment defaults, without mutating @@ -852,7 +855,7 @@ impl DistributedConfig { ) -> Result { let request_plane = match request_plane { Some(value) => value.parse()?, - None => RequestPlaneMode::from_env(), + None => RequestPlaneMode::from_env()?, }; // Determine the discovery backend first — we need it to compute the NATS default below. @@ -919,11 +922,15 @@ impl DistributedConfig { } pub fn for_cli() -> DistributedConfig { + Self::try_for_cli().unwrap_or_else(|err| panic!("{err}")) + } + + pub fn try_for_cli() -> Result { let etcd_config = etcd::ClientOptions { attach_lease: false, ..Default::default() }; - let request_plane = RequestPlaneMode::from_env(); + let request_plane = RequestPlaneMode::from_env()?; let discovery_backend = DiscoveryBackend::KvStore(kv::Selector::Etcd(Box::new(etcd_config))); let event_transport_kind = discovery_backend.resolve_event_transport_kind(); @@ -933,7 +940,7 @@ impl DistributedConfig { event_transport_kind, crate::discovery::EventTransportKind::Nats ); - DistributedConfig { + Ok(DistributedConfig { discovery_backend, nats_config: if nats_enabled { Some(nats::ClientOptions::default()) @@ -943,7 +950,7 @@ impl DistributedConfig { request_plane, response_plane: None, event_transport_kind, - } + }) } /// A DistributedConfig that isn't distributed, for when the frontend and backend are in the @@ -1000,13 +1007,22 @@ impl std::str::FromStr for RequestPlaneMode { } 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(crate::config::environment_names::request_plane::DYN_REQUEST_PLANE) - .ok() - .and_then(|s| s.parse().ok()) - .unwrap_or_default() + fn from_env() -> Result { + Self::from_env_result(std::env::var( + crate::config::environment_names::request_plane::DYN_REQUEST_PLANE, + )) + } + + fn from_env_result(value: std::result::Result) -> Result { + match value { + Err(std::env::VarError::NotPresent) => Ok(Self::default()), + Ok(s) => s.parse(), + Err(std::env::VarError::NotUnicode(raw)) => Err(anyhow::anyhow!( + "Invalid request plane mode: '{}' is not valid Unicode. \ + Valid options are: 'nats', 'tcp'", + raw.to_string_lossy() + )), + } } pub fn is_nats(&self) -> bool { @@ -1014,6 +1030,66 @@ impl RequestPlaneMode { } } +#[cfg(test)] +mod request_plane_env_tests { + use super::RequestPlaneMode; + + #[test] + fn absent_request_plane_defaults_to_tcp() { + let mode = RequestPlaneMode::from_env_result(Err(std::env::VarError::NotPresent)) + .expect("an absent DYN_REQUEST_PLANE must not be an error"); + assert_eq!(mode, RequestPlaneMode::Tcp); + } + + #[test] + fn empty_request_plane_is_an_error() { + RequestPlaneMode::from_env_result(Ok(String::new())) + .expect_err("an empty DYN_REQUEST_PLANE must not silently fall back to TCP"); + } + + #[test] + fn valid_request_plane_value_resolves() { + let mode = RequestPlaneMode::from_env_result(Ok("nats".to_string())) + .expect("DYN_REQUEST_PLANE=nats should resolve"); + assert_eq!(mode, RequestPlaneMode::Nats); + } + + #[test] + fn invalid_request_plane_is_an_error_naming_value_and_options() { + let err = RequestPlaneMode::from_env_result(Ok("nat".to_string())) + .expect_err("a misspelled DYN_REQUEST_PLANE must not silently fall back to TCP"); + let message = err.to_string(); + assert!( + message.contains("nat"), + "error should name the offending value, got: {message}" + ); + assert!( + message.contains("'nats'") && message.contains("'tcp'"), + "error should list the valid options, got: {message}" + ); + } + + #[cfg(unix)] + #[test] + fn non_unicode_request_plane_is_an_error_not_a_default() { + use std::ffi::OsString; + use std::os::unix::ffi::OsStringExt; + + let raw = OsString::from_vec(b"nat\xff".to_vec()); + let err = RequestPlaneMode::from_env_result(Err(std::env::VarError::NotUnicode(raw))) + .expect_err("a non-Unicode DYN_REQUEST_PLANE must not silently fall back to TCP"); + let message = err.to_string(); + assert!( + message.contains("not valid Unicode"), + "error should say the value was not valid Unicode, got: {message}" + ); + assert!( + message.contains("'nats'") && message.contains("'tcp'"), + "error should list the valid options, got: {message}" + ); + } +} + pub mod distributed_test_utils { //! Common test helper functions for DistributedRuntime tests