boxmaker/crates/loopd/src/broker_port.rs

//! The tool port to `brokerd`: one connection per call, a deadline per frame, and plain failures.
//!
//! Whatever goes wrong reaching `brokerd` or reading its frames, the model gets a plain failure and
//! the turn goes on. Every failure prints one line ending at `#broker-unavailable`.

use std::os::unix::net::UnixStream;
use std::path::PathBuf;
use std::time::{Duration, Instant};

use crate::tools::{Pending, ToolPort};
use proto::{
    Envelope, FrameError, Message, PROTOCOL_VERSION, Timestamp, ToolRequest, ToolResponse,
    read_frame, write_frame,
};

/// The failure every outage reaches the model as.
pub const UNAVAILABLE: &str = "the tool broker is unavailable";
/// What the model is told when no broker is configured.
pub const NOT_CONFIGURED: &str = "no tool broker is configured";
/// The failure a request too big for one frame reaches.
pub const TOO_LARGE: &str = "the request is too large for the tool broker";
/// The longest `BrokerPort` waits for an answer after a pending frame, whatever its `expires`
/// says, so a turn cannot be parked for ever by a far expiry.
pub const MAX_PENDING_WAIT: Duration = Duration::from_secs(24 * 60 * 60);
/// The runbook entry every outage line ends with.
pub const POINTER: &str = "see docs/runbook.md#broker-unavailable";

/// "loopd: {UNAVAILABLE}: {why}; {POINTER}"
pub fn unavailable_line(why: &str) -> String {
    format!("loopd: {UNAVAILABLE}: {why}; {POINTER}")
}

/// "loopd: {NOT_CONFIGURED}: every tool call except the core ones will fail; {POINTER}"
pub fn not_configured_line() -> String {
    format!("loopd: {NOT_CONFIGURED}: every tool call except the core ones will fail; {POINTER}")
}

/// Reads through `stream`, but only what is left before `until`: a read past the deadline fails
/// with a timeout, so a peer that sends one byte at a time cannot hold the turn.
struct Deadline<'a> {
    stream: &'a UnixStream,
    until: Instant,
}

impl std::io::Read for Deadline<'_> {
    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
        let left = self.until.saturating_duration_since(Instant::now());
        if left.is_zero() {
            return Err(std::io::Error::from(std::io::ErrorKind::TimedOut));
        }
        crate::socket::set_read_timeout(self.stream, left)?;
        let mut stream = self.stream; // `Read` is implemented for `&UnixStream`
        stream.read(buf)
    }
}

/// The tool port to `brokerd`. One connection per call, with a deadline per frame.
pub struct BrokerPort {
    socket: PathBuf,
    timeout: Duration,
    pending_cap: Duration,
    log: Box<dyn Fn(&str) + Send + Sync>,
}

impl BrokerPort {
    pub fn new(socket: PathBuf, timeout: Duration) -> BrokerPort {
        BrokerPort {
            socket,
            timeout,
            pending_cap: MAX_PENDING_WAIT,
            log: Box::new(|l| eprintln!("{l}")),
        }
    }

    /// The same port with another cap on the wait after a pending frame (`MAX_PENDING_WAIT`).
    pub fn with_pending_cap(self, pending_cap: Duration) -> BrokerPort {
        BrokerPort {
            pending_cap,
            ..self
        }
    }

    pub fn with_log(
        socket: PathBuf,
        timeout: Duration,
        log: Box<dyn Fn(&str) + Send + Sync>,
    ) -> BrokerPort {
        BrokerPort {
            socket,
            timeout,
            pending_cap: MAX_PENDING_WAIT,
            log,
        }
    }
}

impl ToolPort for BrokerPort {
    fn call(&self, request: &ToolRequest, on_pending: &mut dyn FnMut(&Pending)) -> ToolResponse {
        let want = request.call.0;

        // The first deadline, taken before connecting.
        let mut until = match Instant::now().checked_add(self.timeout) {
            Some(until) => until,
            None => return self.unavailable("the timeout is too large"),
        };

        // Exit 2.
        let mut stream = match UnixStream::connect(&self.socket) {
            Ok(stream) => stream,
            Err(e) => {
                return self
                    .unavailable(&format!("cannot connect to {}: {e}", self.socket.display()));
            }
        };

        // Exit 3. `set_write_timeout` fails for a zero timeout.
        if let Err(e) = stream.set_write_timeout(Some(self.timeout)) {
            return self.unavailable(&format!("cannot set a timeout: {e}"));
        }

        // Exit 4. One request frame; the broker is fine if it is too large to send.
        let request_frame = Envelope {
            v: PROTOCOL_VERSION,
            id: want,
            r#final: true,
            msg: Message::ToolRequest(request.clone()),
        };
        if let Err(e) = write_frame(&mut stream, &request_frame) {
            return match e {
                FrameError::TooLarge(_) => ToolResponse::Failed {
                    message: TOO_LARGE.to_string(),
                },
                _ => self.unavailable(&format!("cannot send the request: {e}")),
            };
        }

        // Exit 5-11, once per frame. A pending frame is reported, then we read again.
        let mut seen_pending = false;
        loop {
            let mut deadline = Deadline {
                stream: &stream,
                until,
            };
            let env = match read_frame(&mut deadline) {
                Ok(env) => env,
                // Exit 5.
                Err(FrameError::Closed) => {
                    return self.unavailable("the connection closed before the final answer");
                }
                Err(FrameError::Io(e))
                    if e.kind() == std::io::ErrorKind::WouldBlock
                        || e.kind() == std::io::ErrorKind::TimedOut =>
                {
                    return self.unavailable("no answer in time");
                }
                Err(e) => return self.unavailable(&format!("bad frame: {e}")),
            };

            // Exit 6.
            if env.id != want {
                return self
                    .unavailable(&format!("an answer for request {}, not {}", env.id, want));
            }

            let r#final = env.r#final;
            // Exit 7.
            let resp = match env.msg {
                Message::ToolResponse(resp) => resp,
                Message::Error(e) => {
                    return self
                        .unavailable(&format!("the broker reported an error: {}", e.detail));
                }
                _ => return self.unavailable("an unexpected message"),
            };

            match resp {
                // Exit 8.
                ToolResponse::PendingApproval { approval, expires } => {
                    if r#final {
                        return self.unavailable("a pending frame marked final");
                    }
                    // Exit 10.
                    if seen_pending {
                        return self.unavailable("a second pending frame");
                    }
                    let wait = expires
                        .unix_millis()
                        .saturating_sub(Timestamp::now().unix_millis());
                    let wait = Duration::from_millis(wait).min(self.pending_cap);
                    until = match Instant::now().checked_add(wait) {
                        Some(until) => until,
                        None => return self.unavailable("an expiry too far away"),
                    };
                    until = match until.checked_add(self.timeout) {
                        Some(until) => until,
                        None => return self.unavailable("an expiry too far away"),
                    };
                    seen_pending = true;
                    on_pending(&Pending { approval, expires });
                }
                // Exit 9 and 11.
                ToolResponse::Result { .. }
                | ToolResponse::Failed { .. }
                | ToolResponse::Denied { .. } => {
                    if r#final {
                        return resp;
                    }
                    return self.unavailable("an answer not marked final");
                }
            }
        }
    }
}

/// The one failure no outage reaches the model as, and the one line printed with it.
impl BrokerPort {
    fn unavailable(&self, why: &str) -> ToolResponse {
        (self.log)(&unavailable_line(why));
        ToolResponse::Failed {
            message: UNAVAILABLE.to_string(),
        }
    }
}

/// The port used when no broker is configured: every call fails with `NOT_CONFIGURED` and prints
/// nothing.
pub struct NoBroker;

impl ToolPort for NoBroker {
    fn call(&self, _request: &ToolRequest, _on_pending: &mut dyn FnMut(&Pending)) -> ToolResponse {
        ToolResponse::Failed {
            message: NOT_CONFIGURED.to_string(),
        }
    }
}