1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
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
172
173
174
175
176
177
178
179
180
//! v2 broker HTTP port mode resolution (slice 9 of #488).
//!
//! Implements the three broker port modes from #483 §3 plus the Docker
//! env overrides from #483 §3 ("Env override for Docker / sidecar
//! deployments"). Single resolution point — `BrokerHttpPort::resolve` —
//! so the env-override surface is visible exactly once in the code path
//! and the rest of the broker handles only the resolved enum.
use std::net::{IpAddr, Ipv4Addr};
/// Broker HTTP port mode declared in `BrokerConfig`.
///
/// Per #483 §3, the v2 broker's HTTP server picks its port via one of
/// these strategies. `BrokerHttpPort::resolve` overlays the env vars
/// from §3's table so container deployments can pin the port from
/// outside the binary.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum BrokerHttpPort {
/// Bind exactly this port; fail if unavailable.
Static {
/// The port the operator wants.
port: u16,
},
/// Always bind whatever the OS gives us (`bind(0)` semantics).
Dynamic,
/// Try `preferred`; if EADDRINUSE, fall back to OS-allocated.
StaticOrFallback {
/// The preferred port. Falls back to OS-allocated when taken.
preferred: u16,
},
}
/// Env var that overrides the configured port — when set & parseable,
/// resolution collapses to [`BrokerHttpPort::Static`] regardless of
/// the surrounding `BrokerConfig`.
pub const PORT_OVERRIDE_ENV: &str = "RUNNING_PROCESS_BROKER_HTTP_PORT";
/// Env var that overrides the bound IP. Defaults to `127.0.0.1`.
pub const BIND_OVERRIDE_ENV: &str = "RUNNING_PROCESS_BROKER_HTTP_BIND";
/// Resolved bind state — single source of truth for the rest of the
/// broker after [`BrokerHttpPort::resolve`] runs.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct ResolvedHttpBind {
/// The port mode after env override.
pub port: BrokerHttpPort,
/// The IP to bind on (defaults to loopback).
pub addr: IpAddr,
}
impl BrokerHttpPort {
/// Resolve config + env into the canonical bind state.
///
/// Precedence:
/// 1. If `RUNNING_PROCESS_BROKER_HTTP_PORT` is set and parses as a
/// `u16` → return [`BrokerHttpPort::Static`] for the override
/// (no silent fallback — defeating the container port-mapping
/// is the user's whole reason for setting it).
/// 2. Otherwise → return `config` unchanged.
/// 3. If `RUNNING_PROCESS_BROKER_HTTP_BIND` is set and parses as
/// an `IpAddr` → use that; otherwise default `127.0.0.1`.
/// 4. Empty / invalid env values are treated as unset (config wins).
pub fn resolve(config: BrokerHttpPort) -> ResolvedHttpBind {
let port = match parse_port_env() {
Some(p) => BrokerHttpPort::Static { port: p },
None => config,
};
let addr = parse_bind_env().unwrap_or(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)));
ResolvedHttpBind { port, addr }
}
}
fn parse_port_env() -> Option<u16> {
// Port 0 is honoured: it is how a caller asks for an ephemeral port.
crate::env_vars::BROKER_HTTP_PORT.port()
}
fn parse_bind_env() -> Option<IpAddr> {
crate::env_vars::BROKER_HTTP_BIND
.text()?
.trim()
.parse::<IpAddr>()
.ok()
}
#[cfg(test)]
mod tests {
use super::*;
use std::env;
use std::sync::Mutex;
// The env mutation tests share global state (`std::env`). Serialize
// them through a mutex so parallel test threads can't trample each
// other's env state.
static ENV_LOCK: Mutex<()> = Mutex::new(());
fn with_env<F: FnOnce()>(port: Option<&str>, bind: Option<&str>, f: F) {
let _g = ENV_LOCK.lock().expect("env mutex poisoned");
// Save + clear.
let prev_port = env::var(PORT_OVERRIDE_ENV).ok();
let prev_bind = env::var(BIND_OVERRIDE_ENV).ok();
match port {
Some(p) => env::set_var(PORT_OVERRIDE_ENV, p),
None => env::remove_var(PORT_OVERRIDE_ENV),
}
match bind {
Some(b) => env::set_var(BIND_OVERRIDE_ENV, b),
None => env::remove_var(BIND_OVERRIDE_ENV),
}
f();
// Restore.
match prev_port {
Some(p) => env::set_var(PORT_OVERRIDE_ENV, p),
None => env::remove_var(PORT_OVERRIDE_ENV),
}
match prev_bind {
Some(b) => env::set_var(BIND_OVERRIDE_ENV, b),
None => env::remove_var(BIND_OVERRIDE_ENV),
}
}
#[test]
fn no_env_returns_config_and_loopback_default() {
with_env(None, None, || {
let r = BrokerHttpPort::resolve(BrokerHttpPort::Dynamic);
assert_eq!(r.port, BrokerHttpPort::Dynamic);
assert_eq!(r.addr, IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)));
});
}
#[test]
fn port_env_set_overrides_to_static() {
with_env(Some("8080"), None, || {
let r = BrokerHttpPort::resolve(BrokerHttpPort::StaticOrFallback { preferred: 12_345 });
assert_eq!(r.port, BrokerHttpPort::Static { port: 8080 });
});
}
#[test]
fn bind_env_set_overrides_addr() {
with_env(None, Some("0.0.0.0"), || {
let r = BrokerHttpPort::resolve(BrokerHttpPort::Static { port: 4242 });
assert_eq!(r.port, BrokerHttpPort::Static { port: 4242 });
assert_eq!(r.addr, IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)));
});
}
#[test]
fn invalid_port_env_falls_back_to_config() {
with_env(Some("not-a-port"), None, || {
let r = BrokerHttpPort::resolve(BrokerHttpPort::Dynamic);
assert_eq!(r.port, BrokerHttpPort::Dynamic);
});
}
#[test]
fn empty_port_env_falls_back_to_config() {
with_env(Some(""), None, || {
let r = BrokerHttpPort::resolve(BrokerHttpPort::Dynamic);
assert_eq!(r.port, BrokerHttpPort::Dynamic);
});
}
#[test]
fn invalid_bind_env_falls_back_to_loopback() {
with_env(None, Some("not-an-ip"), || {
let r = BrokerHttpPort::resolve(BrokerHttpPort::Dynamic);
assert_eq!(r.addr, IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)));
});
}
#[test]
fn both_env_overrides_compose() {
with_env(Some("9999"), Some("0.0.0.0"), || {
let r = BrokerHttpPort::resolve(BrokerHttpPort::Dynamic);
assert_eq!(r.port, BrokerHttpPort::Static { port: 9999 });
assert_eq!(r.addr, IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)));
});
}
}