liminal-server 0.25.4

//! Carrier order, loss recovery, silence, and broker independence.

use liminal_server::config::types::OutboundLinkConfig;
use liminal_server::link_carrier::Carrier;
use liminal_server::link_open::{Backend, ChoiceConfig, Deadline, OpenError, Packet, Path, open};
use std::sync::{Arc, Condvar, Mutex};
use std::time::Duration;

#[path = "link_open/broker.rs"]
mod broker;
#[path = "link_open/network.rs"]
mod network;
#[path = "link_open/policy.rs"]
mod policy;
#[path = "link_open/sockets.rs"]
mod sockets;
#[path = "link_open/wire.rs"]
mod wire;

pub mod support;
use support::agent_pass;

type TestResult = Result<(), Box<dyn std::error::Error>>;
type TopologyFact = (Option<Carrier>, Option<String>, Option<String>);

#[derive(Debug, Default)]
struct Facts {
    calls: Vec<String>,
    delivered: Vec<Vec<u8>>,
    sends: Vec<u64>,
    topology: Vec<TopologyFact>,
    stopped: bool,
    misses: usize,
    refuse_relay: bool,
    end_next: bool,
    addresses: Vec<String>,
    datagrams: usize,
    changed: Arc<Condvar>,
}

#[derive(Debug)]
struct Fixture {
    facts: Arc<Mutex<Facts>>,
    wire: bool,
    punch: bool,
    broker: Option<Arc<broker::Control>>,
}

#[derive(Debug)]
struct FixturePath {
    carrier: Carrier,
    facts: Arc<Mutex<Facts>>,
    next: u64,
}

impl Path for FixturePath {
    fn carrier(&self) -> Carrier {
        self.carrier
    }
    fn relay_device(&self) -> Option<String> {
        (self.carrier == Carrier::Relay).then(|| "relay-device".to_owned())
    }
    fn prepare(&mut self, subject: &str, plain: &[u8]) -> Result<Packet, OpenError> {
        let packet = Packet {
            subject: subject.to_owned(),
            counter: self.next,
            bytes: plain.to_vec(),
            deadline: None,
        };
        self.next += 1;
        Ok(packet)
    }
    fn transmit(&mut self, packet: &Packet) -> Result<(), OpenError> {
        let mut facts = self
            .facts
            .lock()
            .map_err(|error| OpenError(error.to_string()))?;
        if facts.end_next {
            facts.end_next = false;
            return Err(OpenError("peer_gone".to_owned()));
        }
        facts.sends.push(packet.counter);
        if facts.misses == 0 {
            facts.delivered.push(packet.bytes.clone());
        }
        drop(facts);
        Ok(())
    }
    fn receipt(&mut self, _counter: u64, bound: Option<Deadline>) -> Result<bool, OpenError> {
        let mut facts = self
            .facts
            .lock()
            .map_err(|error| OpenError(error.to_string()))?;
        if facts.misses > 0 {
            assert_eq!(
                bound.map(Deadline::budget),
                Some(Duration::from_millis(2000))
            );
            facts.misses -= 1;
            drop(facts);
            return Ok(false);
        }
        drop(facts);
        Ok(true)
    }
}

impl Fixture {
    fn path(&self, carrier: Carrier) -> Result<Box<dyn Path>, OpenError> {
        if carrier == Carrier::Direct {
            return Ok(Box::new(sockets::SocketPath::new(Arc::clone(&self.facts))?));
        }
        Ok(Box::new(FixturePath {
            carrier,
            facts: Arc::clone(&self.facts),
            next: 0,
        }))
    }
    fn called(&self, call: &str) -> Result<(), OpenError> {
        let mut facts = self
            .facts
            .lock()
            .map_err(|error| OpenError(error.to_string()))?;
        facts.calls.push(call.to_owned());
        if facts.stopped && matches!(call, "lookup" | "punch" | "repunch") {
            return Err(OpenError("broker_stopped".to_owned()));
        }
        drop(facts);
        Ok(())
    }
}

impl Backend for Fixture {
    fn wire(
        &mut self,
        _config: &OutboundLinkConfig,
        address: &str,
    ) -> Result<Box<dyn Path>, OpenError> {
        self.called(address)?;
        if self.wire {
            self.path(Carrier::Wire)
        } else {
            Err(OpenError("wire_unreachable".to_owned()))
        }
    }
    fn lookup(&mut self, _config: &OutboundLinkConfig) -> Result<Vec<String>, OpenError> {
        self.called("lookup")?;
        if let Some(broker) = &self.broker {
            return broker.lookup();
        }
        let facts = self
            .facts
            .lock()
            .map_err(|error| OpenError(error.to_string()))?;
        Ok(if facts.addresses.is_empty() {
            vec!["registered-address".to_owned()]
        } else {
            facts.addresses.clone()
        })
    }
    fn punch(&mut self, _config: &OutboundLinkConfig) -> Result<Box<dyn Path>, OpenError> {
        self.called("punch")?;
        if let Some(broker) = &self.broker {
            broker.punch()?;
        }
        if self.punch {
            self.path(Carrier::Direct)
        } else {
            Err(OpenError("punch_refused".to_owned()))
        }
    }
    fn relay(&mut self, _config: &OutboundLinkConfig) -> Result<Box<dyn Path>, OpenError> {
        self.called("relay")?;
        if self
            .facts
            .lock()
            .map_err(|error| OpenError(error.to_string()))?
            .refuse_relay
        {
            return Err(OpenError("relay_refused".to_owned()));
        }
        self.path(Carrier::Relay)
    }
    fn repunch(
        &mut self,
        _config: &OutboundLinkConfig,
        _path: &mut dyn Path,
    ) -> Result<(), OpenError> {
        self.called("repunch")?;
        if let Some(broker) = &self.broker {
            broker.punch()?;
        }
        Ok(())
    }
    fn topology(
        &mut self,
        carrier: Option<Carrier>,
        relay: Option<String>,
        refusal: Option<&str>,
    ) -> Result<(), OpenError> {
        self.facts
            .lock()
            .map_err(|error| OpenError(error.to_string()))?
            .topology
            .push((carrier, relay, refusal.map(str::to_owned)));
        Ok(())
    }
}

fn fixture(
    address: Option<&str>,
    wire: bool,
    punch: bool,
) -> (OutboundLinkConfig, Fixture, Arc<Mutex<Facts>>) {
    let facts = Arc::new(Mutex::new(Facts::default()));
    let config = OutboundLinkConfig {
        link_id: "orders-link".to_owned(),
        peer_holder: "importer".to_owned(),
        peer_address: address.map(str::to_owned),
        exports: vec!["orders".to_owned()],
    };
    (
        config,
        Fixture {
            facts: Arc::clone(&facts),
            wire,
            punch,
            broker: None,
        },
        facts,
    )
}

#[test]
fn a_link_with_a_configured_address_opens_over_the_wire_and_asks_no_broker() -> TestResult {
    let federation = wire::federation::Federation::start(wire::federation::LiveOpen::Outright)?;
    let config = federation.outbound();
    let address = config
        .peer_address
        .clone()
        .ok_or("fixture has no wire address")?;
    let facts = Arc::new(Mutex::new(Facts::default()));
    let backend = wire::WireBackend::new(&federation, Arc::clone(&facts));
    let link = open(config, ChoiceConfig::default(), backend)?;
    assert_eq!(link.carrier(), Some(Carrier::Wire));
    assert_eq!(
        facts.lock().map_err(|_| "fixture facts poisoned")?.calls,
        [address]
    );
    drop(link);
    federation.shutdown()?;
    Ok(())
}

#[test]
fn a_link_without_an_address_punches_when_the_wire_is_unreachable() -> TestResult {
    let (config, backend, facts) = fixture(None, false, true);
    let link = open(config, ChoiceConfig::default(), backend)?;
    assert_eq!(link.carrier(), Some(Carrier::Direct));
    assert_eq!(
        facts.lock().map_err(|_| "fixture facts poisoned")?.calls,
        ["lookup", "registered-address", "punch"]
    );
    Ok(())
}

#[test]
fn a_link_that_cannot_punch_relays_and_the_topology_says_so() -> TestResult {
    let (config, backend, facts) = fixture(None, false, false);
    let link = open(config, ChoiceConfig::default(), backend)?;
    assert_eq!(link.carrier(), Some(Carrier::Relay));
    let facts = facts.lock().map_err(|_| "fixture facts poisoned")?;
    assert!(
        facts
            .topology
            .iter()
            .any(|fact| fact.2.as_deref() == Some("punch_refused"))
    );
    assert_eq!(
        facts.topology.last(),
        Some(&(Some(Carrier::Relay), Some("relay-device".to_owned()), None))
    );
    Ok(())
}

#[test]
fn a_lost_datagram_repunches_once_at_its_deadline_and_the_message_arrives_once() -> TestResult {
    let (config, backend, facts) = fixture(None, false, true);
    let mut link = open(config, ChoiceConfig::default(), backend)?;
    facts.lock().map_err(|_| "fixture facts poisoned")?.misses = 1;
    link.send("orders.created", b"one message")?;
    let facts = facts.lock().map_err(|_| "fixture facts poisoned")?;
    assert_eq!(
        facts.calls.iter().filter(|call| *call == "repunch").count(),
        1
    );
    assert_eq!(facts.sends, [0, 0]);
    assert_eq!(facts.delivered, [b"one message".to_vec()]);
    assert_eq!(link.armed_timers(), 0);
    Ok(())
}

#[test]
fn an_idle_direct_link_sends_zero_datagrams_and_arms_zero_timers() -> TestResult {
    let (config, backend, facts) = fixture(None, false, true);
    let link = open(config, ChoiceConfig::default(), backend)?;
    assert_eq!(link.carrier(), Some(Carrier::Direct));
    assert!(
        facts
            .lock()
            .map_err(|_| "fixture facts poisoned")?
            .sends
            .is_empty()
    );
    assert_eq!(link.armed_timers(), 0);
    let changed = Arc::clone(&facts.lock().map_err(|_| "fixture facts poisoned")?.changed);
    // This bounds an observation of silence; no action is scheduled by its expiry.
    let (observation, elapsed) = changed
        .wait_timeout_while(
            facts.lock().map_err(|_| "fixture facts poisoned")?,
            Duration::from_millis(25),
            |facts| facts.datagrams == 0,
        )
        .map_err(|_| "idle observation poisoned")?;
    assert!(elapsed.timed_out());
    assert_eq!(observation.datagrams, 0);
    assert_eq!(link.armed_timers(), 0);
    Ok(())
}

#[test]
fn with_the_broker_stopped_an_up_link_delivers_m_of_m() -> TestResult {
    let (config, mut backend, facts) = fixture(None, false, true);
    let broker = broker::Control::new()?;
    backend.broker = Some(Arc::clone(&broker));
    let mut link = open(config, ChoiceConfig::default(), backend)?;
    assert_eq!(link.carrier(), Some(Carrier::Direct));
    broker.stop()?;
    assert_eq!(
        broker.lookup().err().ok_or("broker must be stopped")?.0,
        "broker_stopped"
    );
    for counter in 0_u8..16 {
        link.send("orders.created", &[counter])?;
    }
    let facts = facts.lock().map_err(|_| "fixture facts poisoned")?;
    assert_eq!(
        facts.delivered,
        (0_u8..16).map(|counter| vec![counter]).collect::<Vec<_>>()
    );
    assert_eq!(facts.calls, ["lookup", "127.0.0.1:9", "punch"]);
    assert_eq!(broker.counters.snapshot().data_bytes_forwarded, 0);
    assert_eq!(broker.counters.snapshot().punches_told, 1);
    Ok(())
}

#[test]
fn a_second_miss_moves_to_relay_without_a_second_repunch() -> TestResult {
    let (config, backend, facts) = fixture(None, false, true);
    let mut link = open(config, ChoiceConfig::default(), backend)?;
    facts.lock().map_err(|_| "fixture facts poisoned")?.misses = 2;
    link.send("orders.created", b"one message")?;
    assert_eq!(link.carrier(), Some(Carrier::Relay));
    let facts = facts.lock().map_err(|_| "fixture facts poisoned")?;
    assert_eq!(
        facts.calls.iter().filter(|call| *call == "repunch").count(),
        1
    );
    assert_eq!(
        facts.calls.iter().filter(|call| *call == "relay").count(),
        1
    );
    assert_eq!(facts.delivered, [b"one message".to_vec()]);
    assert_eq!(link.armed_timers(), 0);
    Ok(())
}

#[test]
fn each_registered_wire_address_is_tried_once() -> TestResult {
    let (config, backend, facts) = fixture(None, false, true);
    facts
        .lock()
        .map_err(|_| "fixture facts poisoned")?
        .addresses = vec!["first".to_owned(), "second".to_owned(), "first".to_owned()];
    let link = open(config, ChoiceConfig::default(), backend)?;
    assert_eq!(link.carrier(), Some(Carrier::Direct));
    assert_eq!(
        facts.lock().map_err(|_| "fixture facts poisoned")?.calls,
        ["lookup", "first", "second", "punch"]
    );
    Ok(())
}

#[test]
fn a_zero_deadline_is_refused_before_any_carrier_is_tried() -> TestResult {
    let (config, backend, facts) = fixture(None, false, true);
    let choice = ChoiceConfig {
        crossing_deadline_ms: 0,
        ..ChoiceConfig::default()
    };
    let error = open(config, choice, backend)
        .err()
        .ok_or("zero deadline must refuse")?;
    assert_eq!(error.0, "crossing_deadline_ms_must_be_positive");
    assert!(
        facts
            .lock()
            .map_err(|_| "fixture facts poisoned")?
            .calls
            .is_empty()
    );
    Ok(())
}

#[test]
fn a_datagram_too_large_for_udp_uses_relay_without_repunching() -> TestResult {
    let (config, backend, facts) = fixture(None, false, true);
    let mut link = open(config, ChoiceConfig::default(), backend)?;
    let payload = vec![7; 1200];
    link.send("orders.created", &payload)?;
    assert_eq!(link.carrier(), Some(Carrier::Relay));
    let facts = facts.lock().map_err(|_| "fixture facts poisoned")?;
    assert_eq!(
        facts.calls.iter().filter(|call| *call == "repunch").count(),
        0
    );
    assert_eq!(facts.delivered, [payload]);
    Ok(())
}

#[test]
fn a_down_link_opens_once_on_its_next_send() -> TestResult {
    let (config, backend, facts) = fixture(Some("configured-address"), true, true);
    let mut link = open(config, ChoiceConfig::default(), backend)?;
    facts.lock().map_err(|_| "fixture facts poisoned")?.end_next = true;
    assert_eq!(
        link.send("orders.created", b"first")
            .err()
            .ok_or("ended carrier must refuse")?
            .0,
        "peer_gone"
    );
    assert_eq!(link.carrier(), None);
    assert_eq!(link.armed_timers(), 0);
    assert_eq!(
        facts.lock().map_err(|_| "fixture facts poisoned")?.calls,
        ["configured-address"]
    );
    link.send("orders.created", b"second")?;
    let facts = facts.lock().map_err(|_| "fixture facts poisoned")?;
    assert_eq!(facts.calls, ["configured-address", "configured-address"]);
    assert_eq!(facts.delivered, [b"second".to_vec()]);
    Ok(())
}

#[test]
fn an_unavailable_broker_records_its_refusal_and_tries_relay_once() -> TestResult {
    let (config, backend, facts) = fixture(None, false, true);
    facts.lock().map_err(|_| "fixture facts poisoned")?.stopped = true;
    let link = open(config, ChoiceConfig::default(), backend)?;
    assert_eq!(link.carrier(), Some(Carrier::Relay));
    let facts = facts.lock().map_err(|_| "fixture facts poisoned")?;
    assert_eq!(facts.calls, ["lookup", "punch", "relay"]);
    assert_eq!(
        facts
            .topology
            .iter()
            .filter(|fact| fact.2.as_deref() == Some("broker_stopped"))
            .count(),
        2
    );
    Ok(())
}

#[test]
fn the_new_choice_defaults_read_an_old_machine_configuration() -> TestResult {
    let machine: liminal_server::config::types::MachineConfig =
        toml::from_str("identity_path = 'fixture.key'\npass_path = 'fixture.pass'\n")?;
    assert_eq!(machine.carrier_choice, ChoiceConfig::default());
    assert_eq!(machine.carrier_choice.crossing_deadline_ms, 2000);
    assert!(toml::from_str::<ChoiceConfig>("keepalive_ms = 1000").is_err());
    Ok(())
}