From 1fd6529376c677c9480665d1a89f09e9704489db Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Iwo=20Strzebo=C5=84ski?= Date: Mon, 7 Sep 2026 14:55:09 +0000 Subject: [PATCH] Add get_calibration; fix a real 10KB packet-size ceiling; revert bad correlation-field experiment Real bug found and fixed live: both rumqttc clients default to a 10KB max packet size in both directions. A real H2C's get_version reply (25625 bytes - it reports a lot of modules: dual AMS, AMS HT, six WTM units, ...) blew straight through that on the v5 client and killed the whole connection outright (a real MQTT protocol error, not a timeout). Raised to 1MB on both clients - print.push_status reports can get large too. Also tried, and reverted after live-testing disproved it: replacing request()'s sequence_id correlation with a homemade continuum_correlation_id field, on the understanding that most replies echo back arbitrary extra request fields. True in general, false for get_version specifically - a real H2C's reply had exactly ["command", "module", "sequence_id"], nothing extra survived, so every get_version call started timing out. Reverted to sequence_id, which is what's actually proven (twice now, on both protocol versions) to round-trip for this command. Added BambuPrinter::get_calibration(option) - print.calibration gets a real, correlated reply (confirmed live: it does echo sequence_id) on both protocol versions, per the schema fixes in continuum-schemas. `option` is the request's own bitmask, left to the caller rather than defaulled to "run everything" - this triggers a real physical calibration sequence on the printer, not a read-only query like get_version, so it hasn't been live-tested here without explicit confirmation first. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01MWQVtBGxHYLkVw2kBupd5g --- src/printer/bambu.rs | 39 +++++- src/printer/bambu_mqtt.rs | 245 ++++++++++++-------------------------- 2 files changed, 114 insertions(+), 170 deletions(-) diff --git a/src/printer/bambu.rs b/src/printer/bambu.rs index d49df78..39fc5a9 100644 --- a/src/printer/bambu.rs +++ b/src/printer/bambu.rs @@ -384,9 +384,9 @@ impl BambuPrinter { /// not through the shared trait. /// /// The first real use of `BambuMqttSession::request` - - /// `info.get_version` is a genuine one-shot, `sequence_id`-correlated - /// reply (unlike `pushing.pushall`, see `request`'s doc comment), so - /// this is exactly the case it's for. + /// `info.get_version` is a genuine one-shot, correlated reply (unlike + /// `pushing.pushall`, see `request`'s doc comment), so this is exactly + /// the case it's for. pub async fn get_version(&self) -> Result { let (session, version) = self.connection.as_ref().ok_or_else(|| PrinterError(format!("{} is not connected", self.base.name)))?; @@ -403,6 +403,39 @@ impl BambuPrinter { .map_err(|err| PrinterError(format!("failed to parse get_version response: {err}"))) } + /// Bambu-only, same reasoning as `get_version` above. Sends + /// `print.calibration` and waits for the correlated `print.calibration` + /// *report* it triggers - confirmed live on both protocol versions, + /// which respond identically here (an earlier, incorrect assumption + /// had v1 folding this into `push_status` instead of replying + /// directly; fixed in continuum-schemas). The printer also keeps + /// sending ordinary `push_status` updates reflecting progress while + /// calibration runs, same as any other job - read those via + /// `print_status()` if you want progress, this call just confirms the + /// calibration sequence started and got acknowledged. + /// + /// `option` is the bitmask this command's own schema defines: + /// `lidarCalibration = 1`, `bedLevelling = 1 << 1`, + /// `vibrationCompensation = 1 << 2`, `motorCancellation = 1 << 3` (OR + /// together whichever steps you want). Deliberately not defaulted to + /// "everything" here - this starts a real, physical calibration + /// sequence on the printer, so the caller states exactly what they + /// mean to trigger rather than this method guessing. + pub async fn get_calibration(&self, option: i64) -> Result { + let (session, version) = + self.connection.as_ref().ok_or_else(|| PrinterError(format!("{} is not connected", self.base.name)))?; + + let mut body = continuum_schema_instance::build_instance_from_file(&get_schema_for_command( + *version, + "print".into(), + "calibration".into(), + )) + .map_err(|err| PrinterError(format!("failed to build calibration request: {err}")))?; + body["option"] = serde_json::Value::from(option); + + session.request("print", body).await + } + /// Bambu-only, same reasoning as `get_version` above. Publishes /// `pushing.pushall`, which - unlike `get_version` - doesn't reply /// under its own category: it triggers/refreshes the ongoing diff --git a/src/printer/bambu_mqtt.rs b/src/printer/bambu_mqtt.rs index 10e27ff..6635d46 100644 --- a/src/printer/bambu_mqtt.rs +++ b/src/printer/bambu_mqtt.rs @@ -20,19 +20,21 @@ //! long-running background job that has to run independently of //! whatever call site is waiting on a response. //! -//! 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. +//! Correlating a reply to its request isn't as uniform as it looks - +//! confirmed live, not assumed. `info.get_version`'s reply echoes back +//! the `sequence_id` you sent; `request` relies on that. A tempting +//! generalization - add your own extra field instead of `sequence_id`, +//! since most replies are supposed to echo back whatever extra fields +//! a request carried - turned out false for this exact command: a real +//! H2C's `get_version` reply had precisely `["command", "module", +//! "sequence_id"]`, nothing else, so a homemade correlation field never +//! survived and every request timed out. `sequence_id` - proven to +//! round-trip, not just assumed to - is what's actually used. +//! `print.push_status` doesn't echo `sequence_id` either (the printer +//! stamps that with its own free-running counter regardless of what +//! triggered it) - but it was never a request/reply case to begin with +//! (`pushing.pushall` triggers an ongoing *stream*, not a single +//! reply) - see `PrintStatusCache` for how that's handled instead. //! 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 @@ -67,29 +69,22 @@ use super::PrinterError; /// 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 +/// `get_version` (see this module's doc comment - a homemade correlation +/// field was tried instead and confirmed *not* to survive on this +/// command's reply). Shared between whichever call registered one (in /// `request`) and the background poll task that completes it. type PendingTable = Arc>>>; -/// 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>>>; - 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); +/// Both rumqttc clients default to a 10KB max packet size in *both* +/// directions - fine for a request, nowhere near enough for a reply: +/// confirmed live against a real H2C, whose `get_version` reply alone +/// (25625 bytes - it reports a lot of modules: dual AMS, AMS HT, six WTM +/// units, ...) blew straight through that limit and killed the whole +/// connection outright (a real MQTT v5 protocol error, not a timeout). +/// `print.push_status` reports (full AMS/xcam state) can get large too. +/// 1MB has real headroom for that. +const MAX_PACKET_SIZE: usize = 1024 * 1024; enum BambuMqttClient { V1(rumqttc::AsyncClient), @@ -106,15 +101,12 @@ enum BambuMqttClient { /// "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. +/// forced/full), confirmed against a real printer. `push_status` also +/// doesn't echo `sequence_id` (see this module's doc comment), so it +/// couldn't be identified via `request`'s correlation even if the +/// timing worked out; `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 @@ -167,7 +159,6 @@ impl PrintStatusCache { pub struct BambuMqttSession { client: BambuMqttClient, pending: PendingTable, - next_category: NextCategoryTable, print_status: Arc>, sn: String, next_sequence_id: AtomicU64, @@ -213,6 +204,7 @@ impl BambuMqttSession { options.set_credentials("bblp", access_code); options.set_transport(Transport::tls_with_config(TlsConfiguration::NativeConnector(connector))); options.set_keep_alive(Duration::from_secs(30)); + options.set_max_packet_size(MAX_PACKET_SIZE, MAX_PACKET_SIZE); let (client, mut event_loop) = rumqttc::AsyncClient::new(options, 10); client @@ -230,22 +222,13 @@ impl BambuMqttSession { } let pending: PendingTable = Arc::new(Mutex::new(HashMap::new())); - 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, - )); + tokio::spawn(poll_loop_v1(event_loop, pending.clone(), print_status.clone(), BambuProtocolVersion::V1, sn_owned)); Ok(Self { client: BambuMqttClient::V1(client), pending, - next_category, print_status, sn: sn.to_string(), next_sequence_id: AtomicU64::new(1), @@ -259,6 +242,7 @@ impl BambuMqttSession { options.set_credentials("bblp", access_code); options.set_transport(Transport::tls_with_config(TlsConfiguration::NativeConnector(connector))); options.set_keep_alive(Duration::from_secs(30)); + options.set_max_packet_size(Some(MAX_PACKET_SIZE as u32)); let (client, mut event_loop) = rumqttc::v5::AsyncClient::new(options, 10); client @@ -272,22 +256,13 @@ impl BambuMqttSession { } let pending: PendingTable = Arc::new(Mutex::new(HashMap::new())); - 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, - )); + tokio::spawn(poll_loop_v2(event_loop, pending.clone(), print_status.clone(), BambuProtocolVersion::V2, sn_owned)); Ok(Self { client: BambuMqttClient::V2(client), pending, - next_category, print_status, sn: sn.to_string(), next_sequence_id: AtomicU64::new(1), @@ -298,20 +273,24 @@ impl BambuMqttSession { /// `{category: body}` (Bambu's top-level key names the *category*, /// e.g. "print"/"info" - `body`'s own `command` field says which /// specific one), and waits up to `REQUEST_TIMEOUT` for a report - /// whose `sequence_id` matches. Overwrites whatever `sequence_id` - /// `body` already had (continuum-schema-instance fills it with an - /// empty placeholder, since it has no way to know a real one) with a - /// fresh one from this session's own counter. + /// whose `sequence_id` matches the one this call sent. /// - /// 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. + /// This crate tried adding its own correlation field + /// (`continuum_correlation_id`) instead of relying on `sequence_id`, + /// on the understanding that extra fields survive into most replies. + /// Confirmed live that this doesn't hold for `get_version` specifically + /// - a real H2C's reply had exactly `["command", "module", + /// "sequence_id"]`, nothing extra at all, so the correlation field + /// never came back and every `get_version` call just timed out. + /// `sequence_id` is what's actually proven to round-trip for it + /// (confirmed earlier, live, on both protocol versions), so that's + /// what this uses. Whether a custom field survives on some *other* + /// command (`calibration`, say) is unverified, not confirmed false - + /// this was only ever checked for `get_version`. + /// + /// Not for `print.push_status` - that's not a single reply, see + /// `PrintStatusCache`. Not for a command with no reply at all either, + /// like `gcode_line` - use `publish` (fire-and-forget) for those. pub async fn request(&self, category: &str, mut body: Value) -> Result { let sequence_id = self.next_sequence_id.fetch_add(1, Ordering::Relaxed).to_string(); body["sequence_id"] = Value::String(sequence_id.clone()); @@ -334,55 +313,11 @@ 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 { - 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 - /// printer just executes it), not answered directly. + /// like `gcode_line` and `pushing.pushall` that don't get a direct + /// correlated reply (an implicit ack the printer just executes, or an + /// ongoing stream instead of a single answer, respectively). pub async fn publish(&self, category: &str, body: &Value) -> Result<(), PrinterError> { self.publish_raw(serde_json::json!({ category: body }).to_string()).await } @@ -429,7 +364,6 @@ impl BambuMqttSession { async fn poll_loop_v1( mut event_loop: rumqttc::EventLoop, pending: PendingTable, - next_category: NextCategoryTable, print_status: Arc>, version: BambuProtocolVersion, sn: String, @@ -437,7 +371,7 @@ async fn poll_loop_v1( loop { match event_loop.poll().await { Ok(Event::Incoming(Packet::Publish(publish))) => { - handle_incoming_report(&pending, &next_category, &print_status, version, &publish.payload).await; + handle_incoming_report(&pending, &print_status, version, &publish.payload).await; } Ok(_) => {} Err(err) => { @@ -451,7 +385,6 @@ async fn poll_loop_v1( async fn poll_loop_v2( mut event_loop: rumqttc::v5::EventLoop, pending: PendingTable, - next_category: NextCategoryTable, print_status: Arc>, version: BambuProtocolVersion, sn: String, @@ -459,7 +392,7 @@ async fn poll_loop_v2( loop { match event_loop.poll().await { Ok(rumqttc::v5::Event::Incoming(PacketV5::Publish(publish))) => { - handle_incoming_report(&pending, &next_category, &print_status, version, &publish.payload).await; + handle_incoming_report(&pending, &print_status, version, &publish.payload).await; } Ok(_) => {} Err(err) => { @@ -472,38 +405,29 @@ async fn poll_loop_v2( /// 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: +/// published as - `request` publishes `body` and wants `body`-shaped data +/// back, not the wrapper around it), and: /// /// 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`. +/// hands it the unwrapped body - but doesn't wake that waiter yet +/// (see 3 for why). +/// 2. If the unwrapped body is a `print.push_status`, folds it into the +/// running `PrintStatusCache` (which reads the message's own `msg` +/// field to decide full-vs-merge, not anything to do with step 1 - see +/// `PrintStatusCache::apply`'s doc comment for why not). +/// 3. Only now - once the cache is guaranteed up to date - actually sends +/// on whichever oneshot matched in step 1, waking `request`. The order +/// of 2 and 3 is load-bearing, caught live on an earlier version of +/// this mechanism: sending on the oneshot first let a waiter's task +/// resume and check `print_status()` before this function had gotten +/// around to updating it. Nothing here awaits between the cache write +/// and the send, so that race stays closed. /// -/// 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 +/// A report matching nothing pending 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>, version: BambuProtocolVersion, payload: &[u8], @@ -512,22 +436,12 @@ async fn handle_incoming_report( let Some(object) = value.as_object() else { return }; let Some((category, body)) = object.iter().next() else { return }; - 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())), + let request_sender = match body.get("sequence_id").and_then(Value::as_str) { + Some(sequence_id) => pending.lock().await.remove(sequence_id), 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 category == "print" && body.get("command").and_then(Value::as_str) == Some("push_status") { if let Ok(report) = serde_json::from_value::(body.clone()) { print_status.lock().await.apply(report, version); } @@ -536,7 +450,4 @@ async fn handle_incoming_report( if let Some(sender) = request_sender { let _ = sender.send(body.clone()); } - if let Some(sender) = next_category_sender { - let _ = sender.send(body.clone()); - } }