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.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

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<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");
+        }
+    }
+}
+
+#[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}"
+    );
+}