runa/crates/runa-server/src/lib.rs

//! Transport-neutral dispatch from protocol requests to image API operations.

use std::io::{self, BufRead, Read, Write};
use std::net::{SocketAddr, TcpListener, TcpStream};
use std::num::NonZeroUsize;
use std::thread;
use std::time::Duration;

use runa_actor::{ActorRef, Event, Message, Status};
use runa_image_api::ImageApi;
use runa_protocol::{
    ActorSnapshotDto, ActorStatusDto, BehaviorDto, BehaviorVersionDto, Command, ErrorCode,
    EventDto, ExecutionSnapshotDto, MessageDto, PROTOCOL_VERSION, ProtocolError, Reply, Request,
    Response,
};

pub const MAX_REQUEST_LINE_BYTES: usize = 1024 * 1024;
const MAX_NETWORK_CLIENTS: usize = 16;

pub struct Server {
    image: ImageApi,
}

/// Serve newline-delimited protocol requests over an already-connected stream.
///
/// Each valid request line receives one JSON response line. Malformed or
/// oversized input terminates the stream with `InvalidData`.
pub fn serve_json_lines<R: BufRead, W: Write>(
    server: &mut Server,
    reader: R,
    writer: W,
) -> io::Result<()> {
    serve_json_lines_with(reader, writer, |request| Ok(server.dispatch(request)))
}

/// Bind the protocol listener to an IPv4 or IPv6 loopback address only.
pub fn bind_loopback_listener(address: SocketAddr) -> io::Result<TcpListener> {
    if !address.ip().is_loopback() {
        return Err(io::Error::new(
            io::ErrorKind::InvalidInput,
            "network listener address must be a loopback address",
        ));
    }

    TcpListener::bind(address)
}

/// Serve JSON-lines requests from loopback clients using one shared image.
///
/// The nonblocking event loop keeps the non-`Send` Rune runtime on one thread.
/// At most `MAX_NETWORK_CLIENTS` are active.
pub fn serve_tcp(listener: TcpListener, mut server: Server) -> io::Result<()> {
    let address = listener.local_addr()?;
    if !address.ip().is_loopback() {
        return Err(io::Error::new(
            io::ErrorKind::InvalidInput,
            "network listener must be bound to a loopback address",
        ));
    }
    listener.set_nonblocking(true)?;
    eprintln!("runa-server: listening on {address}");
    let mut clients = Vec::new();
    loop {
        loop {
            match listener.accept() {
                Ok((stream, peer_address)) => {
                    if clients.len() >= MAX_NETWORK_CLIENTS {
                        eprintln!("runa-server: rejecting {peer_address}; client limit reached");
                        continue;
                    }
                    stream.set_nonblocking(true)?;
                    clients.push(NetworkClient::new(stream, peer_address));
                }
                Err(error) if error.kind() == io::ErrorKind::WouldBlock => break,
                Err(error) if error.kind() == io::ErrorKind::Interrupted => continue,
                Err(error) => return Err(error),
            }
        }

        let mut index = 0;
        while index < clients.len() {
            match clients[index].poll(&mut server) {
                Ok(true) => index += 1,
                Ok(false) => {
                    clients.swap_remove(index);
                }
                Err(error) => {
                    eprintln!("runa-server: client {}: {error}", clients[index].address);
                    clients.swap_remove(index);
                }
            }
        }

        thread::sleep(Duration::from_millis(5));
    }
}

struct NetworkClient {
    stream: TcpStream,
    address: SocketAddr,
    input: Vec<u8>,
    output: Vec<u8>,
    output_offset: usize,
    read_closed: bool,
}

impl NetworkClient {
    fn new(stream: TcpStream, address: SocketAddr) -> Self {
        Self {
            stream,
            address,
            input: Vec::new(),
            output: Vec::new(),
            output_offset: 0,
            read_closed: false,
        }
    }

    fn poll(&mut self, server: &mut Server) -> io::Result<bool> {
        if self.output_offset == self.output.len() {
            self.output.clear();
            self.output_offset = 0;
            self.process_pending_requests(server, self.read_closed)?;

            if self.output.is_empty() && !self.read_closed {
                let mut buffer = [0; 8192];
                match self.stream.read(&mut buffer) {
                    Ok(0) => {
                        self.read_closed = true;
                        self.process_pending_requests(server, true)?;
                    }
                    Ok(length) => {
                        self.input.extend_from_slice(&buffer[..length]);
                        self.process_pending_requests(server, false)?;
                    }
                    Err(error) if error.kind() == io::ErrorKind::WouldBlock => {}
                    Err(error) if error.kind() == io::ErrorKind::Interrupted => {}
                    Err(error) => return Err(error),
                }
            }
        }

        if self.output_offset < self.output.len() {
            match self.stream.write(&self.output[self.output_offset..]) {
                Ok(0) => {
                    return Err(io::Error::new(
                        io::ErrorKind::WriteZero,
                        "failed to write protocol response",
                    ));
                }
                Ok(length) => {
                    self.output_offset += length;
                    if self.output_offset == self.output.len() {
                        self.output.clear();
                        self.output_offset = 0;
                    }
                }
                Err(error) if error.kind() == io::ErrorKind::WouldBlock => {}
                Err(error) if error.kind() == io::ErrorKind::Interrupted => {}
                Err(error) => return Err(error),
            }
        }

        Ok(!self.read_closed || !self.input.is_empty() || self.output_offset < self.output.len())
    }

    fn process_pending_requests(
        &mut self,
        server: &mut Server,
        final_line: bool,
    ) -> io::Result<()> {
        let Some(line_end) = self.input.iter().position(|byte| *byte == b'\n') else {
            if self.input.len() > MAX_REQUEST_LINE_BYTES {
                return Err(invalid_data(format!(
                    "request line exceeds {MAX_REQUEST_LINE_BYTES} bytes"
                )));
            }
            if !final_line || self.input.is_empty() {
                return Ok(());
            }
            let line = self.input.clone();
            self.input.clear();
            self.dispatch_line(server, &line)?;
            return Ok(());
        };

        if line_end > MAX_REQUEST_LINE_BYTES {
            return Err(invalid_data(format!(
                "request line exceeds {MAX_REQUEST_LINE_BYTES} bytes"
            )));
        }
        let line = self.input[..line_end].to_vec();
        self.input.drain(..=line_end);
        self.dispatch_line(server, &line)
    }

    fn dispatch_line(&mut self, server: &mut Server, line: &[u8]) -> io::Result<()> {
        let request: Request = serde_json::from_slice(line)
            .map_err(|error| invalid_data(format!("invalid protocol request JSON: {error}")))?;
        let response = server.dispatch(request);
        let mut encoded = serde_json::to_vec(&response).map_err(io::Error::other)?;
        if encoded.len() > MAX_REQUEST_LINE_BYTES {
            return Err(invalid_data(format!(
                "response exceeds {MAX_REQUEST_LINE_BYTES} bytes"
            )));
        }
        encoded.push(b'\n');
        self.output.extend_from_slice(&encoded);
        Ok(())
    }
}

fn invalid_data(message: String) -> io::Error {
    io::Error::new(io::ErrorKind::InvalidData, message)
}

fn serve_json_lines_with<R, W, F>(mut reader: R, mut writer: W, mut dispatch: F) -> io::Result<()>
where
    R: BufRead,
    W: Write,
    F: FnMut(Request) -> io::Result<Response>,
{
    let mut line = Vec::new();
    while read_bounded_line(&mut reader, &mut line)? {
        let request: Request = serde_json::from_slice(&line)
            .map_err(|error| invalid_data(format!("invalid protocol request JSON: {error}")))?;
        let response = dispatch(request)?;
        serde_json::to_writer(&mut writer, &response).map_err(io::Error::other)?;
        writer.write_all(b"\n")?;
        writer.flush()?;
    }
    Ok(())
}

fn read_bounded_line<R: BufRead>(reader: &mut R, line: &mut Vec<u8>) -> io::Result<bool> {
    line.clear();
    loop {
        let available = reader.fill_buf()?;
        if available.is_empty() {
            return Ok(!line.is_empty());
        }

        let newline = available.iter().position(|byte| *byte == b'\n');
        let payload_len = newline.unwrap_or(available.len());
        if line.len().saturating_add(payload_len) > MAX_REQUEST_LINE_BYTES {
            return Err(io::Error::new(
                io::ErrorKind::InvalidData,
                format!("request line exceeds {MAX_REQUEST_LINE_BYTES} bytes"),
            ));
        }

        line.extend_from_slice(&available[..payload_len]);
        let consumed = payload_len + usize::from(newline.is_some());
        reader.consume(consumed);
        if newline.is_some() {
            return Ok(true);
        }
    }
}

impl Default for Server {
    fn default() -> Self {
        Self::new()
    }
}

impl Server {
    pub fn new() -> Self {
        Self {
            image: ImageApi::new(),
        }
    }

    pub fn with_image_api(image: ImageApi) -> Self {
        Self { image }
    }

    pub fn dispatch(&mut self, request: Request) -> Response {
        let request_id = request.request_id.clone();
        let result = request
            .validate()
            .and_then(|()| self.dispatch_command(request.command));

        Response {
            version: PROTOCOL_VERSION,
            request_id,
            result,
        }
    }

    fn dispatch_command(&mut self, command: Command) -> Result<Reply, ProtocolError> {
        match command {
            Command::PublishBehavior {
                name,
                expected_revision,
                source,
            } => self
                .image
                .publish_behavior(&name, expected_revision, &source)
                .map(behavior_dto)
                .map(Reply::Behavior)
                .map_err(api_error),
            Command::RollbackBehavior {
                name,
                expected_revision,
                target_revision,
            } => self
                .image
                .rollback_behavior(&name, expected_revision, target_revision)
                .map(behavior_dto)
                .map(Reply::Behavior)
                .map_err(api_error),
            Command::GetBehavior { name } => self
                .image
                .behavior(&name)
                .map(behavior_dto)
                .map(Reply::Behavior)
                .ok_or_else(|| ProtocolError::new(ErrorCode::NotFound, "unknown behavior")),
            Command::GetBehaviors => Ok(Reply::Behaviors(
                self.image
                    .behaviors()
                    .into_iter()
                    .map(behavior_dto)
                    .collect(),
            )),
            Command::GetBehaviorSource { name, revision } => self
                .image
                .behavior_source(&name, revision)
                .map(|source| Reply::BehaviorSource {
                    source: source.to_owned(),
                })
                .map_err(api_error),
            Command::Spawn {
                behavior_name,
                state,
                mailbox_capacity,
            } => {
                let capacity = NonZeroUsize::new(mailbox_capacity)
                    .ok_or_else(|| invalid_request("mailbox_capacity must be positive"))?;
                self.image
                    .spawn(&behavior_name, state, capacity)
                    .map(|actor_ref| Reply::ActorSpawned {
                        actor_ref: actor_ref.as_str().to_owned(),
                    })
                    .map_err(api_error)
            }
            Command::Send { actor_ref, message } => {
                let actor_ref = parse_actor_ref(&actor_ref)?;
                self.image
                    .send(actor_ref, message_from_dto(message)?)
                    .map(|()| Reply::Acknowledged)
                    .map_err(api_error)
            }
            Command::Tick {
                ticks,
                instruction_budget,
            } => {
                let budget = NonZeroUsize::new(instruction_budget)
                    .ok_or_else(|| invalid_request("instruction_budget must be positive"))?;
                self.image.tick(ticks, budget);
                Ok(Reply::TickCompleted { ticks })
            }
            Command::InspectActor { actor_ref } => {
                let actor_ref = parse_actor_ref(&actor_ref)?;
                self.image
                    .inspect(actor_ref)
                    .map(|snapshot| actor_snapshot_dto(&snapshot))
                    .map(Reply::Actor)
                    .map_err(api_error)
            }
            Command::ListActors => Ok(Reply::Actors(
                self.image.actors().iter().map(actor_snapshot_dto).collect(),
            )),
            Command::GetEvents => Ok(Reply::Events(
                self.image.events().iter().map(event_dto).collect(),
            )),
            Command::StopActor { actor_ref } => {
                let actor_ref = parse_actor_ref(&actor_ref)?;
                self.image
                    .stop(actor_ref)
                    .map(|()| Reply::Acknowledged)
                    .map_err(api_error)
            }
            Command::RestartActor { actor_ref } => {
                let actor_ref = parse_actor_ref(&actor_ref)?;
                self.image
                    .restart(actor_ref)
                    .map(|()| Reply::Acknowledged)
                    .map_err(api_error)
            }
            Command::PauseActor { actor_ref } => {
                let actor_ref = parse_actor_ref(&actor_ref)?;
                self.image
                    .pause(actor_ref)
                    .map(|()| Reply::Acknowledged)
                    .map_err(api_error)
            }
            Command::ResumeActor { actor_ref } => {
                let actor_ref = parse_actor_ref(&actor_ref)?;
                self.image
                    .resume(actor_ref)
                    .map(|()| Reply::Acknowledged)
                    .map_err(api_error)
            }
            Command::CancelActor { actor_ref } => {
                let actor_ref = parse_actor_ref(&actor_ref)?;
                self.image
                    .cancel(actor_ref)
                    .map(|()| Reply::Acknowledged)
                    .map_err(api_error)
            }
            Command::InstallBehavior {
                actor_ref,
                behavior_name,
            } => {
                let actor_ref = parse_actor_ref(&actor_ref)?;
                self.image
                    .install(actor_ref, &behavior_name)
                    .map(|implementation_id| Reply::BehaviorInstalled {
                        implementation_id: implementation_id.to_string(),
                    })
                    .map_err(api_error)
            }
        }
    }
}

fn behavior_dto(behavior: runa_code::BehaviorSnapshot) -> BehaviorDto {
    BehaviorDto {
        name: behavior.name,
        revision: behavior.revision,
        current_implementation_id: behavior.current.to_string(),
        history: behavior
            .history
            .into_iter()
            .map(|version| BehaviorVersionDto {
                revision: version.revision,
                implementation_id: version.implementation_id.to_string(),
            })
            .collect(),
    }
}

fn actor_snapshot_dto(snapshot: &runa_actor::ActorSnapshot) -> ActorSnapshotDto {
    ActorSnapshotDto {
        actor_ref: snapshot.actor_ref.as_str().to_owned(),
        mailbox_id: snapshot.mailbox.0,
        implementation_id: snapshot.implementation.to_string(),
        incarnation: snapshot.incarnation,
        state: snapshot.state,
        status: match &snapshot.status {
            Status::Idle => ActorStatusDto::Idle,
            Status::Running => ActorStatusDto::Running,
            Status::Awaiting => ActorStatusDto::Awaiting,
            Status::Paused => ActorStatusDto::Paused,
            Status::Stopped => ActorStatusDto::Stopped,
            Status::Failed(message) => ActorStatusDto::Failed(message.clone()),
        },
        execution: snapshot.execution.map(|execution| ExecutionSnapshotDto {
            vm_steps: execution.vm_steps,
            instruction_pointer: execution.instruction_pointer,
            call_depth: execution.call_depth,
        }),
        queued: snapshot.queued.iter().copied().map(message_dto).collect(),
        reserved: snapshot.reserved.map(message_dto),
    }
}

fn message_dto(message: Message) -> MessageDto {
    MessageDto {
        kind: message.kind,
        value: message.value,
        target_ref: message
            .target_ref
            .map(|actor_ref| actor_ref.as_str().to_owned()),
    }
}

fn message_from_dto(message: MessageDto) -> Result<Message, ProtocolError> {
    Ok(Message {
        kind: message.kind,
        value: message.value,
        target_ref: message
            .target_ref
            .as_deref()
            .map(parse_actor_ref)
            .transpose()?,
    })
}

fn event_dto(event: &Event) -> EventDto {
    match event {
        Event::Spawned(actor_ref) => EventDto::Spawned {
            actor_ref: actor_ref.as_str().to_owned(),
        },
        Event::Committed(actor_ref) => EventDto::Committed {
            actor_ref: actor_ref.as_str().to_owned(),
        },
        Event::Failed(actor_ref, message) => EventDto::Failed {
            actor_ref: actor_ref.as_str().to_owned(),
            message: message.clone(),
        },
        Event::Stopped(actor_ref) => EventDto::Stopped {
            actor_ref: actor_ref.as_str().to_owned(),
        },
        Event::Restarted(actor_ref) => EventDto::Restarted {
            actor_ref: actor_ref.as_str().to_owned(),
        },
        Event::Paused(actor_ref) => EventDto::Paused {
            actor_ref: actor_ref.as_str().to_owned(),
        },
        Event::Resumed(actor_ref) => EventDto::Resumed {
            actor_ref: actor_ref.as_str().to_owned(),
        },
        Event::InvocationCancelled(actor_ref) => EventDto::InvocationCancelled {
            actor_ref: actor_ref.as_str().to_owned(),
        },
        Event::BehaviorInstalled(actor_ref, implementation_id) => EventDto::BehaviorInstalled {
            actor_ref: actor_ref.as_str().to_owned(),
            implementation_id: implementation_id.to_string(),
        },
        Event::StateMigrated(actor_ref, previous_state, current_state) => EventDto::StateMigrated {
            actor_ref: actor_ref.as_str().to_owned(),
            previous_state: *previous_state,
            current_state: *current_state,
        },
    }
}

fn api_error(error: impl std::fmt::Display) -> ProtocolError {
    let message = error.to_string();
    let code = if message.starts_with("behavior revision conflict:") {
        ErrorCode::Conflict
    } else if matches!(
        message.as_str(),
        "unknown actor"
            | "unknown behavior"
            | "unknown behavior revision"
            | "unknown target behavior revision"
            | "unknown implementation"
    ) {
        ErrorCode::NotFound
    } else if matches!(
        message.as_str(),
        "mailbox full" | "effect exceeds mailbox capacity"
    ) {
        ErrorCode::CapacityExceeded
    } else if message == "invalid behavior name" {
        ErrorCode::InvalidRequest
    } else {
        ErrorCode::RuntimeFailure
    };
    ProtocolError::new(code, message)
}

fn invalid_request(message: impl Into<String>) -> ProtocolError {
    ProtocolError::new(ErrorCode::InvalidRequest, message)
}

fn parse_actor_ref(value: &str) -> Result<ActorRef, ProtocolError> {
    ActorRef::parse(value).ok_or_else(|| invalid_request("actor_ref must be a valid Nano ID"))
}

#[cfg(test)]
mod tests {
    use super::*;
    use runa_protocol::{Command, PROTOCOL_VERSION};
    use std::io::{BufRead, BufReader, Cursor};
    use std::net::Shutdown;

    fn request(command: Command) -> Request {
        Request {
            version: PROTOCOL_VERSION,
            request_id: "test-request".to_owned(),
            command,
        }
    }

    fn publish(server: &mut Server) -> BehaviorDto {
        let response = server.dispatch(request(Command::PublishBehavior {
            name: "counter".to_owned(),
            expected_revision: None,
            source: "pub fn handle(state, kind, value, target_ref) { (state + value, []) }"
                .to_owned(),
        }));
        match response.result.unwrap() {
            Reply::Behavior(behavior) => behavior,
            other => panic!("unexpected reply: {other:?}"),
        }
    }

    #[test]
    fn dispatches_behavior_actor_and_event_operations() {
        let mut server = Server::new();
        let behavior = publish(&mut server);
        assert_eq!(behavior.revision, 1);
        assert_eq!(behavior.history.len(), 1);

        let spawned = server.dispatch(request(Command::Spawn {
            behavior_name: "counter".to_owned(),
            state: 5,
            mailbox_capacity: 2,
        }));
        let actor_ref = match spawned.result.unwrap() {
            Reply::ActorSpawned { actor_ref } => actor_ref,
            other => panic!("unexpected reply: {other:?}"),
        };
        assert_eq!(actor_ref.len(), 21);
        assert!(!actor_ref.parse::<u64>().is_ok());

        let sent = server.dispatch(request(Command::Send {
            actor_ref: actor_ref.clone(),
            message: MessageDto {
                kind: 0,
                value: 7,
                target_ref: None,
            },
        }));
        assert_eq!(sent.result, Ok(Reply::Acknowledged));
        server.dispatch(request(Command::Tick {
            ticks: 100,
            instruction_budget: 10,
        }));

        let inspected = server.dispatch(request(Command::InspectActor {
            actor_ref: actor_ref.clone(),
        }));
        match inspected.result.unwrap() {
            Reply::Actor(snapshot) => assert_eq!(snapshot.state, 12),
            other => panic!("unexpected reply: {other:?}"),
        }

        let events = server.dispatch(request(Command::GetEvents));
        match events.result.unwrap() {
            Reply::Events(events) => assert!(events.iter().any(
                |event| matches!(event, EventDto::Committed { actor_ref: id } if id == &actor_ref)
            )),
            other => panic!("unexpected reply: {other:?}"),
        }
        assert_eq!(
            server.dispatch(request(Command::GetEvents)).result,
            Ok(Reply::Events(Vec::new()))
        );
    }

    #[test]
    fn dispatch_preserves_request_id_and_maps_validation_and_domain_errors() {
        let mut server = Server::new();
        let mut invalid = request(Command::GetEvents);
        invalid.version += 1;
        let response = server.dispatch(invalid);
        assert_eq!(response.request_id, "test-request");
        assert_eq!(
            response.result.unwrap_err().code,
            ErrorCode::UnsupportedVersion
        );

        let missing = server.dispatch(request(Command::InspectActor {
            actor_ref: "000000000000000000000".to_owned(),
        }));
        assert_eq!(missing.result.unwrap_err().code, ErrorCode::NotFound);

        publish(&mut server);
        let conflict = server.dispatch(request(Command::PublishBehavior {
            name: "counter".to_owned(),
            expected_revision: Some(99),
            source: "pub fn handle(state, kind, value, target_ref) { (state, []) }".to_owned(),
        }));
        assert_eq!(conflict.result.unwrap_err().code, ErrorCode::Conflict);
    }

    #[test]
    fn dispatch_supports_behavior_history_source_and_safe_install() {
        let mut server = Server::new();
        let first = publish(&mut server);
        let updated = server.dispatch(request(Command::PublishBehavior {
            name: "counter".to_owned(),
            expected_revision: Some(first.revision),
            source: "pub fn handle(state, kind, value, target_ref) { (state * value, []) }"
                .to_owned(),
        }));
        assert!(matches!(updated.result, Ok(Reply::Behavior(_))));

        let source = server.dispatch(request(Command::GetBehaviorSource {
            name: "counter".to_owned(),
            revision: 1,
        }));
        assert_eq!(
            source.result,
            Ok(Reply::BehaviorSource {
                source: "pub fn handle(state, kind, value, target_ref) { (state + value, []) }"
                    .to_owned()
            })
        );

        let actor_ref = match server
            .dispatch(request(Command::Spawn {
                behavior_name: "counter".to_owned(),
                state: 3,
                mailbox_capacity: 1,
            }))
            .result
            .unwrap()
        {
            Reply::ActorSpawned { actor_ref } => actor_ref,
            other => panic!("unexpected reply: {other:?}"),
        };
        let installed = server.dispatch(request(Command::InstallBehavior {
            actor_ref,
            behavior_name: "counter".to_owned(),
        }));
        assert!(matches!(
            installed.result,
            Ok(Reply::BehaviorInstalled { .. })
        ));
    }

    #[test]
    fn json_lines_transport_dispatches_multiple_requests_and_accepts_final_line_without_newline() {
        let publish_request = request(Command::PublishBehavior {
            name: "counter".to_owned(),
            expected_revision: None,
            source: "pub fn handle(state, kind, value, target_ref) { (state + value, []) }"
                .to_owned(),
        });
        let mut spawn_request = request(Command::Spawn {
            behavior_name: "counter".to_owned(),
            state: 4,
            mailbox_capacity: 2,
        });
        spawn_request.request_id = "spawn-request".to_owned();
        let mut list_request = request(Command::ListActors);
        list_request.request_id = "list-request".to_owned();
        let mut input = [publish_request, spawn_request]
            .iter()
            .map(serde_json::to_string)
            .collect::<Result<Vec<_>, _>>()
            .unwrap()
            .join("\n");
        input.push('\n');
        input.push_str(&serde_json::to_string(&list_request).unwrap());

        let mut output = Vec::new();
        serve_json_lines(&mut Server::new(), Cursor::new(input), &mut output).unwrap();

        let responses = output
            .split(|byte| *byte == b'\n')
            .filter(|line| !line.is_empty())
            .map(serde_json::from_slice::<Response>)
            .collect::<Result<Vec<_>, _>>()
            .unwrap();
        assert_eq!(responses.len(), 3);
        assert_eq!(
            responses
                .iter()
                .map(|response| response.request_id.as_str())
                .collect::<Vec<_>>(),
            vec!["test-request", "spawn-request", "list-request"]
        );
        assert!(matches!(responses[0].result, Ok(Reply::Behavior(_))));
        assert!(matches!(
            &responses[1].result,
            Ok(Reply::ActorSpawned { actor_ref }) if actor_ref.len() == 21
        ));
        assert!(matches!(
            &responses[2].result,
            Ok(Reply::Actors(actors)) if actors.len() == 1 && actors[0].state == 4
        ));
    }

    #[test]
    fn json_lines_transport_rejects_malformed_and_oversized_input() {
        let mut server = Server::new();
        let malformed =
            serve_json_lines(&mut server, Cursor::new(b"{not-json}\n"), Vec::new()).unwrap_err();
        assert_eq!(malformed.kind(), io::ErrorKind::InvalidData);

        let mut oversized = vec![b'a'; MAX_REQUEST_LINE_BYTES + 1];
        oversized.push(b'\n');
        let error = serve_json_lines(&mut server, Cursor::new(oversized), Vec::new()).unwrap_err();
        assert_eq!(error.kind(), io::ErrorKind::InvalidData);
    }

    #[test]
    fn loopback_listener_rejects_non_loopback_addresses() {
        let error = bind_loopback_listener("0.0.0.0:0".parse().unwrap()).unwrap_err();
        assert_eq!(error.kind(), io::ErrorKind::InvalidInput);

        let listener = bind_loopback_listener("127.0.0.1:0".parse().unwrap()).unwrap();
        assert!(listener.local_addr().unwrap().ip().is_loopback());
    }

    #[test]
    fn tcp_transport_dispatches_requests_to_the_shared_server() {
        let listener = bind_loopback_listener("127.0.0.1:0".parse().unwrap()).unwrap();
        let address = listener.local_addr().unwrap();
        let mut client = TcpStream::connect(address).unwrap();
        let (server_stream, peer_address) = listener.accept().unwrap();
        let mut network_client = NetworkClient::new(server_stream, peer_address);

        let first_request = request(Command::ListActors);
        let mut second_request = request(Command::ListActors);
        second_request.request_id = "second-request".to_owned();
        writeln!(
            client,
            "{}\n{}",
            serde_json::to_string(&first_request).unwrap(),
            serde_json::to_string(&second_request).unwrap()
        )
        .unwrap();
        client.shutdown(Shutdown::Write).unwrap();

        assert!(network_client.poll(&mut Server::new()).unwrap());
        let mut response_line = String::new();
        let mut client_reader = BufReader::new(client);
        client_reader.read_line(&mut response_line).unwrap();
        let first_response: Response = serde_json::from_str(&response_line).unwrap();
        assert_eq!(first_response.request_id, first_request.request_id);

        assert!(network_client.poll(&mut Server::new()).unwrap());
        response_line.clear();
        client_reader.read_line(&mut response_line).unwrap();
        let second_response: Response = serde_json::from_str(&response_line).unwrap();
        assert_eq!(second_response.request_id, second_request.request_id);
        assert_eq!(second_response.result, Ok(Reply::Actors(Vec::new())));

        assert!(!network_client.poll(&mut Server::new()).unwrap());
    }
}