Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
102 changes: 89 additions & 13 deletions lib/runtime/src/distributed.rs
Original file line number Diff line number Diff line change
Expand Up @@ -413,7 +413,7 @@ impl DistributedRuntime {
}

pub async fn from_settings(runtime: Runtime) -> Result<Self> {
let config = DistributedConfig::from_settings();
let config = DistributedConfig::try_from_settings()?;
Self::new(runtime, config).await
}

Expand Down Expand Up @@ -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<DistributedConfig> {
Self::from_settings_with_overrides(None, None, None)
.unwrap_or_else(|error| panic!("{error}"))
}

/// Resolve per-worker options before environment defaults, without mutating
Expand All @@ -852,7 +855,7 @@ impl DistributedConfig {
) -> Result<DistributedConfig> {
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.
Expand Down Expand Up @@ -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<DistributedConfig> {
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();
Expand All @@ -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())
Expand All @@ -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
Expand Down Expand Up @@ -1000,20 +1007,89 @@ 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> {
Self::from_env_result(std::env::var(
crate::config::environment_names::request_plane::DYN_REQUEST_PLANE,
))
}

fn from_env_result(value: std::result::Result<String, std::env::VarError>) -> Result<Self> {
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()
)),
}
Comment thread
MatejKosec marked this conversation as resolved.
}

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

#[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

Expand Down
Loading