liminal-server 0.25.2

//! Configured production carriers settle imported publishes and retain their opening refusals.
#[path = "support/agent_pass.rs"]
mod agent_pass;
#[path = "link_open_live/fixture.rs"]
mod fixture;
use fixture::{Fixture, wire};
use liminal::protocol::Frame;
use std::time::Duration;
type Result<T = ()> = std::result::Result<T, Box<dyn std::error::Error>>;
const OBSERVATION: Duration = Duration::from_secs(5);

fn federated(document: &serde_json::Value) -> Result<&serde_json::Value> {
    document["links"]
        .as_array()
        .ok_or("topology links absent")?
        .iter()
        .find(|link| link["kind"] == "federated")
        .ok_or_else(|| "configured link missing from topology".into())
}
#[test]
fn a_configured_link_opens_through_the_chooser_in_production() -> Result {
    let mut fixture = Fixture::new()?;
    let address = fixture.servers[1].address()?.to_string();
    fixture.configure(Some(address))?;
    fixture.servers[0].start_links()?;
    let topology = fixture.servers[0].topology()?;
    let link = federated(&topology)?;
    assert_eq!(link["carrier"], "wire");
    assert_eq!(link["state"], "up");
    assert_eq!(
        link["carrier_refusals"],
        serde_json::json!([]),
        "the configured open must pass through the chooser's topology record"
    );
    let broker = fixture.servers[3]
        .bindings
        .as_ref()
        .and_then(|bindings| bindings.broker())
        .ok_or("broker bindings absent")?
        .counters()
        .snapshot();
    assert_eq!(broker.registered, 0);
    assert_eq!(broker.lookups, 0);
    assert_eq!(broker.punches_told, 0);
    Ok(())
}
#[test]
fn with_the_broker_stopped_an_up_link_delivers_m_of_m() -> Result {
    let mut fixture = Fixture::new()?;
    fixture.configure(None)?;
    let broker = fixture.servers[3].address()?.to_string();
    let echo = fixture.echo;
    {
        let machine = fixture.servers[1].machine()?;
        machine.carrier_choice.broker_address = Some(broker);
        machine.carrier_choice.broker_echo = Some(echo);
    }
    fixture.servers[1].start_links()?;
    let mut imported = fixture.servers[1].subscribe("orders.created")?;
    fixture.servers[0].start_links()?;
    assert_eq!(
        federated(&fixture.servers[0].topology()?)?["carrier"],
        "direct"
    );
    fixture.servers[3].stop()?;
    for number in 0..16 {
        fixture.servers[0].publish(number)?;
    }
    for number in 0..16 {
        let Frame::Deliver { envelope, .. } = wire::read_frame(&mut imported)? else {
            return Err("imported publish absent".into());
        };
        assert_eq!(
            serde_json::from_slice::<serde_json::Value>(&envelope.payload)?,
            serde_json::json!({"n":number})
        );
    }
    assert!(
        fixture.servers[0]
            .links
            .as_ref()
            .ok_or("configured links absent")?
            .observe_answers(16, OBSERVATION)
    );
    assert_eq!(
        fixture.servers[0].registry.snapshot()["orders-link"].crossed,
        16
    );
    Ok(())
}
#[test]
fn a_link_that_cannot_punch_relays_and_the_topology_keeps_the_refusal_names() -> Result {
    let mut fixture = Fixture::new()?;
    let refused = std::net::TcpListener::bind("127.0.0.1:0")?;
    let address = refused.local_addr()?.to_string();
    drop(refused);
    fixture.configure(Some(address))?;
    fixture.servers[0].machine()?.carrier_choice.broker_address = None;
    let remote = fixture.servers[0].host.local().machine();
    let local = liminal_server::link_handshake::LocalMachine::new(
        lys_core::keys::Ed25519Identity::from_seed(&zeroize::Zeroizing::new([2; 32])),
    );
    let importer_pass = fixture.servers[1].machine()?.pass_path.clone();
    let mut socket = fixture.servers[2].connect_as(&importer_pass)?;
    // The importing device marks its reciprocal relay stream before the opening signal.
    wire::write_frame(
        &mut socket,
        &Frame::ConversationOpen {
            flags: 0,
            stream_id: 11,
            conversation_id: 11,
            subject: format!("net.relay.{}", fixture.devices[0]),
        },
    )?;
    wire::write_frame(&mut socket, &Frame::Ping { flags: 0 })?;
    assert!(matches!(wire::read_frame(&mut socket)?, Frame::Pong { .. }));
    let relay_identity = fixture.servers[2].host.local().machine();
    let endpoint = std::thread::spawn(move || -> std::result::Result<_, String> {
        let carrier = liminal_server::link_carrier::RelayCarrier::new(socket, 11, relay_identity)
            .map_err(|error| error.to_string())?;
        liminal_server::link_relay::accept(carrier, &local, remote)
            .map_err(|error| error.to_string())
    });
    let opened = fixture.servers[0].start_links();
    if opened.is_err() {
        fixture.servers[2].stop()?;
    }
    let receiving = endpoint.join().map_err(|_| "relay receiver panicked")?;
    opened?;
    let receiving = receiving.map_err(std::io::Error::other)?;
    let topology = fixture.servers[0].topology()?;
    let link = federated(&topology)?;
    assert_eq!(link["state"], "up");
    assert_eq!(link["carrier"], "relay");
    assert_eq!(link["relayed"], fixture.devices[2]);
    assert_eq!(link["carried_by"], fixture.devices[2]);
    assert!(
        link["carrier_refusals"]
            .as_array()
            .ok_or("carrier refusals missing")?
            .iter()
            .any(|refusal| refusal == "broker_not_configured")
    );
    drop(receiving);
    Ok(())
}

#[test]
fn stopping_one_configured_bridge_leaves_the_other_carrier_up() -> Result {
    let mut fixture = Fixture::new()?;
    fixture.configure(Some(fixture.servers[1].address()?.to_string()))?;
    let mut second = fixture.servers[0].machine()?.links[0].clone();
    second.link_id = "second-link".to_owned();
    fixture.servers[0].machine()?.links.push(second);
    let mut imported = fixture.servers[1].subscribe("orders.created")?;
    fixture.servers[0].start_links()?;
    fixture.servers[0]
        .links
        .as_mut()
        .ok_or("configured links absent")?
        .finish_first()?;
    fixture.servers[0].publish(37)?;
    let Frame::Deliver { envelope, .. } = wire::read_frame(&mut imported)? else {
        return Err("remaining carrier delivered no import".into());
    };
    assert_eq!(
        serde_json::from_slice::<serde_json::Value>(&envelope.payload)?,
        serde_json::json!({"n":37})
    );
    assert!(
        fixture.servers[0]
            .links
            .as_ref()
            .ok_or("configured links absent")?
            .observe_answers(1, OBSERVATION)
    );
    Ok(())
}

#[test]
fn another_configured_link_to_the_same_peer_opens_over_udp() -> Result {
    let mut fixture = Fixture::new()?;
    fixture.configure(None)?;
    let broker = fixture.servers[3].address()?.to_string();
    let echo = fixture.echo;
    {
        let machine = fixture.servers[1].machine()?;
        machine.carrier_choice.broker_address = Some(broker);
        machine.carrier_choice.broker_echo = Some(echo);
    }
    fixture.servers[1].start_links()?;
    let mut second = fixture.servers[0].machine()?.links[0].clone();
    second.link_id = "second-link".to_owned();
    fixture.servers[0].machine()?.links.push(second);
    fixture.servers[0].start_links()?;
    assert_eq!(fixture.servers[0].registry.snapshot().len(), 2);
    assert_eq!(
        federated(&fixture.servers[0].topology()?)?["carrier"],
        "direct"
    );
    let mut imported = fixture.servers[1].subscribe("orders.created")?;
    fixture.servers[0]
        .links
        .as_mut()
        .ok_or("configured links absent")?
        .finish_first()?;
    fixture.servers[0].publish(41)?;
    let Frame::Deliver { envelope, .. } = wire::read_frame(&mut imported)? else {
        return Err("remaining UDP carrier delivered no import".into());
    };
    assert_eq!(
        serde_json::from_slice::<serde_json::Value>(&envelope.payload)?,
        serde_json::json!({"n":41})
    );
    assert!(
        fixture.servers[0]
            .links
            .as_ref()
            .ok_or("configured links absent")?
            .observe_answers(1, OBSERVATION)
    );
    Ok(())
}