liminal-server 0.25.4

//! Existing topology state and station facts remain pinned.

use super::*;

/// A peer appears as a device and an up link whose ends are both listed
/// devices, sorted; when it leaves, both are gone. This device carries its
/// version and its start in wall milliseconds; a peer carries neither.
#[test]
fn a_peer_joins_and_leaves_the_document() -> TestResult {
    let started = Instant::now();
    let mut model = model(started);
    model.apply(TopologyChange::Peer {
        name: "edge".to_owned(),
        present: true,
    });
    let joined = document(&model, started + Duration::from_secs(3_600))?;
    assert_eq!(joined["rev"], 2);
    assert_eq!(joined["devices"][0]["started"], WALL_MS - 3_600_000);
    assert_eq!(joined["devices"][0]["liminal"], "1.2.3");
    assert_eq!(joined["devices"][1]["id"], "edge");
    assert!(joined["devices"][1].get("liminal").is_none());
    assert_eq!(
        joined["links"][0]["ends"],
        serde_json::json!(["edge", "local"])
    );
    assert_eq!(joined["links"][0]["state"], "up");
    assert_eq!(joined["connections"][0]["id"], "7");
    assert_eq!(joined["stations"][0]["device"], "local");

    model.apply(TopologyChange::Peer {
        name: "edge".to_owned(),
        present: false,
    });
    let left = document(&model, started)?;
    assert_eq!(left["rev"], 3);
    assert_eq!(left["devices"].as_array().map(Vec::len), Some(1));
    assert_eq!(left["links"].as_array().map(Vec::len), Some(0));
    Ok(())
}

/// R7: a registered worker is a station, and a line's stops are its
/// registered publishers then its registered subscribers, each once; an
/// unregistered connection is no stop, and a closed one leaves the line.
#[test]
fn a_line_runs_from_its_publisher_to_its_subscriber() -> TestResult {
    let started = Instant::now();
    let mut model = model(started);
    let flow = |connection, role| TopologyChange::Flow {
        connection,
        line: "orders".to_owned(),
        role,
        present: true,
    };
    for (connection, identity) in [(11, "ping"), (12, "pong")] {
        assert!(model.apply(TopologyChange::Worker {
            connection,
            identity: identity.to_owned(),
            subjects: Vec::new(),
            activities: Vec::new(),
        }));
    }
    assert!(model.apply(flow(12, FlowRole::Subscribes)));
    assert!(model.apply(flow(11, FlowRole::Publishes)));
    assert!(
        !model.apply(flow(11, FlowRole::Publishes)),
        "a repeat is no change"
    );
    assert!(
        !model.apply(flow(7, FlowRole::Subscribes)),
        "an unregistered connection"
    );
    let value = document(&model, started)?;
    assert_eq!(value["lines"][0]["id"], "orders");
    assert_eq!(
        value["lines"][0]["stops"],
        serde_json::json!(["ping", "local", "pong"])
    );
    let stations: Vec<&str> = value["stations"]
        .as_array()
        .ok_or("stations")?
        .iter()
        .filter_map(|station| station["id"].as_str())
        .collect();
    assert_eq!(stations, ["local", "orders.reply", "ping", "pong"]);
    let kinds: Vec<&str> = value["stations"]
        .as_array()
        .ok_or("stations")?
        .iter()
        .filter_map(|station| station["kind"].as_str())
        .collect();
    assert_eq!(kinds, ["exchange", "service", "worker", "worker"]);
    assert_eq!(
        value["devices"][0]["box"],
        serde_json::json!({"x": -140, "y": 60, "w": 1340, "h": 520})
    );

    assert!(model.apply(TopologyChange::ConnectionClosed { id: 12 }));
    let value = document(&model, started)?;
    assert_eq!(
        value["lines"][0]["stops"],
        serde_json::json!(["ping", "local"])
    );
    Ok(())
}

/// A worker that stops subscribing leaves the line's stops at once, without
/// waiting for its connection to close.
#[test]
fn a_flow_withdrawn_leaves_the_stops() -> TestResult {
    let started = Instant::now();
    let mut model = model(started);
    let flow = |present| TopologyChange::Flow {
        connection: 12,
        line: "orders".to_owned(),
        role: FlowRole::Subscribes,
        present,
    };
    assert!(model.apply(TopologyChange::Worker {
        connection: 12,
        identity: "pong".to_owned(),
        subjects: Vec::new(),
        activities: Vec::new(),
    }));
    assert!(model.apply(flow(true)));
    assert!(model.apply(flow(false)));
    assert!(!model.apply(flow(false)), "a repeat is no change");
    let value = document(&model, started)?;
    assert_eq!(value["lines"][0]["stops"], serde_json::json!(["local"]));
    Ok(())
}

/// This device's start is its incarnation's, in wall milliseconds, and the
/// same whenever the document is built: one built at T+0 and read at T+30 s
/// lets the reader show 30 s more than at T+0, where an uptime stamped at
/// T+0 would still read T+0 (Hermes 7 Oct 01:49).
#[test]
fn the_start_is_the_incarnations_whenever_the_document_is_built() -> TestResult {
    let built = Instant::now();
    let mut model = model(built);
    let incarnation = built.checked_sub(Duration::from_secs(5)).ok_or("clock")?;
    assert!(model.incarnation_started(incarnation));

    let (at_zero, fault) = model.document(&WallBase::fixed(WALL_MS, built));
    assert!(fault.is_none(), "{fault:?}");
    // The same wall clock, read 30 s later.
    let later = built + Duration::from_secs(30);
    let (at_thirty, _) = model.document(&WallBase::fixed(WALL_MS + 30_000, later));
    let started = at_zero.devices[0]
        .started
        .ok_or("no start on this device")?;
    assert_eq!(started, WALL_MS - 5_000);
    assert_eq!(at_thirty.devices[0].started, Some(started));
    let read_at_thirty = WALL_MS + 30_000;
    assert_eq!(
        read_at_thirty - started,
        35_000,
        "the uptime read at T+30 s"
    );
    let value = document(&model, built)?;
    assert!(
        value["devices"][0].get("uptime").is_none(),
        "no stale uptime field"
    );
    Ok(())
}

/// Buttercup's finding 3 (Hermes 03:05): this device carries its start or
/// the fault that kept it, never both and never neither. A start the wall
/// clock can give is `started` alone; one before the epoch is
/// `started_fault` alone, naming the clock, and the fault comes back beside
/// the document for the hub to count.
#[test]
fn a_device_has_its_start_or_its_fault_never_both() -> TestResult {
    let built = Instant::now();
    let model = model(built);
    let (good, none) = model.document(&WallBase::fixed(WALL_MS, built));
    assert!(none.is_none(), "{none:?}");
    assert_eq!(
        (
            good.devices[0].started.is_some(),
            good.devices[0].started_fault.is_none()
        ),
        (true, true)
    );
    // A wall base one second after the epoch, read with a start two seconds
    // before it.
    let mut early = model;
    let incarnation = built.checked_sub(Duration::from_secs(2)).ok_or("clock")?;
    early.incarnation_started(incarnation);
    let (faulted, fault) = early.document(&WallBase::fixed(1_000, built));
    assert!(
        matches!(fault, Some(ConsoleError::Clock { .. })),
        "{fault:?}"
    );
    assert_eq!(faulted.devices[0].started, None);
    assert_eq!(
        faulted.devices[0].started_fault.as_deref(),
        Some("console clock: a time is outside u64 milliseconds since the epoch")
    );
    for peer in &faulted.devices[1..] {
        assert_eq!((peer.started, peer.started_fault.as_deref()), (None, None));
    }
    Ok(())
}

/// Lane brief ruling 1 (Hermes 7 Oct 03:51): the document names this device
/// once, at the top level, as one of its devices' ids, whatever peers it
/// holds and however they sort.
#[test]
fn the_document_names_this_device_once() -> TestResult {
    let started = Instant::now();
    let mut model = model(started);
    model.apply(TopologyChange::Peer {
        name: "alpha".to_owned(),
        present: true,
    });
    let value = document(&model, started)?;
    assert_eq!(value["this_device"], "local");
    let devices = value["devices"]
        .as_array()
        .ok_or("the document has no devices")?;
    let named = devices
        .iter()
        .filter(|device| device["id"] == value["this_device"])
        .count();
    assert_eq!(named, 1, "this_device names exactly one device: {value}");
    Ok(())
}