boxmaker/crates/brokerd/src/courier.rs

//! The courier: one pending approval carried to `gatewayd` over `gateway.sock`, and its answer
//! back (M4b spec, section 5). It runs beside the connection thread that waits for the verdict,
//! and is one more taker of the table's entry, answering it exactly as `bxctl` does.

use std::os::unix::net::UnixStream;
use std::path::Path;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::mpsc::{self, Receiver, RecvTimeoutError};
use std::time::{Duration, Instant};

use bxcore::{
    ApprovalOutcome, ApprovalReply, ApprovalRequest, DecisionRecord, DenyReason, Envelope,
    FrameError, GatewayOutcome, Message, PendingApproval, Timestamp, read_frame,
};

use crate::admin;
use crate::broker::{Broker, send};
use crate::ledger::Answer;

pub const RUNBOOK: &str = "see docs/runbook.md#gateway-unreachable";

/// How long to wait before trying again, and how often to look at the table meanwhile.
#[derive(Debug, Clone, Copy)]
pub struct Timing {
    pub retry: Duration,
    pub tick: Duration,
}

impl Default for Timing {
    fn default() -> Self {
        Timing {
            retry: Duration::from_secs(5),
            tick: Duration::from_secs(1),
        }
    }
}

/// How one attempt ended.
enum Attempt {
    /// The approval is answered or no longer pending: the courier is done.
    Done,
    /// It was never posted: `gatewayd` could not be reached, or could not post.
    Unreachable(String),
    /// It was posted, then `gatewayd` went away or broke the protocol.
    Lost(String),
}

/// Run `wait` (the connection thread's wait for the verdict) with a courier beside it, when
/// `[gateway]` is set. The courier stops when `wait` returns.
pub fn alongside<T>(broker: &Broker, info: &PendingApproval, wait: impl FnOnce() -> T) -> T {
    let Some(socket) = broker.cfg().gateway_socket() else {
        return wait();
    };
    let stop = AtomicBool::new(false);
    std::thread::scope(|scope| {
        let started = std::thread::Builder::new().spawn_scoped(scope, || {
            carry(broker, &socket, info, &stop, Timing::default())
        });
        if let Err(e) = started {
            broker.log(&format!(
                "brokerd: approval {}: cannot start its courier: {e}; bxctl can still answer it",
                info.approval
            ));
        }
        let out = wait();
        stop.store(true, Ordering::SeqCst);
        out
    })
}

/// Carry approval `info` to `gatewayd` until it is answered or no longer pending.
pub fn carry(
    broker: &Broker,
    socket: &Path,
    info: &PendingApproval,
    stop: &AtomicBool,
    timing: Timing,
) {
    let id = info.approval;
    let mut warned = false;
    loop {
        if finished(broker, id, stop) {
            return;
        }
        match attempt(broker, socket, info, stop, timing, warned) {
            Attempt::Done => return,
            Attempt::Unreachable(why) => {
                if !warned {
                    warned = true;
                    broker.log(&format!(
                        "brokerd: approval {id}: gatewayd is not reachable at {}: {why}; bxctl can still answer it\n{RUNBOOK}",
                        socket.display()
                    ));
                }
                pause(broker, id, stop, timing);
            }
            Attempt::Lost(why) => {
                broker.log(&format!("brokerd: approval {id}: {why}; posting it again"));
                pause(broker, id, stop, timing);
            }
        }
    }
}

/// The courier's work is over: stopped, or the entry has left the table.
fn finished(broker: &Broker, id: u64, stop: &AtomicBool) -> bool {
    stop.load(Ordering::SeqCst) || !broker.table().contains(id)
}

/// Wait `timing.retry`, in ticks, unless the courier's work ends first.
fn pause(broker: &Broker, id: u64, stop: &AtomicBool, timing: Timing) {
    let until = Instant::now() + timing.retry;
    while Instant::now() < until && !finished(broker, id, stop) {
        std::thread::sleep(
            timing
                .tick
                .min(until.saturating_duration_since(Instant::now())),
        );
    }
}

/// Every frame `gatewayd` sends, read whole on a thread of its own, so the courier can wait with
/// a timeout without losing half a frame. Shutting the stream down ends the thread.
fn frames(stream: &UnixStream) -> std::io::Result<Receiver<Result<Envelope, FrameError>>> {
    // 1. `let mut reader = stream.try_clone()?;` and `let (tx, rx) = mpsc::channel();`.
    // 2. `std::thread::Builder::new().spawn(move || loop { ... })?`, where the loop body is: `let
    //    frame = read_frame(&mut reader); let last = frame.is_err(); if tx.send(frame).is_err() ||
    //    last { return; }`.
    // 3. `Ok(rx)`.
    let mut reader = stream.try_clone()?;
    let (tx, rx) = mpsc::channel();
    std::thread::Builder::new().spawn(move || {
        loop {
            let frame = read_frame(&mut reader);
            let last = frame.is_err();
            if tx.send(frame).is_err() || last {
                return;
            }
        }
    })?;
    Ok(rx)
}

/// One connection: post, then wait for the answer, looking at the table every tick.
fn attempt(
    broker: &Broker,
    socket: &Path,
    info: &PendingApproval,
    stop: &AtomicBool,
    timing: Timing,
    warned: bool,
) -> Attempt {
    let mut stream = match UnixStream::connect(socket) {
        Ok(stream) => stream,
        Err(e) => return Attempt::Unreachable(e.to_string()),
    };
    let rx = match frames(&stream) {
        Ok(rx) => rx,
        Err(e) => return Attempt::Unreachable(e.to_string()),
    };
    let ended = exchange(broker, &mut stream, &rx, info, stop, timing, warned);
    let _ = stream.shutdown(std::net::Shutdown::Both);
    ended
}

/// Frames 1 to 4 of section 4 on an open connection.
fn exchange(
    broker: &Broker,
    stream: &mut UnixStream,
    rx: &Receiver<Result<Envelope, FrameError>>,
    info: &PendingApproval,
    stop: &AtomicBool,
    timing: Timing,
    warned: bool,
) -> Attempt {
    let id = info.approval;
    let request = Message::ApprovalRequest(ApprovalRequest {
        approval: info.clone(),
    });
    if !send(stream, 1, true, request) {
        return Attempt::Unreachable("the request could not be sent".to_string());
    }
    let mut posted = false;
    loop {
        let frame = match rx.recv_timeout(timing.tick) {
            Ok(Ok(frame)) => frame,
            Err(RecvTimeoutError::Timeout) => {
                if finished(broker, id, stop) {
                    return Attempt::Done;
                }
                continue;
            }
            Ok(Err(_)) | Err(RecvTimeoutError::Disconnected) => {
                return lost(posted, "gatewayd went away");
            }
        };
        match (frame.id, frame.r#final, frame.msg) {
            (1, false, Message::ApprovalPosted(_)) if !posted => {
                posted = true;
                if warned {
                    broker.log(&format!("brokerd: approval {id} posted in Mattermost"));
                }
            }
            (1, true, Message::Error(e)) if !posted => {
                return Attempt::Unreachable(format!("gatewayd could not post it: {}", e.detail));
            }
            (1, true, Message::ApprovalReply(reply)) if posted => {
                if reply.approval != id {
                    return lost(posted, "gatewayd answered another approval");
                }
                if let Err(why) = reply.check() {
                    return lost(posted, &format!("gatewayd sent a bad answer ({why})"));
                }
                let outcome = answer(broker, id, reply);
                let _ = send(
                    stream,
                    2,
                    true,
                    Message::ApprovalOutcome(ApprovalOutcome { outcome }),
                );
                return Attempt::Done;
            }
            _ => return lost(posted, "gatewayd broke the protocol"),
        }
    }
}

fn lost(posted: bool, why: &str) -> Attempt {
    if posted {
        Attempt::Lost(why.to_string())
    } else {
        Attempt::Unreachable(why.to_string())
    }
}

/// Take the entry and answer it as `bxctl` would, with who answered and in which post.
fn answer(broker: &Broker, id: u64, reply: ApprovalReply) -> GatewayOutcome {
    // 1. `let Some(entry) = broker.table().take(id) else { return GatewayOutcome::Gone {}; };`
    // 2. `let (by, post) = (Some(reply.by), Some(reply.post));` and the `Answer`: `Answer::Approved
    //    { by, post }` when `reply.approved`, else `Answer::Refused { by, post, reason:
    //    reply.reason }`.
    // 3. `match (reply.approved, admin::answer(broker, entry, answer, Timestamp::now()))`: (true,
    //    DecisionRecord::Denied { reason }) -> `GatewayOutcome::Denied { reason }`; (true, _) ->
    //    `GatewayOutcome::Runs {}`; (false, DecisionRecord::Denied { reason:
    //    DenyReason::ApprovalRefused }) -> `GatewayOutcome::Refused {}`; (false,
    //    DecisionRecord::Denied { reason }) -> `GatewayOutcome::Denied { reason }` (the refusal
    //    could not be recorded); (false, _) -> `GatewayOutcome::Refused {}`.
    let Some(entry) = broker.table().take(id) else {
        return GatewayOutcome::Gone {};
    };
    let (by, post) = (Some(reply.by), Some(reply.post));
    let answer = if reply.approved {
        Answer::Approved { by, post }
    } else {
        Answer::Refused {
            by,
            post,
            reason: reply.reason,
        }
    };
    match (
        reply.approved,
        admin::answer(broker, entry, answer, Timestamp::now()),
    ) {
        (true, DecisionRecord::Denied { reason }) => GatewayOutcome::Denied { reason },
        (true, _) => GatewayOutcome::Runs {},
        (
            false,
            DecisionRecord::Denied {
                reason: DenyReason::ApprovalRefused,
            },
        ) => GatewayOutcome::Refused {},
        (false, DecisionRecord::Denied { reason }) => GatewayOutcome::Denied { reason },
        (false, _) => GatewayOutcome::Refused {},
    }
}