From 91fbcaf665d0d9ac978c9a180b075e5458f8ddd2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Iwo=20Strzebo=C5=84ski?= Date: Sat, 3 Oct 2026 10:22:22 +0000 Subject: [PATCH] Route telemetry up and commands down over the cloud uplink The proxy did not actually proxy. `fleet` was loaded and connected in main.rs and then never touched again, while the uplink said hello and sent heartbeats into the void - two halves of the program running side by side with nothing between them. Both directions are wired now. Up: every 5s, one vendor-neutral `PrinterTelemetry` snapshot per printer that has reported, batched into a single `Telemetry` message. The snapshot is a translation rather than a forwarded `print.push_status`, because it has to describe a Prusa too, and because Bambu's wire format has traps (0-15 fan gears, ints-or-floats temperatures) that should not reach the rest of the system. Down: a `Command` is looked up by our own printer id, executed, and answered with a `CommandResult`. `PrinterCommand` is deliberately small and deliberately has no "start a print" variant - `start_uploaded_file` has never been run against hardware, and the first thing to exercise it should be a human watching the machine, not a message from the internet. The fleet becomes `Arc>>` so main and the uplink can share it. One lock for the whole Vec is coarse - a slow command to one printer stalls telemetry for all of them - which is fine at a handful of printers and is the shape to revisit (a lock per printer, or an actor task each) before a farm gets large. Two things found by running this against the live P1P and H2C: - A v1 Bambu reached the cloud as a row of nulls with one real bed temperature in it, while the v2 H2C beside it was fully populated. v1 only sends a *complete* status when asked, and `pushall`'s own doc comment said to call it periodically - nothing ever did. Each telemetry tick now collects, then requests the next refresh, so v1 telemetry runs one tick behind instead of staying stuck on fragments. The quirk lives in `PrinterHandle::refresh_status` so the uplink does not have to know it. - The third instance of the silent-swallow parse bug, at the old uplink/mod.rs:63: `if let Ok(server_msg) = serde_json::from_str(..)` with no else. Same shape that hid broken `print.push_status` parsing for weeks, except here it would mean commands vanishing with the cloud never being told. Now a `match` with a loud `Err` arm. Verified end to end against both printers with a throwaway WebSocket server: H2C `RUNNING 88% layer 48/50 13min nozzle 230/230 bed 35/35`, P1P `IDLE 0% nozzle 23.5/0 bed 22.5/0`, both fan commands `ok=true`, zero warnings over 24s. --- README.md | 33 +++++--- src/main.rs | 30 ++++---- src/printer/mod.rs | 46 +++++++++++ src/uplink/mod.rs | 171 ++++++++++++++++++++++++++++++++++++++--- src/uplink/protocol.rs | 71 +++++++++++++++++ 5 files changed, 316 insertions(+), 35 deletions(-) diff --git a/README.md b/README.md index cbe08f1..46e83c1 100644 --- a/README.md +++ b/README.md @@ -48,22 +48,37 @@ cloud control plane (`continuum-backend`). (FTPS upload, fan speed). Requires an explicit printer id and refuses to run unless that printer reports `gcode_state: "IDLE"`. -## What's still a stub (cont.) +## What routes between the cloud and the printers + +The uplink and the printer fleet are wired together, in both directions +(`uplink/mod.rs`): + +- **Up:** every 5s, one `PrinterTelemetry` snapshot per printer that has + reported, batched into a single `Telemetry` message. Verified live against + a P1P and an H2C. +- **Down:** a `Command` from the cloud is looked up by our own printer id, + executed, and answered with a `CommandResult`. The set of commands is + deliberately small - a fan speed and a single gcode line - and deliberately + has no "start a print" variant; see `PrinterCommand`. + +One quirk worth knowing: a Bambu on protocol v1 only sends a *complete* +status when asked, so each telemetry tick also fires a `pushing.pushall` for +the next one. A v1 printer's telemetry is therefore one tick behind, which +is why it shows up on the second frame rather than the first. + +## What's still a stub `start_uploaded_file` - the MQTT command that actually *starts* a print from an uploaded file - is written but has deliberately never been fired, since that commits a machine to hours of work. Its doc comment lists the three -details to confirm on the first real attempt. - -## What's still a stub +details to confirm on the first real attempt. `upload_file` is proven, but +only with a small text file, not a multi-megabyte sliced `.3mf`. PrusaLink (REST) and Klipper/Moonraker (WebSocket) - `printer/prusa.rs` and `printer/klipper.rs` just `println!` what they'd send. Also not built -yet: dispatching a cloud-issued command to a connected printer (the uplink -and the printer fleet run side by side in `main.rs`, but nothing routes -between them yet), local buffering for telemetry across connectivity gaps, -LAN printer discovery (SSDP/mDNS), and the mechanical plate-changer -interface. +yet: local buffering for telemetry across connectivity gaps (so telemetry +produced while the uplink is down is simply lost), LAN printer discovery +(SSDP/mDNS), and the mechanical plate-changer interface. ## Getting started diff --git a/src/main.rs b/src/main.rs index 8b2df10..c2dc816 100644 --- a/src/main.rs +++ b/src/main.rs @@ -21,25 +21,27 @@ async fn main() -> anyhow::Result<()> { tracing::info!(gateway_id = %config.gateway_id, farm_id = %config.farm_id, "starting continuum-proxy"); // Loaded and connected here, once, at startup - nothing yet re-reads - // printers.toml on a running daemon (adding/removing a printer means - // a restart), and nothing yet lets the uplink dispatch a command to - // one of these once connected - see PrinterHandle's own doc comment - // for what's built and what's still a stub per vendor. Keeping - // `fleet` alive as a local variable for the rest of `main` (which - // never returns, below) is what keeps every connection open; it's - // not otherwise touched yet. + // printers.toml on a running daemon, so adding or removing a printer + // still means a restart. let mut fleet = fleet::load(std::path::Path::new(&config.printers_path))?; connect_fleet(&mut fleet).await; - // Three things run at once for now: the cloud uplink, the go2rtc - // camera-restreaming process, and (implicitly, via the connections - // opened above) every configured printer's live session. More will - // join them later - printer discovery, the plate changer - as you - // build past src/printer/'s stubs. `tokio::select!` runs both of the - // tasks below and stops as soon as either one returns; both loop + // The fleet is now shared with the uplink rather than just parked here + // keeping connections alive: the uplink reads each printer's status to + // send telemetry up, and looks printers up by id to dispatch commands + // down. `Arc>` is how two tasks share one mutable thing - + // see `uplink::SharedFleet` for why that type and what it costs. + let fleet: uplink::SharedFleet = std::sync::Arc::new(tokio::sync::Mutex::new(fleet)); + + // Two long-running tasks: the cloud uplink and the go2rtc + // camera-restreaming watchdog. Every printer's MQTT session is a third + // thing running concurrently, but implicitly - each one has its own + // background poll task spawned by `connect()`, kept alive by the fleet + // above. More will join these later (printer discovery, the plate + // changer). `tokio::select!` stops as soon as either returns; both loop // forever, so in practice this runs until the process is killed. tokio::select! { - _ = uplink::run(config.clone()) => {} + _ = uplink::run(config.clone(), fleet.clone()) => {} _ = go2rtc::watchdog(config.go2rtc_bin.clone(), config.go2rtc_config.clone()) => {} } diff --git a/src/printer/mod.rs b/src/printer/mod.rs index cadb128..936afea 100644 --- a/src/printer/mod.rs +++ b/src/printer/mod.rs @@ -129,6 +129,52 @@ pub enum PrinterHandle { Klipper(KlipperPrinter), } +impl PrinterHandle { + /// The latest status this process has for the printer, if it has one. + /// + /// An inherent method rather than a `GenericPrinter` trait method, + /// because right now only Bambu has a real status path - the other three + /// return `None` not because they have nothing to say but because their + /// protocol clients are still stubs. Putting it on the trait would make + /// every vendor claim to implement something three of them don't. Move it + /// onto the trait once a second vendor can genuinely answer. + pub async fn print_status(&self) -> Option { + match self { + PrinterHandle::Bambu(p) => p.print_status().await, + PrinterHandle::PrusaLink(_) | PrinterHandle::PrusaSerial(_) | PrinterHandle::Klipper(_) => None, + } + } + + /// Asks the printer to report its full state, for the printers that only + /// report when asked. + /// + /// This exists because of one specific vendor quirk that should not leak + /// any further than this: a Bambu on protocol v1 pushes only *partial* + /// `print.push_status` objects unprompted - often a single temperature - + /// and sends a complete one only in response to `pushing.pushall`. Without + /// a periodic call here, `print_status()` on a v1 printer stays stuck on + /// whatever fragments it happened to volunteer. Found exactly that way: a + /// live P1P reached the cloud as a row of nulls with one real bed + /// temperature in it, while a v2 H2C beside it was fully populated. + /// + /// Nothing arrives by the time this returns - the refreshed status comes + /// back over MQTT as an ordinary push some time later, so a caller that + /// refreshes and immediately reads gets the *previous* refresh's data. On + /// a timer that is simply a one-tick lag, which is why this is + /// fire-and-forget rather than something to await a result from. + /// + /// A no-op for everyone else: `BambuPrinter::pushall` already no-ops on + /// v2 (which pushes complete objects unprompted), and the other three + /// vendors are still stubs. Same reasoning as `print_status` for why this + /// is inherent rather than a `GenericPrinter` method. + pub async fn refresh_status(&self) -> Result<(), PrinterError> { + match self { + PrinterHandle::Bambu(p) => p.pushall().await, + PrinterHandle::PrusaLink(_) | PrinterHandle::PrusaSerial(_) | PrinterHandle::Klipper(_) => Ok(()), + } + } +} + // Hand-written delegation: each trait method just matches on the variant // and forwards to that variant's own implementation. This one block is the // only "boilerplate tax" for not having inheritance - everything else reads diff --git a/src/uplink/mod.rs b/src/uplink/mod.rs index efc6f33..8656265 100644 --- a/src/uplink/mod.rs +++ b/src/uplink/mod.rs @@ -1,33 +1,68 @@ //! The cloud connection: a WebSocket to `continuum-backend`'s `/ws/edge/v1`. //! -//! This is intentionally the simplest version of "a client that reconnects -//! when it drops": no backoff curve, no channels routing messages in from -//! other parts of the program yet - just connect, say hello, send a -//! heartbeat on a timer, and print whatever comes back. Once you're -//! comfortable with this, growing it (real backoff, forwarding printer -//! telemetry, dispatching cloud-issued commands) is additive, not a rewrite. +//! This is where the proxy earns its name. Until now it connected, said +//! hello, and sent heartbeats into the void while the printer fleet lived in +//! a completely separate part of the program with nothing routing between +//! them. Both directions are wired up now: +//! +//! - **Up:** every `TELEMETRY_INTERVAL`, one snapshot per printer, batched +//! into a single `Telemetry` message. +//! - **Down:** a `Command` from the cloud is looked up by printer id, +//! executed, and answered with a `CommandResult`. +//! +//! Still deliberately the simplest version of a reconnecting client: a fixed +//! retry delay rather than a backoff curve, and no local buffering, so +//! telemetry produced while the uplink is down is simply lost (re-adding the +//! SQLite buffer cut in the simplification pass is the fix, and it is a +//! separate piece of work - see PLAN.md). mod protocol; -pub use protocol::{ClientMessage, ServerMessage}; +pub use protocol::{ClientMessage, PrinterCommand, PrinterTelemetry, ServerMessage}; +use std::sync::Arc; use std::time::Duration; use futures_util::{SinkExt, StreamExt}; +use tokio::sync::Mutex; use tokio_tungstenite::tungstenite::Message; use tracing::{info, warn}; use crate::config::Config; +use crate::printer::{GenericPrinter, PrinterHandle}; const RETRY_DELAY: Duration = Duration::from_secs(5); const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(15); +/// Faster than the heartbeat, slower than the printers push. A Bambu on v2 +/// emits a full status roughly every second; forwarding at that rate would be +/// a lot of traffic to describe a machine whose state changes on the scale of +/// a layer, so this samples the cached report instead of chasing every push. +const TELEMETRY_INTERVAL: Duration = Duration::from_secs(5); + +/// The printer fleet, shared between `main` (which owns it and keeps every +/// connection alive) and this module (which reads status and dispatches +/// commands). +/// +/// `Arc` is shared ownership - several places hold the same value and the +/// last one to let go drops it, roughly how every TS object reference +/// already behaves. `Mutex` is what makes mutation across tasks safe: Rust +/// will not let two tasks hold `&mut` to the same data, so a lock is how you +/// take turns. `tokio::sync::Mutex` rather than `std::sync::Mutex` +/// specifically because the code below holds the lock across an `.await`, +/// which the std one is not designed for. +/// +/// One lock for the whole `Vec` is coarse: a slow command to one printer +/// stalls telemetry for all of them. Fine at the handful-of-printers scale +/// this runs at now, and the shape to revisit (a lock per printer, or an +/// actor task per printer owning its own handle) before a farm gets large. +pub type SharedFleet = Arc>>; /// Runs forever: connect, serve the connection until it drops, wait, retry. -pub async fn run(config: Config) { +pub async fn run(config: Config, fleet: SharedFleet) { loop { info!(url = %config.uplink_url, "connecting to cloud uplink"); - if let Err(err) = connect_and_serve(&config).await { + if let Err(err) = connect_and_serve(&config, &fleet).await { warn!(?err, "uplink connection failed, retrying in {RETRY_DELAY:?}"); } @@ -35,7 +70,7 @@ pub async fn run(config: Config) { } } -async fn connect_and_serve(config: &Config) -> anyhow::Result<()> { +async fn connect_and_serve(config: &Config, fleet: &SharedFleet) -> anyhow::Result<()> { let (ws_stream, _response) = tokio_tungstenite::connect_async(&config.uplink_url).await?; let (mut write, mut read) = ws_stream.split(); @@ -47,6 +82,7 @@ async fn connect_and_serve(config: &Config) -> anyhow::Result<()> { write.send(Message::Text(serde_json::to_string(&hello)?)).await?; let mut heartbeat = tokio::time::interval(HEARTBEAT_INTERVAL); + let mut telemetry = tokio::time::interval(TELEMETRY_INTERVAL); let mut seq: u64 = 0; loop { @@ -57,11 +93,50 @@ async fn connect_and_serve(config: &Config) -> anyhow::Result<()> { write.send(Message::Text(serde_json::to_string(&msg)?)).await?; } + _ = telemetry.tick() => { + // Read first, then ask for the next one. A v1 Bambu answers + // `refresh_status` over MQTT some time after this returns, so + // collecting first means every tick reports the previous + // tick's refresh - one interval of lag, rather than a + // guaranteed miss from reading before the answer could + // possibly have arrived. + let printers = collect_telemetry(fleet).await; + refresh_fleet(fleet).await; + // Nothing to say is not worth a frame. Happens whenever no + // printer has reported yet - every vendor but Bambu is still + // a stub, so a Prusa-only fleet is silent here by design. + if !printers.is_empty() { + let msg = ClientMessage::Telemetry { printers }; + write.send(Message::Text(serde_json::to_string(&msg)?)).await?; + } + } + frame = read.next() => { match frame { Some(Ok(Message::Text(text))) => { - if let Ok(server_msg) = serde_json::from_str::(&text) { - info!(?server_msg, "received from cloud"); + match serde_json::from_str::(&text) { + Ok(ServerMessage::Command { command_id, printer_id, command }) => { + let result = dispatch(fleet, &printer_id, command).await; + let msg = match result { + Ok(()) => ClientMessage::CommandResult { + command_id, ok: true, error: None, + }, + Err(err) => { + warn!(%printer_id, %err, "command failed"); + ClientMessage::CommandResult { + command_id, ok: false, error: Some(err), + } + } + }; + write.send(Message::Text(serde_json::to_string(&msg)?)).await?; + } + 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 + // hid a real `print.push_status` failure for + // weeks - and here it would mean commands + // vanishing with the cloud never being told. + Err(err) => warn!(%err, payload = %text, "unparseable message from cloud, ignoring"), } } Some(Ok(Message::Close(_))) | None => return Ok(()), @@ -72,3 +147,75 @@ async fn connect_and_serve(config: &Config) -> anyhow::Result<()> { } } } + +/// Asks every printer to report in full, for the next tick to pick up. Failure +/// is logged and otherwise ignored: a printer that cannot be asked right now +/// is a printer whose telemetry goes stale, not a reason to tear down the +/// cloud connection for the whole farm. +async fn refresh_fleet(fleet: &SharedFleet) { + let fleet = fleet.lock().await; + for printer in fleet.iter() { + if let Err(err) = printer.refresh_status().await { + warn!(printer = %printer.base().id, %err, "failed to request a status refresh"); + } + } +} + +/// One snapshot per printer that has something to report. Translates Bambu's +/// `print.push_status` into the vendor-neutral shape the cloud consumes - see +/// `PrinterTelemetry` for why this isn't just forwarded raw. +async fn collect_telemetry(fleet: &SharedFleet) -> Vec { + let fleet = fleet.lock().await; + let mut out = Vec::new(); + + for printer in fleet.iter() { + let base = printer.base(); + let Some(report) = printer.print_status().await else { + continue; + }; + + out.push(PrinterTelemetry { + printer_id: base.id.clone(), + name: base.name.clone(), + state: report.gcode_state.clone(), + percent: report.mc_percent, + layer: report.layer_num, + total_layers: report.total_layer_num, + remaining_minutes: report.mc_remaining_time, + // An idle printer reports `""` rather than omitting the field; + // empty-string-as-absent is worth normalising here so the cloud + // doesn't have to know that. + job_name: report.subtask_name.clone().filter(|s| !s.is_empty()), + nozzle_temp: report.nozzle_temper, + nozzle_target: report.nozzle_target_temper, + bed_temp: report.bed_temper, + bed_target: report.bed_target_temper, + print_error: report.print_error, + active_alerts: report.hms.as_ref().map(|h| h.len()), + }); + } + + out +} + +/// Finds the printer by our own id and runs the command. The error is a +/// `String` rather than a typed error because its only destination is a +/// `CommandResult` going back to the cloud as text. +async fn dispatch(fleet: &SharedFleet, printer_id: &str, command: PrinterCommand) -> Result<(), String> { + let mut fleet = fleet.lock().await; + let printer = fleet + .iter_mut() + .find(|p| p.base().id == printer_id) + .ok_or_else(|| format!("no printer with id '{printer_id}' in this gateway's fleet"))?; + + info!(%printer_id, ?command, "dispatching cloud command"); + + match command { + PrinterCommand::SetFanSpeed { percent } => { + printer.set_fan_speed(percent).await.map_err(|err| err.to_string()) + } + PrinterCommand::SendGcodeLine { line } => { + printer.send_gcode_line(line).await.map_err(|err| err.to_string()) + } + } +} diff --git a/src/uplink/protocol.rs b/src/uplink/protocol.rs index 2fe1315..a2ecb02 100644 --- a/src/uplink/protocol.rs +++ b/src/uplink/protocol.rs @@ -6,6 +6,56 @@ use serde::{Deserialize, Serialize}; pub enum ClientMessage { Hello { gateway_id: String, farm_id: String, token: String }, Heartbeat { seq: u64 }, + /// One snapshot per printer, on a timer. Sent as a batch rather than a + /// message per printer: a farm has one uplink and many printers, and the + /// cloud wants a coherent picture of the farm at an instant more than it + /// wants the lowest possible latency per machine. + Telemetry { printers: Vec }, + /// The reply to a `ServerMessage::Command`. Correlated by `command_id`, + /// 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 }, +} + +/// A flattened snapshot of one printer, in the terms an operator console +/// actually displays - deliberately not Bambu's `print.push_status` shape. +/// +/// Two reasons for the translation rather than forwarding the raw report: +/// this has to describe a Prusa or a Klipper printer too, and Bambu's wire +/// format has traps that should not reach the rest of the system (fan +/// speeds are 0-15 gears, temperatures arrive as either integers or floats, +/// and v2 reuses one `sequence_id` forever - see `PrintReport`). +/// +/// Every field is optional because an unreachable or just-connected printer +/// genuinely has nothing to report, and "no data yet" is different from +/// "zero". +#[derive(Debug, Clone, Serialize)] +pub struct PrinterTelemetry { + /// Our own id from printers.toml, not the vendor serial - this is what + /// the cloud's `printers` table keys on. + pub printer_id: String, + pub name: String, + /// `"IDLE"`, `"RUNNING"`, `"PAUSE"`, `"FINISH"`, `"FAILED"` - passed + /// 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. + pub state: Option, + pub percent: Option, + pub layer: Option, + pub total_layers: Option, + pub remaining_minutes: Option, + pub job_name: Option, + pub nozzle_temp: Option, + pub nozzle_target: Option, + pub bed_temp: Option, + pub bed_target: Option, + /// Non-zero means the printer is reporting a fault. + 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. + pub active_alerts: Option, } /// What `continuum-backend` sends back over the same connection. @@ -14,4 +64,25 @@ pub enum ClientMessage { pub enum ServerMessage { HelloAck { session_id: String }, HeartbeatAck { seq: u64 }, + /// Do something to one printer. `printer_id` is our own id from + /// printers.toml. + Command { command_id: String, printer_id: String, command: PrinterCommand }, +} + +/// The closed set of things the cloud is allowed to ask a printer to do. +/// +/// Deliberately small, and deliberately missing the obvious one: there is no +/// "start a print" variant. `BambuPrinter::start_uploaded_file` exists but +/// has never been run against real hardware, and the first thing to exercise +/// it should be a human watching the machine - not a message arriving from +/// the internet. Add it here once it's been confirmed by hand. +#[derive(Debug, Clone, Deserialize)] +#[serde(tag = "action", rename_all = "snake_case")] +pub enum PrinterCommand { + /// 0-100. Each vendor converts to whatever its own mechanism wants (for + /// Bambu, an `M106` PWM value over MQTT). + SetFanSpeed { percent: u8 }, + /// A single raw gcode line. The most powerful variant by far, and the + /// one to be most careful about ever exposing to a UI. + SendGcodeLine { line: String }, }