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<Mutex<Vec<PrinterHandle>>>` 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.

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

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.
+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`.
-## What's still a stub
 
 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
+yet: local buffering for telemetry across connectivity gaps (so telemetry
-and the printer fleet run side by side in `main.rs`, but nothing routes
+produced while the uplink is down is simply lost), LAN printer discovery
-between them yet), local buffering for telemetry across connectivity gaps,
+(SSDP/mDNS), and the mechanical plate-changer interface.
-LAN printer discovery (SSDP/mDNS), and the mechanical plate-changer
-interface.
 
 ## Getting started
 

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
+    // printers.toml on a running daemon, so adding or removing a printer
-    // a restart), and nothing yet lets the uplink dispatch a command to
+    // still means a restart.
-    // 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.
     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
+    // The fleet is now shared with the uplink rather than just parked here
-    // camera-restreaming process, and (implicitly, via the connections
+    // keeping connections alive: the uplink reads each printer's status to
-    // opened above) every configured printer's live session. More will
+    // send telemetry up, and looks printers up by id to dispatch commands
-    // join them later - printer discovery, the plate changer - as you
+    // down. `Arc<Mutex<..>>` is how two tasks share one mutable thing -
-    // build past src/printer/'s stubs. `tokio::select!` runs both of the
+    // see `uplink::SharedFleet` for why that type and what it costs.
-    // tasks below and stops as soon as either one returns; both loop
+    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()) => {}
     }
 

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<PrintReport> {
+        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

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
+//! This is where the proxy earns its name. Until now it connected, said
-//! when it drops": no backoff curve, no channels routing messages in from
+//! hello, and sent heartbeats into the void while the printer fleet lived in
-//! other parts of the program yet - just connect, say hello, send a
+//! a completely separate part of the program with nothing routing between
-//! heartbeat on a timer, and print whatever comes back. Once you're
+//! them. Both directions are wired up now:
-//! comfortable with this, growing it (real backoff, forwarding printer
+//!
-//! telemetry, dispatching cloud-issued commands) is additive, not a rewrite.
+//! - **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<Mutex<Vec<PrinterHandle>>>;
 
 /// 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::<ServerMessage>(&text) {
+                        match serde_json::from_str::<ServerMessage>(&text) {
-                            info!(?server_msg, "received from cloud");
+                            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<PrinterTelemetry> {
+    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())
+        }
+    }
+}

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<PrinterTelemetry> },
+    /// 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<String> },
+}
+
+/// 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<String>,
+    pub percent: Option<i32>,
+    pub layer: Option<i32>,
+    pub total_layers: Option<i32>,
+    pub remaining_minutes: Option<i32>,
+    pub job_name: Option<String>,
+    pub nozzle_temp: Option<f32>,
+    pub nozzle_target: Option<f32>,
+    pub bed_temp: Option<f32>,
+    pub bed_target: Option<f32>,
+    /// Non-zero means the printer is reporting a fault.
+    pub print_error: Option<i32>,
+    /// 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<usize>,
 }
 
 /// 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 },
 }