liminal-server 0.20.0

//! The console records against their golden bytes. The same two files are
//! what the console's own `parseRecord` reads in
//! `apps/liminal-console/tests/server-records.test.ts`, so a renamed or
//! missing field fails on one side of the language boundary or the other.

use std::num::NonZeroU32;

use super::records::{
    Connection, Device, DeviceKind, Hop, Line, LineKind, LineLife, LineState, Link, Mount, Outcome,
    Station, StationKind, StationState, TapBatch, TapMessage, TopologyDocument, canonical_bytes,
};

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

const TOPOLOGY_GOLDEN: &[u8] = include_bytes!("../../tests/fixtures/console-records/topology.json");
const TAP_BATCH_GOLDEN: &[u8] =
    include_bytes!("../../tests/fixtures/console-records/tap-batch.json");
const TOPOLOGY_CLOCK_FAULT_GOLDEN: &[u8] =
    include_bytes!("../../tests/fixtures/console-records/topology-clock-fault.json");

fn device(id: &str, liminal: Option<&str>, started: Option<u64>) -> Device {
    Device {
        id: id.to_owned(),
        name: id.to_owned(),
        kind: DeviceKind::Device,
        liminal: liminal.map(str::to_owned),
        started,
        started_fault: None,
        zone: None,
    }
}

/// The topology document as the model sends it when the wall clock gave
/// no time for this device's start: the fault named in its place
/// (Buttercup's finding 3, Hermes 03:05).
fn topology_clock_fault() -> TopologyDocument {
    let mut document = topology();
    if let Some(this) = document.devices.first_mut() {
        this.started = None;
        this.started_fault =
            Some("console clock: a time is outside u64 milliseconds since the epoch".to_owned());
    }
    document
}

fn worker(id: &str) -> Station {
    Station {
        id: id.to_owned(),
        name: id.to_owned(),
        device: "local".to_owned(),
        kind: StationKind::Worker,
        state: StationState::Live,
    }
}

fn topology() -> TopologyDocument {
    TopologyDocument::new(
        3,
        "local".to_owned(),
        vec![
            device("local", Some("1.2.3"), Some(1_791_268_400_000)),
            device("edge", None, None),
        ],
        vec![
            // The exchange: every line's stop, its id the device's (lane
            // brief ruling 2).
            Station {
                id: "local".to_owned(),
                name: "local".to_owned(),
                device: "local".to_owned(),
                kind: StationKind::Exchange,
                state: StationState::Live,
            },
            // The persistent store: the durable line's stop before the
            // exchange (lane brief ruling 5).
            Station {
                id: "local.depot".to_owned(),
                name: "local.depot".to_owned(),
                device: "local".to_owned(),
                kind: StationKind::Depot,
                state: StationState::Live,
            },
            Station {
                id: "orders.reply".to_owned(),
                name: "orders.reply".to_owned(),
                device: "local".to_owned(),
                kind: StationKind::Service,
                state: StationState::Live,
            },
            // The orders line's publisher and its two subscribers, so every
            // tap hop names a listed station.
            worker("orders.worker"),
            worker("orders.ledger"),
            worker("orders.audit"),
        ],
        vec![
            Line {
                id: "orders".to_owned(),
                kind: LineKind::Stream,
                life: LineLife::Kept,
                state: LineState::Live,
                stops: [
                    "orders.worker",
                    "local.depot",
                    "local",
                    "orders.ledger",
                    "orders.audit",
                ]
                .map(str::to_owned)
                .to_vec(),
            },
            Line {
                id: "audit".to_owned(),
                kind: LineKind::Stream,
                life: LineLife::Passing,
                state: LineState::Quiesced,
                stops: vec!["local".to_owned()],
            },
        ],
        vec![Link::to_peer("local", "edge")],
        vec![Connection {
            id: "7".to_owned(),
            mount: Mount::Websocket,
            participant: Some("alice".to_owned()),
        }],
    )
}

fn message(
    id: &str,
    size: u64,
    outcome: Outcome,
    t: u64,
    reply_to: Option<&str>,
    from_station: Option<&str>,
) -> TapMessage {
    TapMessage {
        id: id.to_owned(),
        subject: "orders".to_owned(),
        size,
        outcome,
        t,
        from: "7".to_owned(),
        from_station: from_station.map(str::to_owned),
        to: "local".to_owned(),
        to_stations: from_station.map(|_| vec!["orders.ledger".to_owned()]),
        hops: Vec::new(),
        reply_to: reply_to.map(str::to_owned),
        held: false,
        diverted: None,
    }
}

fn hops(chain: &[(&str, u64)]) -> Vec<Hop> {
    chain
        .iter()
        .map(|(node, at_ms)| Hop {
            node: (*node).to_owned(),
            at_ms: *at_ms,
        })
        .collect()
}

/// The tap-batch golden. Each `t` is the receipt instant (lane brief ruling
/// 6), and its hops are measured from it; the fixtures' README says what
/// each message stands for.
fn tap_batch() -> Result<TapBatch, Box<dyn std::error::Error>> {
    Ok(TapBatch::new(
        "orders".to_owned(),
        7,
        NonZeroU32::new(4).ok_or("4 is not zero")?,
        2,
        vec![
            // One publisher, two subscribers, on a durable line: its hops
            // run publisher, depot, exchange, receivers (lane brief rulings
            // 3 and 8), each measured from its receipt.
            TapMessage {
                to_stations: Some(vec!["orders.ledger".to_owned(), "orders.audit".to_owned()]),
                hops: hops(&[
                    ("orders.worker", 0),
                    ("local.depot", 2),
                    ("local", 2),
                    ("orders.ledger", 4),
                    ("orders.audit", 6),
                ]),
                ..message(
                    "local:41",
                    12,
                    Outcome::Delivered,
                    1_791_272_000_123,
                    Some("q-1"),
                    Some("orders.worker"),
                )
            },
            // Recorded, and it reached no registered worker.
            TapMessage {
                to_stations: Some(vec![]),
                ..message(
                    "local:x9",
                    3,
                    Outcome::Refused,
                    1_791_272_000_456,
                    None,
                    None,
                )
            },
            // The server's own row: never a station, its fan-out recorded,
            // its append measured.
            TapMessage {
                from: "liminal.local".to_owned(),
                to_stations: Some(vec![]),
                hops: hops(&[("local.depot", 1), ("local", 1)]),
                ..message(
                    "local:42",
                    5,
                    Outcome::Delivered,
                    1_791_272_000_789,
                    None,
                    None,
                )
            },
            // Not recorded: whom it reached is not known.
            message(
                "local:43",
                4,
                Outcome::Delivered,
                1_791_272_000_901,
                None,
                None,
            ),
        ],
    ))
}

/// The bytes a subscriber receives: the channel's normaliser over the payload.
fn normalized(bytes: &[u8]) -> Result<Vec<u8>, Box<dyn std::error::Error>> {
    let normalizer = liminal::channel::Schema::new(serde_json::json!({}))?;
    Ok(normalizer.validate_and_apply_defaults(bytes)?)
}

/// Each record's encoding, normalised as the channel delivers it, equals its
/// golden file byte for byte, and the encoder's own bytes are already that
/// normal form (so the publisher sends exactly what a subscriber receives).
#[test]
fn console_records_match_their_golden_bytes() -> TestResult {
    for (name, encoded, golden) in [
        (
            "topology",
            canonical_bytes(&topology(), "topology")?,
            TOPOLOGY_GOLDEN,
        ),
        (
            "tap-batch",
            canonical_bytes(&tap_batch()?, "tap.batch")?,
            TAP_BATCH_GOLDEN,
        ),
        (
            "topology-clock-fault",
            canonical_bytes(&topology_clock_fault(), "topology")?,
            TOPOLOGY_CLOCK_FAULT_GOLDEN,
        ),
    ] {
        let delivered = normalized(&encoded)?;
        assert_eq!(
            delivered,
            golden,
            "{name}: delivered bytes differ from the golden file; delivered:\n{}",
            String::from_utf8_lossy(&delivered)
        );
        assert_eq!(encoded, delivered, "{name}: the encoder is not canonical");
    }
    Ok(())
}

/// The keys the console reads by name are present with the console's
/// spellings, and the absent ones are absent rather than null: a renamed
/// field in the encoder fails here even before the golden is regenerated.
#[test]
fn console_records_carry_the_console_spellings() -> TestResult {
    let topology: serde_json::Value = serde_json::from_slice(TOPOLOGY_GOLDEN)?;
    assert_eq!(topology["v"], 1);
    assert_eq!(topology["type"], "topology");
    assert_eq!(
        topology["links"][0]["ends"],
        serde_json::json!(["edge", "local"])
    );
    assert!(topology["links"][0]["rtt_ms"].is_null());
    assert!(topology["links"][0].get("refusals").is_none());
    assert!(topology.get("proposals").is_none());
    assert!(topology["stations"][0].get("sub").is_none());
    assert!(topology["stations"][0].get("product").is_none());
    assert!(topology["devices"][1].get("started").is_none());
    assert!(topology["devices"][1].get("liminal").is_none());

    let batch: serde_json::Value = serde_json::from_slice(TAP_BATCH_GOLDEN)?;
    assert_eq!(batch["v"], 1);
    assert_eq!(batch["type"], "tap.batch");
    assert_eq!(batch["sampled"], 4);
    assert_eq!(batch["messages"][0]["reply_to"], "q-1");
    assert!(batch["messages"][1].get("reply_to").is_none());
    assert!(batch["messages"][1].get("diverted").is_none());
    assert_eq!(batch["messages"][0]["hops"][0]["at_ms"], 0);
    Ok(())
}

/// A link's ends are sorted whichever device publishes it, so both ends of
/// one link send the same bytes and the console's reader accepts them.
#[test]
fn link_ends_are_sorted_from_either_end() {
    let from_local = Link::to_peer("local", "edge");
    let from_edge = Link::to_peer("edge", "local");
    assert_eq!(from_local.ends, ["edge".to_owned(), "local".to_owned()]);
    assert_eq!(from_local.ends, from_edge.ends);
}

/// Every server outcome lands on its console word, and only words the
/// console reads: delivered, nobody, unaccepted, unresolved, refused.
#[test]
fn each_server_outcome_has_its_console_word() -> TestResult {
    use crate::tap::TapOutcome;
    for (outcome, word) in [
        (TapOutcome::Delivered, "delivered"),
        (TapOutcome::NoSubscriber, "nobody"),
        (TapOutcome::NoneAccepted, "unaccepted"),
        (TapOutcome::Unresolved, "unresolved"),
        (TapOutcome::Refused, "refused"),
    ] {
        assert_eq!(serde_json::to_value(Outcome::from(outcome))?, word);
    }
    Ok(())
}

/// A tap batch becomes its record: ids from the message id or the
/// observation, `from` the connection or `liminal.<device>`, `to` this
/// device, `t` the wall base plus the monotonic time to first sight, and
/// the batch's own seq, divisor and dropped delta.
#[test]
fn a_tap_batch_becomes_its_record() -> TestResult {
    use std::sync::Arc;
    use std::time::{Duration, Instant};

    use super::records::tap_batch_record;
    use super::wall::WallBase;
    use crate::tap::{self, TapIdent, TapOutcome, TapSource};

    let base = Instant::now();
    let wall = WallBase::fixed(1_791_272_000_000, base);
    let message =
        |ident, source, station: Option<&str>, outcome, after_ms, reply_to: Option<&str>| {
            tap::TapMessage {
                ident,
                subject: "orders".to_owned(),
                size: 5,
                outcome,
                seen_at: base + Duration::from_millis(after_ms),
                source,
                station: station.map(Arc::from),
                to_stations: station.map(|_| {
                    Arc::from([tap::Reached {
                        station: Arc::from("orders.ledger"),
                        queued_at: base + Duration::from_millis(after_ms),
                    }])
                }),
                hops: vec![tap::Hop {
                    node: Arc::from("local"),
                    at_ms: 0,
                }],
                reply_to: reply_to.map(str::to_owned),
                held: false,
                payload: Some(b"never on the wire".to_vec()),
            }
        };
    let batch = tap::TapBatch {
        seq: 3,
        sampled: NonZeroU32::new(2).ok_or("2 is not zero")?,
        dropped: 1,
        messages: vec![
            message(
                TapIdent::Message(41),
                TapSource::Connection(7),
                Some("orders.worker"),
                TapOutcome::Delivered,
                123,
                Some("q-1"),
            ),
            tap::TapMessage {
                to_stations: Some(Arc::from(Vec::new())),
                ..message(
                    TapIdent::Observation(9),
                    TapSource::Server,
                    None,
                    TapOutcome::Refused,
                    456,
                    None,
                )
            },
        ],
    };

    let record = tap_batch_record("local", "orders", &batch, &wall)?;
    let bytes = canonical_bytes(&record, "tap.batch")?;
    let value: serde_json::Value = serde_json::from_slice(&bytes)?;
    assert_eq!(value["type"], "tap.batch");
    assert_eq!(value["target"], "orders");
    assert_eq!(
        (
            value["seq"].as_u64(),
            value["sampled"].as_u64(),
            value["dropped"].as_u64()
        ),
        (Some(3), Some(2), Some(1))
    );
    let first = &value["messages"][0];
    assert_eq!(first["id"], "local:41");
    assert_eq!(first["from"], "7");
    assert_eq!(first["from_station"], "orders.worker");
    assert_eq!(first["to_stations"], serde_json::json!(["orders.ledger"]));
    assert_eq!(first["to"], "local");
    assert_eq!(first["t"], 1_791_272_000_123_u64);
    assert_eq!(first["reply_to"], "q-1");
    assert_eq!(first["outcome"], "delivered");
    let second = &value["messages"][1];
    assert_eq!(second["id"], "local:x9");
    assert_eq!(second["from"], "liminal.local");
    assert_eq!(second["t"], 1_791_272_000_456_u64);
    assert_eq!(second["outcome"], "refused");
    assert!(second.get("reply_to").is_none());
    assert!(second.get("from_station").is_none());
    assert_eq!(
        second["to_stations"],
        serde_json::json!([]),
        "a recorded fan-out that reached no worker"
    );
    assert!(
        !String::from_utf8_lossy(&bytes).contains("never on the wire"),
        "a payload never rides the record"
    );
    Ok(())
}

/// Lane brief ruling 2 (Hermes 7 Oct 03:51), on the golden bytes the console
/// reads: every line's stops hold this device's exchange exactly once, and
/// the exchange is a listed station of kind `exchange` on this device.
#[test]
fn every_golden_line_passes_the_exchange_once() -> TestResult {
    let topology: serde_json::Value = serde_json::from_slice(TOPOLOGY_GOLDEN)?;
    let this = topology["this_device"]
        .as_str()
        .ok_or("the golden names no device")?;
    let lines = topology["lines"].as_array().ok_or("no lines")?;
    assert!(!lines.is_empty());
    for line in lines {
        let stops = line["stops"].as_array().ok_or("a line with no stops")?;
        let exchanges = stops
            .iter()
            .filter(|stop| stop.as_str() == Some(this))
            .count();
        assert_eq!(exchanges, 1, "{line}");
    }
    let exchange = topology["stations"]
        .as_array()
        .and_then(|stations| stations.iter().find(|station| station["id"] == this))
        .ok_or("the exchange is not a listed station")?;
    assert_eq!(
        (&exchange["kind"], &exchange["device"]),
        (&serde_json::json!("exchange"), &topology["this_device"])
    );
    Ok(())
}

/// Lane brief rulings 3, 4 and 8 (Hermes 7 Oct 03:51 to 03:57), on the
/// golden bytes the console reads: every tap hop names a listed station; a
/// message's hops never go back in time; a durable message passes the depot
/// before the exchange and reaches its receivers after it; a message that
/// appended nothing has no exchange hop (Jules's F1, Hermes 07:22); and no
/// hop carries `dwell_ms` (dwell waits, so the console's delay reading stays
/// off).
#[test]
fn every_golden_hop_names_a_station_in_order_without_dwell() -> TestResult {
    let topology: serde_json::Value = serde_json::from_slice(TOPOLOGY_GOLDEN)?;
    let batch: serde_json::Value = serde_json::from_slice(TAP_BATCH_GOLDEN)?;
    let stations: Vec<&str> = topology["stations"]
        .as_array()
        .ok_or("no stations")?
        .iter()
        .filter_map(|station| station["id"].as_str())
        .collect();
    let this = topology["this_device"].as_str().ok_or("no device")?;
    let depot = format!("{this}.depot");
    let messages = batch["messages"].as_array().ok_or("no messages")?;
    let mut durable = 0;
    let mut unmeasured = 0;
    for message in messages {
        let hops = message["hops"].as_array().ok_or("no hops")?;
        let nodes: Vec<&str> = hops.iter().filter_map(|hop| hop["node"].as_str()).collect();
        assert_eq!(nodes.len(), hops.len(), "{message}");
        for node in &nodes {
            assert!(stations.contains(node), "{node} is not a listed station");
        }
        let times: Vec<u64> = hops
            .iter()
            .filter_map(|hop| hop["at_ms"].as_u64())
            .collect();
        assert_eq!(times.len(), hops.len(), "{message}");
        assert!(
            times.is_sorted(),
            "{message}: a hop before the one ahead of it"
        );
        assert!(
            hops.iter().all(|hop| hop.get("dwell_ms").is_none()),
            "{message}"
        );
        let exchange = nodes.iter().position(|node| *node == this);
        assert!(nodes.iter().filter(|node| **node == this).count() <= 1);
        unmeasured += usize::from(exchange.is_none());
        if let Some(at) = nodes.iter().position(|node| *node == depot) {
            durable += 1;
            assert_eq!(
                Some(at + 1),
                exchange,
                "{message}: the depot, then the exchange"
            );
        }
        let receivers = message["to_stations"].as_array().map_or(0, Vec::len);
        let publisher = usize::from(message.get("from_station").is_some());
        assert_eq!(
            exchange.map_or(nodes.len() - publisher, |at| nodes.len() - at - 1),
            receivers,
            "{message}: each receiver after the exchange, or after the publisher"
        );
    }
    assert!(durable > 0, "a durable golden message passes the depot");
    assert!(messages.len() > durable, "and another does not");
    assert!(
        unmeasured > 0,
        "a golden message whose exchange nothing measured"
    );
    Ok(())
}