lernie 0.1.70

//! **The far end of a punched wire**, so the ladder can be tested against
//! something that speaks the protocol on a HELD line: an engine that serves
//! every request that lands on one connection, pings its silence, and ends
//! the connection every way a real one can end (DESIGN §4.40).
//!
//! It stands in two ways. [`Roving::listen`] serves what lands on a listener
//! the test bound — the endpoint a presence item names, where the seat's
//! SYN lands. [`Roving::call_back`] is the engine's inbox loop in miniature:
//! it watches a fake DHT node's store for a call under the pairing, opens
//! it, and connects TO the seat's endpoints — the case where the engine's
//! SYN lands on the seat's own punch listener.
//!
//! Like [`engine`](super::engine) it listens, which a seat never does, and
//! that is why it lives here.

use std::collections::VecDeque;
use std::io::Write;
use std::net::{Shutdown, TcpListener, TcpStream};
use std::path::Path;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::{Arc, Mutex, PoisonError};
use std::time::Duration;

use rustls::{ServerConfig, ServerConnection, StreamOwned};
use serde_json::{Value, json};

use crate::channel::frame;
use crate::channel::line;
use crate::channel::rendezvous::Pairing;
use crate::channel::rendezvous::item::Call;
use crate::dht::Mutable;

/// What the connection does once an answer is written.
#[derive(Clone, Copy, Debug)]
pub(crate) enum Fate {
    /// Hold the line, as a real engine does.
    Stay,
    /// Hold the line and ping its silence this many times, as a real engine
    /// does at 25-second intervals — here at once, so the pings sit in the
    /// seat's socket before its next ask.
    Pinged(usize),
    /// Close without a word: a FIN and nothing else.
    Fin,
    /// Say `close_notify`, then close — the engine's own hang-up.
    Farewell,
    /// Reset: linger zero, so the peer reads a connection reset — once the
    /// test says the answer was taken ([`Roving::taken`]).
    Reset,
    /// Write bytes that are not TLS, then hold — a peer gone wrong — once the
    /// test says the answer was taken ([`Roving::taken`]).
    Garbage,
}

/// One request's answer: pings written into the line first, the frames,
/// and what the connection does afterwards.
#[derive(Clone)]
pub(crate) struct Reply {
    pub(crate) pings: usize,
    pub(crate) frames: Vec<Value>,
    pub(crate) fate: Fate,
}

impl Reply {
    /// The ordinary shape: one frame that says yes, the line held.
    pub(crate) fn yes() -> Reply {
        Reply {
            pings: 0,
            frames: vec![json!({"ok": true, "kind": "workspaces"})],
            fate: Fate::Stay,
        }
    }

    pub(crate) fn then(self, fate: Fate) -> Reply {
        Reply { fate, ..self }
    }

    pub(crate) fn pinged(self, pings: usize) -> Reply {
        Reply { pings, ..self }
    }
}

/// What every connection of one stand-in shares with the test.
struct Shared {
    script: Mutex<VecDeque<Reply>>,
    seen: Mutex<Vec<Value>>,
    connections: AtomicUsize,
    /// Pings written, in answers and into silences alike.
    pings: AtomicUsize,
    /// The test's word that the answer is in its hands.
    taken: AtomicBool,
}

/// The stand-in, and what it was told.
pub(crate) struct Roving {
    shared: Arc<Shared>,
}

impl Roving {
    /// Serve every connection that lands on `listener`, one [`Reply`] per
    /// request across all of them, in order.
    pub(crate) fn listen(dir: &Path, listener: TcpListener, script: Vec<Reply>) -> Roving {
        let config = super::engine::server_config(dir);
        let roving = Roving::new(script);
        let shared = Arc::clone(&roving.shared);
        std::thread::spawn(move || {
            while let Ok((tcp, _)) = listener.accept() {
                shared.connections.fetch_add(1, Ordering::Relaxed);
                let (config, shared) = (Arc::clone(&config), Arc::clone(&shared));
                std::thread::spawn(move || serve(&config, tcp, &shared));
            }
        });
        roving
    }

    /// Watch `store` for a call under `pairing`, connect to what it names,
    /// and serve. One call, one connection: the engine remembers the nonce.
    pub(crate) fn call_back(
        dir: &Path,
        store: Arc<Mutex<Vec<Mutable>>>,
        pairing: Pairing,
        script: Vec<Reply>,
    ) -> Roving {
        let config = super::engine::server_config(dir);
        let roving = Roving::new(script);
        let shared = Arc::clone(&roving.shared);
        std::thread::spawn(move || {
            let inbox = pairing.inbox_keypair().unwrap().public();
            // Sleep first: the call is never there before the seat has
            // walked, and a loop that could break on its first pass leaves
            // its sleep uncovered whenever the timing says it did.
            let call = loop {
                std::thread::sleep(Duration::from_millis(20));
                let found = store
                    .lock()
                    .unwrap_or_else(PoisonError::into_inner)
                    .iter()
                    .find(|item| item.key == inbox)
                    .and_then(|item| Call::open(&pairing.seal_key(), &item.value));
                if let Some(call) = found {
                    break call;
                }
            };
            for endpoint in call.endpoints {
                if let Ok(tcp) = TcpStream::connect(endpoint) {
                    shared.connections.fetch_add(1, Ordering::Relaxed);
                    serve(&config, tcp, &shared);
                    return;
                }
            }
        });
        roving
    }

    fn new(script: Vec<Reply>) -> Roving {
        Roving {
            shared: Arc::new(Shared {
                script: Mutex::new(script.into()),
                seen: Mutex::new(Vec::new()),
                connections: AtomicUsize::new(0),
                pings: AtomicUsize::new(0),
                taken: AtomicBool::new(false),
            }),
        }
    }

    /// Every frame it has been handed, in order and across connections.
    pub(crate) fn heard(&self) -> Vec<Value> {
        self.shared
            .seen
            .lock()
            .unwrap_or_else(PoisonError::into_inner)
            .clone()
    }

    /// How many connections have reached it.
    pub(crate) fn connections(&self) -> usize {
        self.shared.connections.load(Ordering::Relaxed)
    }

    /// How many pings it has written. On loopback a write that returned is
    /// in the seat's socket, so this is what says the pings arrived.
    pub(crate) fn pinged(&self) -> usize {
        self.shared.pings.load(Ordering::Relaxed)
    }

    /// The test has the answer: a [`Fate::Reset`] or [`Fate::Garbage`] may
    /// now end the line. A reset discards what the peer has not yet read, and
    /// bytes that are not TLS poison the record the answer rode in with, so
    /// both wait on this word — a pause in its place was a bet against the
    /// box's load, and a loaded one lost it before the seat had read the
    /// answer (bl-73f2).
    pub(crate) fn taken(&self) {
        self.shared.taken.store(true, Ordering::Relaxed);
    }
}

/// One connection: state a version, then per request record what the seat
/// wrote and answer the next [`Reply`]. The seat's own preface is the first
/// frame of a fresh connection and is recorded like any other.
fn serve(config: &Arc<ServerConfig>, tcp: TcpStream, shared: &Shared) {
    let conn = ServerConnection::new(Arc::clone(config)).expect("the stand-in's own config");
    let mut tls = StreamOwned::new(conn, tcp);
    let _ = frame::write_value(
        &mut tls,
        &json!({ "protocol": crate::channel::hello::PROTOCOL }),
    );
    let mut preface_owed = true;
    while let Ok(Some(said)) = frame::read_value(&mut tls) {
        shared
            .seen
            .lock()
            .unwrap_or_else(PoisonError::into_inner)
            .push(said);
        if preface_owed {
            preface_owed = false;
            continue;
        }
        let Some(reply) = shared
            .script
            .lock()
            .unwrap_or_else(PoisonError::into_inner)
            .pop_front()
        else {
            return;
        };
        ping(&mut tls, reply.pings, shared);
        for value in &reply.frames {
            let _ = frame::write_value(&mut tls, value);
        }
        let _ = frame::write_end(&mut tls);
        match reply.fate {
            Fate::Stay => {}
            Fate::Pinged(n) => ping(&mut tls, n, shared),
            Fate::Fin => return,
            Fate::Farewell => {
                tls.conn.send_close_notify();
                let _ = tls.flush();
                let _ = tls.sock.shutdown(Shutdown::Both);
                return;
            }
            Fate::Reset => {
                taken(shared);
                let _ = socket2::SockRef::from(&tls.sock).set_linger(Some(Duration::ZERO));
                return;
            }
            Fate::Garbage => {
                taken(shared);
                let _ = tls.sock.write_all(b"this is not a TLS record");
                std::thread::sleep(Duration::from_secs(1));
                return;
            }
        }
    }
}

/// Write `n` pings, counted once each is in the socket.
fn ping(tls: &mut StreamOwned<ServerConnection, TcpStream>, n: usize, shared: &Shared) {
    for _ in 0..n {
        let _ = frame::write_value(tls, &line::ping());
        shared.pings.fetch_add(1, Ordering::Relaxed);
    }
}

/// Wait for the test's word that the answer was taken ([`Roving::taken`]).
/// Sleep first, so the sleep is run.
fn taken(shared: &Shared) {
    loop {
        std::thread::sleep(Duration::from_millis(5));
        if shared.taken.load(Ordering::Relaxed) {
            break;
        }
    }
}