continuum-proxy/tests/uplink_late_result.rs

//! 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}"
    );
}