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

Add pushall + print_status cache; fix envelope, race, and matching bugs

Browse Source 
PrintReport now flattens CommandEnvelope (sequence_id/command) -
push_status.schema.json composes the shared envelope like every other
Bambu message; the struct's old doc comment concluded otherwise from a
capture that turned out to be an incomplete example (also missing dozens
of other real fields), not proof the message lacks them.

Added BambuMqttSession's PrintStatusCache: the in-memory
print.push_status object, replaced wholesale on v2 (always full) or on a
v1 message with msg != 1 (forced/full), merged field-by-field otherwise
(v1's organic partial updates) via PrintReport::merge_from. Tracks
last_sequence_id to warn on a gap in the organic stream. BambuPrinter
gets pushall() (v1-only; pushing.pushall doesn't exist in v2's schema at
all) and print_status().

Three real bugs found and fixed via live testing against an actual P1P,
none of them guessable from the schema alone:
- pushall's reply doesn't echo the sequence_id we send - the printer
  stamps its own free-running counter regardless of what triggered it.
  request()'s ID correlation silently can't work here; first fix attempt
  added request_next_of_category (wait for the next message under a
  category, no ID needed).
- That undershot too: matching by category alone let an unrelated
  "print" message (that category carries many commands - gcode_line,
  pause, push_status, ...) satisfy the wait before the real push_status
  reply arrived, so pushall() returned Ok having consumed the wrong
  message - now keyed by (category, command).
- A real ordering race: handle_incoming_report woke a waiting oneshot
  before updating the print_status cache, so a caller could resume and
  check print_status() before the write landed - pushall() returned Ok
  immediately followed by print_status() still showing None. Fixed by
  applying the cache update before sending on any matched oneshot.

Superseded by real protocol knowledge before any of the above needed to
ship, though: push_status carries its own msg field (1 = automatic/
partial, 0 = forced/full) - the actual, reliable, self-describing signal
for which kind a message is, regardless of timing or which client (if
any) triggered it. PrintStatusCache now reads that directly instead of
inferring it from request correlation, and pushall() is plain fire-and-
forget (publish, no wait) since there's nothing left to correlate for -
confirmed live: msg reads correctly off a real P1P, and the cache
populates/updates correctly with no gap warnings across several organic
pushes. request_next_of_category is kept (a future case may still need
"no correlation ID at all" for real), just no longer used here.

Also: the poll loop now logs when it exits instead of returning silently
- that silence is exactly what made a real dropped MQTT connection to a
live H2C indistinguishable from "nothing happening" during testing.
This commit is contained in:
octoturge
2026-09-07 14:39:55 +00:00
parent 835a4ae778
commit 6ce60542c6
4 changed files with 452 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/printer/bambu_mqtt.rs
+325 -72
View File
@@ -1,6 +1,6 @@
//! The real MQTT wiring for Bambu printers - the piece `connect`/
//! `send_gcode_line` in `bambu.rs` used to just stub out with a
//! `println!`. Two things make this more than "call `publish`, get a
//! `println!`. Three things make this more than "call `publish`, get a
//! `String` back":
//!
//! 1. MQTT v3.1.1 (BambuMQTT v1) and MQTT v5 (BambuMQTT v2) are two
@@ -11,21 +11,39 @@
//! struct you're holding.
//! 2. A response to a published request doesn't come back from that same
//! call - it arrives later, asynchronously, on a *subscription* to
//! `device/{sn}/report`, correlated only by matching `sequence_id`.
//! That needs: a background task continuously polling the connection
//! (rumqttc doesn't do this for you - nothing happens until you poll
//! the `EventLoop` yourself, repeatedly, forever), and a shared table
//! of "who's waiting for which `sequence_id`" that task completes as
//! reports arrive. This is the first real use of `tokio::spawn` in
//! this crate - flagged as deferred since `printer/mod.rs`'s very
//! first doc comment, and this is why: it only makes sense once there's
//! a long-running background job (polling forever) that has to run
//! independently of whatever call site is waiting on a response.
//! `device/{sn}/report`. That needs a background task continuously
//! polling the connection (rumqttc doesn't do this for you - nothing
//! happens until you poll the `EventLoop` yourself, repeatedly,
//! forever) - this is the first real use of `tokio::spawn` in this
//! crate, flagged as deferred since `printer/mod.rs`'s very first doc
//! comment, and this is why: it only makes sense once there's a
//! long-running background job that has to run independently of
//! whatever call site is waiting on a response.
//!
//! Unverified against a live printer (no broker to test against here):
//! the exact `ConnectReturnCode`/error shape a real Bambu printer's
//! embedded broker gives when it doesn't understand MQTT v5's CONNECT at
//! all (rather than a clean version-rejection CONNACK) - `open()` below
//! Two genuinely different correlation problems live here, confirmed
//! against real printers, not assumed:
//! - `info.get_version`'s reply *does* echo the `sequence_id` you sent
//! - `request`/`pending` (keyed by sequence_id) is correct for this,
//! and for anything else empirically confirmed to behave the same
//! way.
//! - `print.push_status` does NOT - the printer stamps it with its own
//! free-running counter regardless of what triggered it (confirmed
//! live: a `pushing.pushall` request's sequence_id never came back on
//! the reply). So there's no ID to match a `pushall` reply against at
//! all - `next_category`, keyed by category name instead of an ID,
//! is "wait for the *next* message under this category, whichever
//! one it is" - the only thing that actually works here.
//! 3. `print.push_status` also needs to be kept as running in-memory
//! state, not just handed to whoever happens to be waiting at the
//! moment it arrives - see `PrintStatusCache` below for why (short
//! version: v1 only ever sends partial updates, so something has to
//! remember what the fields *were* to produce a coherent whole
//! object).
//!
//! Unverified against a live printer (no way to force it in a test): the
//! exact `ConnectReturnCode`/error shape a real Bambu printer's embedded
//! broker gives when it doesn't understand MQTT v5's CONNECT at all
//! (rather than a clean version-rejection CONNACK) - `open()` below
//! treats *any* failure of the v5 attempt as "fall back to v3.1.1", not
//! just a precisely-matched version-rejection code, since a broker that
//! can't parse a v5 CONNECT packet may just drop the connection instead
@@ -43,26 +61,114 @@ use serde_json::Value;
use tokio::sync::{oneshot, Mutex};
use super::bambu::BambuTls;
use super::PrinterError;
use super::bambu_commands::PrintReport;
use super::BambuProtocolVersion;
use super::PrinterError;
/// Requests waiting on a report with a matching `sequence_id`. Shared
/// between whichever call registered one (in `request`) and the
/// background poll task that completes it when the report arrives.
/// Requests waiting on a report with a matching `sequence_id` - only
/// valid for a command empirically confirmed to echo it back, like
/// `get_version`. Shared between whichever call registered one (in
/// `request`) and the background poll task that completes it.
type PendingTable = Arc<Mutex<HashMap<String, oneshot::Sender<Value>>>>;
/// Callers waiting on the *next* `(category, command)` message,
/// regardless of its `sequence_id` - for a reply with no usable
/// correlation ID at all, like `pushall`'s. Keyed by the pair, not just
/// category: a category like "print" carries many different commands
/// (`gcode_line`, `pause`, `push_status`, ...), and matching on category
/// alone means some unrelated "print" message arriving first would
/// silently satisfy the wait instead of the one actually meant - caught
/// live, see `request_next_of_category`'s doc comment. One slot per
/// pair: a second wait registered before the first resolves silently
/// replaces it (the first caller's `rx` then just sees the channel
/// close) - fine for now since nothing calls `pushall` concurrently with
/// itself.
type NextCategoryTable = Arc<Mutex<HashMap<(String, String), oneshot::Sender<Value>>>>;
const REQUEST_TIMEOUT: Duration = Duration::from_secs(10);
/// Old firmware has been observed taking noticeably longer than
/// `REQUEST_TIMEOUT` to answer `pushall` - confirmed live (a real P1P's
/// first reply arrived after that 10s window had already elapsed).
const NEXT_CATEGORY_TIMEOUT: Duration = Duration::from_secs(30);
enum BambuMqttClient {
V1(rumqttc::AsyncClient),
V2(rumqttc::v5::AsyncClient),
}
/// The running in-memory `print.push_status` object, plus enough state to
/// know how to fold the next update into it:
///
/// - BambuMQTT v2 sends the *entire* object every time - always replace.
/// - BambuMQTT v1 only sends the fields that changed since last time -
/// merge into the existing object, or the picture goes incomplete the
/// instant anything not present in the latest push gets treated as
/// "unset". The exception is a `pushing.pushall`-triggered push, which
/// is always the full object - and v1 self-describes which kind a given
/// message is via its own `msg` field (`1` automatic/partial, `0`
/// forced/full), confirmed against a real printer. Two earlier attempts
/// at inferring this from request/reply *timing* instead (matching
/// `sequence_id`, then "wait for the next print/push_status message")
/// both turned out wrong in practice - the printer stamps `sequence_id`
/// from its own counter regardless of what triggered a push, and
/// category-only matching couldn't tell one "print" message from
/// another. `msg` needs none of that: it's on the message itself,
/// correct regardless of which client (or nothing at all) triggered it,
/// and regardless of timing.
/// - `last_sequence_id` exists to notice when v1's organic stream skipped
/// an update (network hiccup, a message this client wasn't fast enough
/// to read): if the incoming `sequence_id` isn't exactly one more than
/// the last one seen, the merged object is missing whatever changed in
/// the gap. There's no way to know what that baseline number even is
/// ahead of time - the printer runs its own counter, not one this
/// client started - so the very first `push_status` received after
/// connecting just seeds `last_sequence_id` rather than being checked
/// against anything.
struct PrintStatusCache {
report: Option<PrintReport>,
last_sequence_id: Option<u64>,
}
impl PrintStatusCache {
fn new() -> Self {
Self { report: None, last_sequence_id: None }
}
fn apply(&mut self, incoming: PrintReport, version: BambuProtocolVersion) {
let incoming_seq = incoming.envelope.sequence_id.parse::<u64>().ok();
// A forced/full update isn't missing anything by definition, so a
// sequence gap right after one (or on v2, where every message is
// full) isn't a real problem worth warning about.
let is_full_object = matches!(version, BambuProtocolVersion::V2) || incoming.msg != Some(1);
if let (Some(last), Some(incoming_seq), false) = (self.last_sequence_id, incoming_seq, is_full_object) {
if incoming_seq != last.wrapping_add(1) {
eprintln!(
"print.push_status sequence gap: expected {}, got {incoming_seq} - the in-memory \
object may be missing whatever changed in between",
last.wrapping_add(1)
);
}
}
if let Some(incoming_seq) = incoming_seq {
self.last_sequence_id = Some(incoming_seq);
}
match (&mut self.report, is_full_object) {
(_, true) => self.report = Some(incoming),
(Some(existing), false) => existing.merge_from(incoming),
(None, false) => self.report = Some(incoming),
}
}
}
/// One live MQTT connection to a Bambu printer - whichever protocol
/// version `open()` settled on.
pub struct BambuMqttSession {
client: BambuMqttClient,
pending: PendingTable,
next_category: NextCategoryTable,
print_status: Arc<Mutex<PrintStatusCache>>,
sn: String,
next_sequence_id: AtomicU64,
}
@@ -124,9 +230,26 @@ impl BambuMqttSession {
}
let pending: PendingTable = Arc::new(Mutex::new(HashMap::new()));
tokio::spawn(poll_loop_v1(event_loop, pending.clone()));
let next_category: NextCategoryTable = Arc::new(Mutex::new(HashMap::new()));
let print_status = Arc::new(Mutex::new(PrintStatusCache::new()));
let sn_owned = sn.to_string();
tokio::spawn(poll_loop_v1(
event_loop,
pending.clone(),
next_category.clone(),
print_status.clone(),
BambuProtocolVersion::V1,
sn_owned,
));
Ok(Self { client: BambuMqttClient::V1(client), pending, sn: sn.to_string(), next_sequence_id: AtomicU64::new(1) })
Ok(Self {
client: BambuMqttClient::V1(client),
pending,
next_category,
print_status,
sn: sn.to_string(),
next_sequence_id: AtomicU64::new(1),
})
}
async fn open_v2(host: &str, access_code: &str, sn: &str, tls: &BambuTls) -> Result<Self, PrinterError> {
@@ -149,9 +272,26 @@ impl BambuMqttSession {
}
let pending: PendingTable = Arc::new(Mutex::new(HashMap::new()));
tokio::spawn(poll_loop_v2(event_loop, pending.clone()));
let next_category: NextCategoryTable = Arc::new(Mutex::new(HashMap::new()));
let print_status = Arc::new(Mutex::new(PrintStatusCache::new()));
let sn_owned = sn.to_string();
tokio::spawn(poll_loop_v2(
event_loop,
pending.clone(),
next_category.clone(),
print_status.clone(),
BambuProtocolVersion::V2,
sn_owned,
));
Ok(Self { client: BambuMqttClient::V2(client), pending, sn: sn.to_string(), next_sequence_id: AtomicU64::new(1) })
Ok(Self {
client: BambuMqttClient::V2(client),
pending,
next_category,
print_status,
sn: sn.to_string(),
next_sequence_id: AtomicU64::new(1),
})
}
/// Publishes `body` under `device/{sn}/request`, wrapped as
@@ -163,26 +303,15 @@ impl BambuMqttSession {
/// empty placeholder, since it has no way to know a real one) with a
/// fresh one from this session's own counter.
///
/// Only correct for commands that genuinely get *one direct,
/// sequence_id-correlated reply* - `info.get_version` is the
/// canonical example (the envelope schema says as much: "Sequence ID
/// matching the original request"). It is NOT a general "send a
/// command, get the response" primitive - Bambu's own mapping isn't
/// 1:1:
/// - `pushing.pushall` doesn't reply at all - it triggers an *ongoing*
/// `print.push_status` stream, unrelated to this request's
/// `sequence_id`. Use `publish` (fire-and-forget) for it, the same
/// way `send_gcode_line` already does for `gcode_line`.
/// - Even where a field genuinely exists, which report carries it can
/// differ *by protocol version*: `print.calibration` data is bundled
/// inside `print.push_status` on BambuMQTT v1, but v2 gives it its
/// own dedicated report. Calling `request` for something like that
/// would just hang until `REQUEST_TIMEOUT` waiting for a reply that
/// was never going to arrive correlated by `sequence_id` - not a
/// crash, but a silent design bug if used on the wrong kind of
/// command. An ongoing-stream subscription (register interest in a
/// *category*, get every report under it as it arrives, not a single
/// `oneshot`) is a separate primitive this doesn't provide yet.
/// Only use this for a command *empirically confirmed* to echo the
/// sequence_id you send back on its reply - `info.get_version` is
/// verified to do this against real hardware. Don't assume every
/// command does just because the field exists on every message (they
/// all compose the same envelope schema): `print.push_status` also
/// has a `sequence_id`, but a live P1P proved the printer stamps it
/// with its own free-running counter regardless of what triggered
/// it - `pushall` uses `next_category` below instead, precisely
/// because this method's assumption doesn't hold for it.
pub async fn request(&self, category: &str, mut body: Value) -> Result<Value, PrinterError> {
let sequence_id = self.next_sequence_id.fetch_add(1, Ordering::Relaxed).to_string();
body["sequence_id"] = Value::String(sequence_id.clone());
@@ -205,6 +334,51 @@ impl BambuMqttSession {
}
}
/// Publishes `body` under `device/{sn}/request`, wrapped as
/// `{category: body}`, then waits up to `NEXT_CATEGORY_TIMEOUT` for
/// the *next* `reply_category`/`reply_command` report to arrive - not
/// matched by any ID, just whichever one shows up next. For a command
/// whose reply carries no usable correlation ID at all (see
/// `request`'s doc comment) - currently just `pushall`, but written
/// generically since nothing about the mechanism is push_status-
/// specific.
///
/// Matching needs *both* `reply_category` and `reply_command`, not
/// category alone - caught live: matching on category only let an
/// unrelated "print" message (that category carries many commands -
/// `gcode_line`, `pause`, `push_status`, ...) satisfy a `pushall` wait
/// before the real `push_status` reply had even arrived, so
/// `pushall()` returned `Ok` having consumed the wrong message
/// entirely, and the in-memory cache never actually got its update.
pub async fn request_next_of_category(
&self,
category: &str,
body: Value,
reply_category: &str,
reply_command: &str,
) -> Result<Value, PrinterError> {
let key = (reply_category.to_string(), reply_command.to_string());
let (tx, rx) = oneshot::channel();
self.next_category.lock().await.insert(key.clone(), tx);
if let Err(err) = self.publish(category, &body).await {
self.next_category.lock().await.remove(&key);
return Err(err);
}
match tokio::time::timeout(NEXT_CATEGORY_TIMEOUT, rx).await {
Ok(Ok(response)) => Ok(response),
Ok(Err(_)) => Err(PrinterError(format!("{} closed before replying", self.sn))),
Err(_) => {
self.next_category.lock().await.remove(&key);
Err(PrinterError(format!(
"{} sent no {reply_category}/{reply_command} within {NEXT_CATEGORY_TIMEOUT:?}",
self.sn
)))
}
}
}
/// Publishes `body` under `device/{sn}/request`, wrapped as
/// `{category: body}`, without waiting for a reply - for commands
/// like `gcode_line` that only get acknowledged implicitly (the
@@ -231,59 +405,138 @@ impl BambuMqttSession {
.map_err(|err| PrinterError(format!("publish failed: {err}"))),
}
}
/// The current in-memory `print.push_status` object, if at least one
/// has arrived since connecting - `None` until then. Cloned out from
/// behind the lock rather than handing out a guard, so callers aren't
/// holding the poll loop's own lock while they use it.
pub async fn print_status(&self) -> Option<PrintReport> {
self.print_status.lock().await.report.clone()
}
}
/// Runs forever, polling the v3.1.1 eventloop and completing pending
/// requests as matching reports arrive. Ends (silently) once the
/// connection is gone for good - dropping the `BambuMqttSession` drops
/// its `AsyncClient`, which is what eventually makes that happen.
async fn poll_loop_v1(mut event_loop: rumqttc::EventLoop, pending: PendingTable) {
/// Runs forever, polling the v3.1.1 eventloop and routing incoming
/// reports (pending-request correlation, `print.push_status` caching).
/// Exits - loudly, not silently like before (that silence is exactly
/// what made a real dropped connection to a live H2C look identical to
/// "the printer just isn't sending anything") - once the connection is
/// gone for good. Dropping the `BambuMqttSession` drops its `AsyncClient`,
/// which is what eventually makes that happen during normal shutdown;
/// this also fires on an unexpected disconnect, which is the case this
/// logging is actually for. No reconnect logic yet - that's the next
/// thing to add here, not something to fake by pretending the loop
/// didn't end.
async fn poll_loop_v1(
mut event_loop: rumqttc::EventLoop,
pending: PendingTable,
next_category: NextCategoryTable,
print_status: Arc<Mutex<PrintStatusCache>>,
version: BambuProtocolVersion,
sn: String,
) {
loop {
match event_loop.poll().await {
Ok(Event::Incoming(Packet::Publish(publish))) => {
complete_pending(&pending, &publish.payload).await;
handle_incoming_report(&pending, &next_category, &print_status, version, &publish.payload).await;
}
Ok(_) => {}
Err(_) => return,
Err(err) => {
eprintln!("[{sn}] MQTT v3.1.1 connection lost, no longer receiving reports: {err}");
return;
}
}
}
}
async fn poll_loop_v2(mut event_loop: rumqttc::v5::EventLoop, pending: PendingTable) {
async fn poll_loop_v2(
mut event_loop: rumqttc::v5::EventLoop,
pending: PendingTable,
next_category: NextCategoryTable,
print_status: Arc<Mutex<PrintStatusCache>>,
version: BambuProtocolVersion,
sn: String,
) {
loop {
match event_loop.poll().await {
Ok(rumqttc::v5::Event::Incoming(PacketV5::Publish(publish))) => {
complete_pending(&pending, &publish.payload).await;
handle_incoming_report(&pending, &next_category, &print_status, version, &publish.payload).await;
}
Ok(_) => {}
Err(_) => return,
Err(err) => {
eprintln!("[{sn}] MQTT v5 connection lost, no longer receiving reports: {err}");
return;
}
}
}
}
/// Parses an incoming report and, if its `sequence_id` matches something
/// `request` is waiting on, hands it the *unwrapped* body. A report with
/// no `sequence_id`, or one nothing's waiting on (an unsolicited status
/// push, not a reply to anything), is silently ignored - most Bambu
/// reports aren't replies to a specific request at all.
/// Parses an incoming report, unwraps its single category key (a report
/// is `{category: {"sequence_id": "...", ...}}`, same shape a request is
/// published as - both `request` and `request_next_of_category` publish
/// `body` and want `body`-shaped data back, not the wrapper around it),
/// and:
///
/// A report is `{category: {"sequence_id": "...", ...}}` - one key, same
/// shape a request is published as. `request` publishes `body` and wants
/// `body`-shaped data back, not the wrapper around it, so this looks
/// inside whichever single category key is present (rather than
/// requiring the caller to already know it) and sends only that inner
/// value through - confirmed against a real printer's `get_version`
/// reply, which failed to parse as `GetVersionResponse` before this
/// unwrapped it (real device, real bug: `serde_json::from_value` on the
/// still-wrapped `{"info": {...}}` reported "missing field `module`",
/// since `module` lives one level deeper than that).
async fn complete_pending(pending: &PendingTable, payload: &[u8]) {
/// 1. If its `sequence_id` matches something `request` is waiting on,
/// hands it the unwrapped body.
/// 2. If its `(category, command)` matches something
/// `request_next_of_category` is waiting on, hands it the unwrapped
/// body too - independent of 1, since the whole reason this table
/// exists is replies with no usable correlation ID.
/// 3. If the unwrapped body is a `print.push_status`, folds it into the
/// running `PrintStatusCache` (which decides full-vs-merge on its own,
/// from the message's own `msg` field - not from whether step 1 or 2
/// matched, see `PrintStatusCache::apply`'s doc comment for why not)
/// - and *before* step 4 wakes anyone up, not after (see 4).
/// 4. Only now - once the cache is guaranteed up to date - actually sends
/// on whichever oneshot(s) matched in steps 1/2, waking `request`/
/// `request_next_of_category`.
///
/// The order of 3 and 4 is load-bearing, caught live: sending on the
/// oneshot first let a waiter's task resume and check `print_status()`
/// before this function had gotten around to updating it - `pushall()`
/// returned `Ok`, and the very next `print_status()` call still saw
/// `None`. Nothing here awaits between the cache write and the sends
/// below, so that race is now closed.
///
/// A report matching neither table isn't an error - most Bambu reports
/// (including every organic `push_status` push) aren't a reply to
/// anything specific.
async fn handle_incoming_report(
pending: &PendingTable,
next_category: &NextCategoryTable,
print_status: &Arc<Mutex<PrintStatusCache>>,
version: BambuProtocolVersion,
payload: &[u8],
) {
let Ok(value) = serde_json::from_slice::<Value>(payload) else { return };
let Some(object) = value.as_object() else { return };
let Some((_, body)) = object.iter().next() else { return };
let Some(sequence_id) = body.get("sequence_id").and_then(Value::as_str) else { return };
let Some((category, body)) = object.iter().next() else { return };
if let Some(sender) = pending.lock().await.remove(sequence_id) {
let command = body.get("command").and_then(Value::as_str);
let request_sender =
match body.get("sequence_id").and_then(Value::as_str) {
Some(sequence_id) => pending.lock().await.remove(sequence_id),
None => None,
};
let next_category_sender = match command {
Some(command) => next_category.lock().await.remove(&(category.clone(), command.to_string())),
None => None,
};
// push_status doesn't need to know whether this client solicited it -
// full-vs-partial is self-described by the message's own `msg` field,
// see `PrintStatusCache::apply`.
if category == "print" && command == Some("push_status") {
if let Ok(report) = serde_json::from_value::<PrintReport>(body.clone()) {
print_status.lock().await.apply(report, version);
}
}
if let Some(sender) = request_sender {
let _ = sender.send(body.clone());
}
if let Some(sender) = next_category_sender {
let _ = sender.send(body.clone());
}
}