liminal-server 0.28.9

//! Real TCP acceptance. No mock server, timeout, polling or clock wait.

use std::io::{Read, Write};
use std::net::{SocketAddr, TcpStream};
use std::sync::Arc;

use liminal::protocol::{Frame, ProtocolError, ProtocolVersion, decode, encode, encoded_len};
use serde_json::json;

use super::records::{AgentEvent, AgentHarness, AgentSpec};
use super::test_support::{Fixture, FixtureIssuer, TestResult, approved, signed_pass};
use crate::server::connection::{
    ConnectionServices, ConnectionSupervisor, LiminalConnectionServices,
};
use crate::server::listener::ServerListener;

pub(super) struct Exchange {
    listener: Option<ServerListener>,
    supervisor: ConnectionSupervisor,
    pub(super) services: Arc<LiminalConnectionServices>,
    fixture: Fixture,
    keys: tempfile::TempDir,
}

impl Exchange {
    pub(super) fn new() -> Result<Self, Box<dyn std::error::Error>> {
        let fixture = Fixture::new()?;
        let keys = tempfile::tempdir()?;
        let keys_path = keys.path().join("issuer.json");
        let (issuer_keys, pass) = signed_pass("a")?;
        drop(pass);
        std::fs::write(&keys_path, issuer_keys)?;
        let config: crate::config::types::ServerConfig = serde_json::from_value(json!({
            "listen_address":"127.0.0.1:0","health_listen_address":"127.0.0.1:0",
            "channels":[],"routing_rules":[],"drain_timeout_ms":0,
            "limits":{"max_channels":64},
            "auth":{"token":"fixture-bearer","lys":{"issuer":"https://agent.fixture",
                "network_id":"fixture-network","keys_path":keys_path,"agents":{}}}}))?;
        let store: Arc<dyn liminal::durability::DurableStore> = fixture.counts.clone();
        let services = Arc::new(
            LiminalConnectionServices::from_config_with_store(&config, store)?
                .with_agent_fixture("fixture-network", Arc::new(FixtureIssuer::default()))?,
        );
        let agents = services.agents().ok_or("agent runtime missing")?;
        for name in ["a", "b"] {
            agents.execute_approved(&approved(
                &format!("create-{name}"),
                name,
                "fixture-owner",
                AgentEvent::Create(AgentSpec {
                    harness: AgentHarness::Cdx,
                    purpose: "Build one service.".into(),
                    service: None,
                }),
            )?)?;
            services.agent_created(name)?;
        }
        let supervisor = ConnectionSupervisor::builder(services.clone())
            .auth(config.auth.as_ref().ok_or("fixture auth missing")?)?
            .limits(config.limits)
            .build()?;
        let listener = ServerListener::bind(&config, supervisor.clone())?;
        Ok(Self {
            listener: Some(listener),
            supervisor,
            services,
            fixture,
            keys,
        })
    }

    pub(super) fn address(&self) -> Result<SocketAddr, Box<dyn std::error::Error>> {
        Ok(self
            .listener
            .as_ref()
            .ok_or("listener absent")?
            .local_addr())
    }

    /// The websocket door on this exchange's supervisor, on a port of its own.
    pub(super) fn websocket(
        &self,
    ) -> Result<crate::server::connection::websocket::WebSocketListener, Box<dyn std::error::Error>>
    {
        Ok(
            crate::server::connection::websocket::WebSocketListener::bind(
                &crate::config::types::WebSocketConfig {
                    listen_address: "127.0.0.1:0".parse()?,
                    path: "/liminal".to_owned(),
                    allowed_origins: Vec::new(),
                    allowed_origin_ports: Vec::new(),
                    ping_interval_ms: None,
                },
                self.supervisor.clone(),
            )?,
        )
    }

    pub(super) fn finish(mut self) -> TestResult {
        self.listener.take().ok_or("listener absent")?.shutdown()?;
        self.supervisor.shutdown();
        self.services.release_store_holders()?;
        drop(self.services);
        drop(self.fixture);
        drop(self.keys);
        Ok(())
    }
}

pub(super) struct Client {
    stream: TcpStream,
    buffer: Vec<u8>,
    next: u32,
}

impl Client {
    pub(super) fn connect(
        address: SocketAddr,
        holder: &str,
    ) -> Result<Self, Box<dyn std::error::Error>> {
        let mut client = Self {
            stream: TcpStream::connect(address)?,
            buffer: Vec::new(),
            next: 1,
        };
        let (keys, pass) = signed_pass(holder)?;
        drop(keys);
        client.write(&Frame::Connect {
            flags: 0,
            min_version: ProtocolVersion::new(1, 0),
            max_version: ProtocolVersion::new(1, 0),
            auth_token: liminal_protocol::credential::WireCredential::lys_record(pass.as_bytes()),
        })?;
        assert!(matches!(client.read()?, Frame::ConnectAck { .. }));
        Ok(client)
    }
    pub(super) fn write(&mut self, frame: &Frame) -> TestResult {
        let mut bytes = vec![0; encoded_len(frame)?];
        let n = encode(frame, &mut bytes)?;
        self.stream
            .write_all(bytes.get(..n).ok_or("encoded frame overrun")?)?;
        Ok(())
    }
    pub(super) fn read(&mut self) -> Result<Frame, Box<dyn std::error::Error>> {
        loop {
            match decode(&self.buffer) {
                Ok((frame, n)) => {
                    self.buffer.drain(..n);
                    return Ok(frame);
                }
                Err(
                    ProtocolError::IncompleteHeader { .. } | ProtocolError::TruncatedPayload { .. },
                ) => {
                    let mut byte = [0; 4096];
                    let n = self.stream.read(&mut byte)?;
                    if n == 0 {
                        return Err("server closed before the expected signal".into());
                    }
                    self.buffer
                        .extend_from_slice(byte.get(..n).ok_or("read buffer overrun")?);
                    if self.buffer.len()
                        > crate::config::types::LimitsConfig::DEFAULT_MAX_CONNECTION_INBOX_BYTES
                    {
                        return Err("fixture frame exceeded the configured inbox bound".into());
                    }
                }
                Err(error) => return Err(error.into()),
            }
        }
    }
    pub(super) fn subscribe(&mut self, subject: &str) -> Result<Frame, Box<dyn std::error::Error>> {
        let stream_id = self.next;
        self.next += 1;
        self.write(&Frame::Subscribe {
            flags: 0,
            stream_id,
            channel: subject.into(),
            accepted_schemas: Vec::new(),
            max_in_flight: 8,
        })?;
        self.read()
    }
    pub(super) fn value(&mut self) -> Result<serde_json::Value, Box<dyn std::error::Error>> {
        let Frame::Deliver { envelope, .. } = self.read()? else {
            return Err("expected metadata Deliver".into());
        };
        Ok(serde_json::from_slice(&envelope.payload)?)
    }
}

#[test]
fn agt002_real_wire_refuses_a_foreign_identity_and_a_duplicate_then_retains_presence_first()
-> TestResult {
    let exchange = Exchange::new()?;
    let address = exchange.address()?;
    let mut a = Client::connect(address, "a")?;
    let Frame::SubscribeError { message, .. } = a.subscribe("agents.b.inbox")? else {
        return Err("foreign identity admitted".into());
    };
    assert_eq!(message.as_deref(), Some("pass holder a is not agent b"));
    assert!(matches!(
        a.subscribe("agents.a.inbox")?,
        Frame::SubscribeAck { .. }
    ));
    let mut watch = Client::connect(address, "b")?;
    assert!(
        matches!(
            watch.subscribe("agents.a.presence")?,
            Frame::SubscribeAck { .. }
        ),
        "ack precedes current value"
    );
    let current = watch.value()?;
    assert_eq!(current["present"], true);
    let connection = current["connection"].as_u64().ok_or("connection missing")?;
    let since_ms = current["at_ms"].as_u64().ok_or("at_ms missing")?;
    let message = liminal::protocol::MessageEnvelope::new(
        liminal::protocol::SchemaId::new([0; liminal::protocol::SchemaId::WIRE_LEN]),
        liminal::protocol::CausalContext::independent(),
        br#"{"text":"synthetic message"}"#.to_vec(),
    );
    watch.write(&Frame::new_publish(9, "agents.a.inbox", message)?)?;
    assert!(matches!(watch.read()?, Frame::PublishAck { .. }));
    let Frame::Deliver { envelope, .. } = a.read()? else {
        return Err("granted message did not arrive".into());
    };
    let body: serde_json::Value = serde_json::from_slice(&envelope.payload)?;
    let matched = body["text"].as_str() == Some("synthetic message");
    assert!(matched, "inbox body differed");
    let metadata = liminal::protocol::MessageEnvelope::new(
        liminal::protocol::SchemaId::new([0; liminal::protocol::SchemaId::WIRE_LEN]),
        liminal::protocol::CausalContext::independent(),
        b"{}".to_vec(),
    );
    a.write(&Frame::new_publish(10, "agents.a.state", metadata)?)?;
    let Frame::PublishError { message, .. } = a.read()? else {
        return Err("server-owned state was writable".into());
    };
    assert_eq!(
        message.as_deref(),
        Some("agent_subject_server_owned: agents.a.state")
    );
    let mut second = Client::connect(address, "a")?;
    let Frame::SubscribeError { message, .. } = second.subscribe("agents.a.inbox")? else {
        return Err("duplicate seat admitted".into());
    };
    assert_eq!(
        message,
        Some(format!(
            "agent a is already present on connection {connection}, held since unix ms \
             {since_ms}; the seat frees when that connection closes or its peer stops \
             answering keepalive (80 s), so give this client a seat of its own"
        ))
    );
    a.write(&Frame::Disconnect { flags: 0 })?;
    let gone = watch.value()?;
    assert_eq!(gone["present"], false);
    assert_eq!(gone["reason"], "client_close");
    assert!(matches!(
        second.subscribe("agents.a.inbox")?,
        Frame::SubscribeAck { .. }
    ));
    let returned = watch.value()?;
    assert_eq!(returned["present"], true);
    second.write(&Frame::Disconnect { flags: 0 })?;
    assert_eq!(watch.value()?["present"], false);
    drop(a);
    drop(second);
    drop(watch);
    exchange.finish()
}

/// A holder whose process ends frees the seat at once: its socket closes with
/// no Disconnect frame, the transport's close releases the seat, and the next
/// connection takes it. No keepalive and no clock is involved.
#[test]
fn a_holder_that_exits_without_a_disconnect_frees_its_seat_at_once() -> TestResult {
    let exchange = Exchange::new()?;
    let address = exchange.address()?;
    let mut a = Client::connect(address, "a")?;
    assert!(matches!(
        a.subscribe("agents.a.inbox")?,
        Frame::SubscribeAck { .. }
    ));
    let mut watch = Client::connect(address, "b")?;
    assert!(matches!(
        watch.subscribe("agents.a.presence")?,
        Frame::SubscribeAck { .. }
    ));
    assert_eq!(watch.value()?["present"], true);
    // The process exits: its socket closes, and no Disconnect frame is sent.
    drop(a);
    let gone = watch.value()?;
    assert_eq!(gone["present"], false, "{gone}");
    let mut second = Client::connect(address, "a")?;
    assert!(
        matches!(
            second.subscribe("agents.a.inbox")?,
            Frame::SubscribeAck { .. }
        ),
        "the seat of a holder that exited was not freed"
    );
    assert_eq!(watch.value()?["present"], true);
    second.write(&Frame::Disconnect { flags: 0 })?;
    drop(second);
    drop(watch);
    exchange.finish()
}

/// A client of the websocket door: one liminal frame per binary message out,
/// every binary message's bytes read as the same frame stream in.
struct WsClient {
    socket: tungstenite::WebSocket<TcpStream>,
    buffer: Vec<u8>,
    next: u32,
}

impl WsClient {
    fn connect(address: SocketAddr, holder: &str) -> Result<Self, Box<dyn std::error::Error>> {
        let stream = TcpStream::connect(address)?;
        let (socket, _response) = tungstenite::client(format!("ws://{address}/liminal"), stream)
            .map_err(|error| format!("websocket handshake: {error}"))?;
        let mut client = Self {
            socket,
            buffer: Vec::new(),
            next: 1,
        };
        let (keys, pass) = signed_pass(holder)?;
        drop(keys);
        client.write(&Frame::Connect {
            flags: 0,
            min_version: ProtocolVersion::new(1, 0),
            max_version: ProtocolVersion::new(1, 0),
            auth_token: liminal_protocol::credential::WireCredential::lys_record(pass.as_bytes()),
        })?;
        assert!(matches!(client.read()?, Frame::ConnectAck { .. }));
        Ok(client)
    }
    fn write(&mut self, frame: &Frame) -> TestResult {
        let mut bytes = vec![0; encoded_len(frame)?];
        let n = encode(frame, &mut bytes)?;
        bytes.truncate(n);
        self.socket
            .send(tungstenite::Message::Binary(bytes.into()))?;
        Ok(())
    }
    fn read(&mut self) -> Result<Frame, Box<dyn std::error::Error>> {
        loop {
            match decode(&self.buffer) {
                Ok((frame, n)) => {
                    self.buffer.drain(..n);
                    return Ok(frame);
                }
                Err(
                    ProtocolError::IncompleteHeader { .. } | ProtocolError::TruncatedPayload { .. },
                ) => match self.socket.read()? {
                    tungstenite::Message::Binary(bytes) => self.buffer.extend_from_slice(&bytes),
                    tungstenite::Message::Close(_) => {
                        return Err("server closed before the expected signal".into());
                    }
                    _ => {}
                },
                Err(error) => return Err(error.into()),
            }
        }
    }
    fn subscribe(&mut self, subject: &str) -> Result<Frame, Box<dyn std::error::Error>> {
        let stream_id = self.next;
        self.next += 1;
        self.write(&Frame::Subscribe {
            flags: 0,
            stream_id,
            channel: subject.into(),
            accepted_schemas: Vec::new(),
            max_in_flight: 8,
        })?;
        self.read()
    }
    /// The browser's way to leave: a close frame, then reading to the
    /// server's answering close.
    fn close(mut self) -> TestResult {
        self.socket.close(None)?;
        loop {
            match self.socket.read() {
                Ok(_) => {}
                Err(tungstenite::Error::ConnectionClosed) => return Ok(()),
                Err(error) => return Err(error.into()),
            }
        }
    }
}

/// Waffles, 26 October: a console page reloaded through the websocket door
/// found its own seat still held minutes later, "already present on
/// connection 9". The websocket process's close paths never released the
/// agent seat the TCP door releases on every exit. A holder over the
/// websocket door that leaves, by a close frame or by dropping the socket,
/// frees its seat at once, and the next connection takes it.
#[test]
fn a_websocket_holder_that_leaves_frees_its_seat_at_once() -> TestResult {
    let exchange = Exchange::new()?;
    let door = exchange.websocket()?;
    let ws = door.local_addr();
    let mut watch = Client::connect(exchange.address()?, "b")?;
    assert!(matches!(
        watch.subscribe("agents.a.presence")?,
        Frame::SubscribeAck { .. }
    ));
    assert_eq!(watch.value()?["present"], false);
    let mut holder = WsClient::connect(ws, "a")?;
    assert!(matches!(
        holder.subscribe("agents.a.inbox")?,
        Frame::SubscribeAck { .. }
    ));
    assert_eq!(watch.value()?["present"], true);
    for leave in ["close frame", "dropped socket"] {
        if leave == "close frame" {
            holder.close()?;
        } else {
            drop(holder);
        }
        // the next holder asks first: a held seat answers at once, never a hang
        let mut next = WsClient::connect(ws, "a")?;
        let taken = next.subscribe("agents.a.inbox")?;
        assert!(
            matches!(taken, Frame::SubscribeAck { .. }),
            "after a {leave}, the seat was still held: {taken:?}"
        );
        assert_eq!(watch.value()?["present"], false, "{leave}");
        assert_eq!(watch.value()?["present"], true, "{leave}");
        holder = next;
    }
    holder.close()?;
    drop(watch);
    door.shutdown()?;
    exchange.finish()
}