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 <noreply@anthropic.com>
This commit is contained in:
+96
-18
@@ -10,11 +10,21 @@
|
|||||||
//! - **Down:** a `Command` from the cloud is looked up by printer id,
|
//! - **Down:** a `Command` from the cloud is looked up by printer id,
|
||||||
//! executed, and answered with a `CommandResult`.
|
//! 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
|
//! Still deliberately the simplest version of a reconnecting client: a fixed
|
||||||
//! retry delay rather than a backoff curve, and no local buffering, so
|
//! retry delay rather than a backoff curve. Telemetry produced while the
|
||||||
//! telemetry produced while the uplink is down is simply lost (re-adding the
|
//! uplink is down is not buffered, and that is now a considered decision
|
||||||
//! SQLite buffer cut in the simplification pass is the fix, and it is a
|
//! rather than a gap - the cloud stores only each printer's latest state and
|
||||||
//! separate piece of work - see PLAN.md).
|
//! 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 download;
|
||||||
mod protocol;
|
mod protocol;
|
||||||
@@ -34,6 +44,15 @@ use crate::config::Config;
|
|||||||
use crate::printer::{GenericPrinter, PrinterHandle};
|
use crate::printer::{GenericPrinter, PrinterHandle};
|
||||||
|
|
||||||
const RETRY_DELAY: Duration = Duration::from_secs(5);
|
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);
|
const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(15);
|
||||||
/// Faster than the heartbeat, slower than the printers push. A Bambu on v2
|
/// 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
|
/// 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.
|
/// Runs forever: connect, serve the connection until it drops, wait, retry.
|
||||||
pub async fn run(config: Config, fleet: SharedFleet) {
|
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 {
|
loop {
|
||||||
info!(url = %config.uplink_url, "connecting to cloud uplink");
|
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:?}");
|
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 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
|
// 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 -
|
// 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() => {
|
frame = read.next() => {
|
||||||
match frame {
|
match frame {
|
||||||
Some(Ok(Message::Text(text))) => {
|
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
|
// `start_print` is the one command that cannot
|
||||||
// be answered inline. See `spawn_long_command`.
|
// be answered inline. See `spawn_long_command`.
|
||||||
let msg = if command.runs_for_minutes() {
|
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 {
|
ClientMessage::CommandResult {
|
||||||
command_id, ok: true, error: None,
|
command_id, ok: true, error: None,
|
||||||
}
|
}
|
||||||
@@ -285,22 +330,55 @@ async fn collect_telemetry(fleet: &SharedFleet) -> Vec<PrinterTelemetry> {
|
|||||||
/// failed.
|
/// failed.
|
||||||
///
|
///
|
||||||
/// So the `ok` this produces means "accepted, and working on it" - which is
|
/// 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
|
/// all the cloud ever trusted a start reply to mean anyway. *That* a failure
|
||||||
/// this point is reported by the job never reaching `printing`, which the
|
/// after this point happened is still noticed by the cloud on its own: the
|
||||||
/// backend reaps; it is not reported as a `CommandResult`, because answering
|
/// job never reaches `printing` and the backend reaps it. What used to be
|
||||||
/// later would need an outbound channel this loop does not have yet. That is
|
/// missing was *why*, which lived only in this gateway's log - on a machine
|
||||||
/// the next thing to build here, and until it exists the gateway's log is
|
/// in someone else's workshop.
|
||||||
/// where the reason lives.
|
///
|
||||||
fn spawn_long_command(fleet: SharedFleet, printer_id: String, command: PrinterCommand) {
|
/// 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 {
|
tokio::spawn(async move {
|
||||||
match dispatch(&fleet, &printer_id, command).await {
|
match dispatch(&fleet, &printer_id, command).await {
|
||||||
Ok(()) => info!(%printer_id, "long-running command finished"),
|
Ok(()) => info!(%printer_id, "long-running command finished"),
|
||||||
// Loudly, and at error level: nothing else will say why. The
|
Err(err) => {
|
||||||
// cloud sees only a job that never started printing.
|
// Still logged locally, and still at error level: the queue
|
||||||
Err(err) => tracing::error!(
|
// only reaches a cloud that is up, and this is the copy that
|
||||||
|
// survives a gateway with no internet.
|
||||||
|
tracing::error!(
|
||||||
%printer_id, %err,
|
%printer_id, %err,
|
||||||
"long-running command failed after the cloud was already told it was accepted"
|
"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");
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user