From ae3195f772d6f1b284dd2e52073c90844cad6858 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Iwo=20Strzebo=C5=84ski?= Date: Sun, 18 Oct 2026 20:00:00 +0000 Subject: [PATCH] Report a long-running command's failure to the cloud, not just the log A `start_print` is acknowledged the moment it is accepted: the transfer takes minutes at the measured 0.2 MiB/s and the backend gives up after ten seconds, so the real outcome can only ever arrive late. Until now there was nowhere for it to arrive - the uplink's `select!` loop was the only writer, and a task outside it could not reach the socket. A failed job start therefore reported success, and the reason lived only in this gateway's log, on a machine in someone else's workshop. Adds a bounded mpsc queue, created outside the reconnect loop so a message produced while the socket is down waits for the next connection rather than being dropped. A late failure goes back as a second `CommandResult` for the same command_id, carrying the reason. Be clear what that buys: the backend is no longer awaiting that command, so the result does not change the job's fate - the job never reaching `printing` is what fails it, and that already worked. What it adds is the reason, next to the farm and gateway that produced it. Attaching it to the job row needs a column `print_jobs` does not have. Nothing is sent on success; the telemetry that follows says it better. Also documents, in the module doc, why telemetry is not buffered across an uplink gap - it is a decision now rather than a gap. See PLAN.md. Co-Authored-By: Claude Opus 5 --- src/uplink/mod.rs | 118 ++++++++++++++++++++++++++++++++++++++-------- 1 file changed, 98 insertions(+), 20 deletions(-) diff --git a/src/uplink/mod.rs b/src/uplink/mod.rs index 5dd9857..e036518 100644 --- a/src/uplink/mod.rs +++ b/src/uplink/mod.rs @@ -10,11 +10,21 @@ //! - **Down:** a `Command` from the cloud is looked up by printer id, //! executed, and answered with a `CommandResult`. //! +//! - **Out of band:** anything that needs to reach the cloud from outside +//! this loop goes through an mpsc channel (`outbound`), which is how a +//! command that takes minutes reports its outcome after the loop has +//! already answered for it. +//! //! 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). +//! retry delay rather than a backoff curve. Telemetry produced while the +//! uplink is down is not buffered, and that is now a considered decision +//! rather than a gap - the cloud stores only each printer's latest state and +//! has no table for a history, and a Bambu's state is level-triggered +//! (`FINISH` persists until the next print starts), so the first snapshot +//! after a reconnect carries the outcome. Replaying a backlog of five-second +//! snapshots into a row that holds one value would churn it with stale +//! readings for no consumer. See PLAN.md for what would actually be worth +//! buffering. mod download; mod protocol; @@ -34,6 +44,15 @@ use crate::config::Config; use crate::printer::{GenericPrinter, PrinterHandle}; const RETRY_DELAY: Duration = Duration::from_secs(5); +/// Depth of the out-of-band outbound queue. +/// +/// Bounded, because an unbounded queue facing a cloud that may be down for +/// hours is just a slow memory leak. Only out-of-band messages go through it, +/// one per long-running command, so 64 is far more than a farm generates +/// between reconnects. A sender waits when it is full rather than dropping, +/// which is the right trade for a message whose whole purpose is to explain a +/// failure. +const OUTBOUND_CAPACITY: usize = 64; 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 @@ -83,10 +102,16 @@ fn host_header(url: &str) -> anyhow::Result { /// Runs forever: connect, serve the connection until it drops, wait, retry. pub async fn run(config: Config, fleet: SharedFleet) { + // Created out here, not per connection, so a message produced while the + // socket is down waits for the next one instead of being dropped. The + // sender is kept for the whole program's life, which is also what stops + // `recv()` below from ever returning `None`. + let (outbound, mut inbox) = tokio::sync::mpsc::channel::(OUTBOUND_CAPACITY); + loop { info!(url = %config.uplink_url, "connecting to cloud uplink"); - if let Err(err) = connect_and_serve(&config, &fleet).await { + if let Err(err) = connect_and_serve(&config, &fleet, &outbound, &mut inbox).await { warn!(?err, "uplink connection failed, retrying in {RETRY_DELAY:?}"); } @@ -94,7 +119,12 @@ pub async fn run(config: Config, fleet: SharedFleet) { } } -async fn connect_and_serve(config: &Config, fleet: &SharedFleet) -> anyhow::Result<()> { +async fn connect_and_serve( + config: &Config, + fleet: &SharedFleet, + outbound: &tokio::sync::mpsc::Sender, + inbox: &mut tokio::sync::mpsc::Receiver, +) -> anyhow::Result<()> { // The edge token goes on the upgrade request, not in the first message. // The backend verifies it before the socket exists, so a bad token is an // HTTP 401 here rather than a connection that opens and is then closed - @@ -154,6 +184,15 @@ async fn connect_and_serve(config: &Config, fleet: &SharedFleet) -> anyhow::Resu } } + // Whatever a task elsewhere wants said. Taken from the queue + // before it is written, so a send that fails here loses the + // message - the connection is going down anyway, and the + // alternative (peek, write, then pop) buys little for a message + // that is already best-effort diagnostics. + Some(msg) = inbox.recv() => { + write.send(Message::Text(serde_json::to_string(&msg)?)).await?; + } + frame = read.next() => { match frame { Some(Ok(Message::Text(text))) => { @@ -162,7 +201,13 @@ async fn connect_and_serve(config: &Config, fleet: &SharedFleet) -> anyhow::Resu // `start_print` is the one command that cannot // be answered inline. See `spawn_long_command`. let msg = if command.runs_for_minutes() { - spawn_long_command(fleet.clone(), printer_id.clone(), command); + spawn_long_command( + fleet.clone(), + printer_id.clone(), + command, + command_id.clone(), + outbound.clone(), + ); ClientMessage::CommandResult { command_id, ok: true, error: None, } @@ -285,22 +330,55 @@ async fn collect_telemetry(fleet: &SharedFleet) -> Vec { /// failed. /// /// So the `ok` this produces means "accepted, and working on it" - which is -/// all the cloud ever trusted a start reply to mean anyway. A failure after -/// this point is reported by the job never reaching `printing`, which the -/// backend reaps; it is not reported as a `CommandResult`, because answering -/// later would need an outbound channel this loop does not have yet. That is -/// the next thing to build here, and until it exists the gateway's log is -/// where the reason lives. -fn spawn_long_command(fleet: SharedFleet, printer_id: String, command: PrinterCommand) { +/// all the cloud ever trusted a start reply to mean anyway. *That* a failure +/// after this point happened is still noticed by the cloud on its own: the +/// job never reaches `printing` and the backend reaps it. What used to be +/// missing was *why*, which lived only in this gateway's log - on a machine +/// in someone else's workshop. +/// +/// So a late failure now goes back over the outbound queue as a second +/// `CommandResult` for the same `command_id`. Be clear about what the backend +/// does with it: that command is long since resolved, so it is not awaiting +/// an answer and the result does not change the job's fate. It is logged with +/// its reason, next to the gateway and farm that produced it, which is the +/// difference between "job failed" and "job failed: filament mapping for +/// plate 1 names a slot the AMS does not have". Attaching it to the job row +/// needs a column `print_jobs` does not have yet. +/// +/// Nothing is sent on success - the telemetry that follows says so better +/// than a message could, and a result for an already-resolved command is +/// noise unless it carries news. +fn spawn_long_command( + fleet: SharedFleet, + printer_id: String, + command: PrinterCommand, + command_id: String, + outbound: tokio::sync::mpsc::Sender, +) { tokio::spawn(async move { match dispatch(&fleet, &printer_id, command).await { Ok(()) => info!(%printer_id, "long-running command finished"), - // Loudly, and at error level: nothing else will say why. The - // cloud sees only a job that never started printing. - Err(err) => tracing::error!( - %printer_id, %err, - "long-running command failed after the cloud was already told it was accepted" - ), + Err(err) => { + // Still logged locally, and still at error level: the queue + // only reaches a cloud that is up, and this is the copy that + // survives a gateway with no internet. + tracing::error!( + %printer_id, %err, + "long-running command failed after the cloud was already told it was accepted" + ); + let late = ClientMessage::CommandResult { + command_id, + ok: false, + error: Some(err), + }; + // `send` waits for room rather than dropping, and only fails + // if the receiver is gone - which means `run` has returned, + // i.e. the process is shutting down and there is nobody left + // to tell. + if outbound.send(late).await.is_err() { + warn!(%printer_id, "uplink is shutting down; the failure reason was not reported"); + } + } } }); }