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, federation and proposed routes, a link's round
//! trip (sent as `null`) and refusals, and a line's product and stops beyond
//! what the roster knows. Nothing is sent as a made-up value in their place.

use std::num::NonZeroU32;

use serde::Serialize;

use super::error::ConsoleError;
use super::wall::WallBase;
use crate::tap::{self, TapIdent, TapOutcome, TapSource};

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

/// A device: this one, or a cluster 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>,
    /// 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,
}

/// 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,
    /// Always a stream: calls, archives and exports have no source.
    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,
}

/// 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 link to a cluster peer, present while the peer is.
#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
pub struct Link {
    /// 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,
    /// 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,
            ends,
        }
    }
}

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

/// A link's state. A departed peer's link is absent, not down: membership
/// keeps no record of why it left.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
#[serde(rename_all = "lowercase")]
pub enum LinkState {
    /// The peer is a member.
    Up,
}

/// 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>,
    /// One per peer.
    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,
        }
    }
}

/// What became of a tapped message, in the console's words.
///
/// Only the words the server has an honest cause for: the console's call words (reply,
/// died, deadline), deferred, dropped and redelivered have none here.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
#[serde(rename_all = "lowercase")]
pub enum Outcome {
    /// Taken by at least one subscriber.
    Delivered,
    /// Admitted, and no subscriber matched.
    Nobody,
    /// Admitted; subscribers matched and none took it.
    Unaccepted,
    /// Refused at admission, by the schema, or by the library.
    Refused,
    /// The store could not say whether the append landed.
    Unresolved,
}

impl From<TapOutcome> for Outcome {
    fn from(outcome: TapOutcome) -> Self {
        match outcome {
            TapOutcome::Delivered => Self::Delivered,
            TapOutcome::NoSubscriber => Self::Nobody,
            TapOutcome::NoneAccepted => Self::Unaccepted,
            TapOutcome::Unresolved => Self::Unresolved,
            TapOutcome::Refused => Self::Refused,
        }
    }
}

/// One point a message crossed.
#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
pub struct Hop {
    /// The device id.
    pub node: String,
    /// Milliseconds after the message's `t`.
    pub at_ms: u64,
}

/// The facts a tap carries about one message. Payloads never ride here.
#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
pub struct TapMessage {
    /// `<device>:<message id>` when admitted, `<device>:x<observation>`
    /// otherwise.
    pub id: String,
    /// The subject (channel name).
    pub subject: String,
    /// Payload size in bytes.
    pub size: u64,
    /// What became of it.
    pub outcome: Outcome,
    /// Wall time it was first seen, in milliseconds since the epoch.
    pub t: u64,
    /// The publishing connection's id, or `liminal.<device>` for the server.
    pub from: String,
    /// The station that published it: the registered worker identity of the
    /// publishing connection. Absent when that connection is not a registered
    /// worker, and for the server's own records.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub from_station: Option<String>,
    /// This device's id.
    pub to: String,
    /// The stations that received it: each registered worker whose
    /// subscription's inbox on this device queued the message, once, in
    /// delivery order. A subscriber whose inbox refused it is not here; nor is
    /// one that is no registered worker (it is counted in the outcome, never
    /// named), nor a receiver on another device. `[]` when the fan-out was
    /// recorded and reached no registered worker here; absent when it was not
    /// recorded (a tap went live while the publish ran, or the tap could not
    /// lend it a buffer), so a reader shows its receivers as not known rather
    /// than as none.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub to_stations: Option<Vec<String>>,
    /// Where it went, in order.
    pub hops: Vec<Hop>,
    /// The id of the message it answers (its causal parent), when it has one.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub reply_to: Option<String>,
    /// Held by a rule. Rules do not exist yet, so always false.
    pub held: bool,
    /// Where a rule diverted it. Rules do not exist yet, so always absent.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub diverted: Option<String>,
}

/// One `console.tap.<target>` record: one batch, every message in it admitted
/// at the one divisor `sampled`.
#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
pub struct TapBatch {
    v: u64,
    #[serde(rename = "type")]
    record_type: &'static str,
    /// The node or stream the tap is on.
    pub target: String,
    /// One more than this tap's previous batch, from 1.
    pub seq: u64,
    /// N in "sampled 1 in N" for every message in this batch.
    pub sampled: NonZeroU32,
    /// Messages evicted from the tap's ring since its previous batch.
    pub dropped: u64,
    /// The messages, oldest first.
    pub messages: Vec<TapMessage>,
}

impl TapBatch {
    /// A batch for `target`.
    #[must_use]
    pub const fn new(
        target: String,
        seq: u64,
        sampled: NonZeroU32,
        dropped: u64,
        messages: Vec<TapMessage>,
    ) -> Self {
        Self {
            v: RECORD_VERSION,
            record_type: "tap.batch",
            target,
            seq,
            sampled,
            dropped,
            messages,
        }
    }
}

/// 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,
        })
}

/// The `console.tap.<target>` record for one tap batch on `device`.
///
/// Every message is named, timed and attributed in the console's terms. A
/// payload, which only a test grant can carry today, does not ride the
/// record: the console's reader names no field for one.
///
/// # Errors
/// Returns [`ConsoleError::Clock`] when a message's time passes u64
/// milliseconds.
pub fn tap_batch_record(
    device: &str,
    target: &str,
    batch: &tap::TapBatch,
    wall: &WallBase,
) -> Result<TapBatch, ConsoleError> {
    let messages = batch
        .messages
        .iter()
        .map(|message| tap_message(device, message, wall))
        .collect::<Result<Vec<_>, _>>()?;
    Ok(TapBatch::new(
        target.to_owned(),
        batch.seq,
        batch.sampled,
        batch.dropped,
        messages,
    ))
}

fn tap_message(
    device: &str,
    message: &tap::TapMessage,
    wall: &WallBase,
) -> Result<TapMessage, ConsoleError> {
    Ok(TapMessage {
        id: match message.ident {
            TapIdent::Message(id) => format!("{device}:{id}"),
            TapIdent::Observation(observation) => format!("{device}:x{observation}"),
        },
        subject: message.subject.clone(),
        size: message.size,
        outcome: message.outcome.into(),
        t: wall.unix_ms_at(message.seen_at)?,
        from: match message.source {
            TapSource::Connection(connection) => connection.to_string(),
            TapSource::Server => message
                .station
                .as_deref()
                .map_or_else(|| format!("liminal.{device}"), str::to_owned),
        },
        from_station: message.station.as_deref().map(str::to_owned),
        to: device.to_owned(),
        to_stations: message.to_stations.as_deref().map(|stations| {
            stations
                .iter()
                .map(|reached| reached.station.to_string())
                .collect()
        }),
        hops: message
            .hops
            .iter()
            .map(|hop| Hop {
                node: hop.node.to_string(),
                at_ms: hop.at_ms,
            })
            .collect(),
        reply_to: message.reply_to.clone(),
        held: message.held,
        diverted: None,
    })
}