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
3 changes: 1 addition & 2 deletions components/src/dynamo/common/utils/runtime.py
Original file line number Diff line number Diff line change
Expand Up @@ -55,8 +55,7 @@ def create_runtime(
discovery_backend: Discovery backend type (kubernetes, etcd, file, mem).
request_plane: Request distribution method (nats, tcp).
event_plane: Event publishing method (nats, zmq). When None, the Rust
runtime auto-detects from the discovery backend (ZMQ for file/mem,
NATS for etcd/kubernetes).
runtime reads DYN_EVENT_PLANE and defaults to ZMQ.
use_kv_events: Deprecated. NATS enablement is now determined automatically
from the event-plane configuration. This parameter is accepted for
backwards compatibility but will be removed in a future release.
Expand Down
5 changes: 1 addition & 4 deletions lib/bindings/python/rust/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1416,7 +1416,6 @@ impl DistributedRuntime {
};
let request_plane: RequestPlaneMode = request_plane.parse().map_err(to_pyerr)?;
let response_plane = resolve_response_plane_mode(response_plane.as_deref())?;
let explicit_event_plane = event_plane.as_deref().filter(|value| !value.is_empty());
let event_transport_kind =
resolve_event_transport_kind(&discovery_backend_config, event_plane.as_deref())?;

Expand Down Expand Up @@ -1448,9 +1447,7 @@ impl DistributedRuntime {
|| matches!(
event_transport_kind,
dynamo_runtime::discovery::EventTransportKind::Nats
)
|| (explicit_event_plane.is_none()
&& std::env::var(config::environment_names::nats::NATS_SERVER).is_ok());
);

let runtime_config = DistributedConfig {
discovery_backend: discovery_backend_config,
Expand Down
13 changes: 13 additions & 0 deletions lib/bindings/python/tests/test_deprecated_enable_nats.py
Original file line number Diff line number Diff line change
Expand Up @@ -130,6 +130,19 @@ async def _run():
asyncio.run(_run())


@pytest.mark.forked
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved
def test_default_event_plane_ignores_ambient_nats(monkeypatch):
"""Without event_plane, file/TCP mode defaults to ZMQ and must not connect to ambient NATS."""
Comment thread
alec-flowers marked this conversation as resolved.
Resolved

async def _run():
monkeypatch.setenv("NATS_SERVER", "nats://127.0.0.1:9")
monkeypatch.delenv("DYN_EVENT_PLANE", raising=False)
runtime = DistributedRuntime(asyncio.get_running_loop(), "file", "tcp")
runtime.shutdown()

asyncio.run(_run())


@pytest.mark.forked
def test_invalid_event_plane_errors():
async def _run():
Expand Down
38 changes: 28 additions & 10 deletions lib/runtime/src/distributed.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1012,16 +1012,9 @@ impl DistributedConfig {
None => discovery_backend.resolve_event_transport_kind()?,
};

// NATS is used for more than just NATS request-plane RPC:
// - KV router events (NATS core event plane)
// - inter-router replica sync (NATS core)
//
// Enable the NATS client when any of these hold:
// 1. Request plane is NATS
// 2. NATS_SERVER is explicitly configured by the user
// 3. The resolved event transport kind is NATS
// Connect to NATS only when a plane uses it. NATS_SERVER alone says where NATS is, not
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved
// that this process needs it: the operator sets it in every pod when NATS is installed.
let nats_enabled = request_plane.is_nats()
|| std::env::var(crate::config::environment_names::nats::NATS_SERVER).is_ok()
|| matches!(
event_transport_kind,
crate::discovery::EventTransportKind::Nats
Expand Down Expand Up @@ -1054,7 +1047,6 @@ impl DistributedConfig {
DiscoveryBackend::KvStore(kv::Selector::Etcd(Box::new(etcd_config)));
let event_transport_kind = discovery_backend.resolve_event_transport_kind()?;
let nats_enabled = request_plane.is_nats()
|| std::env::var(crate::config::environment_names::nats::NATS_SERVER).is_ok()
|| matches!(
event_transport_kind,
crate::discovery::EventTransportKind::Nats
Expand Down Expand Up @@ -1268,6 +1260,32 @@ mod event_plane_env_tests {
}
}

#[cfg(test)]
mod nats_enablement_tests {
use super::DistributedConfig;
use crate::config::environment_names::{event_plane::DYN_EVENT_PLANE, nats::NATS_SERVER};

#[test]
fn ambient_nats_server_does_not_enable_nats_on_tcp_and_zmq() {
if crate::test_utils::run_isolated(
concat!(
module_path!(),
"::ambient_nats_server_does_not_enable_nats_on_tcp_and_zmq"
),
&[(NATS_SERVER, "nats://127.0.0.1:9")],
) {
return;
}

temp_env::with_var_unset(DYN_EVENT_PLANE, || {
let config =
DistributedConfig::from_settings_with_overrides(Some("file"), Some("tcp"), None)
.unwrap();
assert!(config.nats_config.is_none());
});
}
}

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

Expand Down
Loading