From 8f7a332956b9efb24e416004b219e8abcc423f7f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Iwo=20Strzebo=C5=84ski?= Date: Sat, 3 Oct 2026 10:37:24 +0000 Subject: [PATCH] Make the uplink match the written protocol, and test that it does Adds tests/uplink_protocol.rs, the gateway half of a contract test whose other half is continuum-backend/tests/edge_protocol.test.ts. It checks that every message the backend sends is one this gateway can read, and that what this gateway writes is what the schema says it writes. It found two things immediately. Telemetry was serializing empty fields as explicit nulls. serde emits `"job_name": null` for an empty Option unless told otherwise, and the backend's validator treats a missing key and an explicit null as different things - so every batch containing an idle printer, which by definition has no job name, would have been rejected whole. Every optional field now has skip_serializing_if, confirmed against a live capture: an idle P1P's snapshot has no nulls in it at all. ServerMessage had no variant for the handshake being rejected. A bad or expired token would have arrived as "unparseable message from cloud, ignoring" and then been retried every five seconds forever, which in the logs is indistinguishable from a network fault. HelloError is now a real variant, logged at error level naming the env var to go fix, and it stops the connection attempt instead of looping on it. Also exposes config and uplink from lib.rs so the test can reach them, same reason fleet and printer are already there. Verified live against a P1P and an H2C for the whole run, including the H2C finishing its print: RUNNING 98% -> FINISH 100% layer 50/50 with the nozzle cooling 84C -> 76C, all of it arriving through the uplink, no warnings. Co-Authored-By: Claude Opus 5 --- src/lib.rs | 2 + src/uplink/mod.rs | 7 +++ src/uplink/protocol.rs | 35 ++++++++++- tests/uplink_protocol.rs | 127 +++++++++++++++++++++++++++++++++++++++ 4 files changed, 170 insertions(+), 1 deletion(-) create mode 100644 tests/uplink_protocol.rs diff --git a/src/lib.rs b/src/lib.rs index 8e3e076..8adebf5 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -3,5 +3,7 @@ //! `examples/`) and future integration tests can `use continuum_proxy::...` //! without duplicating module declarations. +pub mod config; pub mod fleet; pub mod printer; +pub mod uplink; diff --git a/src/uplink/mod.rs b/src/uplink/mod.rs index 8656265..7226913 100644 --- a/src/uplink/mod.rs +++ b/src/uplink/mod.rs @@ -130,6 +130,13 @@ async fn connect_and_serve(config: &Config, fleet: &SharedFleet) -> anyhow::Resu }; write.send(Message::Text(serde_json::to_string(&msg)?)).await?; } + // Retrying will not help: the token is wrong, not + // the network. Logged at error level and returned + // so it is obvious which of the two it is. + Ok(ServerMessage::HelloError { message }) => { + tracing::error!(%message, "cloud rejected this gateway's handshake - check CONTINUUM_GATEWAY_TOKEN"); + return Ok(()); + } Ok(other) => info!(?other, "received from cloud"), // Loudly. This used to be an `if let Ok(..)` with // no else, which is the same silent-drop bug that diff --git a/src/uplink/protocol.rs b/src/uplink/protocol.rs index a2ecb02..2a00090 100644 --- a/src/uplink/protocol.rs +++ b/src/uplink/protocol.rs @@ -15,7 +15,12 @@ pub enum ClientMessage { /// which the *cloud* generates - unlike Bambu's `sequence_id`, we're /// both ends of this protocol, so the side that needs the correlation /// gets to choose it. - CommandResult { command_id: String, ok: bool, error: Option }, + CommandResult { + command_id: String, + ok: bool, + #[serde(skip_serializing_if = "Option::is_none")] + error: Option, + }, } /// A flattened snapshot of one printer, in the terms an operator console @@ -30,6 +35,14 @@ pub enum ClientMessage { /// Every field is optional because an unreachable or just-connected printer /// genuinely has nothing to report, and "no data yet" is different from /// "zero". +/// +/// Absent fields are *omitted* rather than serialized as `null`, which is +/// what `skip_serializing_if` below is for. Caught by the contract test +/// against real captures: serde's default is to emit `"job_name": null`, and +/// the backend's validator reads a missing key and an explicit null as +/// different things, so an idle printer's entire telemetry batch was being +/// rejected. The backend now accepts either, but this end sends only the one +/// the schema describes. #[derive(Debug, Clone, Serialize)] pub struct PrinterTelemetry { /// Our own id from printers.toml, not the vendor serial - this is what @@ -40,21 +53,33 @@ pub struct PrinterTelemetry { /// through as the printer's own string rather than narrowed to an enum, /// so an unrecognised state reaches the cloud as itself instead of being /// dropped on the floor here. + #[serde(skip_serializing_if = "Option::is_none")] pub state: Option, + #[serde(skip_serializing_if = "Option::is_none")] pub percent: Option, + #[serde(skip_serializing_if = "Option::is_none")] pub layer: Option, + #[serde(skip_serializing_if = "Option::is_none")] pub total_layers: Option, + #[serde(skip_serializing_if = "Option::is_none")] pub remaining_minutes: Option, + #[serde(skip_serializing_if = "Option::is_none")] pub job_name: Option, + #[serde(skip_serializing_if = "Option::is_none")] pub nozzle_temp: Option, + #[serde(skip_serializing_if = "Option::is_none")] pub nozzle_target: Option, + #[serde(skip_serializing_if = "Option::is_none")] pub bed_temp: Option, + #[serde(skip_serializing_if = "Option::is_none")] pub bed_target: Option, /// Non-zero means the printer is reporting a fault. + #[serde(skip_serializing_if = "Option::is_none")] pub print_error: Option, /// How many Health-Management-System entries the printer currently has /// active. Just the count for now - decoding the `{attr, code}` pairs /// into messages needs Bambu's HMS tables, which we don't have yet. + #[serde(skip_serializing_if = "Option::is_none")] pub active_alerts: Option, } @@ -63,6 +88,14 @@ pub struct PrinterTelemetry { #[serde(tag = "type", rename_all = "snake_case")] pub enum ServerMessage { HelloAck { session_id: String }, + /// The handshake was rejected - a bad, expired or wrong-farm token. The + /// backend closes the connection right after sending this. + /// + /// Worth a variant of its own rather than letting it fall into the + /// unparseable-message branch: retrying a rejected token every few seconds + /// forever looks exactly like a network problem in the logs otherwise, and + /// it is the one uplink failure a human has to go fix by hand. + HelloError { message: String }, HeartbeatAck { seq: u64 }, /// Do something to one printer. `printer_id` is our own id from /// printers.toml. diff --git a/tests/uplink_protocol.rs b/tests/uplink_protocol.rs new file mode 100644 index 0000000..4b73bda --- /dev/null +++ b/tests/uplink_protocol.rs @@ -0,0 +1,127 @@ +//! The other half of the `/ws/edge/v1` contract test. +//! +//! `continuum-backend/tests/edge_protocol.test.ts` checks that what this +//! gateway sends is what the backend accepts. This checks the reverse: that +//! every message the backend sends is one this gateway can actually read, and +//! that what this gateway writes matches the schema it claims to follow. +//! +//! Worth having on both sides because the two ends had drifted into entirely +//! different protocols - a handshake the gateway never sent, telemetry shaped +//! one printer per message instead of a batch, printer ids a gateway has no +//! way to know - and nothing failed anywhere. Both test files are hand-kept +//! mirrors; the canonical shape is +//! `continuum-schemas/schemas/continuum/v1/edge/`. + +use continuum_proxy::uplink::{ClientMessage, PrinterCommand, PrinterTelemetry, ServerMessage}; + +/// Verbatim from the backend's `src/routes/ws/edge.ts`. If the backend +/// changes one of these, this test is where it should break. +const BACKEND_MESSAGES: &[&str] = &[ + r#"{"type":"hello_ack","session_id":"3f2b1c5e-0000-4000-8000-000000000000"}"#, + r#"{"type":"hello_error","message":"invalid edge token"}"#, + r#"{"type":"heartbeat_ack","seq":7}"#, + r#"{"type":"command","command_id":"9a7d2f10-0000-4000-8000-000000000000","printer_id":"p1p","command":{"action":"set_fan_speed","percent":40}}"#, + r#"{"type":"command","command_id":"9a7d2f11-0000-4000-8000-000000000000","printer_id":"p1p","command":{"action":"send_gcode_line","line":"M106 S0"}}"#, +]; + +#[test] +fn every_backend_message_deserializes() { + for raw in BACKEND_MESSAGES { + let parsed: Result = serde_json::from_str(raw); + assert!(parsed.is_ok(), "backend sends this and the gateway cannot read it:\n {raw}\n {:?}", parsed.err()); + } +} + +#[test] +fn a_command_arrives_with_its_payload_intact() { + let raw = r#"{"type":"command","command_id":"cmd-1","printer_id":"p1p","command":{"action":"set_fan_speed","percent":40}}"#; + match serde_json::from_str::(raw).expect("should parse") { + ServerMessage::Command { command_id, printer_id, command } => { + assert_eq!(command_id, "cmd-1"); + // The gateway's own id from printers.toml, not a cloud UUID - the + // backend translates on the way out. + assert_eq!(printer_id, "p1p"); + match command { + PrinterCommand::SetFanSpeed { percent } => assert_eq!(percent, 40), + other => panic!("wrong command variant: {other:?}"), + } + } + other => panic!("wrong message variant: {other:?}"), + } +} + +/// The variant that is deliberately absent. When this test starts failing +/// because someone added it, that is the moment to check the print-start path +/// has actually been run against real hardware first. +#[test] +fn there_is_no_start_print_command() { + let raw = r#"{"type":"command","command_id":"x","printer_id":"p1p","command":{"action":"start_print","file_name":"plate_1.3mf"}}"#; + assert!( + serde_json::from_str::(raw).is_err(), + "a start-print command parsed - has the hardware test been done?" + ); +} + +#[test] +fn unknown_messages_are_rejected_rather_than_silently_accepted() { + // The old protocol. A gateway that accepted this would ack a handshake + // that never happened. + assert!(serde_json::from_str::(r#"{"type":"auth_ok","farmId":"f"}"#).is_err()); +} + +/// The bug this test exists for: serde emits `"job_name": null` for an empty +/// `Option` unless told not to, and the backend's validator reads a missing +/// key and an explicit null as different things. An idle printer has no job +/// name, so every batch containing one was rejected whole. +#[test] +fn empty_telemetry_fields_are_omitted_not_null() { + let msg = ClientMessage::Telemetry { + printers: vec![PrinterTelemetry { + printer_id: "p1p".into(), + name: "P1P".into(), + state: Some("IDLE".into()), + percent: Some(0), + layer: Some(0), + total_layers: Some(0), + remaining_minutes: Some(0), + job_name: None, + nozzle_temp: Some(23.375), + nozzle_target: Some(0.0), + bed_temp: Some(22.40625), + bed_target: Some(0.0), + print_error: Some(0), + active_alerts: Some(0), + }], + }; + + let json = serde_json::to_string(&msg).expect("should serialize"); + assert!(!json.contains("null"), "telemetry should omit empty fields, got:\n {json}"); + assert!(!json.contains("job_name"), "an absent job name should not appear at all"); + // The fields that *do* have values must still be there - an assertion on + // the absence of "null" alone would pass on an empty object. + assert!(json.contains(r#""state":"IDLE""#)); + assert!(json.contains(r#""bed_temp":22.40625"#)); +} + +#[test] +fn a_successful_command_result_omits_the_error() { + let json = serde_json::to_string(&ClientMessage::CommandResult { + command_id: "cmd-1".into(), + ok: true, + error: None, + }) + .expect("should serialize"); + assert_eq!(json, r#"{"type":"command_result","command_id":"cmd-1","ok":true}"#); +} + +#[test] +fn a_failed_command_result_carries_the_reason() { + let json = serde_json::to_string(&ClientMessage::CommandResult { + command_id: "cmd-1".into(), + ok: false, + error: Some("no printer with id 'nope' in this gateway's fleet".into()), + }) + .expect("should serialize"); + assert!(json.contains(r#""ok":false"#)); + assert!(json.contains("no printer with id 'nope'")); +}