mx-remote 5.1.0

// Author: Lars Op den Kamp (lars@opdenkamp-it.nl)
// Copyright (c) 2026 Op den Kamp IT Solutions

//! The runtime: what drives the announcement timer, and what the receive entry
//! point does with a datagram.

use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};

use crate::testing::{bay_config_rec, datagram, fixed_str, uid_n};
use crate::wire::{
    op, protocol_for, BayFeatures, BayStatus, DeviceFeature, DeviceUid, HEADER_LEN,
    PROTOCOL_VERSION,
};

use super::schedule::{next_hello_interval, HELLO_BASE, HELLO_JITTER};
use super::*;

/// The address every fixture datagram appears to come from.
const FROM: Ipv4Addr = Ipv4Addr::new(10, 8, 8, 9);

/// Records every frame that passes the protocol gate.
///
/// Frames are assembled inside the method that sends them, so this is the only
/// way to read back what the client would put on the wire.
#[derive(Default)]
pub(super) struct Tap(Mutex<Vec<Vec<u8>>>);

impl Tap {
    /// The opcodes captured so far.
    pub(super) fn opcodes(&self) -> Vec<u16> {
        self.0
            .lock()
            .expect("the tap is only ever locked to push or read")
            .iter()
            .filter_map(|f| f.get(20..22))
            .filter_map(|b| <[u8; 2]>::try_from(b).ok())
            .map(u16::from_le_bytes)
            .collect()
    }

    pub(super) fn frames(&self) -> Vec<Vec<u8>> {
        self.0
            .lock()
            .expect("the tap is only ever locked to push or read")
            .clone()
    }

    /// Forgets what was captured, so the next call is read on its own.
    pub(super) fn clear(&self) {
        self.0
            .lock()
            .expect("the tap is only ever locked to push or read")
            .clear();
    }
}

/// A client with no socket, so every send fails after passing the gate.
///
/// That is the interesting case for the timer: the announcement must not be
/// recorded when the frame never left.
fn client(n: u8) -> (Remote, Arc<Tap>) {
    client_with(n, Arc::new(()))
}

/// A client as [`client`] builds one, delivering its events to `handler`.
pub(super) fn client_with(n: u8, handler: Arc<dyn EventHandler>) -> (Remote, Arc<Tap>) {
    let remote = Remote::new(
        Config {
            uid: Some(uid_n(n)),
            ..Config::default()
        },
        handler,
    )
    .expect("a client given its own uid reads nothing from disk");
    let tap = Arc::new(Tap::default());
    let sink = Arc::clone(&tap);
    lock(&remote.shared.tx).set_tap(Arc::new(move |frame: &[u8]| {
        sink.0
            .lock()
            .expect("the tap is only ever locked to push or read")
            .push(frame.to_vec())
    }));
    (remote, tap)
}

/// Assembles a hello datagram the way a device does, header and all.
pub(super) fn hello_datagram(sender: DeviceUid, name: &str, serial: &str) -> Vec<u8> {
    let mut payload = Vec::new();
    payload.extend_from_slice(&0x28u16.to_le_bytes());
    fixed_str(&mut payload, name, 16);
    fixed_str(&mut payload, serial, 16);
    fixed_str(&mut payload, "4.8.0", 16);
    payload.extend_from_slice(&DeviceFeature::VIDEO_ROUTING.bits().to_le_bytes());

    let mut out = Vec::with_capacity(HEADER_LEN + payload.len());
    out.extend_from_slice(b"P8");
    out.extend_from_slice(&protocol_for(op::SYS_HELLO).to_le_bytes());
    out.extend_from_slice(sender.as_bytes());
    out.extend_from_slice(&op::SYS_HELLO.0.to_le_bytes());
    out.extend_from_slice(&(payload.len() as u16).to_le_bytes());
    out.extend_from_slice(&payload);
    out
}

/// `process_datagram` is the real receive entry point; every other test enters
/// one level below it. Testing here is what pins the negative half: hello is
/// announced on a clock, so no amount of arriving traffic may provoke one.
#[test]
fn a_datagram_is_decoded_and_does_not_provoke_an_announcement() {
    let (remote, tap) = client(200);
    let peer = uid_n(201);

    remote.shared.process_datagram(
        &hello_datagram(peer, "Peer", "PR0001"),
        Ipv4Addr::new(10, 8, 8, 9),
    );

    let device = remote.device(peer).expect("the datagram was not processed");
    assert_eq!(device.serial, "PR0001");
    assert_eq!(device.address, Some(Ipv4Addr::new(10, 8, 8, 9)));
    assert!(
        !tap.opcodes().contains(&op::SYS_HELLO.0),
        "a received datagram triggered a hello; announcement is a timer, not a reply"
    );
}

/// A client must not decode its own frames back into the registry.
#[test]
fn a_clients_own_frame_is_not_taken_for_a_peers() {
    let (remote, _) = client(205);
    remote.shared.process_datagram(
        &hello_datagram(remote.uid(), "Self", "SF0001"),
        Ipv4Addr::new(10, 8, 8, 9),
    );
    assert!(
        remote.devices().is_empty(),
        "the client registered itself as a device"
    );
}

/// A device announces itself on a schedule whether or not anything is talking
/// to it. A client that only re-announced on arriving traffic went silent on a
/// quiet network and stayed unknown to every peer that started after it.
#[test]
fn the_announcement_is_driven_by_a_timer() {
    let (remote, _) = client(202);
    let now = Instant::now();
    lock(&remote.shared.schedule).set_hello_timer(Some(now), Duration::from_secs(3));

    assert!(
        !remote.shared.announce_due(now + Duration::from_secs(2)),
        "announced early"
    );
    assert!(
        remote.shared.announce_due(now + Duration::from_secs(3)),
        "not due at the interval; a silent network would never announce"
    );
    assert!(
        remote.shared.announce_due(now + Duration::from_secs(3600)),
        "not due long after the interval"
    );

    remote.shared.closing.store(true, Ordering::SeqCst);
    assert!(
        !remote.shared.announce_due(now + Duration::from_secs(3600)),
        "announced while closing"
    );
}

/// The interval is re-drawn on each send, so a mesh full of clients started
/// together does not stay in step.
#[test]
fn the_announcement_interval_is_jittered() {
    let mut seen = std::collections::HashSet::new();
    for _ in 0..200 {
        let interval = next_hello_interval();
        assert!(
            (HELLO_BASE..=HELLO_BASE + HELLO_JITTER).contains(&interval),
            "interval {interval:?} outside {HELLO_BASE:?}..{:?}",
            HELLO_BASE + HELLO_JITTER
        );
        seen.insert(interval);
    }
    assert!(
        seen.len() >= 50,
        "only {} distinct intervals in 200 draws; the jitter is not varying",
        seen.len()
    );
}

/// A send that fails must not consume the interval: the firmware re-arms only
/// inside the branch where the transmit succeeded, so a failure is retried on
/// the next tick rather than costing a whole interval of silence. This client
/// has no socket, so the send fails.
#[test]
fn a_failed_announcement_does_not_consume_the_interval() {
    let (remote, tap) = client(203);
    lock(&remote.shared.schedule).set_hello_timer(None, Duration::ZERO);

    remote.shared.announce();

    assert_eq!(
        tap.opcodes(),
        vec![op::SYS_HELLO.0],
        "the frame did not reach the gate at all, so the timer proves nothing"
    );
    let (last, interval) = lock(&remote.shared.schedule).hello_timer();
    assert_eq!(
        (last, interval),
        (None, Duration::ZERO),
        "a failed send re-armed the timer; it should retry on the next tick"
    );
}

/// The decision and the send are tested above; this drives the loop that joins
/// them. Without it, deleting the call from the probe leaves every other
/// announcement test green - the pieces work and nothing announces.
#[test]
fn the_probe_loop_announces() {
    let (remote, tap) = client(204);
    lock(&remote.shared.schedule).set_hello_timer(Some(Instant::now()), Duration::from_millis(1));

    remote
        .spawn_workers()
        .expect("the worker threads could not start");
    let deadline = Instant::now() + Duration::from_secs(5);
    while Instant::now() < deadline && !tap.opcodes().contains(&op::SYS_HELLO.0) {
        std::thread::sleep(Duration::from_millis(50));
    }
    remote.close();

    assert!(
        tap.opcodes().contains(&op::SYS_HELLO.0),
        "the probe loop never announced, though the announcement was overdue"
    );
}

/// What the client says about itself has to parse as a hello, or no peer will
/// know it is there.
#[test]
fn the_announcement_describes_this_client() {
    let (remote, tap) = client(206);
    remote.shared.announce();

    let frame = tap.frames().pop().expect("nothing was announced");
    let peer = Remote::new(
        Config {
            uid: Some(uid_n(207)),
            ..Config::default()
        },
        Arc::new(()),
    )
    .expect("a client given its own uid reads nothing from disk");
    // A peer is what reads this frame; the sender drops its own as an echo.
    peer.shared
        .process_datagram(&frame, Ipv4Addr::new(10, 8, 8, 1));

    let seen = peer
        .device(remote.uid())
        .expect("a peer could not decode this client's announcement");
    assert_eq!(seen.name, "MXR Rust");
    assert_eq!(seen.serial, CLIENT_SERIAL);
    assert_eq!(seen.version, VERSION);
    assert_eq!(seen.supported_protocol, PROTOCOL_VERSION);
    assert!(seen.features.has(DeviceFeature::MANAGER));
    peer.close();
}

/// An event reaches the handler after the state lock is released, so a handler
/// may read the state the event describes without deadlocking.
#[test]
fn a_handler_may_read_the_state_its_event_describes() {
    struct Reentrant {
        remote: Mutex<Option<Arc<Remote>>>,
        seen: AtomicUsize,
    }

    impl EventHandler for Reentrant {
        fn on_device_update(&self, device: DeviceUid) {
            let remote = self.remote.lock().expect("set before any frame arrives");
            let remote = remote.as_ref().expect("set before any frame arrives");
            assert!(
                remote.device(device).is_some(),
                "the handler could not read the device its own event named"
            );
            self.seen.fetch_add(1, Ordering::SeqCst);
        }
    }

    let handler = Arc::new(Reentrant {
        remote: Mutex::new(None),
        seen: AtomicUsize::new(0),
    });
    let remote = Arc::new(
        Remote::new(
            Config {
                uid: Some(uid_n(208)),
                ..Config::default()
            },
            Arc::clone(&handler) as Arc<dyn EventHandler>,
        )
        .expect("a client given its own uid reads nothing from disk"),
    );
    *handler.remote.lock().expect("nothing else holds this") = Some(Arc::clone(&remote));

    // A device is not announced when it is first heard from: the hello it was
    // built from is the hello being applied, so nothing about it changed. The
    // second one renames it, which is a change.
    let peer = uid_n(209);
    remote.shared.process_datagram(
        &hello_datagram(peer, "Peer", "PR0002"),
        Ipv4Addr::new(10, 8, 8, 9),
    );
    assert_eq!(handler.seen.load(Ordering::SeqCst), 0);
    remote.shared.process_datagram(
        &hello_datagram(peer, "Renamed", "PR0002"),
        Ipv4Addr::new(10, 8, 8, 9),
    );
    assert!(
        handler.seen.load(Ordering::SeqCst) > 0,
        "no event reached the handler"
    );
}

/// Starting introduces this client before the caller can command anything.
///
/// A device drops every frame from a uid it has no record of, hello and
/// discover excepted, so a command sent before the hello is discarded by each
/// peer that has not met us - silently, since nothing answers a frame it
/// dropped and the send reports the bytes it wrote. Leaving this to the probe
/// loop would put the hello a tick late, which is invisible to a caller that
/// holds the client open for days and fatal to a script that starts and
/// immediately commands.
///
/// The discover matters as much: it is the other opcode a stranger may send,
/// and it makes every unit answer at once rather than at its own announcement
/// interval.
#[test]
fn starting_announces_and_solicits_before_it_returns() {
    let (remote, tap) = client(205);
    if remote.start().is_err() {
        // No usable multicast interface here; the ordering is untested rather
        // than shown to hold.
        return;
    }
    let opcodes = tap.opcodes();
    remote.close();

    assert_eq!(
        opcodes.first().copied(),
        Some(op::SYS_HELLO.0),
        "start returned without introducing this client"
    );
    assert!(
        opcodes.contains(&op::SYS_DISCOVER.0),
        "start returned without asking the network to describe itself"
    );
}

/// Collects the device events a probe pass produces.
#[derive(Default)]
struct Watching {
    completed: Mutex<Vec<DeviceUid>>,
    offline: Mutex<Vec<DeviceUid>>,
}

impl EventHandler for Watching {
    fn on_device_config_complete(&self, device: DeviceUid) {
        self.completed.lock().expect("test handler").push(device);
    }

    fn on_device_online_changed(&self, device: DeviceUid, online: bool) {
        if !online {
            self.offline.lock().expect("test handler").push(device);
        }
    }
}

/// Announces `peer` to `remote` as a video matrix, and gives it one bay.
///
/// A matrix is behind the link-configuration gate, so what it still owes after
/// this is its links and nothing else.
fn matrix_with_one_bay(remote: &Remote, peer: DeviceUid) {
    remote
        .shared
        .process_datagram(&hello_datagram(peer, "FF88", "PB0001"), FROM);
    remote.shared.process_datagram(
        &datagram(
            peer,
            op::SYS_BAY_CONFIG,
            PROTOCOL_VERSION,
            &bay_config_rec(
                1,
                1,
                0,
                "Output 1",
                "TV",
                BayStatus::NONE,
                BayFeatures::HDMI_OUT,
            ),
        ),
        FROM,
    );
}

/// Moves `peer`'s registration `age` into the past.
fn age(remote: &Remote, peer: DeviceUid, age: Duration) {
    remote.shared.mutate(|state, _| {
        let device = state.device_mut(peer).expect("peer not registered");
        device.first_seen = device
            .first_seen
            .checked_sub(age)
            .expect("the test clock cannot predate the process");
    });
}

/// A device that stops waiting for its links is announced from the probe pass.
///
/// Nothing arrives when a window closes, so no frame can carry the news: the
/// pass that re-tests completion is the whole of the announcement path. Driven
/// through that pass rather than through the device method, so the two cannot
/// come apart.
#[test]
fn the_probe_pass_announces_a_device_its_window_completed() {
    let seen = Arc::new(Watching::default());
    let (remote, _) = client_with(210, Arc::clone(&seen) as Arc<dyn EventHandler>);
    let peer = uid_n(211);
    matrix_with_one_bay(&remote, peer);

    remote.shared.probe_once(Instant::now());
    assert!(
        seen.completed.lock().expect("test handler").is_empty(),
        "inside the window the pass announced a device still owing its links"
    );

    age(&remote, peer, CONFIG_GRACE + Duration::from_secs(1));
    remote.shared.probe_once(Instant::now());
    assert_eq!(
        *seen.completed.lock().expect("test handler"),
        vec![peer],
        "the pass did not announce a device no further frame will complete"
    );
}

/// A device that drops off is noticed whether or not anything else finished.
///
/// Nothing on the wire marks a device gone - it stops pinging, and comparing
/// its last ping against the clock is the only thing that can tell. A device
/// drops off in whatever state it reached, including before it ever described
/// itself, so a client whose devices are all in that state is precisely the one
/// that must still be checking.
#[test]
fn a_device_that_never_described_itself_is_still_reported_gone() {
    let seen = Arc::new(Watching::default());
    let (remote, _) = client_with(212, Arc::clone(&seen) as Arc<dyn EventHandler>);
    let peer = uid_n(213);
    remote
        .shared
        .process_datagram(&hello_datagram(peer, "FF88", "PB0002"), FROM);

    // Long enough for the silence to count, and nothing has completed: the
    // device sent a hello and never a bay.
    remote
        .shared
        .probe_once(Instant::now() + Duration::from_secs(600));

    assert_eq!(
        *seen.offline.lock().expect("test handler"),
        vec![peer],
        "a device that stopped answering was never reported gone"
    );
}