Keep job outcomes until the cloud acknowledges them

The gateway now follows the job each printer was handed. Started, finished
and failed are written to its SQLite store and sent as job_event on every
connection until a job_event_ack clears them, so an outcome reached during
an outage, or before a restart, is still delivered.

A start that fails after being accepted goes through the same outbox
instead of a one-shot message to whoever was connected. Nothing is
concluded from a printer's state until the job has been seen running,
because a printer told to start still reports the previous job's FINISH.

The backend still owns the queue; this only reports.

src/jobs.rs

@@ -0,0 +1,185 @@
+//! Following the jobs the cloud has handed this gateway, so their outcomes
+//! reach the cloud even if it was unreachable when they happened.
+//!
+//! The cloud owns the queue; this only reports. When a `start_print` is
+//! accepted the job is recorded against its printer, and each tick the
+//! printer's state is compared with it. What the cloud should hear -
+//! started, finished, failed - goes into the store's outbox, and the uplink
+//! re-sends the outbox on every connection until the cloud acknowledges each
+//! event. An outcome reached during an outage, or before a restart, is
+//! therefore still delivered, instead of the cloud guessing from whatever
+//! telemetry it happens to see next.
+use std::time::Duration;
+
+use tracing::{info, warn};
+
+use crate::printer::GenericPrinter;
+use crate::store::{ActiveJob, Store};
+use crate::uplink::SharedFleet;
+
+const TICK: Duration = Duration::from_secs(5);
+/// A job whose printer never reported printing within this long is given up on.
+const NEVER_STARTED_AFTER_MS: i64 = 30 * 60 * 1000;
+
+pub fn now_ms() -> i64 {
+    std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .map(|d| d.as_millis() as i64)
+        .unwrap_or(0)
+}
+
+#[derive(Debug, PartialEq)]
+pub enum Step {
+    Nothing,
+    Started,
+    Finished,
+    Failed(String),
+}
+
+/// What a printer's reported state means for the job it was handed.
+///
+/// Nothing is concluded from a state until the job has been seen running: a
+/// printer that has just been told to start still reports the *previous*
+/// job's `FINISH` (or `FAILED`) for a while, and reading that as this job's
+/// outcome would report a print that has not begun as done.
+pub fn next_step(job: &ActiveJob, state: Option<&str>, now_ms: i64) -> Step {
+    let state = state.map(str::to_ascii_uppercase);
+    let running = matches!(state.as_deref(), Some("RUNNING" | "PAUSE" | "PREPARE" | "SLICING"));
+
+    if !job.started {
+        if running {
+            return Step::Started;
+        }
+        if now_ms - job.since_ms > NEVER_STARTED_AFTER_MS {
+            return Step::Failed("the printer never started printing it".into());
+        }
+        return Step::Nothing;
+    }
+
+    match state.as_deref() {
+        Some("FINISH") => Step::Finished,
+        Some("FAILED") => Step::Failed("the printer reported a fault".into()),
+        // Not a finish, not a fault, and no longer printing: stopped on the
+        // machine itself. Klipper and OctoPrint report a cancelled print as
+        // idle.
+        Some("IDLE") => Step::Failed("the printer stopped without finishing it".into()),
+        _ => Step::Nothing,
+    }
+}
+
+/// The gateway accepted a `start_print` for `job_id`.
+pub fn accepted(store: &Store, printer_id: &str, job_id: &str) {
+    if let Err(err) = store.active_job_set(printer_id, job_id, now_ms()) {
+        warn!(%err, %printer_id, %job_id, "could not record the job against its printer");
+    }
+}
+
+/// The accepted start then failed (download, transfer, start).
+pub fn start_failed(store: &Store, printer_id: &str, job_id: &str, reason: &str) {
+    if let Err(err) = store.outbox_push(job_id, "failed", Some(reason), now_ms()) {
+        warn!(%err, %job_id, "could not queue the failure for the cloud");
+    }
+    let _ = store.active_job_clear(printer_id);
+}
+
+pub async fn run(fleet: SharedFleet, store: std::sync::Arc<Store>) {
+    let mut ticker = tokio::time::interval(TICK);
+    loop {
+        ticker.tick().await;
+        let Ok(active) = store.active_jobs() else { continue };
+        if active.is_empty() {
+            continue;
+        }
+        let states: Vec<(ActiveJob, Option<String>)> = {
+            let fleet = fleet.lock().await;
+            let mut out = Vec::new();
+            for job in active {
+                let state = match fleet.iter().find(|p| p.base().id == job.printer_id) {
+                    Some(printer) => printer.status().await.and_then(|s| s.state),
+                    None => None,
+                };
+                out.push((job, state));
+            }
+            out
+        };
+        for (job, state) in states {
+            apply(&store, &job, next_step(&job, state.as_deref(), now_ms()));
+        }
+    }
+}
+
+fn apply(store: &Store, job: &ActiveJob, step: Step) {
+    let result = match &step {
+        Step::Nothing => return,
+        Step::Started => store
+            .outbox_push(&job.job_id, "started", None, now_ms())
+            .and_then(|_| store.active_job_started(&job.printer_id)),
+        Step::Finished => store
+            .outbox_push(&job.job_id, "finished", None, now_ms())
+            .and_then(|_| store.active_job_clear(&job.printer_id)),
+        Step::Failed(reason) => store
+            .outbox_push(&job.job_id, "failed", Some(reason), now_ms())
+            .and_then(|_| store.active_job_clear(&job.printer_id)),
+    };
+    match result {
+        Ok(_) => info!(printer = %job.printer_id, job = %job.job_id, ?step, "job event queued for the cloud"),
+        Err(err) => warn!(%err, job = %job.job_id, "could not queue a job event"),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn job(started: bool, since_ms: i64) -> ActiveJob {
+        ActiveJob { printer_id: "p1".into(), job_id: "j1".into(), since_ms, started }
+    }
+
+    #[test]
+    fn a_stale_finish_from_the_previous_job_is_not_this_jobs_outcome() {
+        assert_eq!(next_step(&job(false, 0), Some("FINISH"), 10_000), Step::Nothing);
+        assert_eq!(next_step(&job(false, 0), Some("FAILED"), 10_000), Step::Nothing);
+        assert_eq!(next_step(&job(false, 0), Some("IDLE"), 10_000), Step::Nothing);
+    }
+
+    #[test]
+    fn running_marks_it_started_and_then_finish_finishes_it() {
+        assert_eq!(next_step(&job(false, 0), Some("RUNNING"), 10_000), Step::Started);
+        assert_eq!(next_step(&job(true, 0), Some("RUNNING"), 10_000), Step::Nothing);
+        assert_eq!(next_step(&job(true, 0), Some("PAUSE"), 10_000), Step::Nothing);
+        assert_eq!(next_step(&job(true, 0), Some("FINISH"), 10_000), Step::Finished);
+    }
+
+    #[test]
+    fn a_fault_or_a_stop_after_starting_is_a_failure() {
+        assert!(matches!(next_step(&job(true, 0), Some("FAILED"), 1), Step::Failed(_)));
+        assert!(matches!(next_step(&job(true, 0), Some("IDLE"), 1), Step::Failed(_)));
+    }
+
+    #[test]
+    fn a_printer_that_goes_quiet_leaves_the_job_alone() {
+        assert_eq!(next_step(&job(true, 0), None, 1), Step::Nothing);
+    }
+
+    #[test]
+    fn a_job_that_never_starts_is_eventually_given_up_on() {
+        assert_eq!(next_step(&job(false, 0), Some("IDLE"), NEVER_STARTED_AFTER_MS), Step::Nothing);
+        assert!(matches!(next_step(&job(false, 0), Some("IDLE"), NEVER_STARTED_AFTER_MS + 1), Step::Failed(_)));
+    }
+
+    #[test]
+    fn applying_steps_fills_the_outbox_and_clears_the_active_job() {
+        let dir = std::env::temp_dir().join(format!("continuum-jobs-test-{}", std::process::id()));
+        let store = Store::open(&dir.join("s.db")).unwrap();
+        accepted(&store, "p1", "j1");
+        let active = store.active_jobs().unwrap().remove(0);
+        apply(&store, &active, Step::Started);
+        let active = store.active_jobs().unwrap().remove(0);
+        assert!(active.started);
+        apply(&store, &active, Step::Finished);
+        assert!(store.active_jobs().unwrap().is_empty());
+        let events: Vec<_> = store.outbox_pending().unwrap().into_iter().map(|e| e.event).collect();
+        assert_eq!(events, ["started", "finished"]);
+        let _ = std::fs::remove_dir_all(dir);
+    }
+}

src/lib.rs

@@ -15,6 +15,7 @@ pub mod config;
 pub mod config_import;
 pub mod discovery;
 pub mod fleet;
+pub mod jobs;
 pub mod printer;
 pub mod provision;
 pub mod store;

src/main.rs

@@ -13,7 +13,7 @@ use continuum_proxy::bootstrap::{self, Bootstrap};
 use continuum_proxy::config::Config;
 use continuum_proxy::config_import;
 use continuum_proxy::printer::{GenericPrinter, PrinterHandle};
-use continuum_proxy::{camera, completion, fleet, uplink};
+use continuum_proxy::{camera, completion, fleet, jobs, uplink};
 
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
@@ -177,6 +177,7 @@ async fn main() -> anyhow::Result<()> {
             restart_go2rtc.clone(),
         ) => {}
         _ = completion::run(fleet.clone()) => {}
+        _ = jobs::run(fleet.clone(), store.clone()) => {}
         // Returns only after installing a new config from a stick inserted
         // while running. Settings are read once at startup, so the process
         // ends here and relies on its supervisor (systemd `Restart=always`

src/store.rs

@@ -42,6 +42,22 @@ impl Store {
             "CREATE TABLE IF NOT EXISTS printers (
                 id   TEXT PRIMARY KEY,
                 json TEXT NOT NULL
+            );
+            -- What happened to jobs the cloud handed this gateway, until the
+            -- cloud has acknowledged it (see `jobs`). Survives a restart.
+            CREATE TABLE IF NOT EXISTS outbox (
+                id     INTEGER PRIMARY KEY AUTOINCREMENT,
+                job_id TEXT NOT NULL,
+                event  TEXT NOT NULL,
+                detail TEXT,
+                at_ms  INTEGER NOT NULL
+            );
+            -- The job each printer is working on, as far as this gateway knows.
+            CREATE TABLE IF NOT EXISTS active_jobs (
+                printer_id TEXT PRIMARY KEY,
+                job_id     TEXT NOT NULL,
+                since_ms   INTEGER NOT NULL,
+                started    INTEGER NOT NULL DEFAULT 0
             );",
         )?;
         Ok(Self { conn: Mutex::new(conn) })
@@ -78,6 +94,97 @@ impl Store {
     }
 }
 
+/// A job outcome waiting for the cloud's acknowledgement.
+#[derive(Debug, Clone, PartialEq)]
+pub struct OutboxEvent {
+    pub id: i64,
+    pub job_id: String,
+    pub event: String,
+    pub detail: Option<String>,
+    pub at_ms: i64,
+}
+
+/// A job a printer was handed and has not yet finished with.
+#[derive(Debug, Clone, PartialEq)]
+pub struct ActiveJob {
+    pub printer_id: String,
+    pub job_id: String,
+    pub since_ms: i64,
+    pub started: bool,
+}
+
+impl Store {
+    /// Queues an outcome to tell the cloud. Returns its id.
+    pub fn outbox_push(&self, job_id: &str, event: &str, detail: Option<&str>, at_ms: i64) -> anyhow::Result<i64> {
+        let conn = self.conn.lock().expect("store lock poisoned");
+        conn.execute(
+            "INSERT INTO outbox (job_id, event, detail, at_ms) VALUES (?1, ?2, ?3, ?4)",
+            params![job_id, event, detail, at_ms],
+        )?;
+        Ok(conn.last_insert_rowid())
+    }
+
+    /// Everything the cloud has not acknowledged, oldest first.
+    pub fn outbox_pending(&self) -> anyhow::Result<Vec<OutboxEvent>> {
+        let conn = self.conn.lock().expect("store lock poisoned");
+        let mut statement =
+            conn.prepare("SELECT id, job_id, event, detail, at_ms FROM outbox ORDER BY id LIMIT 500")?;
+        let rows = statement.query_map([], |row| {
+            Ok(OutboxEvent {
+                id: row.get(0)?,
+                job_id: row.get(1)?,
+                event: row.get(2)?,
+                detail: row.get(3)?,
+                at_ms: row.get(4)?,
+            })
+        })?;
+        Ok(rows.collect::<Result<_, _>>()?)
+    }
+
+    pub fn outbox_ack(&self, id: i64) -> anyhow::Result<()> {
+        let conn = self.conn.lock().expect("store lock poisoned");
+        conn.execute("DELETE FROM outbox WHERE id = ?1", params![id])?;
+        Ok(())
+    }
+
+    /// Records that `printer_id` was handed `job_id`, replacing any earlier job.
+    pub fn active_job_set(&self, printer_id: &str, job_id: &str, now_ms: i64) -> anyhow::Result<()> {
+        let conn = self.conn.lock().expect("store lock poisoned");
+        conn.execute(
+            "INSERT INTO active_jobs (printer_id, job_id, since_ms, started) VALUES (?1, ?2, ?3, 0)
+             ON CONFLICT(printer_id) DO UPDATE SET job_id = excluded.job_id, since_ms = excluded.since_ms, started = 0",
+            params![printer_id, job_id, now_ms],
+        )?;
+        Ok(())
+    }
+
+    pub fn active_job_started(&self, printer_id: &str) -> anyhow::Result<()> {
+        let conn = self.conn.lock().expect("store lock poisoned");
+        conn.execute("UPDATE active_jobs SET started = 1 WHERE printer_id = ?1", params![printer_id])?;
+        Ok(())
+    }
+
+    pub fn active_job_clear(&self, printer_id: &str) -> anyhow::Result<()> {
+        let conn = self.conn.lock().expect("store lock poisoned");
+        conn.execute("DELETE FROM active_jobs WHERE printer_id = ?1", params![printer_id])?;
+        Ok(())
+    }
+
+    pub fn active_jobs(&self) -> anyhow::Result<Vec<ActiveJob>> {
+        let conn = self.conn.lock().expect("store lock poisoned");
+        let mut statement = conn.prepare("SELECT printer_id, job_id, since_ms, started FROM active_jobs")?;
+        let rows = statement.query_map([], |row| {
+            Ok(ActiveJob {
+                printer_id: row.get(0)?,
+                job_id: row.get(1)?,
+                since_ms: row.get(2)?,
+                started: row.get::<_, i64>(3)? != 0,
+            })
+        })?;
+        Ok(rows.collect::<Result<_, _>>()?)
+    }
+}
+
 /// Creates the file empty and owner-only if it does not exist yet.
 #[cfg(unix)]
 fn create_private(path: &Path) -> anyhow::Result<()> {
@@ -196,4 +303,36 @@ mod tests {
         std::fs::write(&path, b"this is not a sqlite database at all, just text").unwrap();
         assert!(Store::open(&path).is_err());
     }
+
+    #[test]
+    fn an_unacknowledged_event_survives_a_restart_and_an_acknowledged_one_does_not() {
+        let (_dir, path) = temp_db();
+        let store = Store::open(&path).unwrap();
+        let first = store.outbox_push("job-1", "finished", None, 10).unwrap();
+        let second = store.outbox_push("job-2", "failed", Some("download failed"), 20).unwrap();
+        store.outbox_ack(first).unwrap();
+        drop(store);
+
+        let reopened = Store::open(&path).unwrap();
+        let pending = reopened.outbox_pending().unwrap();
+        assert_eq!(pending.len(), 1);
+        assert_eq!(pending[0].id, second);
+        assert_eq!(pending[0].detail.as_deref(), Some("download failed"));
+    }
+
+    #[test]
+    fn a_printer_has_one_active_job_and_a_new_one_replaces_it() {
+        let (_dir, path) = temp_db();
+        let store = Store::open(&path).unwrap();
+        store.active_job_set("p1", "job-1", 100).unwrap();
+        store.active_job_started("p1").unwrap();
+        assert!(store.active_jobs().unwrap()[0].started);
+
+        store.active_job_set("p1", "job-2", 200).unwrap();
+        let jobs = store.active_jobs().unwrap();
+        assert_eq!(jobs, vec![ActiveJob { printer_id: "p1".into(), job_id: "job-2".into(), since_ms: 200, started: false }]);
+
+        store.active_job_clear("p1").unwrap();
+        assert!(store.active_jobs().unwrap().is_empty());
+    }
 }

src/uplink/mod.rs

@@ -33,6 +33,7 @@ pub mod protocol;
 
 pub use protocol::{ClientMessage, DiscoveredPrinter, PrinterCommand, PrinterTelemetry, ServerMessage};
 
+use std::collections::HashSet;
 use std::sync::Arc;
 use std::time::Duration;
 
@@ -82,6 +83,8 @@ const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(15);
 /// a lot of traffic to describe a machine whose state changes on the scale of
 /// a layer, so this samples the cached report instead of chasing every push.
 const TELEMETRY_INTERVAL: Duration = Duration::from_secs(5);
+/// How often the job-event outbox is looked at for anything new to send.
+const OUTBOX_INTERVAL: Duration = Duration::from_secs(3);
 
 /// The printer fleet, shared between `main` (which owns it and keeps every
 /// connection alive) and this module (which reads status and dispatches
@@ -225,9 +228,34 @@ async fn serve(
     let mut heartbeat = tokio::time::interval(HEARTBEAT_INTERVAL);
     let mut telemetry = tokio::time::interval(TELEMETRY_INTERVAL);
     let mut seq: u64 = 0;
+    // Outbox events already written to THIS socket. Per connection on
+    // purpose: after a reconnect everything the cloud has not acknowledged
+    // goes out again, which is the point of keeping them.
+    let mut outbox_tick = tokio::time::interval(OUTBOX_INTERVAL);
+    let mut sent: HashSet<i64> = HashSet::new();
 
     loop {
         tokio::select! {
+            _ = outbox_tick.tick() => {
+                let Some(store) = store else { continue };
+                let pending = match store.outbox_pending() {
+                    Ok(pending) => pending,
+                    Err(err) => { warn!(%err, "could not read the job-event outbox"); continue }
+                };
+                let unsent: Vec<_> = pending.into_iter().filter(|e| !sent.contains(&e.id)).collect();
+                for event in unsent {
+                    let msg = ClientMessage::JobEvent {
+                        event_id: event.id,
+                        job_id: event.job_id,
+                        event: event.event,
+                        detail: event.detail,
+                        at_ms: event.at_ms,
+                    };
+                    write.send(Message::Text(serde_json::to_string(&msg)?)).await?;
+                    sent.insert(event.id);
+                }
+            }
+
             _ = heartbeat.tick() => {
                 seq += 1;
                 let msg = ClientMessage::Heartbeat { seq };
@@ -275,6 +303,7 @@ async fn serve(
                                         command,
                                         command_id.clone(),
                                         outbound.clone(),
+                                        store.cloned(),
                                     );
                                     ClientMessage::CommandResult {
                                         command_id, ok: true, error: None,
@@ -328,6 +357,14 @@ async fn serve(
                             Ok(ServerMessage::ProvisionPrinter { request_id, printer }) => {
                                 spawn_provision(fleet.clone(), store.cloned(), request_id, Some(printer), outbound.clone());
                             }
+                            Ok(ServerMessage::JobEventAck { event_id }) => {
+                                sent.remove(&event_id);
+                                if let Some(store) = store
+                                    && let Err(err) = store.outbox_ack(event_id)
+                                {
+                                    warn!(%err, event_id, "could not clear an acknowledged job event");
+                                }
+                            }
                             Ok(ServerMessage::RemovePrinter { request_id, printer_id }) => {
                                 spawn_remove(fleet.clone(), store.cloned(), request_id, printer_id, outbound.clone());
                             }
@@ -460,10 +497,31 @@ fn spawn_long_command(
     command: PrinterCommand,
     command_id: String,
     outbound: tokio::sync::mpsc::Sender<ClientMessage>,
+    store: Option<Arc<Store>>,
 ) {
+    // The job is followed from here on, so its outcome is reported even if
+    // the cloud is unreachable when it happens (see `jobs`).
+    let job_id = match (&command, &store) {
+        (PrinterCommand::StartPrint { job_id, .. }, Some(store)) => {
+            crate::jobs::accepted(store, &printer_id, job_id);
+            Some(job_id.clone())
+        }
+        _ => None,
+    };
     tokio::spawn(async move {
         match dispatch(&fleet, &printer_id, command).await {
             Ok(()) => info!(%printer_id, "long-running command finished"),
+            Err(err) if job_id.is_some() => {
+                // Still logged locally; the cloud hears of it from the
+                // outbox, which survives it being down right now.
+                tracing::error!(
+                    %printer_id, %err,
+                    "long-running command failed after the cloud was already told it was accepted"
+                );
+                if let (Some(store), Some(job_id)) = (&store, &job_id) {
+                    crate::jobs::start_failed(store, &printer_id, job_id, &err);
+                }
+            }
             Err(err) => {
                 // Still logged locally, and still at error level: the queue
                 // only reaches a cloud that is up, and this is the copy that

src/uplink/protocol.rs

@@ -32,6 +32,17 @@ pub enum ClientMessage {
         stream_base_url: Option<String>,
     },
     Heartbeat { seq: u64 },
+    /// What happened to a job the cloud handed this gateway. Kept in the
+    /// store and re-sent on every connection until a `JobEventAck`.
+    JobEvent {
+        event_id: i64,
+        job_id: String,
+        /// `started`, `finished` or `failed`.
+        event: String,
+        #[serde(skip_serializing_if = "Option::is_none")]
+        detail: Option<String>,
+        at_ms: i64,
+    },
     /// One snapshot per printer, on a timer. Sent as a batch rather than a
     /// message per printer: a farm has one uplink and many printers, and the
     /// cloud wants a coherent picture of the farm at an instant more than it
@@ -213,6 +224,8 @@ pub struct PrinterTelemetry {
 #[serde(tag = "type", rename_all = "snake_case")]
 pub enum ServerMessage {
     HelloAck { session_id: String },
+    /// The cloud has dealt with a `JobEvent`; the gateway can forget it.
+    JobEventAck { event_id: i64 },
     /// The handshake was rejected - a bad, expired or wrong-farm token. The
     /// backend closes the connection right after sending this.
     ///

tests/uplink_outbox.rs

@@ -0,0 +1,183 @@
+//! Job outcomes survive the cloud being away: an event in the gateway's store
+//! is sent on connect, sent again after a reconnect if nobody acknowledged it,
+//! and gone once acknowledged. A start that fails after being accepted is
+//! queued the same way instead of being sent once to whoever is listening.
+//!
+//! Real `uplink::run` against a real websocket server; no printer is touched
+//! (the fleet is empty, so the start fails on its lookup before any download).
+
+use std::sync::Arc;
+use std::time::Duration;
+
+use futures_util::{SinkExt, StreamExt};
+use serde_json::{json, Value};
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::net::TcpListener;
+use tokio::sync::Mutex;
+use tokio_tungstenite::WebSocketStream;
+use tokio_tungstenite::tungstenite::Message;
+use tokio_tungstenite::tungstenite::handshake::server::{Request, Response};
+
+use continuum_proxy::config::Config;
+use continuum_proxy::store::Store;
+use continuum_proxy::uplink;
+
+/// Long enough for a loopback round trip, short enough that a regression
+/// fails the test instead of hanging the suite.
+const WAIT: Duration = Duration::from_secs(10);
+
+fn config(port: u16) -> Config {
+    Config {
+        gateway_id: "test-gw".into(),
+        farm_id: "test-farm".into(),
+        gateway_token: "test-token".into(),
+        uplink_url: format!("ws://127.0.0.1:{port}/ws/edge/v1"),
+        // Nothing here starts go2rtc or reads the fleet file - `uplink::run`
+        // is handed its fleet rather than loading one - but Config has no
+        // Default, so these have to say something.
+        go2rtc_bin: "/nonexistent-go2rtc".into(),
+        go2rtc_config: "/nonexistent/go2rtc.yaml".into(),
+        go2rtc_api_listen: "127.0.0.1:1984".into(),
+        stream_base_url: None,
+        printers_path: "/nonexistent/printers.toml".into(),
+        store_path: "/nonexistent/continuum.db".into(),
+        // Off, which is the default and also the only safe thing in a test:
+        // an enabled one would sweep whatever network the test host is on.
+        discovery: Default::default(),
+    }
+}
+
+/// The next frame as JSON. Heartbeats and telemetry share this socket, so
+/// callers filter; `interval` fires on its first tick, which means a
+/// heartbeat arrives immediately after the hello.
+async fn next_json<S>(ws: &mut WebSocketStream<S>) -> Value
+where
+    S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin,
+{
+    loop {
+        let frame = tokio::time::timeout(WAIT, ws.next())
+            .await
+            .expect("timed out waiting for a frame from the gateway")
+            .expect("the gateway closed the connection")
+            .expect("websocket error");
+        if let Message::Text(text) = frame {
+            return serde_json::from_str(&text).expect("the gateway sent invalid JSON");
+        }
+    }
+}
+
+/// Plays the cloud's `/auth/edge-token`: reads one HTTP request, checks that
+/// the gateway sent its farm and long-lived token, and answers with a JWT.
+/// Returns the request text so the caller can assert on it.
+async fn answer_token_exchange(listener: &TcpListener) -> String {
+    let (mut stream, _) = listener.accept().await.unwrap();
+    let mut request = Vec::new();
+    let mut buf = [0u8; 4096];
+    loop {
+        let n = tokio::time::timeout(WAIT, stream.read(&mut buf)).await.unwrap().unwrap();
+        assert!(n > 0, "the gateway hung up before finishing its token request");
+        request.extend_from_slice(&buf[..n]);
+        let text = String::from_utf8_lossy(&request).into_owned();
+        if let Some((head, body)) = text.split_once("\r\n\r\n") {
+            let length = head
+                .lines()
+                .find_map(|l| l.to_ascii_lowercase().strip_prefix("content-length:").map(|v| v.trim().parse::<usize>().unwrap()))
+                .unwrap_or(0);
+            if body.len() >= length {
+                break;
+            }
+        }
+    }
+    let body = r#"{"token":"minted-edge-jwt","expiresInSeconds":3600}"#;
+    stream
+        .write_all(format!("HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", body.len()).as_bytes())
+        .await
+        .unwrap();
+    String::from_utf8_lossy(&request).into_owned()
+}
+
+
+/// Answers the token exchange and accepts the websocket, returning it after
+/// the gateway's hello.
+async fn accept(listener: &TcpListener) -> WebSocketStream<tokio::net::TcpStream> {
+    answer_token_exchange(listener).await;
+    let (stream, _) = listener.accept().await.unwrap();
+    let mut ws = tokio_tungstenite::accept_hdr_async(stream, |_: &Request, response: Response| Ok(response)).await.unwrap();
+    let hello = next_json(&mut ws).await;
+    assert_eq!(hello["type"], "hello");
+    ws
+}
+
+/// Reads frames until `want` job events have arrived, returning them by job id.
+async fn job_events(ws: &mut WebSocketStream<tokio::net::TcpStream>, want: usize) -> Vec<Value> {
+    let mut events = Vec::new();
+    while events.len() < want {
+        let msg = next_json(ws).await;
+        if msg["type"] == "job_event" {
+            events.push(msg);
+        }
+    }
+    events.sort_by_key(|e| e["event_id"].as_i64());
+    events
+}
+
+#[tokio::test]
+async fn unacknowledged_job_events_are_resent_after_a_reconnect_and_stop_once_acknowledged() {
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let port = listener.local_addr().unwrap().port();
+
+    let dir = std::env::temp_dir().join(format!("continuum-outbox-test-{}", std::process::id()));
+    let store = Arc::new(Store::open(&dir.join("s.db")).unwrap());
+    // An outcome reached while the cloud was away.
+    store.outbox_push("11111111-1111-4111-8111-111111111111", "finished", None, 1_000).unwrap();
+
+    let fleet: uplink::SharedFleet = Arc::new(Mutex::new(Vec::new()));
+    let (outbound, inbox) = uplink::channel();
+    tokio::spawn(uplink::run(config(port), fleet, Some(store.clone()), outbound, inbox));
+
+    let mut ws = accept(&listener).await;
+    let first = job_events(&mut ws, 1).await;
+    assert_eq!(first[0]["job_id"], "11111111-1111-4111-8111-111111111111");
+    assert_eq!(first[0]["event"], "finished");
+    assert_eq!(first[0]["at_ms"], 1_000);
+
+    // A start that fails after being accepted goes through the same outbox.
+    ws.send(Message::Text(
+        json!({
+            "type": "command", "command_id": "cmd-1", "printer_id": "not-in-this-fleet",
+            "command": { "action": "start_print", "job_id": "22222222-2222-4222-8222-222222222222",
+                         "file_name": "cube.gcode", "url": "http://127.0.0.1:1/never-fetched" }
+        })
+        .to_string(),
+    ))
+    .await
+    .unwrap();
+    // Only the new one: the first was already sent on this connection.
+    let mut both = first.clone();
+    both.extend(job_events(&mut ws, 1).await);
+    assert_eq!(both[1]["job_id"], "22222222-2222-4222-8222-222222222222");
+    assert_eq!(both[1]["event"], "failed");
+    assert!(both[1]["detail"].as_str().unwrap().contains("not-in-this-fleet"), "{}", both[1]);
+
+    // The cloud never answers; the connection drops.
+    drop(ws);
+
+    let mut ws = accept(&listener).await;
+    let again = job_events(&mut ws, 2).await;
+    assert_eq!(
+        again.iter().map(|e| e["event_id"].clone()).collect::<Vec<_>>(),
+        both.iter().map(|e| e["event_id"].clone()).collect::<Vec<_>>(),
+        "after a reconnect everything unacknowledged is sent again"
+    );
+
+    ws.send(Message::Text(json!({ "type": "job_event_ack", "event_id": again[0]["event_id"] }).to_string()))
+        .await
+        .unwrap();
+    let deadline = tokio::time::Instant::now() + WAIT;
+    while store.outbox_pending().unwrap().len() != 1 {
+        assert!(tokio::time::Instant::now() < deadline, "the acknowledged event was not cleared");
+        tokio::time::sleep(Duration::from_millis(50)).await;
+    }
+    assert_eq!(store.outbox_pending().unwrap()[0].job_id, "22222222-2222-4222-8222-222222222222");
+    let _ = std::fs::remove_dir_all(dir);
+}