majordomus_cli/mesh/

broadcast.rs

//! The UDP broadcast provider: the controlled fallback for segments where multicast is
//! filtered. Same protocol, same port, no second serialization — only the destination
//! differs: the limited broadcast address in `auto` mode, or the directed broadcast
//! addresses the declaration lists in `explicit` mode. Never `auto` by default, and it
//! does not spray interfaces: one socket, the destinations named above, nothing more.
//!
//! Transmit-first by design: when the multicast provider runs, its socket already hears
//! everything aimed at the port and a second bound socket would fight it for the
//! address. This provider binds a listener only when it is the sole UDP provider.
//!
//! ```
//! use majordomus_cli::mesh::broadcast::BroadcastProvider;
//! use majordomus_cli::mesh::config::{BroadcastConfig, BroadcastMode};
//! use majordomus_cli::mesh::provider::{MeshProvider, MeshProviderState};
//!
//! // A disabled declaration yields a provider that starts nothing and says so.
//! let provider = BroadcastProvider::new(BroadcastConfig::default(), false);
//! let status = provider.status();
//! assert_eq!(status.id, "udp_broadcast");
//! assert_eq!(status.state, MeshProviderState::Stopped);
//! assert_eq!(status.sent, 0);
//! # let _ = BroadcastMode::Auto;
//! ```

use std::net::{Ipv4Addr, SocketAddrV4, UdpSocket};
use std::sync::atomic::Ordering;
use std::sync::{Arc, Mutex};
use std::time::Duration;

use super::config::{BroadcastConfig, BroadcastMode};
use super::protocol::{encode, MAX_DATAGRAM};
use super::provider::{
    jitter_ms, Counters, MeshProvider, MeshProviderState, Observation, ProviderContext,
    ProviderStatus,
};
use super::registry::MeshSource;
use super::MeshError;

const READ_TIMEOUT: Duration = Duration::from_millis(500);

/// The broadcast provider: the declared destinations, the counters its threads share,
/// and the state its status reports. Constructed by the manager from the declaration.
pub struct BroadcastProvider {
    config: BroadcastConfig,
    /// Whether this provider should also listen (no multicast provider is running).
    listen: bool,
    counters: Arc<Counters>,
    state: Arc<Mutex<(MeshProviderState, Option<String>)>>,
}

impl BroadcastProvider {
    /// A provider over its declaration; `listen` when no other UDP socket hears the
    /// port — with the multicast provider running, that socket already hears every
    /// datagram aimed at it, and this provider only transmits.
    pub fn new(config: BroadcastConfig, listen: bool) -> Self {
        BroadcastProvider {
            config,
            listen,
            counters: Arc::new(Counters::default()),
            state: Arc::new(Mutex::new((MeshProviderState::Stopped, None))),
        }
    }

    fn destinations(&self) -> Result<Vec<SocketAddrV4>, MeshError> {
        match self.config.mode {
            BroadcastMode::Disabled => Ok(Vec::new()),
            BroadcastMode::Auto => Ok(vec![SocketAddrV4::new(
                Ipv4Addr::BROADCAST,
                self.config.port,
            )]),
            BroadcastMode::Explicit => self
                .config
                .networks
                .iter()
                .map(|n| {
                    n.parse::<Ipv4Addr>()
                        .map(|ip| SocketAddrV4::new(ip, self.config.port))
                        .map_err(|_| {
                            MeshError::Provider(format!("'{n}' is not an IPv4 broadcast address"))
                        })
                })
                .collect(),
        }
    }
}

impl MeshProvider for BroadcastProvider {
    fn id(&self) -> &'static str {
        "udp_broadcast"
    }

    fn start(&mut self, ctx: &ProviderContext) -> Result<(), MeshError> {
        let destinations = match self.destinations() {
            Ok(d) => d,
            Err(e) => {
                *self.state.lock().expect("broadcast state") =
                    (MeshProviderState::Failed, Some(e.to_string()));
                return Err(e);
            }
        };
        if destinations.is_empty() {
            let reason = "mode is disabled".to_string();
            *self.state.lock().expect("broadcast state") =
                (MeshProviderState::Stopped, Some(reason.clone()));
            return Err(MeshError::Provider(reason));
        }
        // The sender binds an ephemeral port (the listener, when this provider has one,
        // owns the well-known port); directed and limited broadcast need the flag.
        let bind_port = if self.listen { self.config.port } else { 0 };
        let socket = match UdpSocket::bind((Ipv4Addr::UNSPECIFIED, bind_port)) {
            Ok(s) => s,
            Err(e) => {
                let reason = format!("cannot bind udp port {bind_port}: {e}");
                *self.state.lock().expect("broadcast state") =
                    (MeshProviderState::Failed, Some(reason.clone()));
                return Err(MeshError::Provider(reason));
            }
        };
        if let Err(e) = socket.set_broadcast(true) {
            let reason = format!("cannot enable broadcast: {e}");
            *self.state.lock().expect("broadcast state") =
                (MeshProviderState::Failed, Some(reason.clone()));
            return Err(MeshError::Provider(reason));
        }
        let _ = socket.set_read_timeout(Some(READ_TIMEOUT));
        *self.state.lock().expect("broadcast state") = (
            MeshProviderState::Running,
            Some(format!(
                "{} destination(s), {}",
                destinations.len(),
                if self.listen {
                    "listening"
                } else {
                    "transmit only"
                }
            )),
        );

        if self.listen {
            let socket = socket.try_clone().map_err(|e| {
                MeshError::Provider(format!("cannot clone the broadcast socket: {e}"))
            })?;
            let tx = ctx.tx.clone();
            let stop = Arc::clone(&ctx.stop);
            let counters = Arc::clone(&self.counters);
            let state = Arc::clone(&self.state);
            let _ = std::thread::Builder::new()
                .name("majordomus-mesh-broadcast-listen".into())
                .spawn(move || {
                    let mut buffer = [0u8; MAX_DATAGRAM + 1];
                    while !stop.load(Ordering::SeqCst) {
                        match socket.recv_from(&mut buffer) {
                            Ok((len, from)) => {
                                counters.received.fetch_add(1, Ordering::Relaxed);
                                let _ = tx.send(Observation {
                                    source: MeshSource::UdpBroadcast,
                                    path: from.to_string(),
                                    bytes: buffer[..len].to_vec(),
                                });
                            }
                            Err(e)
                                if e.kind() == std::io::ErrorKind::WouldBlock
                                    || e.kind() == std::io::ErrorKind::TimedOut => {}
                            Err(e) => {
                                *state.lock().expect("broadcast state") =
                                    (MeshProviderState::Failed, Some(format!("recv: {e}")));
                                return;
                            }
                        }
                    }
                    *state.lock().expect("broadcast state") = (MeshProviderState::Stopped, None);
                });
        }

        {
            let stop = Arc::clone(&ctx.stop);
            let beacon = Arc::clone(&ctx.beacon);
            let counters = Arc::clone(&self.counters);
            let interval = Duration::from_secs(self.config.interval_seconds);
            let _ = std::thread::Builder::new()
                .name("majordomus-mesh-broadcast-announce".into())
                .spawn(move || {
                    while !stop.load(Ordering::SeqCst) {
                        if let Ok(bytes) = encode(&beacon.next_envelope()) {
                            for destination in &destinations {
                                if socket.send_to(&bytes, destination).is_ok() {
                                    counters.sent.fetch_add(1, Ordering::Relaxed);
                                }
                            }
                        }
                        let total = interval + Duration::from_millis(jitter_ms(2000));
                        let mut slept = Duration::ZERO;
                        while slept < total && !stop.load(Ordering::SeqCst) {
                            let slice = READ_TIMEOUT.min(total - slept);
                            std::thread::sleep(slice);
                            slept += slice;
                        }
                    }
                });
        }
        Ok(())
    }

    fn status(&self) -> ProviderStatus {
        let (state, detail) = self.state.lock().expect("broadcast state").clone();
        ProviderStatus {
            id: self.id().into(),
            state,
            detail,
            sent: self.counters.sent.load(Ordering::Relaxed),
            received: self.counters.received.load(Ordering::Relaxed),
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn destinations_follow_the_declared_mode() {
        let auto = BroadcastProvider::new(
            BroadcastConfig {
                mode: BroadcastMode::Auto,
                ..BroadcastConfig::default()
            },
            false,
        );
        assert_eq!(auto.destinations().unwrap().len(), 1);
        let explicit = BroadcastProvider::new(
            BroadcastConfig {
                mode: BroadcastMode::Explicit,
                networks: vec!["192.168.1.255".into(), "10.0.0.255".into()],
                ..BroadcastConfig::default()
            },
            false,
        );
        assert_eq!(explicit.destinations().unwrap().len(), 2);
        let disabled = BroadcastProvider::new(BroadcastConfig::default(), false);
        assert!(disabled.destinations().unwrap().is_empty());
    }

    /// How many fresh ports a listening start may try before the test gives up.
    const BIND_ATTEMPTS: usize = 8;

    /// The text the host gives a bind that finds its address taken, read from a real
    /// collision rather than written down: it differs between operating systems.
    fn address_in_use() -> String {
        let held = UdpSocket::bind((Ipv4Addr::UNSPECIFIED, 0)).unwrap();
        let refused = UdpSocket::bind(held.local_addr().unwrap()).unwrap_err();
        assert_eq!(refused.kind(), std::io::ErrorKind::AddrInUse, "{refused}");
        refused.to_string()
    }

    /// Start a listening provider on a port nobody holds, and hand back the provider with
    /// the port it holds.
    ///
    /// A free port can only be found by binding port 0 and letting it go, and between the
    /// release and the provider's own bind any process on the host may take it: under load
    /// one did, and the test failed with "cannot bind udp port 55932: Address already in
    /// use". The provider binding the port it was configured with is what is under test,
    /// so the port stays a configured one; a start that lost that race, and only that
    /// race, is retried with a fresh port a bounded number of times. Any other failure of
    /// the start fails the test at once.
    fn start_listening_on_a_free_port(
        ctx: &crate::mesh::provider::ProviderContext,
    ) -> (BroadcastProvider, u16) {
        let in_use = address_in_use();
        let mut lost = Vec::new();
        for _ in 0..BIND_ATTEMPTS {
            let port = {
                let probe = UdpSocket::bind((Ipv4Addr::UNSPECIFIED, 0)).unwrap();
                probe.local_addr().unwrap().port()
            };
            let mut provider = BroadcastProvider::new(
                BroadcastConfig {
                    mode: BroadcastMode::Auto,
                    port,
                    interval_seconds: 1,
                    ..BroadcastConfig::default()
                },
                true,
            );
            match provider.start(ctx) {
                Ok(()) => {
                    // Hard: the sole UDP provider binds its listening socket and reaches
                    // Running.
                    assert_eq!(provider.status().state, MeshProviderState::Running);
                    return (provider, port);
                }
                Err(e)
                    if e.to_string()
                        .ends_with(&format!("cannot bind udp port {port}: {in_use}")) =>
                {
                    lost.push(port);
                }
                Err(e) => panic!("the listening provider did not start: {e}"),
            }
        }
        panic!("every port probed was taken before the provider could bind it: {lost:?}");
    }

    #[test]
    fn the_sole_udp_provider_listens_and_counts_what_it_hears() {
        let (tx, rx) = crate::mesh::provider::ObservationSender::bounded(64);
        let stop = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
        let ctx = crate::mesh::provider::ProviderContext {
            tx,
            stop: std::sync::Arc::clone(&stop),
            beacon: std::sync::Arc::new(crate::mesh::provider::Beacon::new(
                std::sync::Arc::new(crate::mesh::identity::NodeIdentity::ephemeral().unwrap()),
                vec![],
                vec![],
                vec![],
                "test",
            )),
        };
        let (_provider, port) = start_listening_on_a_free_port(&ctx);
        // Best-effort: a host that delivers the loopback datagram hands it up as a
        // broadcast observation; one that does not route it leaves the channel silent.
        // The bound socket was the invariant under test; delivery is the network's to
        // grant, and the round-trip is proven host-independently in tests/mesh.rs.
        let sender = UdpSocket::bind((Ipv4Addr::LOCALHOST, 0)).unwrap();
        let _ = sender.send_to(b"aimed at the port", (Ipv4Addr::LOCALHOST, port));
        if let Ok(heard) = rx.recv_timeout(Duration::from_secs(2)) {
            assert_eq!(heard.source, MeshSource::UdpBroadcast);
        }
        stop.store(true, std::sync::atomic::Ordering::SeqCst);
    }

    #[test]
    fn a_malformed_network_is_an_error_naming_it() {
        let provider = BroadcastProvider::new(
            BroadcastConfig {
                mode: BroadcastMode::Explicit,
                networks: vec!["not-an-address".into()],
                ..BroadcastConfig::default()
            },
            false,
        );
        let error = provider.destinations().unwrap_err().to_string();
        assert!(error.contains("not-an-address"), "{error}");
    }
}