liminal-server 0.20.0

//! The console topology and the tap's own exclusions over the wire (SRV-012
//! R4, R6, R7, R8; Ringo's d2c41c0 `tests/topology.rs`, ported in-crate).
//!
//! Every test starts a server the way the runtime does (services from config
//! with a participant, then the connection supervisor, then the self-stream
//! publisher that also runs the console pump, then the listener), speaks the
//! ordinary wire as a client, and waits on a frame: a `SubscribeAck`, a
//! `Deliver`, a `PublishAck` or an error frame. No test sleeps, polls or reads
//! a clock. A count is read from the services' [`StoreCountsHandle`] or from
//! [`SignalCounts`], and each pin is a ceiling that may only go down.
//!
//! A topology record is canonical JSON with `v` first, in the field names of
//! the console's `src/live/types.ts`: `rev`, `devices`, `stations`, `lines`,
//! `links` and `connections`. A connection is not a station, and no
//! `console.*` channel is ever a line.
//!
//! Ported in-crate because the console streams ride R3's publisher owner,
//! which only `crate::server` can start. Tests that count delivered work keep
//! an ordinary subscriber (`taker`) on the tapped stream.

use std::collections::{BTreeMap, BTreeSet};
use std::error::Error;
use std::io::{Read, Write};
use std::net::{SocketAddr, TcpStream};
use std::path::Path;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Condvar, Mutex, MutexGuard};
use std::time::Duration;

use liminal::Envelope;
use liminal::channel::ClusterObserver;
use liminal::conversation::EchoBehaviour;
use liminal::durability::{StoreCounts, StoreCountsHandle};
use liminal::protocol::{
    CausalContext, Frame, MessageEnvelope, ProtocolError, ProtocolVersion, SchemaId,
    WorkerRegisterOutcome, WorkerRegistration, decode, encode, encoded_len,
};
use liminal_protocol::reason_code::{
    CONSOLE_NAMESPACE_REFUSED_CODE, CONSOLE_TAP_TARGET_REFUSED_CODE,
};
use serde_json::Value;
use tempfile::TempDir;

use crate::config::types::{ChannelDef, LimitsConfig, ServerConfig, ServicesConfig};
use crate::server::connection::{ChannelRegistration, LiminalConnectionServices};
use crate::server::self_streams::events::PublishOrigin;
use crate::server::self_streams::owner::PublisherRuntime;
use crate::server::self_streams::sink::SelfStreams;
use crate::server::shutdown::{ShutdownHandle, run_shutdown_sequence};
use crate::server::signals::SignalCounts;
use crate::server::{ConnectionSupervisor, ServerListener};

type TestResult<T = ()> = Result<T, Box<dyn Error>>;

const TOPOLOGY: &str = "console.topology";
const TAP_PREFIX: &str = "console.tap.";

/// The ordinary stream every test publishes to: ephemeral, so any store write a
/// test sees is not the ordinary stream's.
const ORDINARY: &str = "orders";

/// A guard on a blocked socket read, never a deadline: the frame is the signal,
/// and a minute only stops a broken server from hanging the battery.
const READ_GUARD: Duration = Duration::from_secs(60);

/// The ordinary stream, `extra_channels` fixture channels and `named`, all
/// ephemeral, with a participant.
fn config_with(
    persistence_path: &Path,
    extra_channels: usize,
    named: &[&str],
) -> TestResult<ServerConfig> {
    let mut channels: Vec<ChannelDef> = std::iter::once(&ORDINARY)
        .chain(named)
        .map(|name| ChannelDef {
            name: (*name).to_owned(),
            schema_ref: None,
            durable: false,
            loaded_schema: None,
        })
        .collect();
    channels.extend((0..extra_channels).map(|index| ChannelDef {
        name: format!("fixture.{index:02}"),
        schema_ref: None,
        durable: false,
        loaded_schema: None,
    }));
    Ok(ServerConfig {
        listen_address: "127.0.0.1:0".parse()?,
        health_listen_address: "127.0.0.1:0".parse()?,
        channels,
        routing_rules: Vec::new(),
        persistence_path: Some(persistence_path.to_path_buf()),
        cluster: None,
        auth: None,
        drain_timeout_ms: 30_000,
        services: ServicesConfig::default(),
        limits: LimitsConfig {
            max_channels: Some(64),
            ..LimitsConfig::default()
        },
        // The console streams ride the self-stream publisher, which needs a
        // participant (R4: without one no console subject is registered).
        participant: Some(crate::server::participant::test_participant_config()),
        websocket: None,
    })
}

/// A server started as the runtime starts one, over an on-disk store.
struct Server {
    services: Arc<LiminalConnectionServices>,
    listener: ServerListener,
    publisher: PublisherRuntime,
    counts: StoreCountsHandle,
    store: TempDir,
}

impl Server {
    fn start(extra_channels: usize) -> TestResult<Self> {
        Self::start_with(extra_channels, &[])
    }

    /// [`Self::start`] with `named` configured as ephemeral channels too.
    fn start_with(extra_channels: usize, named: &[&str]) -> TestResult<Self> {
        Self::start_verifying(extra_channels, named, None)
    }

    /// [`Self::start_with`], admitting connections by `pass` when given.
    fn start_verifying(
        extra_channels: usize,
        named: &[&str],
        pass: Option<crate::auth_pass::PassVerifier>,
    ) -> TestResult<Self> {
        let store = TempDir::new()?;
        let config = config_with(store.path(), extra_channels, named)?;
        let services = Arc::new(LiminalConnectionServices::from_config(&config)?);
        // The server's cost rows come on its owner's 10 s schedule, and no
        // console pin measures them: they are held out of every window.
        services.signals().wake().hold();
        let counts = services.store_counts();
        let shutdown = ShutdownHandle::new();
        let supervisor = ConnectionSupervisor::with_fatal_shutdown(
            services.clone(),
            None,
            pass,
            config.limits,
            shutdown.clone(),
        )?;
        let publisher = crate::server::runtime::start_self_stream_publisher(
            &services,
            &supervisor,
            &config,
            &shutdown,
        )?
        .ok_or("no self-stream publisher: the console streams need one")?;
        let listener = ServerListener::bind(&config, supervisor)?;
        Ok(Self {
            services,
            listener,
            publisher,
            counts,
            store,
        })
    }

    fn signals(&self) -> SignalCounts {
        self.services.signals().counts()
    }

    /// Waits until every connection opened or closed so far has had its
    /// `liminal.events` row written, and every row that writing queued in
    /// turn (the row of the store flush its write made), so a window opened
    /// next holds none of those rows.
    ///
    /// The connections' rows are counted when queued; the owner, released
    /// if a test held it, then drains the queue until nothing more is
    /// queued.
    fn settled(&self) -> TestResult {
        let signals = self.signals();
        let events = signals.connections_opened + signals.connections_closed;
        if self
            .services
            .signals()
            .await_events(events, Duration::from_secs(30))
        {
            let streams = self.streams()?;
            streams.release_drainer()?;
            Ok(streams.drain_all()?)
        } else {
            Err(format!("{events} connection events not queued in 30 s").into())
        }
    }

    fn connect(&self) -> TestResult<Client> {
        Client::connect(self.listener.local_addr())
    }

    /// The self-stream publisher's streams, whose drainer a test holds.
    fn streams(&self) -> TestResult<Arc<SelfStreams>> {
        Ok(Arc::clone(
            self.services
                .signals()
                .streams()
                .ok_or("no self-stream publisher is installed")?,
        ))
    }

    /// Stops the server the way the runtime does, then removes its store.
    ///
    /// The shutdown sequence runs first, then the self-stream publisher; a
    /// failed removal fails the test instead of leaving residue.
    fn stop(self) -> TestResult {
        let Self {
            services,
            mut listener,
            mut publisher,
            counts,
            store,
        } = self;
        let supervisor = listener.supervisor();
        run_shutdown_sequence(&mut listener, None, &supervisor, Duration::ZERO)?;
        publisher.finish()?;
        // Everything holding the store goes before its directory does, so no
        // late write lands in a directory being removed.
        drop((supervisor, listener, publisher, counts, services));
        store.close()?;
        Ok(())
    }
}

/// A raw wire client past `Connect` -> `ConnectAck`.
struct Client {
    stream: TcpStream,
    buffer: Vec<u8>,
    next_stream: u32,
}

impl Client {
    fn connect(address: SocketAddr) -> TestResult<Self> {
        Self::connect_presenting(address, Vec::new())
    }

    /// [`Self::connect`] presenting `auth_token` (a pass, for one).
    fn connect_presenting(address: SocketAddr, auth_token: Vec<u8>) -> TestResult<Self> {
        let stream = TcpStream::connect(address)?;
        stream.set_nodelay(true)?;
        stream.set_read_timeout(Some(READ_GUARD))?;
        let mut client = Self {
            stream,
            buffer: Vec::new(),
            next_stream: 1,
        };
        client.write(&Frame::Connect {
            flags: 0,
            min_version: ProtocolVersion::new(1, 0),
            max_version: ProtocolVersion::new(1, 0),
            auth_token,
        })?;
        match client.read()? {
            Frame::ConnectAck { .. } => Ok(client),
            other => Err(format!("expected ConnectAck, got {:?}", other.frame_type()).into()),
        }
    }

    fn write(&mut self, frame: &Frame) -> TestResult {
        let mut bytes = vec![0_u8; encoded_len(frame)?];
        let written = encode(frame, &mut bytes)?;
        self.stream.write_all(
            bytes
                .get(..written)
                .ok_or("encoded length overran the buffer")?,
        )?;
        self.stream.flush()?;
        Ok(())
    }

    fn read(&mut self) -> TestResult<Frame> {
        loop {
            match decode(&self.buffer) {
                Ok((frame, consumed)) => {
                    self.buffer.drain(..consumed);
                    return Ok(frame);
                }
                Err(
                    ProtocolError::IncompleteHeader { .. } | ProtocolError::TruncatedPayload { .. },
                ) => {
                    let mut chunk = [0_u8; 8192];
                    let read = self.stream.read(&mut chunk)?;
                    if read == 0 {
                        return Err("the server closed the connection mid-frame".into());
                    }
                    self.buffer
                        .extend_from_slice(chunk.get(..read).ok_or("read past the chunk")?);
                }
                Err(error) => return Err(error.into()),
            }
        }
    }

    const fn take_stream(&mut self) -> u32 {
        let stream_id = self.next_stream;
        self.next_stream += 1;
        stream_id
    }

    /// Subscribes and answers the subscription id, or the refusal's reason code.
    fn subscribe(&mut self, channel: &str) -> TestResult<Result<u64, u16>> {
        self.subscribe_accepting(channel, Vec::new())
    }

    /// [`Self::subscribe`] accepting only `accepted_schemas`.
    fn subscribe_accepting(
        &mut self,
        channel: &str,
        accepted_schemas: Vec<SchemaId>,
    ) -> TestResult<Result<u64, u16>> {
        let stream_id = self.take_stream();
        self.write(&Frame::Subscribe {
            flags: 0,
            stream_id,
            channel: channel.to_owned(),
            accepted_schemas,
            max_in_flight: 1_024,
        })?;
        match self.read()? {
            Frame::SubscribeAck {
                subscription_id, ..
            } => Ok(Ok(subscription_id)),
            Frame::SubscribeError { reason_code, .. } => Ok(Err(reason_code)),
            other => {
                Err(format!("expected a subscribe answer, got {:?}", other.frame_type()).into())
            }
        }
    }

    fn unsubscribe(&mut self, subscription_id: u64) -> TestResult {
        let stream_id = self.take_stream();
        self.write(&Frame::Unsubscribe {
            flags: 0,
            stream_id,
            subscription_id,
        })
    }

    /// Publishes one JSON row and answers `Ok` on the ack, or the reason code.
    fn publish(&mut self, channel: &str, row: usize) -> TestResult<Result<(), u16>> {
        let stream_id = self.take_stream();
        let envelope = MessageEnvelope::new(
            SchemaId::new([0_u8; SchemaId::WIRE_LEN]),
            CausalContext::independent(),
            format!("{{\"row\":{row}}}").into_bytes(),
        );
        self.write(&Frame::new_publish(stream_id, channel, envelope)?)?;
        match self.read()? {
            Frame::PublishAck { .. } => Ok(Ok(())),
            Frame::PublishError { reason_code, .. } => Ok(Err(reason_code)),
            other => Err(format!("expected a publish answer, got {:?}", other.frame_type()).into()),
        }
    }

    /// Registers this connection as the worker `identity`, failing unless
    /// the server accepts it.
    fn register_worker(&mut self, identity: &str) -> TestResult {
        self.write(&Frame::WorkerRegister {
            flags: 0,
            registration: WorkerRegistration {
                namespaces: vec!["default".to_owned()],
                task_queue: "orders".to_owned(),
                node: None,
                activity_types: Vec::new(),
                identity: identity.to_owned(),
                activities: Vec::new(),
            },
        })?;
        match self.read()? {
            Frame::WorkerRegisterAck {
                outcome: WorkerRegisterOutcome::Accepted,
                ..
            } => Ok(()),
            other => Err(format!("expected an accepted registration, got {other:?}").into()),
        }
    }

    /// The next delivered record's JSON body.
    fn next_record(&mut self) -> TestResult<Value> {
        Ok(serde_json::from_slice(&self.next_payload()?)?)
    }

    /// The first delivered record whose `list` holds `len` entries. A
    /// connection's own open reaches the topology after its subscribe can,
    /// so the document a subscriber first sees may trail it by a change.
    fn record_with(&mut self, list: &str, len: usize) -> TestResult<Value> {
        loop {
            let record = self.next_record()?;
            if count(&record, list)? == len {
                return Ok(record);
            }
        }
    }

    /// The next delivered record's bytes, as the wire carried them.
    fn next_payload(&mut self) -> TestResult<Vec<u8>> {
        match self.read()? {
            Frame::Deliver { envelope, .. } => Ok(envelope.payload),
            other => Err(format!("expected Deliver, got {:?}", other.frame_type()).into()),
        }
    }
}

/// Subscribes `client` to `channel`, failing the test on a refusal.
fn subscribed(client: &mut Client, channel: &str) -> TestResult<u64> {
    client
        .subscribe(channel)?
        .map_err(|code| format!("{channel} refused: {code}").into())
}

/// One of the server's own rows on the ordinary stream, as its publisher
/// owner writes them on its own schedule: the interleaving a count of every
/// observation, or of every append, loses under load.
fn server_row(server: &Server) -> TestResult {
    server.services.publish_with_origin(
        ORDINARY,
        &MessageEnvelope::new(
            SchemaId::new([0_u8; SchemaId::WIRE_LEN]),
            CausalContext::independent(),
            b"{\"row\":0}".to_vec(),
        ),
        None,
        PublishOrigin::Server { device: "local" },
    )?;
    Ok(())
}

/// The tap observations of the clients' publishes alone: every observation
/// less the server's own rows.
const fn client_observations(signals: &SignalCounts) -> u64 {
    signals.tap_observations - signals.tap_server_observations
}

/// Publishes row `row` to the ordinary stream, failing the test on a refusal.
fn published(client: &mut Client, row: usize) -> TestResult {
    client
        .publish(ORDINARY, row)?
        .map_err(|code| format!("{ORDINARY} refused: {code}").into())
}

fn rev(record: &Value) -> TestResult<u64> {
    record
        .get("rev")
        .and_then(Value::as_u64)
        .ok_or_else(|| format!("topology record without a rev: {record}").into())
}

fn names<'a>(record: &'a Value, list: &str, key: &str) -> TestResult<Vec<&'a str>> {
    record
        .get(list)
        .and_then(Value::as_array)
        .ok_or_else(|| format!("topology record without {list}: {record}"))?
        .iter()
        .map(|item| {
            item.get(key)
                .and_then(Value::as_str)
                .ok_or_else(|| format!("{list} entry without {key}: {item}").into())
        })
        .collect()
}

fn count(record: &Value, list: &str) -> TestResult<usize> {
    record
        .get(list)
        .and_then(Value::as_array)
        .map(Vec::len)
        .ok_or_else(|| format!("topology record without {list}: {record}").into())
}

fn line_state<'a>(record: &'a Value, line: &str) -> TestResult<&'a str> {
    record
        .get("lines")
        .and_then(Value::as_array)
        .and_then(|lines| {
            lines
                .iter()
                .find(|item| item.get("id").and_then(Value::as_str) == Some(line))
        })
        .and_then(|item| item.get("state"))
        .and_then(Value::as_str)
        .ok_or_else(|| format!("no line {line} in {record}").into())
}

const fn store_writes(counts: &StoreCounts) -> u64 {
    counts.append_calls + counts.append_all_calls + counts.cas_calls + counts.flushes
}

const fn store_reads(counts: &StoreCounts) -> u64 {
    counts.read_from_calls
        + counts.read_from_rows
        + counts.read_at_calls
        + counts.read_value_calls
        + counts.scan_page_calls
}

/// The snapshot a subscriber receives is built from held state, so a server
/// with twenty channels and ten connections reads no store row to answer it.
#[test]
fn the_snapshot_on_subscribe_reads_no_store_row() -> TestResult {
    let server = Server::start(19)?;
    let held: Vec<Client> = (0..10)
        .map(|_| server.connect())
        .collect::<TestResult<_>>()?;
    let mut console = server.connect()?;
    let before = server.counts.snapshot();

    subscribed(&mut console, TOPOLOGY)?;
    let record = console.record_with("connections", held.len() + 1)?;

    let delta = server.counts.snapshot().since(&before);
    assert_eq!(
        store_reads(&delta),
        0,
        "store reads for one snapshot: {delta}"
    );
    // Twenty configured channels and the two self streams.
    assert_eq!(names(&record, "lines", "id")?.len(), 22);
    assert_eq!(count(&record, "connections")?, held.len() + 1);
    assert_eq!(held.len(), 10);
    drop(held);
    server.stop()
}

/// One responder registration and one removal are two changes and exactly two
/// records, each the whole latest document.
#[test]
fn each_responder_change_publishes_exactly_one_record() -> TestResult {
    let server = Server::start(0)?;
    let mut console = server.connect()?;
    subscribed(&mut console, TOPOLOGY)?;
    let first = console.next_record()?;
    let records_before = server.signals().console.topology_records_published;

    server
        .services
        .register_responder("billing.quote", Arc::new(EchoBehaviour))?;
    let registered = console.next_record()?;
    server.services.unregister_responder("billing.quote")?;
    let removed = console.next_record()?;

    let records = server.signals().console.topology_records_published - records_before;
    assert_eq!(records, 2);
    assert_eq!(rev(&registered)?, rev(&first)? + 1);
    assert_eq!(rev(&removed)?, rev(&first)? + 2);
    assert!(names(&registered, "stations", "id")?.contains(&"billing.quote"));
    assert!(!names(&removed, "stations", "id")?.contains(&"billing.quote"));
    server.stop()
}

/// A runtime channel and its quiesce are one record each; the line goes from
/// `live` to `quiesced`.
#[test]
fn a_channel_registered_then_quiesced_is_two_records() -> TestResult {
    let server = Server::start(0)?;
    let mut console = server.connect()?;
    subscribed(&mut console, TOPOLOGY)?;
    console.next_record()?;
    let records_before = server.signals().console.topology_records_published;

    server.services.register_channel(&ChannelRegistration {
        name: "invoices".to_owned(),
        schema_bytes: None,
        durable: false,
    })?;
    let registered = console.next_record()?;
    server
        .services
        .quiesce_channel("invoices", "retired by test")?;
    let quiesced = console.next_record()?;

    let records = server.signals().console.topology_records_published - records_before;
    assert_eq!(records, 2);
    assert_eq!(line_state(&registered, "invoices")?, "live");
    assert_eq!(line_state(&quiesced, "invoices")?, "quiesced");
    server.stop()
}

/// Participants are the live connections.
///
/// One connection opened and closed is exactly two records, and the same two
/// calls give R3 exactly one open and one close event.
#[test]
fn a_connection_opened_and_closed_is_two_records_and_two_events() -> TestResult {
    let server = Server::start(0)?;
    let mut console = server.connect()?;
    subscribed(&mut console, TOPOLOGY)?;
    let first = console.record_with("connections", 1)?;
    let before = server.signals();

    let visitor = server.connect()?;
    let opened = console.record_with("connections", 2)?;
    drop(visitor);
    let closed = console.record_with("connections", 1)?;

    let after = server.signals();
    assert_eq!(
        count(&opened, "connections")?,
        count(&first, "connections")? + 1
    );
    assert_eq!(
        count(&closed, "connections")?,
        count(&first, "connections")?
    );
    assert_eq!(count(&opened, "stations")?, count(&first, "stations")?);
    assert_eq!(
        after.console.topology_records_published - before.console.topology_records_published,
        2
    );
    assert_eq!(after.connections_opened - before.connections_opened, 1);
    assert_eq!(after.connections_closed - before.connections_closed, 1);
    server.stop()
}

/// Every call the channel library makes on its cluster observer, by channel.
///
/// Each call wakes the ledger, so a test waits on the call itself.
#[derive(Debug, Default)]
struct ObserverLedger {
    calls: Mutex<ObserverCalls>,
    changed: Condvar,
    /// A callback cannot answer an error, so one that met a poisoned lock
    /// says so here and every later question fails by name.
    poisoned: AtomicBool,
}

#[derive(Debug, Default)]
struct ObserverCalls {
    /// Subscriber pids the library says are present, per channel.
    subscribers: BTreeMap<String, BTreeSet<u64>>,
    unsubscribes: u64,
    /// `on_unsubscribe` calls, per channel.
    departures: BTreeMap<String, u64>,
    publishes: BTreeMap<String, u64>,
    published_bytes: BTreeMap<String, usize>,
}

const LEDGER_POISONED: &str = "the observer ledger's lock was poisoned";

impl ObserverLedger {
    fn calls(&self) -> TestResult<MutexGuard<'_, ObserverCalls>> {
        if self.poisoned.load(Ordering::Acquire) {
            return Err(LEDGER_POISONED.into());
        }
        self.calls.lock().map_err(|_| LEDGER_POISONED.into())
    }

    /// Records one callback, or marks the ledger poisoned when it cannot.
    fn record(&self, change: impl FnOnce(&mut ObserverCalls)) {
        match self.calls.lock() {
            Ok(mut calls) => {
                change(&mut calls);
                drop(calls);
            }
            Err(_) => self.poisoned.store(true, Ordering::Release),
        }
        self.changed.notify_all();
    }

    /// Waits until `on_unsubscribe` has been called `target` times and answers
    /// the count; the guard is a minute, never a deadline.
    fn wait_for_unsubscribes(&self, target: u64) -> TestResult<u64> {
        let (calls, guard) = self
            .changed
            .wait_timeout_while(self.calls()?, READ_GUARD, |calls| {
                calls.unsubscribes < target
            })
            .map_err(|_| LEDGER_POISONED)?;
        if guard.timed_out() {
            return Err(format!(
                "on_unsubscribe called {} times, expected {target}",
                calls.unsubscribes
            )
            .into());
        }
        Ok(calls.unsubscribes)
    }

    /// Waits until `on_unsubscribe` has been called `target` times for
    /// `channel`, which the console's detach precedes; the guard is a minute,
    /// never a deadline.
    fn wait_for_departures(&self, channel: &str, target: u64) -> TestResult {
        let (calls, guard) = self
            .changed
            .wait_timeout_while(self.calls()?, READ_GUARD, |calls| {
                calls.departures.get(channel).copied().unwrap_or_default() < target
            })
            .map_err(|_| LEDGER_POISONED)?;
        if guard.timed_out() {
            return Err(format!(
                "on_unsubscribe called {} times for {channel}, expected {target}",
                calls.departures.get(channel).copied().unwrap_or_default()
            )
            .into());
        }
        Ok(())
    }

    fn console_publishes(&self) -> TestResult<u64> {
        Ok(self
            .calls()?
            .publishes
            .iter()
            .filter(|(channel, _)| channel.starts_with("console."))
            .map(|(_, count)| count)
            .sum())
    }

    fn subscribers_of(&self, channel: &str) -> TestResult<usize> {
        Ok(self
            .calls()?
            .subscribers
            .get(channel)
            .map_or(0, BTreeSet::len))
    }

    fn bytes_to(&self, channel: &str) -> TestResult<usize> {
        Ok(self
            .calls()?
            .published_bytes
            .get(channel)
            .copied()
            .unwrap_or_default())
    }

    fn publishes_to(&self, channel: &str) -> TestResult<u64> {
        Ok(self
            .calls()?
            .publishes
            .get(channel)
            .copied()
            .unwrap_or_default())
    }
}

impl ClusterObserver for ObserverLedger {
    fn on_subscribe(&self, channel: &str, subscriber_pid: u64) {
        self.record(|calls| {
            calls
                .subscribers
                .entry(channel.to_owned())
                .or_default()
                .insert(subscriber_pid);
        });
    }

    fn on_unsubscribe(&self, channel: &str, subscriber_pid: u64) {
        self.record(|calls| {
            calls.unsubscribes += 1;
            *calls.departures.entry(channel.to_owned()).or_default() += 1;
            if let Some(present) = calls.subscribers.get_mut(channel) {
                present.remove(&subscriber_pid);
            }
        });
    }

    fn on_publish(&self, channel: &str, envelope: &Envelope) {
        self.record(|calls| {
            *calls.publishes.entry(channel.to_owned()).or_default() += 1;
            *calls.published_bytes.entry(channel.to_owned()).or_default() += envelope.payload.len();
        });
    }
}

/// Departures from one channel, held before the console hears them until
/// [`Self::release`], or until this is dropped when a failed assertion
/// unwinds first.
struct HeldDepartures<'a> {
    server: &'a Server,
    released: bool,
}

impl<'a> HeldDepartures<'a> {
    fn hold(server: &'a Server, channel: &str) -> TestResult<Self> {
        server
            .services
            .channel_cluster()
            .observer()
            .hold_departures(channel)?;
        Ok(Self {
            server,
            released: false,
        })
    }

    /// Hands the held departures on, failing the test if it cannot.
    fn release(mut self) -> TestResult {
        self.released = true;
        Ok(self.hand_on()?)
    }

    fn hand_on(&self) -> Result<(), crate::ServerError> {
        self.server
            .services
            .channel_cluster()
            .observer()
            .release_departures()
    }
}

impl Drop for HeldDepartures<'_> {
    /// Only an unwinding test reaches here; its own failure is the one
    /// reported, so this one is written beside it.
    fn drop(&mut self) {
        if !self.released
            && let Err(error) = self.hand_on()
        {
            eprintln!("held departures were not released: {error}");
        }
    }
}

/// Installs the ledger as the observer's witness, which hears every
/// subscribe and departure after the console has, console subjects included.
fn install_witness(server: &Server) -> TestResult<Arc<ObserverLedger>> {
    let ledger = Arc::new(ObserverLedger::default());
    server
        .services
        .channel_cluster()
        .observer()
        .set_witness(Arc::clone(&ledger) as Arc<dyn ClusterObserver>)?;
    Ok(ledger)
}

/// Installs the ledger where the cluster fan-out sits: behind the
/// supervisor's one `ServerObserver`, whose cell takes no second observer.
fn install_ledger(server: &Server) -> TestResult<Arc<ObserverLedger>> {
    let ledger = Arc::new(ObserverLedger::default());
    server
        .services
        .channel_cluster()
        .observer()
        .set_cluster(Arc::clone(&ledger) as Arc<dyn ClusterObserver>)?;
    Ok(ledger)
}

/// `on_unsubscribe` fires from the real drop site. N explicit unsubscribes and
/// one connection closing with N subscriptions give exactly N and N calls.
#[test]
fn every_server_side_unsubscribe_reaches_the_observer() -> TestResult {
    const N: u64 = 4;
    let server = Server::start(0)?;
    let ledger = install_ledger(&server)?;

    let mut explicit = server.connect()?;
    for _ in 0..N {
        let id = subscribed(&mut explicit, ORDINARY)?;
        explicit.unsubscribe(id)?;
    }
    assert_eq!(ledger.wait_for_unsubscribes(N)?, N);

    let mut closing = server.connect()?;
    for _ in 0..N {
        subscribed(&mut closing, ORDINARY)?;
    }
    drop(closing);
    assert_eq!(ledger.wait_for_unsubscribes(2 * N)?, 2 * N);
    assert_eq!(ledger.subscribers_of(ORDINARY)?, 0);
    server.stop()
}

/// `console.*` never crosses to a peer.
///
/// The observer's `on_publish` is the cluster fan-out, and it is called for
/// every ordinary publish and for no topology record or tap batch.
#[test]
fn console_subjects_never_reach_the_cluster_fan_out() -> TestResult {
    const N: usize = 10;
    let server = Server::start(0)?;
    let ledger = install_ledger(&server)?;
    let mut publisher = server.connect()?;
    let mut taker = server.connect()?;
    let mut tap = server.connect()?;
    let mut console = server.connect()?;
    subscribed(&mut taker, ORDINARY)?;
    subscribed(&mut console, TOPOLOGY)?;
    console.next_record()?;
    subscribed(&mut tap, &format!("{TAP_PREFIX}{ORDINARY}"))?;

    server
        .services
        .register_responder("billing.quote", Arc::new(EchoBehaviour))?;
    console.next_record()?;
    for row in 0..N {
        published(&mut publisher, row)?;
    }
    tap.next_record()?;

    assert_eq!(ledger.publishes_to(ORDINARY)?, u64::try_from(N)?);
    assert!(
        ledger.bytes_to(ORDINARY)? >= N,
        "each publish carried its row"
    );
    assert_eq!(ledger.console_publishes()?, 0);
    assert_eq!(ledger.subscribers_of(TOPOLOGY)?, 0);
    assert_eq!(
        ledger.subscribers_of(&format!("{TAP_PREFIX}{ORDINARY}"))?,
        0
    );
    assert_eq!(ledger.subscribers_of(ORDINARY)?, 1);
    server.stop()
}

/// Idle is silent: N ordinary publishes change no topology, so they publish
/// zero topology records, while all N were acknowledged (keepalive-honest).
#[test]
fn ordinary_publishes_publish_no_topology_record() -> TestResult {
    const N: usize = 50;
    let server = Server::start(0)?;
    let mut publisher = server.connect()?;
    let before = server.signals().console.topology_records_published;

    let mut acknowledged = 0_usize;
    for row in 0..N {
        published(&mut publisher, row)?;
        acknowledged += 1;
    }

    assert_eq!(acknowledged, N);
    assert_eq!(
        server.signals().console.topology_records_published - before,
        0
    );
    server.stop()
}

/// The tap never observes its own output.
///
/// N publishes to an ordinary stream under an active tap are exactly N
/// observations, the batches carrying them add none, and no batch names a
/// `console.` subject.
#[test]
fn the_tap_does_not_tap_its_own_batches() -> TestResult {
    const N: usize = 40;
    let server = Server::start(0)?;
    let mut taker = server.connect()?;
    subscribed(&mut taker, ORDINARY)?;
    let mut tap = server.connect()?;
    subscribed(&mut tap, &format!("{TAP_PREFIX}{ORDINARY}"))?;
    // Under load the publisher owner falls behind: writing a connection's
    // `liminal.events` row flushes the store, and the flush's own row is
    // queued while that write is in flight, counted by nothing settled()
    // waits on. Forced: the owner writes the last connection's row alone
    // and is held with its flush's row still queued.
    let streams = server.streams()?;
    server.settled()?;
    streams.drain_all()?;
    streams.hold_drainer(0)?;
    let mut publisher = server.connect()?;
    let opened = server.signals().connections_opened;
    if !server
        .services
        .signals()
        .await_events(opened, Duration::from_secs(30))
    {
        return Err(format!("{opened} connection rows not queued in 30 s").into());
    }
    streams.hold_drainer(1)?;
    streams.drain()?;
    // The server's own rows (a connection's `liminal.events` row, a cost row)
    // are tapped too, when its publisher owner writes them; only
    // the clients' publishes are counted here.
    server.settled()?;
    let all_before = server.signals().tap_observations;
    let before = client_observations(&server.signals());

    for row in 0..N {
        published(&mut publisher, row)?;
    }
    // The interleaving the old count lost under load, forced: one of the
    // server's own rows lands inside the window.
    server_row(&server)?;
    // The owner catches up, and has written the held row before it is
    // counted.
    streams.release_drainer()?;
    streams.drain_all()?;
    let mut carried = 0_usize;
    while carried < N {
        let batch = tap.next_record()?;
        let messages = batch
            .get("messages")
            .and_then(Value::as_array)
            .ok_or_else(|| format!("tap batch without messages: {batch}"))?;
        for message in messages {
            let subject = message
                .get("subject")
                .and_then(Value::as_str)
                .ok_or_else(|| format!("tap message without a subject: {message}"))?;
            assert!(!subject.starts_with("console."), "the tap saw {subject}");
        }
        let dropped = batch
            .get("dropped")
            .and_then(Value::as_u64)
            .ok_or_else(|| format!("tap batch without dropped: {batch}"))?;
        carried += messages.len() + usize::try_from(dropped)?;
    }

    assert_eq!(
        server.signals().tap_observations - all_before,
        u64::try_from(N + 1)?,
        "the old count, every observation, reads the server's row as a client's"
    );
    assert_eq!(
        client_observations(&server.signals()) - before,
        u64::try_from(N)?
    );
    server.stop()
}

/// A client may not write into the console namespace: the publish is refused
/// at admission by name and the tap observes nothing.
#[test]
fn a_client_publish_under_console_is_refused_at_admission() -> TestResult {
    let server = Server::start(0)?;
    let mut client = server.connect()?;
    let before = server.signals();

    let answer = client.publish(TOPOLOGY, 0)?;

    assert_eq!(answer, Err(CONSOLE_NAMESPACE_REFUSED_CODE));
    let after = server.signals();
    assert_eq!(after.tap_observations - before.tap_observations, 0);
    assert_eq!(
        after.console.topology_records_published - before.console.topology_records_published,
        0
    );
    server.stop()
}

/// A tap on the console namespace is refused by name, never answered with an
/// empty tap.
#[test]
fn a_tap_target_under_console_is_refused_by_name() -> TestResult {
    let server = Server::start(0)?;
    let mut client = server.connect()?;

    let answer = client.subscribe(&format!("{TAP_PREFIX}{TOPOLOGY}"))?;

    assert_eq!(answer, Err(CONSOLE_TAP_TARGET_REFUSED_CODE));
    server.stop()
}

/// A tap subscribe refused after admission takes no tap slot and leaves no
/// roster entry: a loop of them leaves the roster size as it found it.
#[test]
fn refused_tap_subscribes_leave_the_roster_unchanged() -> TestResult {
    let server = Server::start(0)?;
    let mut client = server.connect()?;
    let tap = format!("{TAP_PREFIX}{ORDINARY}");
    let before = server.services.registered_channels()?.len();

    for _ in 0..(crate::console::MAX_LIVE_TAPS + 2) {
        let answer =
            client.subscribe_accepting(&tap, vec![SchemaId::new([0xAB; SchemaId::WIRE_LEN])])?;
        assert!(answer.is_err(), "an unknown schema is refused: {answer:?}");
    }

    assert_eq!(server.services.registered_channels()?.len(), before);
    assert_eq!(server.services.console().reserved_taps(), Some(0));
    subscribed(&mut client, &tap)?;
    server.stop()
}

/// With no tap and no console subscriber, a connection's repeated publishes
/// reach no console lock: an unregistered publisher emits no flow at all, so N
/// publishes make zero topology emits.
#[test]
fn repeated_publishes_take_no_topology_lock() -> TestResult {
    const N: usize = 100;
    let server = Server::start(0)?;
    let mut taker = server.connect()?;
    subscribed(&mut taker, ORDINARY)?;
    let mut client = server.connect()?;
    published(&mut client, 0)?;
    let console = server.services.console();
    // The two connections' own opens are applied first, by signal.
    assert!(console.wait_for_connections(2), "both opens applied");
    let (emits, locks) = (console.emits(), console.topology_locks());

    for row in 1..=N {
        published(&mut client, row)?;
    }

    assert_eq!(console.emits() - emits, 0, "topology emits");
    assert_eq!(
        console.topology_locks() - locks,
        0,
        "topology lock acquisitions"
    );
    server.stop()
}

/// A publish that reached nobody is tapped as 'nobody', so a stream with
/// publishers and no readers does not look idle on the console.
#[test]
fn a_publish_to_nobody_is_tapped_as_nobody() -> TestResult {
    let server = Server::start(0)?;
    let mut tap = server.connect()?;
    subscribed(&mut tap, &format!("{TAP_PREFIX}{ORDINARY}"))?;
    let mut publisher = server.connect()?;
    let mut taker = server.connect()?;

    published(&mut publisher, 0)?;
    subscribed(&mut taker, ORDINARY)?;
    published(&mut publisher, 1)?;

    // The delivered publish is the signal: the tap keeps order, so once it
    // arrives everything before it has, and a publish to nobody that was not
    // tapped shows as its absence, at once.
    let mut outcomes = Vec::new();
    let mut receivers = Vec::new();
    while !outcomes.iter().any(|outcome| outcome == "delivered") {
        let batch = tap.next_record()?;
        let messages = batch["messages"]
            .as_array()
            .ok_or_else(|| format!("tap batch without messages: {batch}"))?;
        for message in messages {
            let outcome = message["outcome"]
                .as_str()
                .ok_or_else(|| format!("tap message without an outcome: {message}"))?;
            outcomes.push(outcome.to_owned());
            receivers.push(message.get("to_stations").cloned());
        }
    }
    assert_eq!(outcomes, ["nobody", "delivered"]);
    assert_eq!(
        receivers,
        [Some(serde_json::json!([])), Some(serde_json::json!([]))],
        "each fan-out was recorded, and neither reached a registered worker"
    );
    server.stop()
}

/// A tapped publish names its publishing station only when its connection
/// is a registered worker: the worker's identity, from its own
/// registration; an unregistered connection's publish carries none.
#[test]
fn a_registered_workers_publish_names_its_station() -> TestResult {
    let server = Server::start(0)?;
    let mut tap = server.connect()?;
    subscribed(&mut tap, &format!("{TAP_PREFIX}{ORDINARY}"))?;
    let mut worker = server.connect()?;
    worker.register_worker("orders.pricer")?;
    let mut plain = server.connect()?;

    published(&mut worker, 0)?;
    published(&mut plain, 1)?;

    let mut stations = Vec::new();
    while stations.len() < 2 {
        let batch = tap.next_record()?;
        let messages = batch["messages"]
            .as_array()
            .ok_or_else(|| format!("tap batch without messages: {batch}"))?;
        for message in messages {
            stations.push(message.get("from_station").cloned());
        }
    }
    assert_eq!(
        stations,
        [Some(Value::from("orders.pricer")), None],
        "the worker's publish names its station; the plain connection's names none"
    );
    server.stop()
}

/// A delivered publish names the stations that received it: a registered
/// worker whose subscription took it; a plain subscriber that took it too is
/// counted in the outcome and never named.
#[test]
fn a_delivery_names_its_receiving_station() -> TestResult {
    let server = Server::start(0)?;
    let mut tap = server.connect()?;
    subscribed(&mut tap, &format!("{TAP_PREFIX}{ORDINARY}"))?;
    let mut ledger = server.connect()?;
    ledger.register_worker("orders.ledger")?;
    subscribed(&mut ledger, ORDINARY)?;
    let mut plain = server.connect()?;
    subscribed(&mut plain, ORDINARY)?;
    let mut publisher = server.connect()?;

    published(&mut publisher, 0)?;

    let batch = tap.next_record()?;
    let message = &batch["messages"][0];
    assert_eq!(message["outcome"], "delivered");
    assert_eq!(message["to_stations"], serde_json::json!(["orders.ledger"]));
    assert!(message.get("from_station").is_none());
    server.stop()
}

/// The shared pass vector's registry key, and its principal re-signed by the
/// vector's seed over a window around now, with the participant it names.
fn minted_pass() -> TestResult<(String, Vec<u8>, Vec<u8>)> {
    use ed25519_dalek::{Signer, SigningKey};
    let vector: Value = serde_json::from_str(include_str!("../../test-vectors/wire-pass-v1.json"))?;
    let field = |name: &str| -> TestResult<&str> {
        vector[name]
            .as_str()
            .ok_or_else(|| format!("vector field {name} missing").into())
    };
    let seed: [u8; 32] = hex::decode(field("private_key_seed_hex")?)?
        .try_into()
        .map_err(|_| "seed length")?;
    let mut pass = crate::auth_pass::WirePassV1::parse(&hex::decode(field("pass_hex")?)?)
        .map_err(|error| format!("vector pass parse failed: {error:?}"))?;
    let now = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)?
        .as_secs();
    pass.issued_at = now.saturating_sub(60);
    pass.expires_at = now.saturating_add(3_600);
    let unsigned = pass
        .canonical_unsigned_bytes()
        .map_err(|error| format!("canonical encode failed: {error:?}"))?;
    pass.signature = SigningKey::from_bytes(&seed).sign(&unsigned).to_bytes();
    let bytes = pass
        .canonical_bytes()
        .map_err(|error| format!("canonical encode failed: {error:?}"))?;
    Ok((
        field("registry_verifying_key_hex")?.to_owned(),
        bytes,
        pass.participant,
    ))
}

/// A pass's participant is set on its connection at attach (SRV-012 R7).
/// The attach happens before the `ConnectAck`, so on the ack the topology
/// already names it: no record is waited for.
#[test]
fn a_pass_names_its_connections_participant() -> TestResult {
    let (registry_key, pass, participant) = minted_pass()?;
    let verifier =
        crate::auth_pass::PassVerifier::from_config(&crate::config::types::PassConfig {
            registry_verifying_key: registry_key,
            maximum_clock_skew_seconds: 60,
        })?;
    let server = Server::start_verifying(0, &[], Some(verifier))?;

    let _passed = Client::connect_presenting(server.listener.local_addr(), pass)?;

    assert_eq!(
        server.services.console().participants(),
        vec![Some(hex::encode(participant))]
    );
    server.stop()
}

/// Every `console.*` subject is ephemeral: M topology records and the tap
/// batches of N publishes make zero store writes.
#[test]
fn console_subjects_never_write_the_store() -> TestResult {
    const M: usize = 8;
    const N: usize = 20;
    let server = Server::start(0)?;
    // Every client connects before the counts are read: a connection is a
    // topology change and (with R3's self streams) one `liminal.events` row,
    // so the window below must hold neither.
    let mut publisher = server.connect()?;
    let mut taker = server.connect()?;
    let mut tap = server.connect()?;
    let mut console = server.connect()?;
    subscribed(&mut taker, ORDINARY)?;
    subscribed(&mut console, TOPOLOGY)?;
    console.next_record()?;
    // Where the old window started: after the first record.
    let records_at_first = server.signals().console.topology_records_published;
    // The interleaving it lost under load, forced: a connection's open
    // reaching the topology after that first record (a synthetic one, which
    // writes no `liminal.events` row).
    server
        .services
        .signals()
        .topology(crate::console::TopologyChange::ConnectionOpened {
            id: u64::MAX,
            mount: crate::console::records::Mount::Tcp,
            participant: None,
        });
    // The window starts once every open is in: the four clients and that one.
    console.record_with("connections", 5)?;
    subscribed(&mut tap, &format!("{TAP_PREFIX}{ORDINARY}"))?;
    server.settled()?;
    let before = server.counts.snapshot();
    let records_before = server.signals().console.topology_records_published;

    for index in 0..M {
        let subject = format!("svc.{index}");
        server
            .services
            .register_responder(subject.as_str(), Arc::new(EchoBehaviour))?;
        console.next_record()?;
    }
    for row in 0..N {
        published(&mut publisher, row)?;
    }
    tap.next_record()?;

    let delta = server.counts.snapshot().since(&before);
    let records_now = server.signals().console.topology_records_published;
    // The old window counts the forced late open's record, and any real one
    // that landed late too.
    assert!(
        records_now - records_at_first > u64::try_from(M)?,
        "the old window counts no late open: {}",
        records_now - records_at_first
    );
    assert_eq!(records_now - records_before, u64::try_from(M)?);
    assert_eq!(
        store_writes(&delta),
        0,
        "store writes for console subjects: {delta}"
    );
    server.stop()
}

/// A tap belongs to its subscribers.
///
/// N connections each tapping one stream and then closing without an
/// unsubscribe leave zero taps attached, so a closed or shed subscriber never
/// keeps a tap sampling.
#[test]
fn a_tap_whose_subscribers_close_is_detached() -> TestResult {
    const N: usize = 6;
    let server = Server::start(N)?;
    let ledger = install_witness(&server)?;
    let mut console = server.connect()?;
    subscribed(&mut console, TOPOLOGY)?;
    let baseline = count(&console.record_with("connections", 1)?, "connections")?;

    let mut tapping = Vec::with_capacity(N);
    for index in 0..N {
        let mut client = server.connect()?;
        subscribed(&mut client, &format!("{TAP_PREFIX}fixture.{index:02}"))?;
        tapping.push(client);
    }
    assert_eq!(server.signals().console.taps_attached, u64::try_from(N)?);

    drop(tapping);
    let mut connections = usize::MAX;
    while connections != baseline {
        connections = count(&console.next_record()?, "connections")?;
    }
    // The departure reaches the console's detach before the ledger, on the
    // channel actor's thread, so the ledger's count orders the read.
    ledger.wait_for_unsubscribes(u64::try_from(N)?)?;

    assert_eq!(server.signals().console.taps_attached, 0);
    server.stop()
}

/// The console's own bookkeeping never fails on ordinary work.
///
/// Every topology emit, tap observe, attach and detach on a mixed workload
/// succeeds, and each failure counter stays at 0, while the workload's own
/// counts moved.
#[test]
fn console_bookkeeping_counts_no_failure_on_ordinary_work() -> TestResult {
    const N: usize = 20;
    let server = Server::start(0)?;
    let ledger = install_witness(&server)?;
    let mut publisher = server.connect()?;
    let mut taker = server.connect()?;
    let mut tap = server.connect()?;
    let mut console = server.connect()?;
    subscribed(&mut taker, ORDINARY)?;
    subscribed(&mut console, TOPOLOGY)?;
    // A connection's own open can reach the topology after its subscribe:
    // the window starts once all four are in.
    console.record_with("connections", 4)?;
    server.settled()?;
    let before = server.signals();

    let tap_id = subscribed(&mut tap, &format!("{TAP_PREFIX}{ORDINARY}"))?;
    server
        .services
        .register_responder("billing.quote", Arc::new(EchoBehaviour))?;
    console.next_record()?;
    server.services.register_channel(&ChannelRegistration {
        name: "invoices".to_owned(),
        schema_bytes: None,
        durable: false,
    })?;
    console.next_record()?;
    server
        .services
        .quiesce_channel("invoices", "retired by test")?;
    console.next_record()?;
    for row in 0..N {
        published(&mut publisher, row)?;
    }
    // Forced: one of the server's own rows inside the window, which the
    // count of every observation would take for a client's.
    server_row(&server)?;
    tap.next_record()?;
    // The unsubscribe is only written; its departure reaches the console's
    // detach later, on the channel actor's thread. Forced late: held until
    // the counts are read.
    let tapped = format!("{TAP_PREFIX}{ORDINARY}");
    let departures = HeldDepartures::hold(&server, &tapped)?;
    tap.unsubscribe(tap_id)?;
    server.services.unregister_responder("billing.quote")?;
    console.next_record()?;
    // The read waits for the detach: the witness hears the departure after
    // the console has.
    departures.release()?;
    ledger.wait_for_departures(&tapped, 1)?;

    let counts = server.signals();
    // Keepalive-honest half: the workload ran (four changes, N observations).
    assert_eq!(
        counts.console.topology_records_published - before.console.topology_records_published,
        4
    );
    assert_eq!(
        counts.tap_observations - before.tap_observations,
        u64::try_from(N + 1)?,
        "every observation, the server's row included"
    );
    assert_eq!(
        client_observations(&counts) - client_observations(&before),
        u64::try_from(N)?
    );
    assert_eq!(counts.console.topology_emit_failures, 0);
    assert_eq!(counts.console.tap_observe_failures, 0);
    assert_eq!(counts.console.tap_attach_failures, 0);
    assert_eq!(counts.console.tap_detach_failures, 0);
    assert_eq!(counts.console.console_wake_failures, 0);
    assert_eq!(counts.console.tap_unsubscribe_failures, 0);
    assert_eq!(counts.console.console_publish_failures, 0);
    assert_eq!(counts.console.console_records_lost, 0);
    assert_eq!(counts.console.taps_attached, 0);
    server.stop()
}

/// A subscriber that joins after K changes receives exactly one record, the
/// latest held document, and the next change follows it with no record between.
#[test]
fn a_late_subscriber_receives_only_the_latest_document() -> TestResult {
    const K: usize = 6;
    let server = Server::start(0)?;
    for index in 0..K {
        let subject = format!("svc.{index}");
        server
            .services
            .register_responder(subject.as_str(), Arc::new(EchoBehaviour))?;
    }
    // Connected before the counts are read: the connection's `liminal.events`
    // row starts that durable stream (R3), which is not the snapshot's work.
    let mut late = server.connect()?;
    server.settled()?;
    let before = server.counts.snapshot();

    subscribed(&mut late, TOPOLOGY)?;
    let snapshot = late.next_record()?;
    server
        .services
        .register_responder("svc.next", Arc::new(EchoBehaviour))?;
    let next = late.next_record()?;

    let stations = names(&snapshot, "stations", "id")?;
    for index in 0..K {
        let subject = format!("svc.{index}");
        assert!(stations.contains(&subject.as_str()), "{subject} missing");
    }
    assert_eq!(rev(&next)?, rev(&snapshot)? + 1);
    assert_eq!(store_reads(&server.counts.snapshot().since(&before)), 0);
    server.stop()
}

/// Without `[cluster]` the device is `local`, running this crate's version.
///
/// It is published as `liminal.local`, the same rule as R3's
/// `ServerMetadata::from_config`.
#[test]
fn an_unclustered_server_is_the_local_device() -> TestResult {
    let server = Server::start(0)?;
    let mut console = server.connect()?;
    subscribed(&mut console, TOPOLOGY)?;
    let record = console.next_record()?;

    let devices = record
        .get("devices")
        .and_then(Value::as_array)
        .ok_or_else(|| format!("topology record without devices: {record}"))?;
    assert_eq!(devices.len(), 1);
    let device = devices.first().ok_or("no device")?;
    assert_eq!(device.get("name").and_then(Value::as_str), Some("local"));
    assert_eq!(
        device.get("liminal").and_then(Value::as_str),
        Some(env!("CARGO_PKG_VERSION"))
    );
    assert!(
        device.get("started").and_then(Value::as_u64).is_some(),
        "no start on {device}"
    );
    server.stop()
}

/// SRV-012 R4, keepalive-honest: with no console subscriber, ten responder
/// registrations and fifty ordinary publishes land, and the console publishes
/// nothing and is never woken.
#[test]
fn console_streams_are_idle_without_subscribers() -> TestResult {
    let server = Server::start(0)?;
    let mut taker = server.connect()?;
    subscribed(&mut taker, ORDINARY)?;
    let mut client = server.connect()?;
    let stations = server.services.responder_subjects()?.len();

    for index in 0..10 {
        server
            .services
            .register_responder(format!("idle.{index:02}"), Arc::new(EchoBehaviour))?;
    }
    for row in 0..50 {
        published(&mut client, row)?;
    }
    let delivered = (0..50)
        .map(|_| taker.next_record())
        .collect::<TestResult<Vec<_>>>()?;

    assert_eq!(server.services.responder_subjects()?.len() - stations, 10);
    assert_eq!(delivered.len(), 50);
    let console = server.signals().console;
    assert_eq!(console.topology_records_published, 0);
    assert_eq!(console.tap_records_published, 0);
    assert_eq!(console.console_pump_wakes, 0);
    server.stop()
}

/// SRV-012 R10: the bytes a subscriber receives on `console.topology` and on
/// `console.tap.<target>` are the channel's normal form of what the encoder
/// made for the same live values, and carry `v` 1.
#[test]
fn delivered_console_bytes_are_the_normalized_record() -> TestResult {
    let server = Server::start(0)?;
    let tap_subject = format!("{TAP_PREFIX}{ORDINARY}");
    let mut console = server.connect()?;
    subscribed(&mut console, TOPOLOGY)?;
    let topology = console.next_payload()?;
    let mut taker = server.connect()?;
    subscribed(&mut taker, ORDINARY)?;
    let mut tap = server.connect()?;
    subscribed(&mut tap, &tap_subject)?;
    let mut client = server.connect()?;
    published(&mut client, 1)?;
    let batch = tap.next_payload()?;

    let sent = server
        .services
        .console()
        .sent()
        .ok_or("the sent-record capture is poisoned")?;
    let normalizer = liminal::channel::Schema::new(serde_json::json!({}))?;
    for (subject, delivered) in [(TOPOLOGY, &topology), (tap_subject.as_str(), &batch)] {
        let (_, encoded) = sent
            .iter()
            .find(|(sent_to, _)| sent_to == subject)
            .ok_or_else(|| format!("nothing was sent to {subject}"))?;
        assert_eq!(
            &normalizer.validate_and_apply_defaults(encoded)?,
            delivered,
            "{subject}"
        );
        let record: Value = serde_json::from_slice(delivered)?;
        assert_eq!(
            record.get("v").and_then(Value::as_u64),
            Some(1),
            "{subject}"
        );
    }
    server.stop()
}

/// SRV-012 R1: a configured ephemeral `console.x` is still refused to a
/// client publish, and the refusal moves no tap observation, no store append
/// and no cost arm; an ordinary publish on the same connection moves the
/// observations by exactly one.
#[test]
fn console_namespace_refusal_holds_for_a_configured_console_channel() -> TestResult {
    let server = Server::start_with(0, &["console.x"])?;
    let wake = server.services.signals().wake();
    let mut taker = server.connect()?;
    subscribed(&mut taker, ORDINARY)?;
    let mut tap = server.connect()?;
    subscribed(&mut tap, &format!("{TAP_PREFIX}{ORDINARY}"))?;
    let mut client = server.connect()?;
    server.settled()?;
    let before = server.signals();
    let appends = server.counts.snapshot().append_calls;
    let arms = wake.arms();

    let refused = client.publish("console.x", 1)?;
    assert_eq!(wake.arms(), arms, "the refusal arms no cost tick");
    assert_eq!(refused, Err(CONSOLE_NAMESPACE_REFUSED_CODE));
    let after = server.signals();
    assert_eq!(client_observations(&after), client_observations(&before));
    // Every append in the window is one of the server's own durable rows
    // (`liminal.events`, `liminal.cost`), each observed: the refused publish
    // appended nothing.
    assert_eq!(
        server.counts.snapshot().append_calls - appends,
        after.tap_server_observations - before.tap_server_observations
    );
    // Forced: one of the server's own rows inside the window. The old count
    // of every observation reads it as the refused publish's.
    server_row(&server)?;
    let forced = server.signals();
    assert!(
        forced.tap_observations > before.tap_observations,
        "every observation counts the server's row"
    );
    assert_eq!(client_observations(&forced), client_observations(&before));

    published(&mut client, 2)?;
    assert_eq!(
        client_observations(&server.signals()),
        client_observations(&before) + 1
    );
    server.stop()
}