Continuum/continuum-proxy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Report a long-running command's failure to the cloud, not just the log

Browse Source 
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.
This commit is contained in:
octoturge
2026-10-18 20:00:00 +00:00
parent 51ff768153
commit 930d117c4e
1 changed files with 98 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/uplink/mod.rs
+96 -18
View File
@@ -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<String> {
/// 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::<ClientMessage>(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<ClientMessage>,
inbox: &mut tokio::sync::mpsc::Receiver<ClientMessage>,
) -> 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<PrinterTelemetry> {
/// 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<ClientMessage>,
) {
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!(
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");
}
}
}
});
}