rigless 0.2.0

// Copyright (c) 2026 Rigless Labs.
// SPDX-License-Identifier: MIT OR Apache-2.0
//! The far end of an asynchronous serial line: a chip, or a host.
//!
//! A line carries frames one way and frames the other, each on its own time. The controller times
//! what it sends and hands the far end each word as its stop bit ends. The far end offers what it
//! has to send and the controller times that too, so the same peer answers a controller at any
//! baud. A peer that's a chip answers on virtual time from what it was sent. A peer that's a host
//! answers with whatever a person or a program has written to it since the controller last asked.
//! That's why the controller asks at every stop of the session and not only at the end of each
//! frame. What's on the far end is the caller's business. A board hangs a chip, `rigless run` hangs
//! a [`Host`], a test hangs a stand-in, and the controller can't tell which.
//!
//! A line can also garble what arrives on it, a board condition: [`Errors`] says how often a word
//! arrives with noise or a framing error, and the controller asks it how each word came.

use std::any::Any;
use std::collections::VecDeque;
use std::io::{self, Read, Write};
use std::net::{TcpListener, TcpStream};

use crate::chip::Place;
use crate::chip::Timeline;
use crate::edge::draw;
use crate::snapshot::{self, Reader, Store, Writer};
use crate::time::{EventId, VirtualNanos};

/// What hangs on the far end of a serial line. A snapshot keeps it through its own [`Store`].
pub trait Peer: Store + std::fmt::Debug + Send + Any {
    /// A word the controller finished sending: eight or nine bits as the controller frames them, at the
    /// instant its stop bit ended, which `context` gives.
    fn receive(&mut self, word: u16, context: &mut Timeline<'_, '_>);

    /// An event this chip scheduled through the [`Timeline`] it was handed, at its instant. The
    /// controller hands every event of its instance to every chip it carries, after its own, so a
    /// chip ignores an id it didn't ask for. A silenced chip hears its events all the same, since
    /// what it does with time goes on while it's cut off. Nothing, for a chip that schedules none.
    fn on_event(&mut self, _event: EventId, _context: &mut Timeline<'_, '_>) {}

    /// The part this chip hangs on restarted at the instant `context` gives, and every event the
    /// chip scheduled through its controller went with the reset, since a chip keeps running
    /// through its part's reset. A chip that keeps time schedules again here. Nothing, for a chip
    /// that schedules none.
    fn restarted(&mut self, _context: &mut Timeline<'_, '_>) {}

    /// The next word this peer has put on the line by `now`, oldest first, or `None` once it has nothing
    /// more due by then, so a caller can take everything that's due.
    ///
    /// The controller asks whenever its receiver is free: at the end of each frame it takes, and at every
    /// stop of the session, so a word a host wrote lands at the stop it was noticed at and one a chip
    /// scheduled lands at the first stop after its instant.
    fn transmit(&mut self, now: VirtualNanos) -> Option<u16>;

    /// Every word this peer has been sent, oldest first, or nothing where it keeps no record.
    fn received(&self) -> &[u16] {
        &[]
    }

    /// Forgets what this peer kept past `now`, since the run, put back to before what it kept to
    /// bring the run back the same way, goes a new way from there.
    fn forget_after(&mut self, now: VirtualNanos) {
        let _ = now;
    }
}

impl Place<dyn Peer> {
    /// While the far end is silent, takes every word it has put on the line by `now` and loses it, as a
    /// cut line does, so it has nothing stale to say once it answers again. Each is a transaction, as a
    /// word it's heard sending is. It does nothing while the far end answers.
    pub fn lose_while_silent(&mut self, now: VirtualNanos) {
        if self.answering().is_none() {
            while self.chip_mut().transmit(now).is_some() {
                self.begin();
                self.release();
            }
        }
    }
}

const HIGH_HALF: u32 = 32;

/// Words arriving on a line garbled, a board condition: how many in a million come with noise, and
/// how many with a framing error. Which ones do is drawn for each word, keyed by the run's seed, the
/// name of the error's condition and how many words the line had carried before it, so the pattern
/// doesn't move with when the firmware reads. [`crate::edge::draw`].
///
/// A controller keeps one for its line and asks [`Errors::next`] how each word came as its frame
/// ends. The board sets the rates and the seed.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(bound(deserialize = "'de: 'static"))]
pub struct Errors {
    /// Words arriving with noise on them.
    pub noise: Rate,
    /// Words arriving with a framing error.
    pub framing: Rate,
    /// The run's seed.
    pub seed: u64,
    /// How many words the line has carried.
    words: u64,
}

/// How often words arrive with one kind of error, and the name its draws are keyed by.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct Rate {
    /// The condition's name, as the board lists it.
    pub name: &'static str,
    /// Words in a million. [`Rate::EVERY_WORD`] or more is every word, and 0 draws nothing.
    pub per_million: u32,
}

/// How one word arrived.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct Arrival {
    /// It came with noise on it.
    pub noise: bool,
    /// It came with a framing error.
    pub framing: bool,
    /// The data bits the errors flipped, none when it came clean and never none when it didn't. The
    /// word that arrives is the one sent with these bits flipped.
    pub flipped: u16,
}

impl Errors {
    /// How the next word came, when its data is `bits` bits wide, from 1 to 16.
    ///
    /// Each kind with a rate draws once for the word, at the count of words before it. The draw's
    /// high 32 bits against the rate say whether the word has that error. When it does, the draw
    /// modulo `2^bits − 1`, plus one, is the data bits it flips, so an errored word never arrives as
    /// it was sent. Noise's draw picks them when the word has both errors.
    pub fn next(&mut self, bits: u32) -> Arrival {
        let count = self.words;
        self.words += 1;
        let noise = self.noise.drawn(self.seed, count);
        let framing = self.framing.drawn(self.seed, count);
        let every_pattern = (1_u64 << bits.clamp(1, u16::BITS)) - 1;
        let flipped = noise.or(framing).map_or(0, |drawn| {
            1 + u16::try_from(drawn % every_pattern).expect("below a u16")
        });
        Arrival {
            noise: noise.is_some(),
            framing: framing.is_some(),
            flipped,
        }
    }
}

impl Rate {
    /// Words in a million at which every word errs.
    pub const EVERY_WORD: u32 = 1_000_000;

    /// The draw for the word after the line's `count`th, when that word comes with this error.
    fn drawn(self, seed: u64, count: u64) -> Option<u64> {
        if self.per_million == 0 {
            return None;
        }
        let drawn = draw(seed, self.name, count);
        let out_of_a_million = ((drawn >> HIGH_HALF) * u64::from(Self::EVERY_WORD)) >> HIGH_HALF;
        (out_of_a_million < u64::from(self.per_million)).then_some(drawn)
    }
}

/// How many bytes one look at the socket takes at most.
const CHUNK: usize = 4096;

/// How much virtual time passes between two looks at the socket for a connection or for bytes. A
/// port asks its far end at every frame and at every stop, tens of thousands of times a virtual
/// second, and a look is a system call. Once a millisecond is a thousand a virtual second, which is
/// nothing, and a host that connects or writes waits at most that long to be noticed.
const LOOK_PERIOD: VirtualNanos = VirtualNanos(1_000_000);

/// A host on the far end, over a TCP socket the runtime listens on.
///
/// Bytes the host writes are taken when the controller asks, and offered oldest first. Bytes the
/// controller sends are written straight through. A word wider than a byte crosses as its low byte.
/// Nothing here waits. A host that hasn't connected, or has gone, is a far end with nothing to say
/// that hears nothing, and the run neither pauses for it nor notices it. A wire has one far end, so
/// a host that connects while another is there takes its place, the way a cable moved from one
/// ground station to another does. What the first had written and the port hadn't yet taken goes
/// with it.
///
/// A snapshot keeps what the port hasn't taken yet and when the socket was last looked at. Past
/// that the host keeps what every look brought, so a run put back to before them hears the same
/// bytes at the same looks until it's past the last look it made, and how many bytes a host has
/// had, so a run put back sends none of those again.
// ceiling: every byte a host sent for the whole run; upgrade when a run's host traffic outgrows
// memory
#[derive(Debug)]
pub struct Host {
    listener: TcpListener,
    stream: Option<TcpStream>,
    incoming: VecDeque<u8>,
    outgoing: VecDeque<u8>,
    /// How many bytes the port has sent along the run, the last of them at the back of `outgoing`.
    sent: u64,
    /// How many of the bytes the port sent a host has had, or nobody was there to have, along the
    /// furthest the run has been.
    written: u64,
    /// When the socket was last looked at, or never.
    looked: Option<VirtualNanos>,
    /// What each look at the socket brought, by when it looked.
    heard: Vec<(VirtualNanos, Heard)>,
    /// The latest look at the socket, which a run put back to before it doesn't make again.
    furthest: Option<VirtualNanos>,
}

/// What a look at the socket brought.
#[derive(Debug)]
enum Heard {
    /// A host connected, which drops what the one before it had written.
    Connected,
    /// Bytes the host wrote.
    Bytes(Vec<u8>),
}

impl Host {
    /// Listens on `port` of the loopback interface for a host. The latest to connect is the one on
    /// the wire.
    pub fn listen(port: u16) -> io::Result<Self> {
        let listener = TcpListener::bind(("127.0.0.1", port))?;
        listener.set_nonblocking(true)?;
        Ok(Self {
            listener,
            stream: None,
            incoming: VecDeque::new(),
            outgoing: VecDeque::new(),
            sent: 0,
            written: 0,
            looked: None,
            heard: Vec::new(),
            furthest: None,
        })
    }

    /// The port the runtime listens on. It's the one you asked for, or whatever the system gave if
    /// you asked for zero.
    pub fn port(&self) -> io::Result<u16> {
        self.listener.local_addr().map(|address| address.port())
    }

    /// Looks at the socket once a [`LOOK_PERIOD`] for a connection and for bytes, and writes what
    /// the port sent whenever there's a host to write to.
    fn serve(&mut self, now: VirtualNanos) {
        let look = self
            .looked
            .is_none_or(|looked| now.0 >= looked.0.saturating_add(LOOK_PERIOD.0));
        if look {
            self.looked = Some(now);
            if self.furthest.is_some_and(|furthest| now <= furthest) {
                self.hear_again(now);
            } else {
                self.furthest = Some(now);
                self.hear(now);
            }
        }
        self.write();
    }

    /// Takes a host that's connecting, and what the one on the wire wrote.
    fn hear(&mut self, now: VirtualNanos) {
        if let Ok((stream, _)) = self.listener.accept()
            && stream.set_nonblocking(true).is_ok()
        {
            self.stream = Some(stream);
            self.incoming.clear();
            self.heard.push((now, Heard::Connected));
        }
        let Some(stream) = &mut self.stream else {
            return;
        };
        let mut buffer = [0; CHUNK];
        loop {
            match stream.read(&mut buffer) {
                Ok(0) => {
                    self.stream = None;
                    break;
                }
                Ok(count) => {
                    self.incoming.extend(&buffer[..count]);
                    self.heard
                        .push((now, Heard::Bytes(buffer[..count].to_vec())));
                }
                Err(error) if error.kind() == io::ErrorKind::WouldBlock => break,
                Err(error) if error.kind() == io::ErrorKind::Interrupted => {}
                Err(_) => {
                    self.stream = None;
                    break;
                }
            }
        }
    }

    /// A run put back to before the latest look at `now`: a look then brings what it brought the
    /// first time.
    fn hear_again(&mut self, now: VirtualNanos) {
        let first = self.heard.partition_point(|&(at, _)| at < now);
        for (_, heard) in self.heard[first..].iter().take_while(|&&(at, _)| at == now) {
            match heard {
                Heard::Connected => self.incoming.clear(),
                Heard::Bytes(bytes) => self.incoming.extend(bytes),
            }
        }
    }

    /// Writes what the port sent that no host has had, as far as the host takes it, and drops it
    /// with nobody there to have it. What a run put back sends again, a host had the first time.
    fn write(&mut self) {
        let mut front = self.sent - self.outgoing.len() as u64;
        while front < self.written && self.outgoing.pop_front().is_some() {
            front += 1;
        }
        match &mut self.stream {
            None => self.outgoing.clear(),
            Some(stream) => {
                while let Some(&byte) = self.outgoing.front() {
                    match stream.write(&[byte]) {
                        Ok(_) => {
                            self.outgoing.pop_front();
                        }
                        Err(error) if error.kind() == io::ErrorKind::Interrupted => {}
                        Err(error) if error.kind() == io::ErrorKind::WouldBlock => break,
                        Err(_) => {
                            self.stream = None;
                            self.outgoing.clear();
                            break;
                        }
                    }
                }
            }
        }
        let had = self.sent - self.outgoing.len() as u64;
        self.written = self.written.max(had);
    }
}

/// A snapshot keeps what the port hasn't taken yet and when the socket was last looked at. Who's on
/// the socket, and what every look brought, stay as they are.
impl Store for Host {
    fn store(&self, to: &mut Writer) -> Result<(), snapshot::Error> {
        to.put(&self.incoming)?;
        to.put(&self.outgoing)?;
        to.put(&self.sent)?;
        to.put(&self.looked)
    }

    fn load(&mut self, from: &mut Reader) -> Result<(), snapshot::Error> {
        self.incoming = from.take()?;
        self.outgoing = from.take()?;
        self.sent = from.take()?;
        self.looked = from.take()?;
        Ok(())
    }
}

impl Peer for Host {
    fn receive(&mut self, word: u16, context: &mut Timeline<'_, '_>) {
        self.outgoing
            .push_back(u8::try_from(word & 0xFF).expect("masked to a byte"));
        self.sent += 1;
        self.serve(context.now());
    }

    fn transmit(&mut self, now: VirtualNanos) -> Option<u16> {
        self.serve(now);
        self.incoming.pop_front().map(u16::from)
    }

    /// What every look past `now` brought goes, and what the port sends from here is the new way's,
    /// none of it had by a host yet.
    fn forget_after(&mut self, now: VirtualNanos) {
        self.heard.retain(|&(at, _)| at <= now);
        self.furthest = self.furthest.map(|furthest| furthest.min(now));
        self.written = self.written.min(self.sent);
    }
}

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

    const BYTE: u32 = 8;
    const EVERY: u32 = Rate::EVERY_WORD;

    fn errors(noise: u32, framing: u32, seed: u64) -> Errors {
        Errors {
            noise: Rate {
                name: "line.noise_errors",
                per_million: noise,
            },
            framing: Rate {
                name: "line.framing_errors",
                per_million: framing,
            },
            seed,
            ..Errors::default()
        }
    }

    fn arrivals(mut line: Errors, words: usize) -> Vec<Arrival> {
        (0..words).map(|_| line.next(BYTE)).collect()
    }

    #[test]
    fn no_rate_is_a_clean_line_and_a_million_errs_every_word_with_some_bit_flipped() {
        assert!(
            arrivals(errors(0, 0, 9), 1_000)
                .iter()
                .all(|arrival| *arrival == Arrival::default())
        );
        let every = arrivals(errors(EVERY, EVERY, 9), 1_000);
        let noise_alone = arrivals(errors(EVERY, 0, 9), 1_000);
        for (both, noise) in every.iter().zip(&noise_alone) {
            assert!(both.noise && both.framing, "{both:?}");
            assert!((1..=0xFF).contains(&both.flipped), "{both:?}");
            assert_eq!(both.flipped, noise.flipped, "noise picks the bits");
        }
    }

    #[test]
    fn the_flipped_bits_stay_inside_the_data_however_wide_it_is() {
        for bits in [1, 5, 7, 8, 9, 16] {
            let data = (1_u32 << bits) - 1;
            let mut line = errors(EVERY, 0, u64::from(bits));
            for _ in 0..1_000 {
                let flipped = u32::from(line.next(bits).flipped);
                assert!(
                    flipped != 0 && flipped & !data == 0,
                    "{bits} bits: {flipped:#x}"
                );
            }
        }
    }

    #[test]
    #[cfg_attr(miri, ignore = "a hundred thousand words is beyond Miri")]
    fn a_rate_errs_about_that_many_words_and_the_pattern_is_the_seeds_the_names_and_the_counts() {
        let noisy = arrivals(errors(100_000, 0, 3), 100_000);
        let erred = noisy.iter().filter(|arrival| arrival.noise).count();
        assert!((9_500..10_500).contains(&erred), "{erred} in 100 000");
        assert_eq!(noisy, arrivals(errors(100_000, 0, 3), 100_000));
        assert_ne!(
            noisy[..64],
            arrivals(errors(100_000, 0, 4), 64),
            "another seed, another pattern"
        );
        let both = arrivals(errors(100_000, 500_000, 3), 64);
        assert!(
            both.iter()
                .zip(&noisy)
                .all(|(both, noise)| both.noise == noise.noise),
            "one kind's rate never moves the other's pattern"
        );
    }
}

#[cfg(test)]
mod host_tests {
    use std::net::TcpStream;

    use super::*;

    /// A byte the host wrote is there by the next look, and not before the period has passed.
    #[test]
    fn the_socket_is_looked_at_once_a_period_and_a_byte_arrives_by_the_next_look() {
        let mut host = Host::listen(0).unwrap();
        let mut client = TcpStream::connect(("127.0.0.1", host.port().unwrap())).unwrap();
        client.write_all(b"a").unwrap();
        // The first look connects and reads. The socket may not have the byte yet, so keep looking
        // one period at a time, the way a run does.
        let mut now = VirtualNanos(0);
        let first = loop {
            if let Some(word) = host.transmit(now) {
                break word;
            }
            now = VirtualNanos(now.0 + LOOK_PERIOD.0);
            assert!(now.0 < 1_000 * LOOK_PERIOD.0, "the byte never arrived");
        };
        assert_eq!(first, u16::from(b'a'));
        let looked_at = now;
        client.write_all(b"b").unwrap();
        for _ in 0..100 {
            assert_eq!(
                host.transmit(looked_at),
                None,
                "within the period the socket is not looked at again"
            );
        }
        let mut now = VirtualNanos(looked_at.0 + LOOK_PERIOD.0);
        let second = loop {
            if let Some(word) = host.transmit(now) {
                break word;
            }
            now = VirtualNanos(now.0 + LOOK_PERIOD.0);
            assert!(
                now.0 < looked_at.0 + 1_000 * LOOK_PERIOD.0,
                "the byte never arrived"
            );
        };
        assert_eq!(second, u16::from(b'b'));
    }

    /// The port sending `bytes` at `now`, as [`Peer::receive`] has it send a word.
    fn send(host: &mut Host, bytes: &[u8], now: VirtualNanos) {
        host.outgoing.extend(bytes);
        host.sent += bytes.len() as u64;
        host.serve(now);
    }

    /// Asserts the host has had nothing more, a tenth of a second on.
    fn nothing_more(client: &mut TcpStream) {
        client
            .set_read_timeout(Some(std::time::Duration::from_millis(100)))
            .unwrap();
        assert!(client.read(&mut [0]).is_err(), "the host had a byte twice");
    }

    /// The look `look` periods into the run.
    fn look(look: u64) -> VirtualNanos {
        VirtualNanos(look * LOOK_PERIOD.0)
    }

    /// A host on the wire, taken at the first look that finds it, and the next look after it.
    fn connected() -> (Host, TcpStream, u64) {
        let mut host = Host::listen(0).unwrap();
        let client = TcpStream::connect(("127.0.0.1", host.port().unwrap())).unwrap();
        let mut at = 0;
        while host.stream.is_none() {
            host.serve(look(at));
            at += 1;
            assert!(at < 1_000, "the host was never taken");
        }
        (host, client, at)
    }

    /// Each word the port takes from the looks from `from` on, with the look it came at, until
    /// `count` have come, and the look after the last.
    fn heard(host: &mut Host, from: u64, count: usize) -> (Vec<(u64, u16)>, u64) {
        let mut words = Vec::new();
        let mut at = from;
        while words.len() < count {
            if let Some(word) = host.transmit(look(at)) {
                words.push((at, word));
            } else {
                at += 1;
            }
            assert!(at < from + 10_000, "only {words:?} came");
        }
        (words, at + 1)
    }

    /// What the host wrote after a save comes again after a restore, at the same looks, however
    /// little the host writes now, and what the port hadn't taken yet when it was saved is there to
    /// take again. What the port sent the host after the save, and sends again after the restore,
    /// the host has once.
    #[test]
    fn a_host_heard_after_a_save_is_heard_the_same_after_a_restore() {
        let (mut host, mut client, first) = connected();
        client.write_all(b"before").unwrap();
        let (_, first) = heard(&mut host, first, 1);
        let saved = snapshot::stored(&host).unwrap();
        let (waiting, next) = heard(&mut host, first, 5);
        assert_eq!(
            waiting.iter().map(|&(_, word)| word).collect::<Vec<_>>(),
            b"efore".map(u16::from),
            "\"efore\" was waiting"
        );
        client.write_all(b"after").unwrap();
        send(&mut host, b"sent", look(next));
        let (after, last) = heard(&mut host, next, 5);
        snapshot::load(&mut host, snapshot::keep(saved)).unwrap();
        client.write_all(b"never").unwrap();
        assert_eq!(heard(&mut host, first, 5).0, waiting);
        send(&mut host, b"sent", look(next));
        assert_eq!(heard(&mut host, next, 5).0, after);
        let (past, _) = heard(&mut host, last, 5);
        assert_eq!(
            past.iter().map(|&(_, word)| word).collect::<Vec<_>>(),
            b"never".map(u16::from),
            "past the last look, the socket again"
        );
        let mut had = [0; 4];
        client.read_exact(&mut had).unwrap();
        assert_eq!(&had, b"sent");
        nothing_more(&mut client);
    }

    /// What the port sends a host that's slow to read is written as the host makes room, however
    /// often the run serves the socket at one instant, so every byte comes, in order.
    #[test]
    #[cfg_attr(miri, ignore = "eight mebibytes through a socket is beyond Miri")]
    #[expect(
        clippy::disallowed_methods,
        reason = "the host reads on a thread of its own while the run writes"
    )]
    fn a_slow_host_has_every_byte_the_port_sent() {
        // More than the socket's buffers hold, so writing it has to wait on the host.
        const BURST: usize = 8 << 20;
        let (mut host, mut client, at) = connected();
        let burst: Vec<u8> = (0..=u8::MAX).cycle().take(BURST).collect();
        send(&mut host, &burst, look(at));
        host.serve(look(at));
        assert!(!host.outgoing.is_empty(), "the host has yet to make room");
        let reading = std::thread::spawn(move || {
            let mut had = vec![0; BURST];
            client.read_exact(&mut had).map(|()| had)
        });
        let mut now = at;
        while !host.outgoing.is_empty() {
            now += 1;
            host.serve(look(now));
            assert!(now < at + 1_000_000, "the host never made room");
        }
        assert!(
            reading.join().unwrap().unwrap() == burst,
            "every byte, in order"
        );
    }

    /// A run that goes a new way after a restore sends the host what it sends from there, and its
    /// looks from there are its own.
    #[test]
    fn a_new_way_sends_the_host_what_it_sends() {
        let (mut host, mut client, first) = connected();
        let saved = snapshot::stored(&host).unwrap();
        client.write_all(b"old").unwrap();
        send(&mut host, b"old", look(first));
        let (old, next) = heard(&mut host, first, 3);
        snapshot::load(&mut host, snapshot::keep(saved)).unwrap();
        host.forget_after(look(first - 1));
        send(&mut host, b"new", look(first));
        client.write_all(b"new").unwrap();
        let (new, _) = heard(&mut host, first, 3);
        assert_eq!(
            new.iter().map(|&(_, word)| word).collect::<Vec<_>>(),
            b"new".map(u16::from),
            "the old way's {old:?}, to look {next}, go"
        );
        let mut had = [0; 6];
        client.read_exact(&mut had).unwrap();
        assert_eq!(&had, b"oldnew");
        nothing_more(&mut client);
    }
}