Files
continuum-proxy/tests/uplink_protocol.rs
T
octoturgeandClaude Opus 5 48672b1f45 Execute start_print, and never await it on the uplink task
The gateway can now run a cloud-dispatched print: download the job file from
the presigned URL in the command, then hand it to the existing
`upload_project_file`, which derives the filament mapping from the archive,
transfers it over FTPS and starts it.

## The download refuses to be a weapon

`src/uplink/download.rs` checks the HTTP status before reading a byte,
because an expired presigned URL answers 403 with an XML error document and
writing *that* to a printer as a project is the worst outcome available. It
rejects a `Content-Length` over 256 MiB before reading, and enforces the cap
while streaming as well, so a server that lies or omits the length cannot
make a small gateway box buffer until the OOM killer arrives. A byte count
that disagrees with `Content-Length` is an error: a `.3mf` is a zip, and a
truncated one loses the central directory the printer needs. HTTPS only, no
redirects, and the URL is stripped from error messages.

`PrinterCommand` no longer derives `Debug` for the same reason - `dispatch`
logs every command, and a presigned URL is a bearer credential until it
expires. It prints as `<redacted>`.

## Why it is spawned

`start_print` is the one command that cannot be answered inline, and the
reason is not latency. Everything else here is an MQTT publish returning in
milliseconds. This downloads tens of megabytes and then transfers them at a
measured 0.2 MiB/s - about four minutes for a 50 MB project - and awaiting
that inside the uplink's `select!` loop stops heartbeats and, far worse,
stops telemetry for the duration.

No telemetry is what makes spawning mandatory rather than tidy. The cloud now
derives a job's state from the telemetry that follows a start, so a gateway
that goes silent for four minutes during every job start withholds exactly
the information the job depends on - while the backend times the command out
after ten seconds and concludes the start failed. The job would be requeued
onto a printer that was busy printing it.

So the `ok` means "accepted, working on it", which is all a start reply was
ever trusted to mean. `runs_for_minutes` is an exhaustive match rather than a
`matches!`, so a new variant has to make that choice instead of defaulting to
the stall.

The gap this leaves is honest and recorded in the code: a failure after the
ack has no `CommandResult` to travel on, because answering later needs an
outbound channel this loop does not have. Until it does, the gateway's log is
where the reason lives and the backend reaps the job.

reqwest 0.12 with `rustls-tls-native-roots`, not 0.13: 0.13 pulls
`webpki-root-certs` under CDLA-Permissive-2.0, which is not on the allow
list, and `aws-lc` brings OpenSSL-licensed C. `cargo deny check licenses`
passes. 54 tests.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-10-15 22:53:47 +00:00

254 lines
12 KiB
Rust

//! 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::printer::PrintControl;
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"}}"#,
r#"{"type":"command","command_id":"9a7d2f12-0000-4000-8000-000000000000","printer_id":"p1p","command":{"action":"control_print","control":"pause"}}"#,
r#"{"type":"command","command_id":"9a7d2f13-0000-4000-8000-000000000000","printer_id":"p1p","command":{"action":"control_print","control":"resume"}}"#,
r#"{"type":"command","command_id":"9a7d2f14-0000-4000-8000-000000000000","printer_id":"p1p","command":{"action":"control_print","control":"stop"}}"#,
];
#[test]
fn every_backend_message_deserializes() {
for raw in BACKEND_MESSAGES {
let parsed: Result<ServerMessage, _> = 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::<ServerMessage>(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:?}"),
}
}
/// Each control verb has to arrive as the right `PrintControl`, not merely
/// parse. A silent mis-mapping here would send a stop where a pause was
/// asked for, which on a real machine is the difference between a print that
/// can be resumed and one that is over.
#[test]
fn each_print_control_verb_maps_to_its_own_action() {
for (verb, expected) in [
("pause", PrintControl::Pause),
("resume", PrintControl::Resume),
("stop", PrintControl::Stop),
] {
let raw = format!(
r#"{{"type":"command","command_id":"c","printer_id":"p1p","command":{{"action":"control_print","control":"{verb}"}}}}"#
);
match serde_json::from_str::<ServerMessage>(&raw).expect("should parse") {
ServerMessage::Command { command: PrinterCommand::ControlPrint { control }, .. } => {
assert_eq!(control, expected, "'{verb}' mapped to the wrong action");
// And the wire word the vendor layer sends has to match the
// wire word that arrived - `as_str` is what builds the Bambu
// command name.
assert_eq!(control.as_str(), verb);
}
other => panic!("wrong variant for '{verb}': {other:?}"),
}
}
}
/// "cancel" is the trap: it is Moonraker's word, and the Klipper
/// implementation behind `control_print` really does send a cancel - but the
/// wire word is "stop". A verb only one vendor recognises must be rejected
/// here rather than reaching a gateway that cannot map it.
#[test]
fn a_vendor_specific_control_verb_is_rejected() {
for verb in ["cancel", "abort", "Pause", ""] {
let raw = format!(
r#"{{"type":"command","command_id":"c","printer_id":"p1p","command":{{"action":"control_print","control":"{verb}"}}}}"#
);
assert!(
serde_json::from_str::<ServerMessage>(&raw).is_err(),
"'{verb}' parsed as a print control verb and should not have"
);
}
}
/// The schema's `StartPrint` variant, in the shape the backend dispatches.
/// Every field has to arrive intact: `url` is what gets downloaded and
/// `file_name` is what the printer is told to call it, so a swapped or
/// dropped one prints the wrong thing or nothing.
#[test]
fn a_start_print_command_arrives_with_its_fields_intact() {
let raw = r#"{"type":"command","command_id":"c","printer_id":"p1p","command":{"action":"start_print","job_id":"5b0e7c1a-0000-4000-8000-000000000000","file_name":"plate_1.3mf","url":"https://objects.example.com/jobs/abc.3mf?X-Amz-Signature=deadbeef&X-Amz-Expires=300"}}"#;
match serde_json::from_str::<ServerMessage>(raw).expect("should parse") {
ServerMessage::Command { command: PrinterCommand::StartPrint { job_id, file_name, url }, .. } => {
assert_eq!(job_id, "5b0e7c1a-0000-4000-8000-000000000000");
assert_eq!(file_name, "plate_1.3mf");
assert_eq!(
url,
"https://objects.example.com/jobs/abc.3mf?X-Amz-Signature=deadbeef&X-Amz-Expires=300"
);
}
other => panic!("wrong variant: {other:?}"),
}
}
/// The URL is a bearer credential until it expires, and `dispatch` logs
/// every command with `{:?}`.
#[test]
fn a_start_print_url_does_not_appear_in_debug_output() {
let raw = r#"{"type":"command","command_id":"c","printer_id":"p1p","command":{"action":"start_print","job_id":"j","file_name":"plate_1.3mf","url":"https://objects.example.com/x?X-Amz-Signature=deadbeef"}}"#;
let parsed = serde_json::from_str::<ServerMessage>(raw).expect("should parse");
let shown = format!("{parsed:?}");
assert!(!shown.contains("deadbeef"), "signature leaked into a log line: {shown}");
assert!(shown.contains("plate_1.3mf"), "the file name should still be logged: {shown}");
}
/// Each field is load-bearing and none has a sensible default: a command
/// missing its URL must be rejected here, not discovered at the printer.
#[test]
fn a_start_print_missing_any_required_field_is_rejected() {
let full = [
("job_id", r#""job_id":"5b0e7c1a-0000-4000-8000-000000000000""#),
("file_name", r#""file_name":"plate_1.3mf""#),
("url", r#""url":"https://objects.example.com/jobs/abc.3mf""#),
];
for missing in 0..full.len() {
let fields: Vec<&str> =
full.iter().enumerate().filter(|(i, _)| *i != missing).map(|(_, (_, f))| *f).collect();
let raw = format!(
r#"{{"type":"command","command_id":"c","printer_id":"p1p","command":{{"action":"start_print",{}}}}}"#,
fields.join(",")
);
assert!(
serde_json::from_str::<ServerMessage>(&raw).is_err(),
"start_print without '{}' parsed and should not have:\n {raw}",
full[missing].0
);
}
}
#[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::<ServerMessage>(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),
// An idle printer is in no stage, so this is the second field
// after `job_name` that must be omitted rather than nulled.
stage: None,
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"#));
assert!(!json.contains("stage"), "an absent stage should not appear at all");
}
/// A stage of 0 means "printing" and must survive the wire, because
/// `skip_serializing_if = "Option::is_none"` on an `Option<i32>` is one
/// `unwrap_or_default` away from turning "printing" into "no stage". The
/// distinction matters: a printer at 14% and layer 0 is mid-bed-levelling if
/// the stage is 54 and genuinely stuck if it is 0.
#[test]
fn a_zero_stage_is_sent_rather_than_treated_as_absent() {
let msg = ClientMessage::Telemetry {
printers: vec![PrinterTelemetry {
printer_id: "p1p".into(),
name: "P1P".into(),
state: Some("RUNNING".into()),
percent: Some(14),
stage: Some(0),
layer: Some(1),
total_layers: Some(150),
remaining_minutes: Some(39),
job_name: Some("cube-test".into()),
nozzle_temp: Some(255.2),
nozzle_target: Some(255.0),
bed_temp: Some(79.3),
bed_target: Some(80.0),
print_error: Some(0),
active_alerts: Some(0),
}],
};
let json = serde_json::to_string(&msg).expect("should serialize");
assert!(json.contains(r#""stage":0"#), "a zero stage must be on the wire, got:\n {json}");
}
#[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'"));
}