liminal-server 0.25.4

use crate::link_handshake::{LinkHandshake, LinkTransport, LocalMachine, NOISE_OVERHEAD};
use crate::link_open::Path;
use crate::link_open::carriers::DirectPath;
use crate::link_udp::{PunchSocket, UdpCarrier, UdpError};
use lys_core::keys::Ed25519Identity;
use std::io;
use std::net::SocketAddr;
use std::sync::{Arc, Mutex, mpsc};
use std::time::Duration;
use zeroize::Zeroizing;

type TestResult = Result<(), Box<dyn std::error::Error>>;
type Arrival = (Vec<u8>, SocketAddr);

struct SignalSocket {
    outgoing: mpsc::Sender<Arrival>,
    incoming: Mutex<mpsc::Receiver<Arrival>>,
    address: SocketAddr,
}

impl SignalSocket {
    fn new() -> Result<Self, io::Error> {
        let (outgoing, incoming) = mpsc::channel();
        Ok(Self {
            outgoing,
            incoming: Mutex::new(incoming),
            address: "127.0.0.1:41000".parse().map_err(io::Error::other)?,
        })
    }
}

impl PunchSocket for SignalSocket {
    fn send_to(&self, bytes: &[u8], target: SocketAddr) -> io::Result<usize> {
        self.outgoing
            .send((bytes.to_vec(), target))
            .map_err(io::Error::other)?;
        Ok(bytes.len())
    }

    fn recv_from(&self, bytes: &mut [u8]) -> io::Result<(usize, SocketAddr)> {
        let (message, address) = self
            .incoming
            .lock()
            .map_err(|error| io::Error::other(error.to_string()))?
            .recv()
            .map_err(io::Error::other)?;
        let output = bytes
            .get_mut(..message.len())
            .ok_or_else(|| io::Error::other("datagram buffer too small"))?;
        output.copy_from_slice(&message);
        Ok((message.len(), address))
    }

    fn local_addr(&self) -> io::Result<SocketAddr> {
        Ok(self.address)
    }

    fn prepare(&self, bound: Option<Duration>) -> io::Result<()> {
        if bound.is_some() {
            return Err(io::Error::other(
                "signal socket requires an unbounded receive",
            ));
        }
        Ok(())
    }
}

fn transport() -> Result<LinkTransport, Box<dyn std::error::Error>> {
    let local = LocalMachine::new(Ed25519Identity::from_seed(&Zeroizing::new([1; 32])));
    let peer = LocalMachine::new(Ed25519Identity::from_seed(&Zeroizing::new([2; 32])));
    let mut initiator = LinkHandshake::initiator(&local, peer.machine())?;
    let mut responder = LinkHandshake::responder(&peer, local.machine())?;
    let mut bytes = [0; NOISE_OVERHEAD];
    let written = initiator.write(&mut bytes)?;
    responder.read(&bytes[..written])?;
    let written = responder.write(&mut bytes)?;
    initiator.read(&bytes[..written])?;
    Ok(initiator.finish()?)
}

fn carrier() -> Result<Arc<UdpCarrier>, Box<dyn std::error::Error>> {
    Ok(Arc::new(UdpCarrier::from_socket(Arc::new(
        SignalSocket::new()?,
    ))?))
}

#[test]
fn retired_direct_interrupt_keeps_the_other_receiver_index_open() -> TestResult {
    let carrier = carrier()?;
    let endpoint = "127.0.0.1:42000".parse()?;
    carrier.add_link(11, 21, endpoint, transport()?)?;
    carrier.add_link(12, 22, endpoint, transport()?)?;
    let first = DirectPath::new(carrier.clone(), 11)?;
    let other = DirectPath::new(carrier.clone(), 12)?;
    let stop = first.interrupt_handle().ok_or("direct interrupt absent")?;
    // Worker retirement may complete before its owner's interruption call.
    drop(first);
    let result = stop();
    assert!(
        result.is_ok(),
        "retired receiver interruption looked up a removed index: {result:?}"
    );
    assert!(matches!(
        carrier.cancel_link(11),
        Err(UdpError::UnknownIndex)
    ));
    other.refuse_unsolicited()?;
    assert_eq!(carrier.seal(12, b"remaining export")?.receiver_index, 22);
    drop(other);
    drop(stop);
    let carrier = Arc::try_unwrap(carrier).map_err(|_| "carrier still shared")?;
    carrier.shutdown()?;
    Ok(())
}

#[test]
fn retired_direct_interrupt_cannot_cancel_a_reused_receiver_index() -> TestResult {
    let carrier = carrier()?;
    let endpoint = "127.0.0.1:42000".parse()?;
    carrier.add_link(11, 21, endpoint, transport()?)?;
    let original = DirectPath::new(carrier.clone(), 11)?;
    let stop = original
        .interrupt_handle()
        .ok_or("direct interrupt absent")?;
    drop(original);
    carrier.add_link(11, 31, endpoint, transport()?)?;
    let replacement = DirectPath::new(carrier.clone(), 11)?;
    stop()?;
    let observed = replacement.refuse_unsolicited();
    assert!(
        observed.is_ok(),
        "retired interruption cancelled the reused receiver index: {observed:?}"
    );
    assert_eq!(carrier.seal(11, b"replacement export")?.receiver_index, 31);
    drop(replacement);
    drop(stop);
    let carrier = Arc::try_unwrap(carrier).map_err(|_| "carrier still shared")?;
    carrier.shutdown()?;
    Ok(())
}