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

Execute start_print, and never await it on the uplink task

Browse Source 
The gateway can now run a cloud-dispatched print: download the job file from
the presigned URL in the command, then hand it to the existing
`upload_project_file`, which derives the filament mapping from the archive,
transfers it over FTPS and starts it.

## The download refuses to be a weapon

`src/uplink/download.rs` checks the HTTP status before reading a byte,
because an expired presigned URL answers 403 with an XML error document and
writing *that* to a printer as a project is the worst outcome available. It
rejects a `Content-Length` over 256 MiB before reading, and enforces the cap
while streaming as well, so a server that lies or omits the length cannot
make a small gateway box buffer until the OOM killer arrives. A byte count
that disagrees with `Content-Length` is an error: a `.3mf` is a zip, and a
truncated one loses the central directory the printer needs. HTTPS only, no
redirects, and the URL is stripped from error messages.

`PrinterCommand` no longer derives `Debug` for the same reason - `dispatch`
logs every command, and a presigned URL is a bearer credential until it
expires. It prints as `<redacted>`.

## Why it is spawned

`start_print` is the one command that cannot be answered inline, and the
reason is not latency. Everything else here is an MQTT publish returning in
milliseconds. This downloads tens of megabytes and then transfers them at a
measured 0.2 MiB/s - about four minutes for a 50 MB project - and awaiting
that inside the uplink's `select!` loop stops heartbeats and, far worse,
stops telemetry for the duration.

No telemetry is what makes spawning mandatory rather than tidy. The cloud now
derives a job's state from the telemetry that follows a start, so a gateway
that goes silent for four minutes during every job start withholds exactly
the information the job depends on - while the backend times the command out
after ten seconds and concludes the start failed. The job would be requeued
onto a printer that was busy printing it.

So the `ok` means "accepted, working on it", which is all a start reply was
ever trusted to mean. `runs_for_minutes` is an exhaustive match rather than a
`matches!`, so a new variant has to make that choice instead of defaulting to
the stall.

The gap this leaves is honest and recorded in the code: a failure after the
ack has no `CommandResult` to travel on, because answering later needs an
outbound channel this loop does not have. Until it does, the gateway's log is
where the reason lives and the backend reaps the job.

reqwest 0.12 with `rustls-tls-native-roots`, not 0.13: 0.13 pulls
`webpki-root-certs` under CDLA-Permissive-2.0, which is not on the allow
list, and `aws-lc` brings OpenSSL-licensed C. `cargo deny check licenses`
passes. 54 tests.
This commit is contained in:
octoturge
2026-10-15 22:53:47 +00:00
parent 76b3894ca9
commit 901d5e9748
5 changed files with 339 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/uplink/mod.rs
+90 -8
View File
@@ -16,6 +16,7 @@
//! SQLite buffer cut in the simplification pass is the fix, and it is a
//! separate piece of work - see PLAN.md).
mod download;
mod protocol;
pub use protocol::{ClientMessage, PrinterCommand, PrinterTelemetry, ServerMessage};
@@ -116,15 +117,23 @@ async fn connect_and_serve(config: &Config, fleet: &SharedFleet) -> anyhow::Resu
Some(Ok(Message::Text(text))) => {
match serde_json::from_str::<ServerMessage>(&text) {
Ok(ServerMessage::Command { command_id, printer_id, command }) => {
let result = dispatch(fleet, &printer_id, command).await;
let msg = match result {
Ok(()) => ClientMessage::CommandResult {
// `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);
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),
}
} else {
match dispatch(fleet, &printer_id, command).await {
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),
}
}
}
};
@@ -206,10 +215,69 @@ async fn collect_telemetry(fleet: &SharedFleet) -> Vec<PrinterTelemetry> {
out
}
/// Runs a command that takes minutes, off this connection's task, and answers
/// the cloud immediately.
///
/// Everything else here is an MQTT publish that returns in milliseconds, so
/// awaiting it inline is fine. `start_print` is not: it downloads tens of
/// megabytes and then transfers them to a printer over FTPS at a measured
/// 0.2 MiB/s, which is about four minutes for a 50 MB project. Awaited
/// inline, that blocks this whole `select!` loop - no heartbeats and, far
/// worse, **no telemetry** for the duration.
///
/// No telemetry is the one that makes this mandatory rather than tidy. The
/// cloud derives a job's state from the telemetry that follows a start, so a
/// gateway that goes silent for four minutes during every job start is
/// withholding exactly the information the job depends on, while the backend
/// is also timing the command out after ten seconds and concluding the start
/// 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) {
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"
),
}
});
}
/// 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> {
// Fetched before the fleet lock is taken: a download is bounded by
// minutes of network time, and holding the lock across it would stall
// telemetry for every other printer. The transfer to the printer that
// follows has to hold it (it needs `&mut` access), but that is no reason
// to hold it through this part too. A `start_print` for a printer that
// does not exist is therefore found out after the download rather than
// before; the lookup below is cheap, so it is done first for that case.
let downloaded = match &command {
PrinterCommand::StartPrint { url, .. } => {
if !fleet.lock().await.iter().any(|p| p.base().id == printer_id) {
return Err(format!("no printer with id '{printer_id}' in this gateway's fleet"));
}
Some(
download::fetch_job_file(url)
.await
.map_err(|err| format!("download failed (nothing was sent to the printer): {err}"))?,
)
}
_ => None,
};
let mut fleet = fleet.lock().await;
let printer = fleet
.iter_mut()
@@ -228,5 +296,19 @@ async fn dispatch(fleet: &SharedFleet, printer_id: &str, command: PrinterCommand
PrinterCommand::ControlPrint { control } => {
printer.control_print(control).await.map_err(|err| err.to_string())
}
PrinterCommand::StartPrint { job_id, file_name, .. } => {
let bytes = downloaded.ok_or_else(|| "internal error: start_print reached dispatch with no downloaded file".to_string())?;
let byte_count = bytes.len();
info!(%printer_id, %job_id, %file_name, byte_count, "job file downloaded, sending to printer");
// One call covers deriving the filament mapping, the FTPS
// transfer, the truncation check and the start, and reports them
// all as one error - so the stage here is "printer", not finer.
// The printer's own message says which part (mapping, upload of
// the file, or start) it was.
printer
.upload_project_file(file_name, bytes)
.await
.map_err(|err| format!("printer upload/start failed (download of {byte_count} bytes succeeded): {err}"))
}
}
}