epics-ca-rs 0.16.2

use std::collections::HashMap;
use std::net::{Ipv4Addr, SocketAddr};
use std::sync::Arc;
use std::time::Duration;
use tokio::net::UdpSocket;
use tokio::sync::Notify;

use crate::protocol::*;
use epics_base_rs::error::CaResult;

/// Run the beacon emitter. Broadcasts CA_PROTO_RSRV_IS_UP at exponentially
/// increasing intervals (starting at 20ms, doubling up to `max_period`).
///
/// `beacon_addrs` lists every destination — usually the per-interface
/// broadcast addresses plus any operator-supplied entries. Unicast routes
/// (e.g. site-wide gateways) are sent the same beacon stream.
///
/// When `reset` is notified (e.g. on TCP connect/disconnect), the interval
/// resets to the initial 20ms, matching C EPICS behavior. This lets clients
/// detect server state changes quickly via beacon anomaly detection.
///
/// `signer` is an opt-in Ed25519 [`signed_beacon::SignedBeaconEmitter`]
/// that emits a companion datagram immediately after each beacon so
/// clients with a configured keyring can authenticate the server
/// identity. C clients ignore the companion (unknown command); the
/// regular beacon stream is unchanged.
pub async fn run_beacon_emitter(
    server_port: u16,
    beacon_addrs: Vec<SocketAddr>,
    max_period: Duration,
    reset: Arc<Notify>,
    #[cfg(feature = "cap-tokens")] signer: Option<
        Arc<crate::server::signed_beacon::SignedBeaconEmitter>,
    >,
) -> CaResult<()> {
    let socket = UdpSocket::bind("0.0.0.0:0").await?;
    socket.set_broadcast(true)?;

    // Resolve the server's local IP via routing. Prefer the first beacon
    // destination as a probe so multi-NIC hosts pick the matching outgoing
    // interface; fall back to limited broadcast.
    let probe_dest = beacon_addrs
        .first()
        .copied()
        .unwrap_or(SocketAddr::from((Ipv4Addr::BROADCAST, CA_REPEATER_PORT)));
    let server_ip: u32 = {
        let probe = std::net::UdpSocket::bind("0.0.0.0:0").ok();
        let ip = probe.and_then(|s| {
            s.connect(probe_dest).ok()?;
            match s.local_addr().ok()? {
                SocketAddr::V4(a) if !a.ip().is_unspecified() => {
                    Some(u32::from_be_bytes(a.ip().octets()))
                }
                _ => None,
            }
        });
        ip.unwrap_or(0)
    };

    let mut beacon_id: u32 = 0;
    let initial_interval = Duration::from_millis(20);
    let max_interval = max_period.max(initial_interval);
    let mut interval = initial_interval;

    // Per-destination last-error dedup. Mirrors rsrv c23012d
    // (`rsrv_online_notify_task` `lastError[]`): a persistently
    // unreachable beacon address emits a sendto() failure every
    // beacon period (~15 s steady-state, faster during ramp-up). We
    // log on first occurrence and on errno change, then once on
    // recovery — repeats are silent.
    let mut send_errors: HashMap<SocketAddr, std::io::ErrorKind> = HashMap::new();

    if beacon_addrs.is_empty() {
        // Nothing to send to — quietly idle but still consume reset
        // notifications so the channel doesn't fill up.
        loop {
            reset.notified().await;
        }
    }

    loop {
        // Build beacon message: CA_PROTO_RSRV_IS_UP
        let mut hdr = CaHeader::new(CA_PROTO_RSRV_IS_UP);
        hdr.data_type = CA_MINOR_VERSION;
        hdr.count = server_port;
        hdr.cid = beacon_id;
        hdr.available = server_ip;
        let bytes = hdr.to_bytes();

        for addr in &beacon_addrs {
            match socket.send_to(&bytes, addr).await {
                Ok(_) => {
                    if let Some(prev) = send_errors.remove(addr) {
                        tracing::info!(
                            target: "epics_ca_rs::beacon",
                            %addr, prev_error = ?prev,
                            "CA beacon send recovered"
                        );
                    }
                }
                Err(e) => {
                    let kind = e.kind();
                    if send_errors.insert(*addr, kind) != Some(kind) {
                        tracing::warn!(
                            target: "epics_ca_rs::beacon",
                            %addr, error = %e,
                            "CA beacon send failed"
                        );
                    }
                }
            }
        }

        // Signed-beacon companion: send a separate Ed25519-signed
        // datagram so clients with a configured keyring can verify
        // server identity. Same destinations.
        #[cfg(feature = "cap-tokens")]
        if let Some(ref s) = signer {
            s.emit(server_ip, server_port, beacon_id).await;
        }

        beacon_id = beacon_id.wrapping_add(1);

        tokio::select! {
            () = epics_base_rs::runtime::task::sleep(interval) => {
                if interval < max_interval {
                    interval = (interval * 2).min(max_interval);
                }
            }
            () = reset.notified() => {
                interval = initial_interval;
            }
        }
    }
}