liminal-server 0.19.0

//! The hub's pins (SRV-012 R6, R7, R9): counts and published records, never
//! clocks. Tap numbers follow the sampling rule (a subscriber keeps a
//! matching publish when its own `seen % N == 0`, counting from 0; N doubles
//! after a push leaves the ring more than half full; a take halves it).

use std::cell::{Cell, RefCell};
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::Instant;

use serde_json::Value;

use super::error::ConsoleError;
use super::hub::{ConsoleHub, MAX_LIVE_TAPS, PumpReport, TOPOLOGY_SUBJECT};
use super::model::{FlowRole, ThisDevice, TopologyChange, TopologyModel};
use super::publisher::ServerPublisher;
use super::records::{LineLife, LineState};
use super::wall::WallBase;
use crate::tap::{PublishFact, TapOutcome, TapSource};

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

const ORDERS_TAP: &str = "console.tap.orders";

fn hub(capacity: usize) -> Result<ConsoleHub, ConsoleError> {
    let started = Instant::now();
    let model = TopologyModel::new(
        ThisDevice {
            id: "local".to_owned(),
            liminal: "1.2.3".to_owned(),
            started,
            depot: false,
        },
        [],
        [],
        [],
        [],
    );
    ConsoleHub::with_capacity("local", model, WallBase::fixed(1_000, started), capacity)
}

fn publish(hub: &ConsoleHub, times: usize) -> TestResult {
    for _ in 0..times {
        hub.tap().observe(&PublishFact {
            subject: "orders",
            payload: b"{}",
            outcome: TapOutcome::Delivered,
            message_id: None,
            source: TapSource::Connection(7),
            reply_to: None,
            recorded: None,
        })?;
    }
    Ok(())
}

/// A publisher that refuses its next `failures` publishes, then records.
#[derive(Default)]
struct Scripted {
    failures: Cell<u32>,
    records: RefCell<Vec<(String, Value)>>,
    retired: RefCell<Vec<String>>,
}

impl Scripted {
    fn take(&self) -> Vec<(String, Value)> {
        self.records.take()
    }
}

fn refused(subject: &str) -> ConsoleError {
    ConsoleError::Server {
        subject: subject.to_owned(),
        source: crate::ServerError::ListenerAccept {
            message: "scripted refusal".to_owned(),
        },
    }
}

impl ServerPublisher for Scripted {
    fn publish_console(&self, subject: &str, payload: &[u8]) -> Result<(), ConsoleError> {
        if self.failures.get() > 0 {
            self.failures.set(self.failures.get() - 1);
            return Err(refused(subject));
        }
        let value = serde_json::from_slice(payload).map_err(|source| ConsoleError::Encode {
            record: "test read-back",
            source,
        })?;
        self.records.borrow_mut().push((subject.to_owned(), value));
        Ok(())
    }

    fn retire_console_channel(&self, subject: &str) -> Result<(), ConsoleError> {
        self.retired.borrow_mut().push(subject.to_owned());
        Ok(())
    }
}

/// (seq, sampled, dropped, messages) of each tap record; a field that is
/// missing or of the wrong type fails the test rather than reading as 0.
fn shape(records: &[(String, Value)]) -> Result<Vec<(u64, u64, u64, usize)>, String> {
    records
        .iter()
        .map(|(_, record)| {
            let number = |field: &str| {
                record[field]
                    .as_u64()
                    .ok_or_else(|| format!("tap record has no numeric {field}"))
            };
            let messages = record["messages"]
                .as_array()
                .ok_or("tap record has no messages")?;
            Ok((
                number("seq")?,
                number("sampled")?,
                number("dropped")?,
                messages.len(),
            ))
        })
        .collect()
}

/// R9: a failed record is kept and published first; the tap's ring is not
/// drained while it is kept, so what came after waits there.
///
/// Capacity 8: p1–p3 kept at N=1 (3 of 8 is not more than half). The pump
/// takes seq 1 (three messages) and its publish fails. p4, p5 kept. The next
/// pump publishes seq 1 only; the one after drains seq 2 (two messages).
#[test]
fn a_failed_tap_publication_is_retried_first() -> TestResult {
    let hub = hub(8)?;
    let out = Scripted::default();
    hub.attach(ORDERS_TAP)?;
    publish(&hub, 3)?;

    out.failures.set(1);
    let failed = hub.pump(&out)?;
    assert_eq!(
        failed,
        PumpReport {
            publish_failures: 1,
            ..PumpReport::default()
        }
    );
    assert_eq!(hub.counts().console_pending_records, 1);

    publish(&hub, 2)?;
    assert_eq!(hub.pump(&out)?.tap_records, 1);
    assert_eq!(
        shape(&out.take())?,
        vec![(1, 1, 0, 3)],
        "the kept record first"
    );
    assert_eq!(hub.pump(&out)?.tap_records, 1);
    let drained = out.take();
    assert_eq!(drained[0].0, ORDERS_TAP);
    assert_eq!(shape(&drained)?, vec![(2, 1, 0, 2)]);

    let counts = hub.counts();
    assert_eq!(counts.console_publish_failures, 1);
    assert_eq!(counts.tap_records_published, 2);
    assert_eq!(counts.console_pending_records, 0);
    assert_eq!(counts.console_records_lost, 0);
    Ok(())
}

/// R9: two failures in a row lose nothing but what the ring evicts, and the
/// eviction is reported in the next batch's `dropped`.
///
/// Capacity 2: p1 (seen 0) kept at N=1; the pump takes seq 1 and fails. The
/// ring is empty again: p2 (seen 1) kept; p3 (seen 2) kept, two of two, N=2;
/// p4 (seen 3) out; p5 (seen 4) kept at 2, evicting p2, N=4; p6–p8 out; p9
/// (seen 8) kept at 4, evicting p3, N=8. The second pump fails on the kept
/// seq 1 and drains nothing. The third publishes seq 1; the fourth drains
/// [p5]@2 carrying both evictions and [p9]@4.
#[test]
fn a_second_failure_loses_only_to_eviction() -> TestResult {
    let hub = hub(2)?;
    let out = Scripted::default();
    hub.attach(ORDERS_TAP)?;
    publish(&hub, 1)?;
    out.failures.set(2);
    assert_eq!(hub.pump(&out)?.publish_failures, 1);
    publish(&hub, 8)?;
    assert_eq!(hub.pump(&out)?.publish_failures, 1);
    assert_eq!(
        hub.counts().console_pending_records,
        1,
        "seq 1 is still kept"
    );

    assert_eq!(hub.pump(&out)?.tap_records, 1);
    assert_eq!(hub.pump(&out)?.tap_records, 2);
    assert_eq!(
        shape(&out.take())?,
        vec![(1, 1, 0, 1), (2, 2, 2, 1), (3, 4, 0, 1)]
    );

    let counts = hub.counts();
    assert_eq!(counts.tap_dropped_total, 2);
    assert_eq!(counts.console_records_lost, 0, "the ring is the only loss");
    assert_eq!(counts.console_publish_failures, 2);
    Ok(())
}

/// R4 and R7: with nobody on `console.topology` a change wakes nobody and
/// publishes nothing; a subscriber is sent the latest document, a second
/// subscriber is sent it again, and a repeat change sends nothing.
#[test]
fn a_new_topology_subscriber_gets_the_latest_document() -> TestResult {
    let hub = hub(8)?;
    let out = Scripted::default();
    let wakes = Arc::new(AtomicU64::new(0));
    let counter = Arc::clone(&wakes);
    hub.set_notifier(Arc::new(move || {
        counter.fetch_add(1, Ordering::Relaxed);
    }))?;
    let opened = TopologyChange::Service {
        subject: "orders.reply".to_owned(),
        present: true,
    };

    hub.emit(opened.clone())?;
    assert_eq!(wakes.load(Ordering::Relaxed), 0, "idle with no subscriber");
    assert_eq!(hub.pump(&out)?, PumpReport::default());

    hub.attach(TOPOLOGY_SUBJECT)?;
    assert_eq!(wakes.load(Ordering::Relaxed), 1);
    assert_eq!(hub.pump(&out)?.topology_records, 1);
    hub.attach(TOPOLOGY_SUBJECT)?;
    assert_eq!(hub.pump(&out)?.topology_records, 1);
    let sent = out.take();
    let revs: Vec<_> = sent.iter().map(|(_, doc)| doc["rev"].as_u64()).collect();
    assert_eq!(revs, vec![Some(2), Some(2)], "start is 1; one change");
    let stations: Vec<&str> = sent[0].1["stations"]
        .as_array()
        .ok_or("no stations")?
        .iter()
        .filter_map(|station| station["id"].as_str())
        .collect();
    assert_eq!(
        stations,
        ["local", "orders.reply"],
        "the exchange and the service"
    );

    hub.emit(opened)?;
    assert_eq!(
        hub.pump(&out)?,
        PumpReport::default(),
        "a repeat is no change"
    );
    assert_eq!(hub.counts().topology_records_published, 2);
    Ok(())
}

/// R9 for the topology: a failed document is due again and the next pump
/// sends the document as it stands then.
#[test]
fn a_failed_topology_publication_is_sent_as_it_stands_next() -> TestResult {
    let hub = hub(8)?;
    let out = Scripted::default();
    hub.attach(TOPOLOGY_SUBJECT)?;
    out.failures.set(1);
    assert_eq!(hub.pump(&out)?.publish_failures, 1);
    hub.emit(TopologyChange::Peer {
        name: "edge".to_owned(),
        present: true,
    })?;
    assert_eq!(hub.pump(&out)?.topology_records, 1);
    let sent = out.take();
    assert_eq!(sent.len(), 1);
    assert_eq!(sent[0].1["rev"].as_u64(), Some(2));
    assert_eq!(hub.pump(&out)?, PumpReport::default());
    Ok(())
}

/// A registered worker's publishes name its station until its connection
/// closes; from then its publishes, and the tap's station table, hold none.
#[test]
fn a_workers_station_is_named_until_its_connection_closes() -> TestResult {
    let hub = hub(8)?;
    let out = Scripted::default();
    hub.attach("console.tap.orders")?;
    publish(&hub, 1)?;
    hub.emit(TopologyChange::Worker {
        connection: 7,
        identity: "orders.pricer".to_owned(),
    })?;
    publish(&hub, 1)?;
    hub.emit(TopologyChange::ConnectionClosed { id: 7 })?;
    publish(&hub, 1)?;
    hub.pump(&out)?;
    let stations: Vec<Option<Value>> = out
        .take()
        .iter()
        .filter(|(subject, _)| subject == "console.tap.orders")
        .flat_map(|(_, record)| record["messages"].as_array().cloned().unwrap_or_default())
        .map(|message| message.get("from_station").cloned())
        .collect();
    assert_eq!(stations, [None, Some(Value::from("orders.pricer")), None]);
    Ok(())
}

/// R6: at most [`MAX_LIVE_TAPS`] taps; one more is refused by name and
/// counted, while another subscriber to a live tap is not a new tap.
#[test]
fn the_live_tap_limit_refuses_one_more() -> TestResult {
    let hub = hub(8)?;
    for target in 0..MAX_LIVE_TAPS {
        hub.attach(&format!("console.tap.t{target}"))?;
    }
    assert!(!hub.tap_room("console.tap.one-more")?);
    assert!(hub.tap_room("console.tap.t0")?);
    assert!(hub.tap_room("orders")?);
    assert!(matches!(
        hub.attach("console.tap.one-more"),
        Err(ConsoleError::TapLimit {
            limit: MAX_LIVE_TAPS,
            ..
        })
    ));
    hub.attach("console.tap.t0")?;

    let counts = hub.counts();
    assert_eq!(counts.tap_attach_failures, 1);
    assert_eq!(counts.taps_attached, 256);
    Ok(())
}

/// R6: the tap stops and its channel leaves the roster when the last
/// subscriber leaves, not before; one attached again before the pump stays.
#[test]
fn the_last_subscriber_retires_the_tap_channel() -> TestResult {
    let hub = hub(8)?;
    let out = Scripted::default();
    hub.attach(ORDERS_TAP)?;
    hub.attach(ORDERS_TAP)?;
    hub.detach(ORDERS_TAP)?;
    assert_eq!(hub.pump(&out)?.retired, 0);
    assert_eq!(hub.attached_subjects()?, vec![ORDERS_TAP.to_owned()]);

    hub.detach(ORDERS_TAP)?;
    assert!(hub.attached_subjects()?.is_empty());
    assert_eq!(hub.tap().subscriber_count(), 0);
    assert_eq!(hub.pump(&out)?.retired, 1);
    assert_eq!(out.retired.take(), vec![ORDERS_TAP.to_owned()]);

    hub.attach(ORDERS_TAP)?;
    hub.detach(ORDERS_TAP)?;
    hub.attach(ORDERS_TAP)?;
    assert_eq!(hub.pump(&out)?.retired, 0, "attached again: it stays");
    assert_eq!(hub.counts().taps_attached, 1);
    Ok(())
}

/// A publisher that notes, at each roster removal, whether the hub's taps
/// lock was held.
struct LockWitness<'hub> {
    hub: &'hub ConsoleHub,
    held: RefCell<Vec<bool>>,
}

impl ServerPublisher for LockWitness<'_> {
    fn publish_console(&self, _subject: &str, _payload: &[u8]) -> Result<(), ConsoleError> {
        Ok(())
    }

    fn retire_console_channel(&self, _subject: &str) -> Result<(), ConsoleError> {
        self.held.borrow_mut().push(self.hub.taps_held());
        Ok(())
    }
}

/// The retire's re-check and its roster removal are one step under the taps
/// lock, so a subscriber attaching in between cannot lose its channel.
#[test]
fn a_tap_channel_leaves_the_roster_under_the_taps_lock() -> TestResult {
    let hub = hub(8)?;
    hub.attach(ORDERS_TAP)?;
    hub.detach(ORDERS_TAP)?;
    let out = LockWitness {
        hub: &hub,
        held: RefCell::new(Vec::new()),
    };

    assert_eq!(hub.pump(&out)?.retired, 1);

    assert_eq!(out.held.take(), vec![true]);
    Ok(())
}

/// What leaves with a tap whose last subscriber went is counted by name:
/// the unsent record it held and the messages still in its ring.
#[test]
fn a_departed_taps_records_are_counted_as_abandoned() -> TestResult {
    let hub = hub(8)?;
    let out = Scripted::default();
    hub.attach(ORDERS_TAP)?;
    publish(&hub, 3)?;
    out.failures.set(1);
    hub.pump(&out)?;
    publish(&hub, 2)?;

    hub.detach(ORDERS_TAP)?;

    let counts = hub.counts();
    assert_eq!(counts.tap_records_abandoned, 1);
    assert_eq!(counts.tap_messages_abandoned, 2);
    assert_eq!(counts.console_records_lost, 0);
    Ok(())
}

/// A publisher whose first publication sees the tap leave and come back,
/// then fails.
struct Resubscribing<'hub> {
    hub: &'hub ConsoleHub,
    calls: Cell<u64>,
}

impl ServerPublisher for Resubscribing<'_> {
    fn publish_console(&self, subject: &str, _payload: &[u8]) -> Result<(), ConsoleError> {
        self.calls.set(self.calls.get() + 1);
        if self.calls.get() > 1 {
            return Ok(());
        }
        self.hub.detach(subject).map_err(ConsoleError::from)?;
        self.hub.attach(subject)?;
        Err(refused(subject))
    }

    fn retire_console_channel(&self, _subject: &str) -> Result<(), ConsoleError> {
        Ok(())
    }
}

/// An earlier tap's unsent record never reaches a tap attached after it, so
/// the new tap's seq starts afresh; the record is counted as abandoned.
#[test]
fn a_resubscribed_tap_gets_none_of_the_old_taps_records() -> TestResult {
    let hub = hub(8)?;
    hub.attach(ORDERS_TAP)?;
    publish(&hub, 3)?;
    let out = Resubscribing {
        hub: &hub,
        calls: Cell::new(0),
    };

    assert_eq!(hub.pump(&out)?.publish_failures, 1);

    let counts = hub.counts();
    assert_eq!(counts.console_pending_records, 0);
    assert_eq!(counts.tap_records_abandoned, 1);
    Ok(())
}

/// One tap's failed take loses nothing of the others: the tap before it is
/// still published in the same wake, the error comes back after, and the
/// failing tap's ring is taken on the next wake.
#[test]
fn a_failed_take_keeps_every_other_taps_records() -> TestResult {
    const BILLING_TAP: &str = "console.tap.billing";
    let hub = hub(8)?;
    let out = Scripted::default();
    hub.attach(BILLING_TAP)?;
    hub.attach(ORDERS_TAP)?;
    hub.tap().observe(&PublishFact {
        subject: "billing",
        payload: b"{}",
        outcome: TapOutcome::Delivered,
        message_id: None,
        source: TapSource::Connection(7),
        reply_to: None,
        recorded: None,
    })?;
    publish(&hub, 2)?;
    hub.fail_next_take(ORDERS_TAP);

    assert!(hub.pump(&out).is_err(), "the failed take is returned");
    let subjects: Vec<String> = out.take().into_iter().map(|(subject, _)| subject).collect();
    assert_eq!(subjects, vec![BILLING_TAP.to_owned()]);

    assert_eq!(hub.pump(&out)?.tap_records, 1);
    assert_eq!(shape(&out.take())?, vec![(1, 1, 0, 2)]);
    assert_eq!(hub.counts().console_records_lost, 0);
    Ok(())
}

/// Buttercup's finding 3 (Hermes 03:05): a clock error building the
/// topology document never stops the feed. The document still goes out,
/// naming the fault on this device where its start would be; the taps pump
/// in the same pump; and the next pump goes on.
#[test]
fn a_clock_error_in_the_document_never_stops_the_feed() -> TestResult {
    let at = Instant::now();
    let started = at
        .checked_sub(std::time::Duration::from_secs(2))
        .ok_or("no instant two seconds back")?;
    let model = TopologyModel::new(
        ThisDevice {
            id: "local".to_owned(),
            liminal: "1.2.3".to_owned(),
            started,
            depot: false,
        },
        [],
        [],
        [],
        [],
    );
    // The wall base reads one second after the epoch, so a start two seconds
    // before it falls before the epoch.
    let hub = ConsoleHub::with_capacity("local", model, WallBase::fixed(1_000, at), 8)?;
    let out = Scripted::default();
    hub.attach(TOPOLOGY_SUBJECT)?;
    hub.attach(ORDERS_TAP)?;
    publish(&hub, 2)?;
    hub.pump(&out)?;
    publish(&hub, 1)?;
    hub.pump(&out)?;
    hub.attach(TOPOLOGY_SUBJECT)?;
    hub.pump(&out)?;

    let records = out.take();
    let subjects: Vec<&str> = records
        .iter()
        .map(|(subject, _)| subject.as_str())
        .collect();
    assert_eq!(
        subjects,
        [TOPOLOGY_SUBJECT, ORDERS_TAP, ORDERS_TAP, TOPOLOGY_SUBJECT]
    );
    for (_, record) in records
        .iter()
        .filter(|(subject, _)| subject == TOPOLOGY_SUBJECT)
    {
        let this = &record["devices"][0];
        assert!(this.get("started").is_none(), "{this}");
        let fault = this["started_fault"]
            .as_str()
            .ok_or_else(|| format!("no started_fault: {this}"))?;
        assert!(fault.starts_with("console clock: "), "{fault}");
    }
    assert_eq!(
        hub.counts().topology_clock_failures,
        2,
        "each document sent with the fault is counted"
    );
    Ok(())
}

/// Buttercup's V1, as Hermes ruled at 04:27: a worker identity equal to this
/// server's exchange id or depot id is kept apart and counted, never merged.
/// The registration stands; the worker is left off the map and out of the
/// tap's stations; the exchange and the depot keep their kinds, and the line
/// it subscribes to passes the exchange once.
#[test]
fn a_worker_named_as_the_exchange_or_the_depot_is_kept_apart_and_counted() -> TestResult {
    let started = Instant::now();
    let model = TopologyModel::new(
        ThisDevice {
            id: "local".to_owned(),
            liminal: "1.2.3".to_owned(),
            started,
            depot: true,
        },
        [],
        [],
        [],
        [],
    );
    let hub = ConsoleHub::with_capacity("local", model, WallBase::fixed(1_000, started), 8)?;
    let out = Scripted::default();
    hub.attach(TOPOLOGY_SUBJECT)?;
    hub.attach(ORDERS_TAP)?;
    hub.emit(TopologyChange::Channel {
        name: "orders".to_owned(),
        life: LineLife::Kept,
        state: LineState::Live,
    })?;
    for (connection, identity) in [(7, "local"), (8, "local.depot")] {
        hub.emit(TopologyChange::Worker {
            connection,
            identity: identity.to_owned(),
        })?;
        hub.emit(TopologyChange::Flow {
            connection,
            line: "orders".to_owned(),
            role: FlowRole::Subscribes,
            present: true,
        })?;
    }
    publish(&hub, 1)?;
    assert_eq!(hub.counts().station_collisions, 2);
    hub.pump(&out)?;
    let sent = out.take();
    let document = &sent
        .iter()
        .rev()
        .find(|(subject, _)| subject == TOPOLOGY_SUBJECT)
        .ok_or("no topology document")?
        .1;
    let kinds: Vec<(&str, &str)> = document["stations"]
        .as_array()
        .ok_or("no stations")?
        .iter()
        .filter_map(|station| Some((station["id"].as_str()?, station["kind"].as_str()?)))
        .collect();
    assert_eq!(kinds, [("local", "exchange"), ("local.depot", "depot")]);
    assert_eq!(
        document["lines"][0]["stops"],
        serde_json::json!(["local.depot", "local"])
    );
    let from: Vec<Option<Value>> = sent
        .iter()
        .filter(|(subject, _)| subject == ORDERS_TAP)
        .flat_map(|(_, record)| record["messages"].as_array().cloned().unwrap_or_default())
        .map(|message| message.get("from_station").cloned())
        .collect();
    assert_eq!(from, [None], "the tap names no station for it");
    Ok(())
}