brokerd: courier, carrying pending approvals to gatewayd and answers back

Implemented-By: OpenCode session (model recorded in docs/implementer-log.md)

crates/brokerd/src/broker.rs

@@ -6,7 +6,7 @@
 use std::io::Read;
 use std::os::unix::net::UnixStream;
 use std::sync::Mutex;
-use std::sync::mpsc::RecvTimeoutError;
+use std::sync::mpsc::{Receiver, RecvTimeoutError};
 use std::time::Duration;
 
 use proto::{
@@ -16,6 +16,7 @@ use proto::{
 
 use crate::approvals::{Table, Verdict};
 use crate::config::Config;
+use crate::courier;
 use crate::grants;
 use crate::ledger::{Call, Decided, Grants, Ledger};
 use crate::policy::{Ask, SessionState};
@@ -166,6 +167,44 @@ pub fn forbid(broker: &Broker, stream: &mut UnixStream, id: u64, msg: &Message,
     );
 }
 
+/// What waiting for the verdict came to.
+enum Waited {
+    Verdict(Result<Verdict, DenyReason>),
+    PeerLeft,
+}
+
+/// Wait for the verdict, or for the peer to leave. If the peer leaves and the entry is still in
+/// the table, taking it ends the approval unanswered; if someone already took it, their verdict
+/// is awaited.
+fn wait_for_verdict(
+    broker: &Broker,
+    stream: &UnixStream,
+    verdict: &Receiver<Verdict>,
+    seq: u64,
+) -> Waited {
+    loop {
+        match verdict.recv_timeout(Duration::from_secs(1)) {
+            Ok(verdict) => return Waited::Verdict(Ok(verdict)),
+            Err(RecvTimeoutError::Timeout) => {
+                if alive(stream) {
+                    continue;
+                }
+                match broker.table().take(seq) {
+                    Some(_) => return Waited::PeerLeft,
+                    None => match verdict.recv() {
+                        Ok(verdict) => return Waited::Verdict(Ok(verdict)),
+                        Err(_) => return Waited::Verdict(Err(DenyReason::AuditUnavailable)),
+                    },
+                }
+            }
+            // The taker dropped it unanswered.
+            Err(RecvTimeoutError::Disconnected) => {
+                return Waited::Verdict(Err(DenyReason::AuditUnavailable));
+            }
+        }
+    }
+}
+
 /// Whether the peer is still there: it sends nothing more and never half-closes, so a byte would
 /// break the protocol and a timeout means it is waiting.
 pub fn alive(stream: &UnixStream) -> bool {
@@ -256,6 +295,7 @@ fn pending(
         created: now,
         expires,
     };
+    let shown = info.clone();
     let verdict = broker.table().insert(info, ask);
 
     // 3. Show the caller the pending call. If the frame does not reach the peer, the entry is
@@ -273,24 +313,14 @@ fn pending(
         return None; // 3. the peer left before the frame went out
     }
 
-    // 4. Wait for the verdict, or for the peer to leave.
-    let outcome = loop {
-        match verdict.recv_timeout(Duration::from_secs(1)) {
-            Ok(verdict) => break Ok(verdict),
-            Err(RecvTimeoutError::Timeout) => {
-                if alive(stream) {
-                    continue;
-                }
-                match broker.table().take(seq) {
-                    Some(_) => return None, // 4. the peer left while we waited
-                    None => match verdict.recv() {
-                        Ok(verdict) => break Ok(verdict),
-                        Err(_) => break Err(DenyReason::AuditUnavailable),
-                    },
-                }
-            }
-            Err(RecvTimeoutError::Disconnected) => break Err(DenyReason::AuditUnavailable), // 4. the taker dropped it unanswered
-        }
+    // 4. Wait for the verdict, or for the peer to leave, with a courier carrying the approval to
+    //    Mattermost meanwhile when `[gateway]` is set.
+    let waited = courier::alongside(broker, &shown, || {
+        wait_for_verdict(broker, stream, &verdict, seq)
+    });
+    let outcome = match waited {
+        Waited::Verdict(outcome) => outcome,
+        Waited::PeerLeft => return None, // 4. the peer left while we waited
     };
 
     // 5. Answer from the verdict.

crates/brokerd/src/courier.rs

@@ -0,0 +1,276 @@
+//! 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 proto::{
+    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 {},
+    }
+}

crates/brokerd/src/lib.rs

@@ -7,6 +7,7 @@ pub mod audit;
 pub mod broker;
 pub mod config;
 pub mod container;
+pub mod courier;
 pub mod grants;
 pub mod ledger;
 pub mod pipes;