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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01MWQVtBGxHYLkVw2kBupd5g

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<super::GetVersionResponse, PrinterError> {
         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<serde_json::Value, PrinterError> {
+        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

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<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);
+/// 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<Mutex<PrintStatusCache>>,
     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<Value, PrinterError> {
         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<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
-    /// 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<Mutex<PrintStatusCache>>,
     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<Mutex<PrintStatusCache>>,
     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<Mutex<PrintStatusCache>>,
     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) {
+    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 category == "print" && body.get("command").and_then(Value::as_str) == Some("push_status") {
         if let Ok(report) = serde_json::from_value::<PrintReport>(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());
-    }
 }