pb_mapper_core/
config.rs

1use std::net::{AddrParseError, SocketAddr};
2use std::sync::{Arc, Once};
3use std::time::Duration;
4
5use clap::ValueEnum;
6use snafu::ResultExt;
7use tracing_subscriber::layer::SubscriberExt;
8use tracing_subscriber::{EnvFilter, Layer, fmt};
9
10use crate::error::{CfgPbServerEnvNotExistSnafu, Result};
11
12#[derive(ValueEnum, Debug, Clone, Copy)]
13pub enum StatusOp {
14    /// Get active remote connection IDs.
15    RemoteId,
16    /// Get registered service keys.
17    Keys,
18}
19
20/// Every address a name resolved to, in the order the resolver returned them.
21///
22/// Resolution used to end in `.next()`, which threw away everything after the
23/// first candidate. That is enough for an `ip:port` literal, but a hostname with
24/// several A/AAAA records — a relay behind round-robin DNS, or a dual-stack host
25/// whose first record is an unreachable IPv6 address — resolved to one address
26/// and then failed against it, with the working candidates never tried.
27///
28/// The dial loops below this were always able to try every address: both
29/// `each_addr` and the `uni-stream` providers iterate a `ToSocketAddrs` and only
30/// report the last error once every candidate has failed. They just never
31/// received more than one. So this type is the whole of the fix: carry the list
32/// to them, and they do the rest.
33///
34/// Guaranteed non-empty, so a caller never has to handle a resolution that
35/// succeeded with nothing in it. Cheap to clone — the addresses are shared, not
36/// copied — because every tunnel worker and every retry needs its own handle.
37#[derive(Clone, Debug, Eq, PartialEq)]
38pub struct ResolvedAddrs {
39    /// Non-empty by construction; see [`ResolvedAddrs::new`].
40    addrs: Arc<[SocketAddr]>,
41}
42
43impl ResolvedAddrs {
44    /// Build from resolver output, rejecting an empty result.
45    ///
46    /// `name` is what was resolved, and `parse_error` the failure from parsing it
47    /// as a literal `ip:port`, so an empty resolution reports the same error a
48    /// malformed address would.
49    fn new(name: &str, addrs: Vec<SocketAddr>, parse_error: AddrParseError) -> Result<Self> {
50        if addrs.is_empty() {
51            return Err(crate::error::Error::CfgParseSockAddr {
52                string: name.to_string(),
53                source: parse_error,
54            });
55        }
56        Ok(Self {
57            addrs: Arc::from(addrs),
58        })
59    }
60
61    /// Build from candidates already resolved elsewhere, rejecting an empty list.
62    ///
63    /// For callers that resolve through another address trait — the tunnel
64    /// internals take a generic `ToSocketAddrs` at their public boundary — and
65    /// need the result in this non-empty form.
66    #[must_use]
67    pub fn from_candidates(addrs: Vec<SocketAddr>) -> Option<Self> {
68        if addrs.is_empty() {
69            return None;
70        }
71        Some(Self {
72            addrs: Arc::from(addrs),
73        })
74    }
75
76    /// The candidates, for handing to `each_addr` or a `ToSocketAddrs` bound.
77    ///
78    /// `&[SocketAddr]` is what both address traits in play accept, and it is
79    /// `Copy`, which the `A: ToSocketAddrs + Copy` bounds on the tunnel internals
80    /// require. A `Vec` is neither.
81    #[inline]
82    #[must_use]
83    pub fn as_slice(&self) -> &[SocketAddr] {
84        &self.addrs
85    }
86
87    /// The first candidate.
88    ///
89    /// For the places that genuinely need one address rather than a list: a
90    /// `SocketAddr` field on a status record, a preflight probe, a log line. Not
91    /// for dialling — that is what [`Self::as_slice`] is for.
92    #[inline]
93    #[must_use]
94    pub fn primary(&self) -> SocketAddr {
95        // Non-empty by construction, so indexing cannot panic. `[0]` rather than
96        // `first().unwrap()`: the invariant is the reason, and `unwrap` is denied.
97        self.addrs[0]
98    }
99}
100
101impl std::fmt::Display for ResolvedAddrs {
102    /// Renders the primary address, with a count when candidates were dropped.
103    ///
104    /// Log lines pass these through `%`, and a bare Debug dump of a one-element
105    /// list reads worse than the address itself.
106    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
107        write!(formatter, "{}", self.primary())?;
108        match self.addrs.len() {
109            1 => Ok(()),
110            more => write!(formatter, " (+{} more)", more - 1),
111        }
112    }
113}
114
115impl From<SocketAddr> for ResolvedAddrs {
116    fn from(addr: SocketAddr) -> Self {
117        Self {
118            addrs: Arc::from(vec![addr]),
119        }
120    }
121}
122
123/// Resolve `addr` to every candidate it names.
124///
125/// A literal `ip:port` short-circuits. Otherwise `localhost:` goes to the system
126/// resolver — the custom DNS path does not answer for it — and any other hostname
127/// tries the configured DNS servers first, falling back to the system resolver.
128#[inline]
129pub fn resolve_addrs(addr: &str) -> Result<ResolvedAddrs> {
130    // The literal case returns here; otherwise the parse error is kept as the
131    // failure reported below, so a caller sees what it asked for rather than a
132    // resolver's internal complaint.
133    let parse_error = match addr.parse::<SocketAddr>() {
134        Ok(socket_addr) => return Ok(ResolvedAddrs::from(socket_addr)),
135        Err(error) => error,
136    };
137
138    let system = || match std::net::ToSocketAddrs::to_socket_addrs(addr) {
139        Ok(addrs) => addrs.collect(),
140        Err(_) => Vec::new(),
141    };
142
143    let addrs = if addr.starts_with("localhost:") {
144        system()
145    } else {
146        match crate::addr::get_socket_addrs(addr) {
147            Ok(addrs) if !addrs.is_empty() => addrs,
148            _ => system(),
149        }
150    };
151    ResolvedAddrs::new(addr, addrs, parse_error)
152}
153
154/// Async counterpart of [`resolve_addrs`], for Tokio contexts.
155pub async fn resolve_addrs_async(addr: &str) -> Result<ResolvedAddrs> {
156    let parse_error = match addr.parse::<SocketAddr>() {
157        Ok(socket_addr) => return Ok(ResolvedAddrs::from(socket_addr)),
158        Err(error) => error,
159    };
160
161    async fn system(addr: &str) -> Vec<SocketAddr> {
162        match tokio::net::lookup_host(addr).await {
163            Ok(addrs) => addrs.collect(),
164            Err(_) => Vec::new(),
165        }
166    }
167
168    let addrs = if addr.starts_with("localhost:") {
169        system(addr).await
170    } else {
171        match crate::addr::get_socket_addrs_async(addr).await {
172            Ok(addrs) if !addrs.is_empty() => addrs,
173            _ => system(addr).await,
174        }
175    };
176    ResolvedAddrs::new(addr, addrs, parse_error)
177}
178
179const PB_MAPPER_SERVER: &str = "PB_MAPPER_SERVER";
180
181/// Env to control whether the keep-alive option of TCP is enabled
182pub const PB_MAPPER_KEEP_ALIVE: &str = "PB_MAPPER_KEEP_ALIVE";
183pub const PB_MAPPER_CONTROL_IO_TIMEOUT: &str = "PB_MAPPER_CONTROL_IO_TIMEOUT";
184pub const PB_MAPPER_STREAM_ACK_TIMEOUT: &str = "PB_MAPPER_STREAM_ACK_TIMEOUT";
185pub const PB_MAPPER_STREAM_READY_TIMEOUT: &str = "PB_MAPPER_STREAM_READY_TIMEOUT";
186pub const PB_MAPPER_STREAM_RECOVERY_TIMEOUT: &str = "PB_MAPPER_STREAM_RECOVERY_TIMEOUT";
187pub const PB_MAPPER_CONTROL_CONN_POOL_SIZE: &str = "PB_MAPPER_CONTROL_CONN_POOL_SIZE";
188pub const PB_MAPPER_CONTROL_HEARTBEAT_INTERVAL: &str = "PB_MAPPER_CONTROL_HEARTBEAT_INTERVAL";
189pub const PB_MAPPER_CONTROL_HEARTBEAT_TOLERANCE: &str = "PB_MAPPER_CONTROL_HEARTBEAT_TOLERANCE";
190pub const PB_MAPPER_CONTROL_SUSPECT_GRACE: &str = "PB_MAPPER_CONTROL_SUSPECT_GRACE";
191pub const PB_MAPPER_REGISTRATION_PROBE_TIMEOUT: &str = "PB_MAPPER_REGISTRATION_PROBE_TIMEOUT";
192pub const PB_MAPPER_REGISTRATION_REJECT_BACKOFF_MIN: &str =
193    "PB_MAPPER_REGISTRATION_REJECT_BACKOFF_MIN";
194pub const PB_MAPPER_REGISTRATION_REJECT_BACKOFF_MAX: &str =
195    "PB_MAPPER_REGISTRATION_REJECT_BACKOFF_MAX";
196pub const PB_MAPPER_SERVER_LEASE_TIMEOUT: &str = "PB_MAPPER_SERVER_LEASE_TIMEOUT";
197pub const PB_MAPPER_SERVER_LEASE_SWEEP_INTERVAL: &str = "PB_MAPPER_SERVER_LEASE_SWEEP_INTERVAL";
198pub const PB_MAPPER_CLIENT_HEALTH_CHECK_INTERVAL: &str = "PB_MAPPER_CLIENT_HEALTH_CHECK_INTERVAL";
199pub const PB_MAPPER_CLIENT_HEALTH_CHECK_TIMEOUT: &str = "PB_MAPPER_CLIENT_HEALTH_CHECK_TIMEOUT";
200pub const PB_MAPPER_CLIENT_HEALTH_FAILURE_THRESHOLD: &str =
201    "PB_MAPPER_CLIENT_HEALTH_FAILURE_THRESHOLD";
202pub const PB_MAPPER_LOG_FORMAT: &str = "PB_MAPPER_LOG_FORMAT";
203const DEFAULT_CONTROL_IO_TIMEOUT: Duration = Duration::from_secs(30);
204const DEFAULT_STREAM_ACK_TIMEOUT: Duration = Duration::from_millis(300);
205const DEFAULT_STREAM_READY_TIMEOUT: Duration = Duration::from_secs(1);
206const DEFAULT_STREAM_RECOVERY_TIMEOUT: Duration = Duration::from_secs(2);
207const DEFAULT_CONTROL_CONN_POOL_SIZE: usize = 2;
208const DEFAULT_CONTROL_HEARTBEAT_INTERVAL: Duration = Duration::from_secs(2);
209const DEFAULT_CONTROL_HEARTBEAT_TOLERANCE: Duration = Duration::from_secs(6);
210const DEFAULT_CONTROL_SUSPECT_GRACE: Duration = Duration::from_secs(2);
211const DEFAULT_REGISTRATION_PROBE_TIMEOUT: Duration = Duration::from_secs(1);
212const DEFAULT_REGISTRATION_REJECT_BACKOFF_MIN: Duration = Duration::from_secs(5);
213const DEFAULT_REGISTRATION_REJECT_BACKOFF_MAX: Duration = Duration::from_secs(80);
214const DEFAULT_SERVER_LEASE_TIMEOUT: Duration = Duration::from_secs(15);
215const DEFAULT_SERVER_LEASE_SWEEP_INTERVAL: Duration = Duration::from_secs(5);
216const DEFAULT_CLIENT_HEALTH_CHECK_INTERVAL: Duration = Duration::from_secs(15);
217const DEFAULT_CLIENT_HEALTH_CHECK_TIMEOUT: Duration = Duration::from_secs(5);
218const DEFAULT_CLIENT_HEALTH_FAILURE_THRESHOLD: usize = 3;
219
220#[derive(Debug, Clone, Copy, PartialEq, Eq)]
221pub enum LogFormat {
222    Pretty,
223    Compact,
224    Json,
225}
226
227pub fn parse_log_format(value: &str) -> LogFormat {
228    match value.trim().to_ascii_lowercase().as_str() {
229        "compact" => LogFormat::Compact,
230        "json" => LogFormat::Json,
231        _ => LogFormat::Pretty,
232    }
233}
234
235fn log_format_from_env() -> LogFormat {
236    std::env::var(PB_MAPPER_LOG_FORMAT)
237        .ok()
238        .map(|value| parse_log_format(&value))
239        .unwrap_or(LogFormat::Pretty)
240}
241
242fn default_env_filter() -> EnvFilter {
243    EnvFilter::builder()
244        .with_default_directive(tracing::level_filters::LevelFilter::INFO.into())
245        .from_env_lossy()
246}
247
248pub fn parse_duration(value: &str) -> Option<Duration> {
249    let value = value.trim();
250    if value.is_empty() {
251        return None;
252    }
253    if let Some(raw) = value.strip_suffix("ms") {
254        return raw.trim().parse::<u64>().ok().map(Duration::from_millis);
255    }
256    if let Some(raw) = value.strip_suffix('s') {
257        return raw.trim().parse::<u64>().ok().map(Duration::from_secs);
258    }
259    if let Some(raw) = value.strip_suffix('m') {
260        return raw
261            .trim()
262            .parse::<u64>()
263            .ok()
264            .and_then(|minutes| minutes.checked_mul(60))
265            .map(Duration::from_secs);
266    }
267    if let Some(raw) = value.strip_suffix('h') {
268        return raw
269            .trim()
270            .parse::<u64>()
271            .ok()
272            .and_then(|hours| hours.checked_mul(60 * 60))
273            .map(Duration::from_secs);
274    }
275    value.parse::<u64>().ok().map(Duration::from_secs)
276}
277
278pub fn duration_from_env(name: &str, default: Duration) -> Duration {
279    std::env::var(name)
280        .ok()
281        .and_then(|value| parse_duration(&value))
282        .unwrap_or(default)
283}
284
285/// Read a duration that has to be positive, falling back to `default` on a zero.
286///
287/// Zero is never a usable value for these settings: a period of zero panics
288/// `tokio::time::interval`, and a zero minimum trips [`RetryBackoff::new`]'s
289/// assertion. Clamping a zero up to a millisecond avoids the panic, but trades it
290/// for a hot loop — a rejected registration retried a thousand times a second, or
291/// a full registration scan queued into the routing loop every millisecond. The
292/// default is the only value that is both safe and useful, so a zero selects it
293/// and says so once.
294///
295/// [`RetryBackoff::new`]: crate::timeout::RetryBackoff::new
296fn positive_duration_from_env(name: &str, default: Duration) -> Duration {
297    let value = duration_from_env(name, default);
298    if value.is_zero() {
299        tracing::warn!(
300            event = "config_zero_duration_ignored",
301            variable = name,
302            default = ?default,
303            "ignoring a zero duration and using the default instead"
304        );
305        return default;
306    }
307    value
308}
309
310pub fn control_io_timeout() -> Duration {
311    duration_from_env(PB_MAPPER_CONTROL_IO_TIMEOUT, DEFAULT_CONTROL_IO_TIMEOUT)
312}
313
314pub fn stream_ack_timeout() -> Duration {
315    duration_from_env(PB_MAPPER_STREAM_ACK_TIMEOUT, DEFAULT_STREAM_ACK_TIMEOUT)
316}
317
318pub fn stream_ready_timeout() -> Duration {
319    duration_from_env(PB_MAPPER_STREAM_READY_TIMEOUT, DEFAULT_STREAM_READY_TIMEOUT)
320}
321
322pub fn stream_recovery_timeout() -> Duration {
323    duration_from_env(
324        PB_MAPPER_STREAM_RECOVERY_TIMEOUT,
325        DEFAULT_STREAM_RECOVERY_TIMEOUT,
326    )
327}
328
329pub fn control_conn_pool_size() -> usize {
330    std::env::var(PB_MAPPER_CONTROL_CONN_POOL_SIZE)
331        .ok()
332        .and_then(|value| value.trim().parse::<usize>().ok())
333        .filter(|size| *size > 0)
334        .map(|size| size.min(16))
335        .unwrap_or(DEFAULT_CONTROL_CONN_POOL_SIZE)
336}
337
338pub fn control_heartbeat_interval() -> Duration {
339    duration_from_env(
340        PB_MAPPER_CONTROL_HEARTBEAT_INTERVAL,
341        DEFAULT_CONTROL_HEARTBEAT_INTERVAL,
342    )
343}
344
345pub fn control_heartbeat_tolerance() -> Duration {
346    duration_from_env(
347        PB_MAPPER_CONTROL_HEARTBEAT_TOLERANCE,
348        DEFAULT_CONTROL_HEARTBEAT_TOLERANCE,
349    )
350}
351
352pub fn control_suspect_grace() -> Duration {
353    duration_from_env(
354        PB_MAPPER_CONTROL_SUSPECT_GRACE,
355        DEFAULT_CONTROL_SUSPECT_GRACE,
356    )
357}
358
359pub fn registration_probe_timeout() -> Duration {
360    duration_from_env(
361        PB_MAPPER_REGISTRATION_PROBE_TIMEOUT,
362        DEFAULT_REGISTRATION_PROBE_TIMEOUT,
363    )
364}
365
366/// The retry ladder for a registration the relay rejected but marked retryable.
367///
368/// Separate from the transport ladder, and far slower. A transport failure means
369/// the relay could not be reached, so retrying quickly is how the tunnel comes
370/// back. A retryable rejection is the opposite: the relay answered, and refused —
371/// a per-service connection quota that is full, a namespace at its service limit.
372/// Reconnecting a few times a second adds load to the very condition that has to
373/// clear, and it produced thousands of identical reject lines a minute in
374/// production. So the ladder starts at 5s and settles at 80s.
375///
376/// Both ends come back together because they are only valid as a pair: a zero
377/// falls back to its default, and the maximum is then raised to the minimum,
378/// rather than letting an environment typo reach [`RetryBackoff::new`]'s
379/// assertions, which would abort the process at start-up.
380///
381/// Raising the maximum, but defaulting a zero minimum, because the two failures
382/// are different. An inverted range is a legible request — wait exactly this long
383/// — and collapsing it to a fixed delay honours it. A zero minimum is not: it
384/// asks for no wait at all, and since [`RetryBackoff`] caps its multiplier at
385/// 1024, a millisecond minimum would never climb past about a second no matter
386/// how high the maximum, hammering the very quota the ladder exists to let clear.
387///
388/// [`RetryBackoff`]: crate::timeout::RetryBackoff
389/// [`RetryBackoff::new`]: crate::timeout::RetryBackoff::new
390pub fn registration_reject_backoff() -> (Duration, Duration) {
391    let min = positive_duration_from_env(
392        PB_MAPPER_REGISTRATION_REJECT_BACKOFF_MIN,
393        DEFAULT_REGISTRATION_REJECT_BACKOFF_MIN,
394    );
395    let max = positive_duration_from_env(
396        PB_MAPPER_REGISTRATION_REJECT_BACKOFF_MAX,
397        DEFAULT_REGISTRATION_REJECT_BACKOFF_MAX,
398    )
399    .max(min);
400    (min, max)
401}
402
403pub fn server_lease_timeout() -> Duration {
404    duration_from_env(PB_MAPPER_SERVER_LEASE_TIMEOUT, DEFAULT_SERVER_LEASE_TIMEOUT)
405}
406
407/// How often the relay looks for registrations that stopped renewing their lease.
408///
409/// A third of [`server_lease_timeout`] by default, which bounds two things at
410/// once: how long an expired registration can keep attracting subscribers, and —
411/// because the sweep needs two passes to retire one — how much grace a live
412/// connection gets to notice its own idle timeout first.
413///
414/// A zero falls back to the default: `tokio::time::interval` panics on a zero
415/// period, and each tick queues a scan over every registration through the single
416/// routing loop, so sweeping as fast as the timer allows would starve the traffic
417/// the sweep exists to protect.
418pub fn server_lease_sweep_interval() -> Duration {
419    positive_duration_from_env(
420        PB_MAPPER_SERVER_LEASE_SWEEP_INTERVAL,
421        DEFAULT_SERVER_LEASE_SWEEP_INTERVAL,
422    )
423}
424
425pub fn client_health_check_interval() -> Duration {
426    duration_from_env(
427        PB_MAPPER_CLIENT_HEALTH_CHECK_INTERVAL,
428        DEFAULT_CLIENT_HEALTH_CHECK_INTERVAL,
429    )
430}
431
432pub fn client_health_check_timeout() -> Duration {
433    duration_from_env(
434        PB_MAPPER_CLIENT_HEALTH_CHECK_TIMEOUT,
435        DEFAULT_CLIENT_HEALTH_CHECK_TIMEOUT,
436    )
437}
438
439pub fn client_health_failure_threshold() -> usize {
440    std::env::var(PB_MAPPER_CLIENT_HEALTH_FAILURE_THRESHOLD)
441        .ok()
442        .and_then(|value| value.trim().parse::<usize>().ok())
443        .filter(|threshold| *threshold > 0)
444        .map(|threshold| threshold.min(100))
445        .unwrap_or(DEFAULT_CLIENT_HEALTH_FAILURE_THRESHOLD)
446}
447
448/// Whether the environment asks for TCP keep-alive, from `PB_MAPPER_KEEP_ALIVE`.
449///
450/// A default for a process to read once at startup — not something the tunnels
451/// consult. Keep-alive is a per-tunnel parameter because one process can run
452/// many tunnels that disagree about it, which is why this is a function and not
453/// the `LazyLock<bool>` it used to be: that froze on the first tunnel to touch
454/// a socket and left every later one, including the UI's own toggle, unable to
455/// change it.
456pub fn keep_alive_from_env() -> bool {
457    match std::env::var(PB_MAPPER_KEEP_ALIVE) {
458        // The documented spelling is `ON`; the previous check was `is_ok()`,
459        // which turned keep-alive on for any value at all — `OFF` included.
460        Ok(value) => matches!(
461            value.trim().to_ascii_lowercase().as_str(),
462            "on" | "1" | "true" | "yes"
463        ),
464        Err(_) => false,
465    }
466}
467
468/// The relay address as configured: `addr` if given, else `PB_MAPPER_SERVER`.
469///
470/// Unresolved on purpose. A caller that hands the address to something which
471/// resolves it itself — the SDK client, for one — should pass the name along
472/// rather than resolve it here and stringify the result, which both duplicates
473/// the lookup and narrows a multi-address name down to one entry on the way.
474pub fn pb_mapper_server_addr(addr: Option<&str>) -> Result<String> {
475    match addr {
476        Some(addr) => Ok(addr.to_string()),
477        None => std::env::var(PB_MAPPER_SERVER).context(CfgPbServerEnvNotExistSnafu),
478    }
479}
480
481/// Every address the relay resolves to: `addr` if given, else `PB_MAPPER_SERVER`.
482#[inline]
483pub fn resolve_pb_mapper_server(addr: Option<&str>) -> Result<ResolvedAddrs> {
484    match addr {
485        Some(addr) => resolve_addrs(addr),
486        None => {
487            let addr = std::env::var(PB_MAPPER_SERVER).context(CfgPbServerEnvNotExistSnafu)?;
488            resolve_addrs(&addr)
489        }
490    }
491}
492
493/// Async counterpart of [`resolve_pb_mapper_server`], for Tokio contexts.
494pub async fn resolve_pb_mapper_server_async(addr: Option<&str>) -> Result<ResolvedAddrs> {
495    match addr {
496        Some(addr) => resolve_addrs_async(addr).await,
497        None => {
498            let addr = std::env::var(PB_MAPPER_SERVER).context(CfgPbServerEnvNotExistSnafu)?;
499            resolve_addrs_async(&addr).await
500        }
501    }
502}
503
504pub fn init_tracing() {
505    static INIT_TRACING: Once = Once::new();
506    INIT_TRACING.call_once(|| {
507        let result = match log_format_from_env() {
508            LogFormat::Pretty => {
509                let subscriber = tracing_subscriber::registry().with(
510                    fmt::layer()
511                        .pretty()
512                        .with_writer(std::io::stdout)
513                        .with_filter(default_env_filter()),
514                );
515                tracing::subscriber::set_global_default(subscriber)
516            }
517            LogFormat::Compact => {
518                let subscriber = tracing_subscriber::registry().with(
519                    fmt::layer()
520                        .compact()
521                        .with_writer(std::io::stdout)
522                        .with_filter(default_env_filter()),
523                );
524                tracing::subscriber::set_global_default(subscriber)
525            }
526            LogFormat::Json => {
527                let subscriber = tracing_subscriber::registry().with(
528                    fmt::layer()
529                        .json()
530                        .flatten_event(true)
531                        .with_writer(std::io::stdout)
532                        .with_filter(default_env_filter()),
533                );
534                tracing::subscriber::set_global_default(subscriber)
535            }
536        };
537
538        if let Err(e) = result {
539            eprintln!("failed to initialize tracing subscriber: {e}");
540        }
541    });
542}
543
544#[cfg(test)]
545mod tests {
546    use super::*;
547
548    #[test]
549    fn parse_log_format_accepts_supported_values() {
550        assert_eq!(parse_log_format("pretty"), LogFormat::Pretty);
551        assert_eq!(parse_log_format("compact"), LogFormat::Compact);
552        assert_eq!(parse_log_format("json"), LogFormat::Json);
553        assert_eq!(parse_log_format(" JSON "), LogFormat::Json);
554        assert_eq!(parse_log_format("unknown"), LogFormat::Pretty);
555    }
556
557    /// One test rather than several, because these share process-wide state and
558    /// the test runner threads them.
559    #[test]
560    fn keep_alive_reads_the_environment_every_time() {
561        let restore = std::env::var(PB_MAPPER_KEEP_ALIVE).ok();
562
563        // SAFETY: mutating the environment is unsafe in edition 2024 because
564        // it is process-global. This is the only test that touches
565        // `PB_MAPPER_KEEP_ALIVE` — which is why it is one test and not several
566        // — and it restores the original value before returning.
567        unsafe {
568            std::env::remove_var(PB_MAPPER_KEEP_ALIVE);
569        }
570        assert!(!keep_alive_from_env(), "absent means off");
571
572        unsafe {
573            std::env::set_var(PB_MAPPER_KEEP_ALIVE, "ON");
574        }
575        assert!(keep_alive_from_env(), "the documented spelling");
576
577        // The regression. This used to be a `LazyLock<bool>`, so the answer was
578        // whatever the first caller in the process saw and could never change —
579        // which is why the UI's per-service toggle did nothing after the first
580        // tunnel started.
581        unsafe {
582            std::env::set_var(PB_MAPPER_KEEP_ALIVE, "OFF");
583        }
584        assert!(
585            !keep_alive_from_env(),
586            "OFF must mean off; the old check was `is_ok()`, so any value at \
587             all — OFF included — turned keep-alive on"
588        );
589
590        for truthy in ["on", "1", "true", "yes", " ON "] {
591            unsafe {
592                std::env::set_var(PB_MAPPER_KEEP_ALIVE, truthy);
593            }
594            assert!(keep_alive_from_env(), "{truthy:?} should enable");
595        }
596        for falsy in ["", "off", "0", "false", "no"] {
597            unsafe {
598                std::env::set_var(PB_MAPPER_KEEP_ALIVE, falsy);
599            }
600            assert!(!keep_alive_from_env(), "{falsy:?} should not enable");
601        }
602
603        unsafe {
604            match restore {
605                Some(value) => std::env::set_var(PB_MAPPER_KEEP_ALIVE, value),
606                None => std::env::remove_var(PB_MAPPER_KEEP_ALIVE),
607            }
608        }
609    }
610
611    /// The point of [`ResolvedAddrs`]: a name with several records keeps them
612    /// all, so the dial loops can try each one. Collapsing to the first is what
613    /// made a multi-record relay unreachable whenever its first address was.
614    #[test]
615    fn resolved_addrs_keeps_every_candidate() {
616        let first: SocketAddr = "127.0.0.1:7666".parse().expect("literal");
617        let second: SocketAddr = "[::1]:7666".parse().expect("literal");
618        let addrs = ResolvedAddrs::from_candidates(vec![first, second]).expect("non-empty");
619
620        assert_eq!(addrs.as_slice(), [first, second]);
621        assert_eq!(addrs.primary(), first, "order is the resolver's order");
622    }
623
624    /// An empty candidate list is not a resolution: nothing could be dialled, so
625    /// it has to be rejected here rather than surface as a connect failure
626    /// against an address the caller never gave.
627    #[test]
628    fn resolved_addrs_rejects_an_empty_candidate_list() {
629        assert!(ResolvedAddrs::from_candidates(Vec::new()).is_none());
630    }
631
632    /// The rendering has to name one address — it goes into connect errors and
633    /// log fields — while still admitting that others were available.
634    #[test]
635    fn resolved_addrs_renders_the_primary_and_the_rest_as_a_count() {
636        let single: SocketAddr = "127.0.0.1:7666".parse().expect("literal");
637        assert_eq!(ResolvedAddrs::from(single).to_string(), "127.0.0.1:7666");
638
639        let second: SocketAddr = "[::1]:7666".parse().expect("literal");
640        let both = ResolvedAddrs::from_candidates(vec![single, second]).expect("non-empty");
641        assert_eq!(both.to_string(), "127.0.0.1:7666 (+1 more)");
642    }
643
644    /// A literal address needs no resolver, and must survive verbatim.
645    #[test]
646    fn resolve_addrs_passes_a_literal_through() {
647        let addrs = resolve_addrs("127.0.0.1:7666").expect("a literal always resolves");
648        assert_eq!(
649            addrs.as_slice(),
650            ["127.0.0.1:7666".parse().expect("literal")]
651        );
652    }
653
654    /// `localhost` is the everyday multi-record name: it resolves to a v4 and a
655    /// v6 loopback on most hosts, and both must reach the dial loop.
656    #[test]
657    fn resolve_addrs_keeps_every_localhost_record() {
658        let addrs = resolve_addrs("localhost:7666").expect("localhost always resolves");
659        assert!(
660            addrs.as_slice().iter().all(|addr| addr.ip().is_loopback()),
661            "localhost must resolve to loopback only, got {:?}",
662            addrs.as_slice()
663        );
664    }
665
666    /// A name that resolves to nothing is an error, not an empty success.
667    ///
668    /// Tested with a missing port rather than an unresolvable host: a resolver
669    /// that answers every query with a wildcard address — WSL's NAT DNS, among
670    /// others — makes "this host does not exist" untestable, while no resolver
671    /// invents a port.
672    #[test]
673    fn resolve_addrs_fails_when_nothing_can_be_resolved() {
674        assert!(resolve_addrs("127.0.0.1").is_err(), "no port");
675        assert!(resolve_addrs("localhost").is_err(), "no port");
676    }
677
678    /// One test for all three variables, for the same reason as
679    /// `keep_alive_reads_the_environment_every_time`: the environment is
680    /// process-global and the runner threads these.
681    #[test]
682    fn reject_backoff_and_sweep_interval_survive_a_bad_environment() {
683        let names = [
684            PB_MAPPER_REGISTRATION_REJECT_BACKOFF_MIN,
685            PB_MAPPER_REGISTRATION_REJECT_BACKOFF_MAX,
686            PB_MAPPER_SERVER_LEASE_SWEEP_INTERVAL,
687        ];
688        let restore = names.map(|name| (name, std::env::var(name).ok()));
689
690        // SAFETY: mutating the environment is unsafe in edition 2024 because it
691        // is process-global. This is the only test that touches these three
692        // names — which is why it is one test and not three — and it restores
693        // their original values before returning.
694        unsafe {
695            for name in names {
696                std::env::remove_var(name);
697            }
698        }
699        assert_eq!(
700            registration_reject_backoff(),
701            (
702                DEFAULT_REGISTRATION_REJECT_BACKOFF_MIN,
703                DEFAULT_REGISTRATION_REJECT_BACKOFF_MAX
704            ),
705            "absent means the defaults"
706        );
707        assert_eq!(
708            server_lease_sweep_interval(),
709            DEFAULT_SERVER_LEASE_SWEEP_INTERVAL
710        );
711
712        unsafe {
713            std::env::set_var(PB_MAPPER_REGISTRATION_REJECT_BACKOFF_MIN, "30s");
714            std::env::set_var(PB_MAPPER_REGISTRATION_REJECT_BACKOFF_MAX, "2m");
715        }
716        assert_eq!(
717            registration_reject_backoff(),
718            (Duration::from_secs(30), Duration::from_secs(120)),
719            "both ends come from the environment"
720        );
721
722        // A zero is not a usable setting at either end: `RetryBackoff::new`
723        // asserts on a zero minimum, and a millisecond minimum would retry a
724        // thousand times a second. Both fall back to their defaults instead.
725        unsafe {
726            std::env::set_var(PB_MAPPER_REGISTRATION_REJECT_BACKOFF_MIN, "0s");
727            std::env::set_var(PB_MAPPER_REGISTRATION_REJECT_BACKOFF_MAX, "0s");
728        }
729        assert_eq!(
730            registration_reject_backoff(),
731            (
732                DEFAULT_REGISTRATION_REJECT_BACKOFF_MIN,
733                DEFAULT_REGISTRATION_REJECT_BACKOFF_MAX
734            ),
735            "a zero at either end selects the default, not a millisecond"
736        );
737
738        // A zero minimum with a usable maximum still defaults the minimum, and
739        // the maximum given is kept.
740        unsafe {
741            std::env::set_var(PB_MAPPER_REGISTRATION_REJECT_BACKOFF_MIN, "0ms");
742            std::env::set_var(PB_MAPPER_REGISTRATION_REJECT_BACKOFF_MAX, "3m");
743        }
744        assert_eq!(
745            registration_reject_backoff(),
746            (
747                DEFAULT_REGISTRATION_REJECT_BACKOFF_MIN,
748                Duration::from_secs(180)
749            )
750        );
751
752        unsafe {
753            std::env::set_var(PB_MAPPER_REGISTRATION_REJECT_BACKOFF_MIN, "1m");
754            std::env::set_var(PB_MAPPER_REGISTRATION_REJECT_BACKOFF_MAX, "1s");
755        }
756        let (min, max) = registration_reject_backoff();
757        assert_eq!(min, Duration::from_secs(60));
758        assert_eq!(max, min, "an inverted range collapses to a fixed delay");
759
760        unsafe {
761            std::env::set_var(PB_MAPPER_SERVER_LEASE_SWEEP_INTERVAL, "0s");
762        }
763        assert_eq!(
764            server_lease_sweep_interval(),
765            DEFAULT_SERVER_LEASE_SWEEP_INTERVAL,
766            "a zero period would panic `tokio::time::interval`, and a millisecond \
767             one would queue a full scan every millisecond"
768        );
769
770        unsafe {
771            for (name, value) in restore {
772                match value {
773                    Some(value) => std::env::set_var(name, value),
774                    None => std::env::remove_var(name),
775                }
776            }
777        }
778    }
779
780    /// The async path is the one the tunnels use, and has to agree with the
781    /// blocking one on a literal.
782    #[tokio::test]
783    async fn resolve_addrs_async_matches_the_blocking_path_on_a_literal() {
784        let expected = resolve_addrs("127.0.0.1:7666").expect("literal");
785        let actual = resolve_addrs_async("127.0.0.1:7666")
786            .await
787            .expect("literal");
788        assert_eq!(actual, expected);
789    }
790}
pb_mapper_core/config.rs

pb_mapper_core/
config.rs
