From 7a0201c0dea7fb4ccd383342a7762531f0f24d07 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Iwo=20Strzebo=C5=84ski?= Date: Sun, 18 Oct 2026 21:52:05 +0000 Subject: [PATCH] Test that a late command failure reaches the cloud The outbound queue had no coverage, and it is exactly the kind of code that looks right and does nothing: the failure path only runs minutes after the command was answered, so nothing in the ordinary flow would notice it was missing. Stands up a real websocket server and runs the real `uplink::run` against it, asserting two results for one command_id in order - ok:true on acceptance, then ok:false carrying a reason that names what went wrong. That the id repeats is the contract, not a bug to deduplicate. It touches no printer and makes no outbound request: the fleet is empty and `dispatch` checks the fleet before downloading, so the command fails on a lookup. The unreachable URL is on port 1 so a regression that did reach the network fails fast rather than hanging on a real host. Verified to fail without the feature, not just to pass with it: stubbing out the send makes it time out waiting for the second result. That check is worth doing by hand here - the go2rtc RTSP test asserted the absence of a config key and passed the whole time the server was open. --- tests/uplink_late_result.rs | 129 ++++++++++++++++++++++++++++++++++++ 1 file changed, 129 insertions(+) create mode 100644 tests/uplink_late_result.rs diff --git a/tests/uplink_late_result.rs b/tests/uplink_late_result.rs new file mode 100644 index 0000000..065f390 --- /dev/null +++ b/tests/uplink_late_result.rs @@ -0,0 +1,129 @@ +//! Does a command that outlives the answer already given for it report back? +//! +//! A `start_print` is acknowledged the moment it is accepted, because the +//! transfer takes minutes and the cloud gives up after ten seconds. So its +//! real outcome can only ever arrive late, and before the outbound queue +//! existed there was nowhere for it to arrive from: the uplink's `select!` +//! loop was the only writer and a spawned task could not reach the socket. +//! A failed start therefore reported success and the reason stayed in this +//! gateway's log, on a machine in someone else's workshop. +//! +//! This stands up a real websocket server and runs the real `uplink::run` +//! against it. **It touches no printer and makes no outbound request**: the +//! fleet is empty, and `dispatch` checks the fleet for the printer id +//! *before* downloading anything, so the command fails on a lookup. + +use std::sync::Arc; +use std::time::Duration; + +use futures_util::{SinkExt, StreamExt}; +use serde_json::{json, Value}; +use tokio::net::TcpListener; +use tokio::sync::Mutex; +use tokio_tungstenite::WebSocketStream; +use tokio_tungstenite::tungstenite::Message; + +use continuum_proxy::config::Config; +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(), + } +} + +/// 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"); + } + } +} + +#[tokio::test] +async fn a_long_command_that_fails_reports_its_reason_after_the_fact() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + + let fleet: uplink::SharedFleet = Arc::new(Mutex::new(Vec::new())); + tokio::spawn(uplink::run(config(port), fleet)); + + let (stream, _) = listener.accept().await.unwrap(); + let mut ws = tokio_tungstenite::accept_async(stream).await.unwrap(); + + let hello = next_json(&mut ws).await; + assert_eq!(hello["type"], "hello", "a gateway introduces itself first"); + + ws.send(Message::Text( + json!({ + "type": "command", + "command_id": "cmd-1", + "printer_id": "not-in-this-fleet", + "command": { + "action": "start_print", + "job_id": "7c9e6679-0000-4000-8000-000000000000", + "file_name": "cube.gcode.3mf", + // Never fetched: the fleet lookup fails first. Port 1 so a + // regression that *did* reach the network fails fast and + // loudly rather than hanging on a real host. + "url": "http://127.0.0.1:1/never-fetched", + } + }) + .to_string(), + )) + .await + .unwrap(); + + let mut results = Vec::new(); + while results.len() < 2 { + let msg = next_json(&mut ws).await; + if msg["type"] == "command_result" { + results.push(msg); + } + } + + // Two results for one command, in this order, and that is the contract - + // not a bug to be deduplicated away. The first is "accepted, working on + // it", which is all the cloud ever trusted a start reply to mean; the + // second is what actually happened. + assert_eq!(results[0]["command_id"], "cmd-1"); + assert_eq!(results[0]["ok"], true, "a start is acknowledged on acceptance"); + assert!(results[0]["error"].is_null()); + + assert_eq!(results[1]["command_id"], "cmd-1", "the late result reuses the id"); + assert_eq!(results[1]["ok"], false); + + // The reason is the entire point of the queue: without it the cloud sees + // a job that never started printing and can say nothing about why. + let reason = results[1]["error"].as_str().expect("a failure carries its reason"); + assert!( + reason.contains("not-in-this-fleet"), + "the reason should name the printer that could not be found, got: {reason}" + ); +}