gatewayd: gateway, the gateway.sock listener and one thread per connection

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

crates/gatewayd/src/gateway.rs

@@ -0,0 +1,298 @@
+//! `gateway.sock`: the listener, and one thread per `brokerd` connection (M4b spec, sections 4
+//! and 6). A connection thread speaks the protocol only; everything that touches Mattermost or the
+//! state file is asked of the serve loop, which alone owns them.
+
+use std::os::unix::fs::{DirBuilderExt, FileTypeExt, PermissionsExt};
+use std::os::unix::net::{UnixListener, UnixStream};
+use std::path::Path;
+use std::sync::Arc;
+use std::sync::mpsc::{self, Receiver, RecvTimeoutError, Sender, TryRecvError};
+use std::time::Duration;
+
+use proto::{
+    ApprovalPosted, ApprovalReply, Envelope, ErrorCode, FrameError, GatewayOutcome, Message,
+    PROTOCOL_VERSION, PendingApproval, WireError, read_frame, write_frame,
+};
+
+/// A log line printer, shared by the threads.
+pub type Log = Arc<dyn Fn(&str) + Send + Sync>;
+
+/// What a connection thread asks of the serve loop.
+#[derive(Debug)]
+pub enum ToLoop {
+    /// Post this approval; an accepted answer comes back on `answer`, and the post ids (or why
+    /// it could not be posted) on `shown`.
+    Show {
+        info: PendingApproval,
+        answer: Sender<ApprovalReply>,
+        shown: Sender<Result<Vec<String>, String>>,
+    },
+    /// It ended: edit its last post, and forget it.
+    Ended { approval: u64, ending: Ending },
+}
+
+/// How an approval ended, as far as this `gatewayd` knows.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum Ending {
+    /// Answered here, by `by`, and `brokerd` said what came of it.
+    Answered {
+        by: String,
+        reason: Option<String>,
+        outcome: GatewayOutcome,
+    },
+    /// `brokerd` closed first, or said nothing after the answer.
+    NoLonger,
+}
+
+/// How long to wait: for the first frame, for the loop to post, for `brokerd`'s outcome; and how
+/// often to look for an answer meanwhile.
+#[derive(Debug, Clone, Copy)]
+pub struct Timing {
+    pub first_frame: Duration,
+    pub posting: Duration,
+    pub outcome: Duration,
+    pub tick: Duration,
+}
+
+impl Default for Timing {
+    fn default() -> Self {
+        Timing {
+            first_frame: Duration::from_secs(10),
+            posting: Duration::from_secs(120),
+            outcome: Duration::from_secs(30),
+            tick: Duration::from_millis(100),
+        }
+    }
+}
+
+/// Listen on `socket`: its directory made 0700, a stale socket removed, the socket made 0600.
+pub fn listen(socket: &Path) -> Result<UnixListener, String> {
+    // 1. The directory: `socket.parent()`, which must not be empty, else "cannot prepare <socket>:
+    //    a socket needs a directory".
+    // 2. `std::fs::DirBuilder::new().recursive(true).mode(0o700).create(dir)`, then
+    //    `set_permissions(dir, 0o700)`; an error is "cannot prepare <dir>: <e>".
+    // 3. `std::fs::symlink_metadata(socket)`: a socket (`file_type().is_socket()`) is removed
+    //    (stale); anything else there is "cannot listen on <socket>: something else is there";
+    //    nothing there is fine.
+    // 4. `UnixListener::bind(socket)`, then `set_permissions(socket, 0o600)`; an error is "cannot
+    //    listen on <socket>: <e>".
+    // 5. Ok(listener). Paths are shown with `.display()`.
+    let dir = match socket.parent() {
+        Some(dir) => dir,
+        None => {
+            return Err(format!(
+                "cannot prepare {}: a socket needs a directory",
+                socket.display()
+            ));
+        }
+    };
+    if let Err(e) = std::fs::DirBuilder::new()
+        .recursive(true)
+        .mode(0o700)
+        .create(dir)
+    {
+        return Err(format!("cannot prepare {}: {e}", dir.display()));
+    }
+    if let Err(e) = std::fs::set_permissions(dir, std::fs::Permissions::from_mode(0o700)) {
+        return Err(format!("cannot prepare {}: {e}", dir.display()));
+    }
+    match std::fs::symlink_metadata(socket) {
+        Ok(meta) if meta.file_type().is_socket() => {
+            if let Err(e) = std::fs::remove_file(socket) {
+                return Err(format!("cannot listen on {}: {e}", socket.display()));
+            }
+        }
+        Ok(_) => {
+            return Err(format!(
+                "cannot listen on {}: something else is there",
+                socket.display()
+            ));
+        }
+        Err(_) => {}
+    }
+    let listener = match UnixListener::bind(socket) {
+        Ok(listener) => listener,
+        Err(e) => return Err(format!("cannot listen on {}: {e}", socket.display())),
+    };
+    if let Err(e) = std::fs::set_permissions(socket, std::fs::Permissions::from_mode(0o600)) {
+        return Err(format!("cannot listen on {}: {e}", socket.display()));
+    }
+    Ok(listener)
+}
+
+/// Accept connections on a thread of their own, each served on its own thread.
+pub fn serve(
+    listener: UnixListener,
+    to_loop: Sender<ToLoop>,
+    timing: Timing,
+    log: Log,
+) -> std::io::Result<()> {
+    // Spawn (with `std::thread::Builder`, `?`) a thread that loops over `listener.incoming()`,
+    // skipping errors. For each stream: clone `to_loop` and `log`, and spawn a thread running
+    // `connection(stream, &to_loop, timing, &log)`; if that spawn fails, log "gatewayd: cannot
+    // start a thread for an approval: <e>". Then Ok(()).
+    std::thread::Builder::new()
+        .name("gateway-accept".to_string())
+        .spawn(move || {
+            for stream in listener.incoming() {
+                match stream {
+                    Ok(stream) => {
+                        let to_loop = to_loop.clone();
+                        let thread_log = log.clone();
+                        if let Err(e) = std::thread::Builder::new()
+                            .name("gateway-connection".to_string())
+                            .spawn(move || connection(stream, &to_loop, timing, &thread_log))
+                        {
+                            log(&format!(
+                                "gatewayd: cannot start a thread for an approval: {e}"
+                            ));
+                        }
+                    }
+                    Err(_) => continue,
+                }
+            }
+        })?;
+    Ok(())
+}
+
+fn frame(id: u64, last: bool, msg: Message) -> Envelope {
+    Envelope {
+        v: PROTOCOL_VERSION,
+        id,
+        r#final: last,
+        msg,
+    }
+}
+
+/// Every frame `brokerd` sends, read whole on a thread of its own; shutting the stream down ends it.
+fn frames(stream: &UnixStream) -> std::io::Result<Receiver<Result<Envelope, FrameError>>> {
+    // As in `brokerd`'s courier: `let mut reader = stream.try_clone()?;`, a channel, and a thread
+    // (`std::thread::Builder`, `?`) looping `let frame = read_frame(&mut reader); let last =
+    // frame.is_err(); if tx.send(frame).is_err() || last { return; }`. Then `Ok(rx)`.
+    let mut reader = stream.try_clone()?;
+    let (tx, rx) = mpsc::channel();
+    std::thread::Builder::new()
+        .name("gateway-frame".to_string())
+        .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 `brokerd` connection, frames 1 to 4 of the spec's section 4.
+pub fn connection(mut stream: UnixStream, to_loop: &Sender<ToLoop>, timing: Timing, log: &Log) {
+    // 1. The request, within `first_frame`.
+    let _ = stream.set_read_timeout(Some(timing.first_frame));
+    let info = match read_frame(&mut stream) {
+        Ok(Envelope {
+            id: 1,
+            r#final: true,
+            msg: Message::ApprovalRequest(req),
+            ..
+        }) => req.approval,
+        _ => return,
+    };
+    let _ = stream.set_read_timeout(None);
+    let id = info.approval;
+    // 2. The loop posts it.
+    let (answer_tx, answer_rx) = mpsc::channel();
+    let (shown_tx, shown_rx) = mpsc::channel();
+    if to_loop
+        .send(ToLoop::Show {
+            info,
+            answer: answer_tx,
+            shown: shown_tx,
+        })
+        .is_err()
+    {
+        return;
+    }
+    let posts = match shown_rx.recv_timeout(timing.posting) {
+        Ok(Ok(posts)) => posts,
+        Ok(Err(why)) => {
+            let e = WireError {
+                code: ErrorCode::Internal,
+                detail: why,
+            };
+            let _ = write_frame(&mut stream, &frame(1, true, Message::Error(e)));
+            return;
+        }
+        Err(_) => {
+            let e = WireError {
+                code: ErrorCode::Internal,
+                detail: "gatewayd did not post it in time".to_string(),
+            };
+            let _ = write_frame(&mut stream, &frame(1, true, Message::Error(e)));
+            return;
+        }
+    };
+    let rx = match frames(&stream) {
+        Ok(rx) => rx,
+        Err(_) => return end(to_loop, id, Ending::NoLonger),
+    };
+    if write_frame(
+        &mut stream,
+        &frame(1, false, Message::ApprovalPosted(ApprovalPosted { posts })),
+    )
+    .is_err()
+    {
+        return end(to_loop, id, Ending::NoLonger);
+    }
+    // 3. An answer from an approver, or brokerd closing first.
+    let mut answers = Some(answer_rx);
+    let reply = loop {
+        if let Some(answers_rx) = &answers {
+            match answers_rx.recv_timeout(timing.tick) {
+                Ok(reply) => break reply,
+                Err(RecvTimeoutError::Timeout) => {}
+                Err(RecvTimeoutError::Disconnected) => answers = None,
+            }
+        } else {
+            std::thread::sleep(timing.tick);
+        }
+        match rx.try_recv() {
+            Err(TryRecvError::Empty) => {}
+            // brokerd closed (the approval ended elsewhere), or sent what it may not.
+            _ => {
+                let _ = stream.shutdown(std::net::Shutdown::Both);
+                return end(to_loop, id, Ending::NoLonger);
+            }
+        }
+    };
+    let (by, reason) = (reply.by.clone(), reply.reason.clone());
+    if write_frame(&mut stream, &frame(1, true, Message::ApprovalReply(reply))).is_err() {
+        return end(to_loop, id, Ending::NoLonger);
+    }
+    // 4. What came of it.
+    let ending = match rx.recv_timeout(timing.outcome) {
+        Ok(Ok(Envelope {
+            id: 2,
+            r#final: true,
+            msg: Message::ApprovalOutcome(o),
+            ..
+        })) => Ending::Answered {
+            by,
+            reason,
+            outcome: o.outcome,
+        },
+        _ => {
+            (log)(&format!(
+                "gatewayd: approval {id}: brokerd said nothing after the answer"
+            ));
+            Ending::NoLonger
+        }
+    };
+    let _ = stream.shutdown(std::net::Shutdown::Both);
+    end(to_loop, id, ending);
+}
+
+fn end(to_loop: &Sender<ToLoop>, approval: u64, ending: Ending) {
+    // `let _ = to_loop.send(ToLoop::Ended { approval, ending });`
+    let _ = to_loop.send(ToLoop::Ended { approval, ending });
+}

crates/gatewayd/src/lib.rs

@@ -4,6 +4,7 @@
 pub mod approvals;
 pub mod config;
 pub mod deliver;
+pub mod gateway;
 pub mod http;
 pub mod mm;
 pub mod net;