liminal-server 0.25.4

use super::{BridgeRegistry, Observation, Wake};
use std::io::Read;
use std::net::{TcpListener, TcpStream};
use std::sync::{Arc, Mutex};

fn observed_carrier() -> Result<(Arc<Observation>, TcpStream), String> {
    let listener = TcpListener::bind("127.0.0.1:0").map_err(|error| error.to_string())?;
    let carrier = TcpStream::connect(listener.local_addr().map_err(|error| error.to_string())?)
        .map_err(|error| error.to_string())?;
    let (peer, _) = listener.accept().map_err(|error| error.to_string())?;
    Ok((Observation::new(carrier), peer))
}

fn closed(peer: &mut TcpStream) -> Result<(), String> {
    let mut byte = [0];
    assert_eq!(peer.read(&mut byte).map_err(|error| error.to_string())?, 0);
    Ok(())
}

fn poison<Value: Send + 'static>(state: Arc<Mutex<Value>>) -> Result<(), String> {
    let joined = std::thread::spawn(move || {
        let guard = state.lock().map_err(|_| "already poisoned")?;
        assert!(std::hint::black_box(false), "poison fixture");
        drop(guard);
        Ok::<(), &str>(())
    })
    .join();
    if joined.is_err() {
        Ok(())
    } else {
        Err("fixture did not poison its mutex".to_owned())
    }
}

#[test]
fn poisoned_wake_stops_instead_of_reading_state() {
    let wake = Arc::new(Wake::default());
    let worker = Arc::clone(&wake);
    assert!(
        std::thread::spawn(move || {
            let guard = worker.state.lock().map_err(|_| "already poisoned")?;
            assert!(std::hint::black_box(false), "poison fixture");
            drop(guard);
            Ok::<(), &str>(())
        })
        .join()
        .is_err()
    );
    wake.notify();
    assert!(wake.stopping(), "a poisoned wake must stop the bridge");
}

#[test]
fn poisoned_counters_are_refused_instead_of_returned() -> Result<(), String> {
    let (observation, mut peer) = observed_carrier()?;
    let worker = Arc::clone(&observation);
    assert!(
        std::thread::spawn(move || {
            let mut counters = worker.counters.lock().map_err(|_| "already poisoned")?;
            counters.crossed = 73;
            assert!(std::hint::black_box(false), "poison fixture");
            drop(counters);
            Ok::<(), &str>(())
        })
        .join()
        .is_err()
    );
    let snapshot = observation.snapshot();
    assert_eq!(snapshot.crossed, 0, "untrusted counters must not escape");
    assert!(snapshot.error.is_some(), "the poison refusal must be named");
    assert!(observation.wake.check().is_err());
    closed(&mut peer)?;
    Ok(())
}

#[test]
fn poisoned_registry_does_not_return_untrusted_observations() -> Result<(), String> {
    let registry = BridgeRegistry::default();
    let (observation, mut peer) = observed_carrier()?;
    registry.install("orders".to_owned(), &observation);
    poison(Arc::clone(&registry.0))?;
    assert!(
        registry.snapshot().is_empty(),
        "a poisoned registry must refuse its map"
    );
    assert!(observation.wake.check().is_err());
    closed(&mut peer)?;
    Ok(())
}

/// ACCESS-009: the retry owner logs a dial refusal whose reason changed, so
/// a dial that fails after Lys issues is told by name; the registry reports
/// the change.
#[test]
fn a_dial_refusal_reports_whether_its_reason_changed() -> Result<(), String> {
    let registry = BridgeRegistry::default();
    let refused = |reason: &str| {
        registry
            .dial_refused("orders", format!("all_link_carriers_refused: {reason}"))
            .map_err(|error| error.to_string())
    };
    let first = refused("agent_dial_refused")?;
    let same = refused("agent_dial_refused")?;
    let other = refused("connect_ack_absent")?;
    assert!(
        first && !same && other,
        "changes reported: {first} {same} {other}"
    );
    assert_eq!(registry.snapshot()["orders"].dial_refused, 3);
    assert_eq!(
        registry.snapshot()["orders"].last_refusal.as_deref(),
        Some("all_link_carriers_refused: connect_ack_absent")
    );
    Ok(())
}

#[test]
fn refused_dials_and_reopened_carriers_retain_their_lifecycle_counts() -> Result<(), String> {
    use crate::link::{LinkEnd, LinkLife};
    use crate::link_handshake::MachineIdentity;
    let registry = BridgeRegistry::default();
    let changed = registry
        .dial_refused("orders", "all_link_carriers_refused: wire=away".to_owned())
        .map_err(|error| error.to_string())?;
    assert!(changed, "the first refusal is a change");
    assert_eq!(registry.snapshot()["orders"].dial_refused, 1);
    assert_eq!(registry.snapshot()["orders"].lifecycle.links_opened, 0);
    for expected in 1..=2 {
        let counts = registry
            .snapshot()
            .remove("orders")
            .ok_or("counts absent")?;
        let observation =
            Observation::retaining(None, counts).map_err(|error| error.to_string())?;
        let life = LinkLife::new(MachineIdentity::new([7; 32]));
        observation
            .life
            .set(life.clone())
            .map_err(|_| "lifecycle already installed")?;
        registry.install("orders".to_owned(), &observation);
        let counts = registry.snapshot();
        assert_eq!(counts["orders"].lifecycle.links_opened, expected);
        assert_eq!(
            counts["orders"].lifecycle.links_closed_by_peer_gone,
            expected - 1
        );
        assert_eq!(counts["orders"].dial_refused, 1);
        if expected == 1 {
            life.end(LinkEnd::PeerGone)?;
        }
    }
    assert_eq!(registry.snapshot().len(), 1);
    Ok(())
}