diff --git a/src/jobs.rs b/src/jobs.rs new file mode 100644 index 0000000..df37c19 --- /dev/null +++ b/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) { + 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)> = { + 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); + } +} diff --git a/src/lib.rs b/src/lib.rs index 9c63491..31015dc 100644 --- a/src/lib.rs +++ b/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; diff --git a/src/main.rs b/src/main.rs index a3673d0..6358af7 100644 --- a/src/main.rs +++ b/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` diff --git a/src/store.rs b/src/store.rs index 082b84f..a27c7a0 100644 --- a/src/store.rs +++ b/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, + 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 { + 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> { + 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::>()?) + } + + 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> { + 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::>()?) + } +} + /// 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()); + } } diff --git a/src/uplink/mod.rs b/src/uplink/mod.rs index 78fe94c..567eb6b 100644 --- a/src/uplink/mod.rs +++ b/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 = 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, + store: Option>, ) { + // 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 diff --git a/src/uplink/protocol.rs b/src/uplink/protocol.rs index 950b35c..b377882 100644 --- a/src/uplink/protocol.rs +++ b/src/uplink/protocol.rs @@ -32,6 +32,17 @@ pub enum ClientMessage { stream_base_url: Option, }, 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, + 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. /// diff --git a/tests/uplink_outbox.rs b/tests/uplink_outbox.rs new file mode 100644 index 0000000..6ffe0e7 --- /dev/null +++ b/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(ws: &mut WebSocketStream) -> 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::().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 { + 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, want: usize) -> Vec { + 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::>(), + both.iter().map(|e| e["event_id"].clone()).collect::>(), + "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); +}