liminal_server/console/

records.rs

//! The console's records, in the field names of the console's own reader.
//!
//! The reader is `apps/liminal-console/src/live/types.ts` (`TopologyDocument`
//! and `TapBatch`), so a record the server publishes is a record
//! `parseRecord` reads.
//!
//! The bytes are the server's canonical form: the record is turned into a JSON
//! value, whose maps keep their keys sorted, and serialised from there. That is
//! the same output the channel's schema normaliser gives, so `v` arrives in
//! sorted position, not first, and a reader takes every field by key.
//!
//! Left out of the topology because nothing in the server holds them yet:
//! archives, exports, gates and proposed routes, a link's round
//! trip (sent as `null`), and a line's product and stops beyond
//! what the roster knows. Nothing is sent as a made-up value in their place.

mod carrier_refusals;

use serde::Serialize;

use super::error::ConsoleError;

/// The record version the console reads.
pub const RECORD_VERSION: u64 = 1;

/// A device: this one, a cluster peer, or a federated peer.
#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
pub struct Device {
    /// The device id (the cluster node name, or `local` without a cluster).
    pub id: String,
    /// The device's name as shown.
    pub name: String,
    /// Always a device: federation has no source.
    pub kind: DeviceKind,
    /// The liminal version running there: known for this device (from the
    /// build), not for a peer, whose version nothing here holds.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub liminal: Option<String>,
    /// When the server's incarnation started, in wall-clock milliseconds
    /// since the epoch: known for this device, not for a peer. A reader takes
    /// the uptime as its own now less this, so the figure never goes stale
    /// between documents (Hermes 7 Oct 01:49, which replaced `uptime`).
    #[serde(skip_serializing_if = "Option::is_none")]
    pub started: Option<u64>,
    /// Why this device's start is missing, in the server's words: the clock
    /// fault met building the document (Buttercup's finding 3, Hermes 03:05).
    /// Present only when `started` is absent on this device, so never both.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub started_fault: Option<String>,
    /// The device this one is a district of, when configured.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub parent: Option<String>,
    /// Where the map draws the device's zone (`box` in types.ts), when this
    /// server places it.
    #[serde(rename = "box", skip_serializing_if = "Option::is_none")]
    pub zone: Option<DeviceBox>,
}

/// A device's zone on the console's map, in the map's own units.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
pub struct DeviceBox {
    /// Left edge.
    pub x: i64,
    /// Top edge.
    pub y: i64,
    /// Width.
    pub w: i64,
    /// Height.
    pub h: i64,
}

impl DeviceBox {
    /// This server's own zone: the map's fitted area inset by 40 on each
    /// side, the zone the console draws for a lone device (`NetworkPage`'s
    /// `FIT` of -180, 20, 1420, 600).
    pub const THIS: Self = Self {
        x: -140,
        y: 60,
        w: 1340,
        h: 520,
    };
}

/// What a device is.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
#[serde(rename_all = "lowercase")]
pub enum DeviceKind {
    /// A liminal server.
    Device,
}

/// A station: a registered service.
#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
pub struct Station {
    /// The station id (the responder's subject).
    pub id: String,
    /// Its name as shown.
    pub name: String,
    /// The device it runs on.
    pub device: String,
    /// What kind of station.
    pub kind: StationKind,
    /// Its state.
    pub state: StationState,
    /// SRV-027 R3: the subjects a registered worker serves (its
    /// registration's task queue). Absent on a station that is not a worker.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub subjects: Option<Vec<String>>,
    /// SRV-027 R3: what a registered worker accepts and returns, each
    /// activity's schemas as the exact strings it registered; empty for a
    /// worker that declared none (raw bytes in and out). Absent on a
    /// station that is not a worker.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub activities: Option<Vec<StationActivity>>,
}

/// One activity a worker declared at registration (SRV-027 R3): its name and
/// its input and output schemas, carried as the strings the registration
/// held, never parsed or normalised here.
#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
pub struct StationActivity {
    /// The activity's name.
    pub name: String,
    /// The input schema as registered.
    pub input_schema_json: String,
    /// The output schema as registered.
    pub output_schema_json: String,
}

/// What a station is. Only services have a source until the services
/// registry exists.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
#[serde(rename_all = "lowercase")]
pub enum StationKind {
    /// A registered conversation responder.
    Service,
    /// A registered worker.
    Worker,
    /// This server's persistent store, which a durable line's messages
    /// reach before the fan-out (lane brief ruling 5).
    Depot,
    /// This server's exchange, the fan-out every line passes; its id is the
    /// device's (lane brief ruling 2).
    Exchange,
}

/// A station's state. A registered responder is live; nothing in the server
/// is asleep, held or retired as a station yet.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
#[serde(rename_all = "lowercase")]
pub enum StationState {
    /// Registered and answering.
    Live,
}

/// A line: a stream on the roster.
#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
pub struct Line {
    /// The channel name.
    pub id: String,
    /// A channel stream or a live serving subject.
    pub kind: LineKind,
    /// How long its messages are kept.
    pub life: LineLife,
    /// Whether it admits.
    pub state: LineState,
    /// The stations it passes, in order.
    pub stops: Vec<String>,
}

/// What a line is.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
#[serde(rename_all = "lowercase")]
pub enum LineKind {
    /// A channel.
    Stream,
    /// A subject served by a live responder or worker.
    Call,
}

/// How long a line keeps its messages, from the channel's mode.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
#[serde(rename_all = "lowercase")]
pub enum LineLife {
    /// Ephemeral: messages pass and are not stored.
    Passing,
    /// Durable: messages are stored.
    Kept,
}

/// Whether a line admits.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
#[serde(rename_all = "lowercase")]
pub enum LineState {
    /// Admitting publishes and subscribes.
    Live,
    /// Quiesced: refusing new publishes and subscribes.
    Quiesced,
}

/// A cluster or federated link, with only observed network facts.
#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
pub struct Link {
    /// Carrier refusals retained across later observations.
    pub carrier_refusals: Vec<String>,
    /// The device carrying a relayed link.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub carried_by: Option<String>,
    /// The link id, `<this device>~<peer>`.
    pub id: String,
    /// The peer's name as shown.
    pub name: String,
    /// How the two devices are joined.
    pub kind: LinkKind,
    /// The link's state.
    pub state: LinkState,
    /// The transport carrying this link, when observed.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub carrier: Option<Carrier>,
    /// The relaying device, only for a relay carrier.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub relayed: Option<String>,
    /// The cause of a down link, when observed.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub reason: Option<String>,
    /// Observed crossings; absent when no source counts them.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub crossed: Option<u64>,
    /// Observed refusals; absent when no source counts them.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub refused: Option<u64>,
    /// Crossings answered from an existing receipt.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub already_crossed: Option<u64>,
    /// Observed opens and ends by carrier cause.
    #[serde(flatten)]
    pub lifecycle: Option<crate::link::LinkLifeCounters>,
    /// Round trip in milliseconds. No source: always `null`, never `0`.
    pub rtt_ms: Option<u64>,
    /// The two device ids it joins, in sorted order, so the same link has
    /// the same bytes whichever end publishes it.
    pub ends: [String; 2],
}

impl Link {
    /// The link from `local` to `peer`, up, with no round trip measured.
    #[must_use]
    pub fn to_peer(local: &str, peer: &str) -> Self {
        let ends = if peer < local {
            [peer.to_owned(), local.to_owned()]
        } else {
            [local.to_owned(), peer.to_owned()]
        };
        Self {
            id: format!("{local}~{peer}"),
            name: peer.to_owned(),
            kind: LinkKind::SameNetwork,
            state: LinkState::Up,
            rtt_ms: None,
            carrier: None,
            carrier_refusals: Vec::new(),
            carried_by: None,
            relayed: None,
            reason: None,
            crossed: None,
            refused: None,
            already_crossed: None,
            lifecycle: None,
            ends,
        }
    }
    /// A federated observation; an up link carries no end reason.
    #[must_use]
    pub fn federated(
        local: &str,
        peer: &str,
        carrier: Option<Carrier>,
        state: LinkState,
        reason: Option<String>,
        counters: LinkCounters,
    ) -> Self {
        let relayed = match &carrier {
            Some(Carrier::Relay(device)) => Some(device.clone()),
            _ => None,
        };
        Self {
            kind: LinkKind::Federated,
            state,
            carrier,
            carried_by: relayed.clone(),
            carrier_refusals: reason.clone().into_iter().collect(),
            relayed,
            reason: if state == LinkState::Down {
                reason
            } else {
                None
            },
            crossed: counters.crossed,
            refused: counters.refused,
            already_crossed: counters.already_crossed,
            lifecycle: counters.lifecycle,
            ..Self::to_peer(local, peer)
        }
    }
}

/// The transport and, for a relay, the device carrying the link.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum Carrier {
    /// A direct wire connection.
    Wire,
    /// A direct UDP connection.
    Direct,
    /// A connection carried by the named relaying device.
    Relay(String),
}

impl Serialize for Carrier {
    fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
        serializer.serialize_str(match self {
            Self::Wire => "wire",
            Self::Direct => "direct",
            Self::Relay(_) => "relay",
        })
    }
}

/// Independently observed counters; unknown is absent, known zero is zero.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct LinkCounters {
    /// Crossings answered from an existing receipt.
    pub already_crossed: Option<u64>,
    /// Opens and ends counted by the lifecycle owner.
    pub lifecycle: Option<crate::link::LinkLifeCounters>,
    /// Crossings counted by the source.
    pub crossed: Option<u64>,
    /// Refusals counted by the source.
    pub refused: Option<u64>,
}

/// How two devices are joined.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
pub enum LinkKind {
    /// Members of one cluster.
    #[serde(rename = "same network")]
    SameNetwork,
    /// Machines joined across networks.
    #[serde(rename = "federated")]
    Federated,
}

/// A link's observed state. A departed cluster peer has no down record.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
#[serde(rename_all = "lowercase")]
pub enum LinkState {
    /// The peer is a member.
    Up,
    /// A federated link ended.
    Down,
}

/// A live connection to this server.
#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
pub struct Connection {
    /// The connection id, as a tap message's `from` names it.
    pub id: String,
    /// The transport it arrived on.
    pub mount: Mount,
    /// The participant its pass names, when it presented one.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub participant: Option<String>,
}

/// The transport a connection arrived on.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
#[serde(rename_all = "lowercase")]
pub enum Mount {
    /// The TCP wire listener.
    Tcp,
    /// The WebSocket listener.
    Websocket,
    /// The in-process transport.
    Loopback,
}

impl From<crate::server::mount::MountKind> for Mount {
    fn from(mount: crate::server::mount::MountKind) -> Self {
        match mount {
            crate::server::mount::MountKind::Tcp => Self::Tcp,
            crate::server::mount::MountKind::WebSocket => Self::Websocket,
            crate::server::mount::MountKind::Loopback => Self::Loopback,
        }
    }
}

/// One `console.topology` record: the whole document as it stands.
#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
pub struct TopologyDocument {
    v: u64,
    #[serde(rename = "type")]
    record_type: &'static str,
    /// One more than the previous document's, from 1 at start.
    pub rev: u64,
    /// The id of the device this server is, one of `devices`' ids (lane
    /// brief ruling 1, Hermes 7 Oct 03:51): named once here, never as a flag
    /// on each device.
    pub this_device: String,
    /// This device, then each peer.
    pub devices: Vec<Device>,
    /// The registered services.
    pub stations: Vec<Station>,
    /// Every roster channel outside `console.`.
    pub lines: Vec<Line>,
    /// Cluster peer links, then known federated links.
    pub links: Vec<Link>,
    /// One per live connection.
    pub connections: Vec<Connection>,
}

impl TopologyDocument {
    /// A document at `rev`, from the server that is device `this_device`.
    #[must_use]
    pub const fn new(
        rev: u64,
        this_device: String,
        devices: Vec<Device>,
        stations: Vec<Station>,
        lines: Vec<Line>,
        peer_links: Vec<Link>,
        connections: Vec<Connection>,
    ) -> Self {
        Self {
            v: RECORD_VERSION,
            record_type: "topology",
            rev,
            this_device,
            devices,
            stations,
            lines,
            links: peer_links,
            connections,
        }
    }
}

mod tap_records;
pub use tap_records::{Hop, Outcome, TapBatch, TapMessage, tap_batch_record};

/// The canonical bytes of a record: through a JSON value, so every key is in
/// sorted order at every depth, as the schema normaliser writes them.
///
/// # Errors
/// Returns [`ConsoleError::Encode`] naming the record when serialisation fails.
pub fn canonical_bytes<Record: Serialize>(
    record: &Record,
    kind: &'static str,
) -> Result<Vec<u8>, ConsoleError> {
    serde_json::to_value(record)
        .and_then(|value| serde_json::to_vec(&value))
        .map_err(|source| ConsoleError::Encode {
            record: kind,
            source,
        })
}