Fix printer send + live telemetry; add hardware-in-the-loop tests

spoolbook-rs/examples/stop_print.rs

@@ -0,0 +1,91 @@
+// Emergency / manual cancel: connect, publish the `stop` command, watch gcode_state settle.
+//
+//   cargo run --example stop_print -- <ip> <access_code> <serial>
+use rumqttc::{AsyncClient, Event, MqttOptions, Packet, QoS, TlsConfiguration, Transport};
+use std::time::Duration;
+
+#[derive(Debug)]
+struct NoCertVerification;
+
+impl rustls::client::danger::ServerCertVerifier for NoCertVerification {
+    fn verify_server_cert(
+        &self,
+        _end_entity: &rustls::pki_types::CertificateDer<'_>,
+        _intermediates: &[rustls::pki_types::CertificateDer<'_>],
+        _server_name: &rustls::pki_types::ServerName<'_>,
+        _ocsp_response: &[u8],
+        _now: rustls::pki_types::UnixTime,
+    ) -> Result<rustls::client::danger::ServerCertVerified, rustls::Error> {
+        Ok(rustls::client::danger::ServerCertVerified::assertion())
+    }
+    fn verify_tls12_signature(
+        &self,
+        _message: &[u8],
+        _cert: &rustls::pki_types::CertificateDer<'_>,
+        _dss: &rustls::DigitallySignedStruct,
+    ) -> Result<rustls::client::danger::HandshakeSignatureValid, rustls::Error> {
+        Ok(rustls::client::danger::HandshakeSignatureValid::assertion())
+    }
+    fn verify_tls13_signature(
+        &self,
+        _message: &[u8],
+        _cert: &rustls::pki_types::CertificateDer<'_>,
+        _dss: &rustls::DigitallySignedStruct,
+    ) -> Result<rustls::client::danger::HandshakeSignatureValid, rustls::Error> {
+        Ok(rustls::client::danger::HandshakeSignatureValid::assertion())
+    }
+    fn supported_verify_schemes(&self) -> Vec<rustls::SignatureScheme> {
+        rustls::crypto::CryptoProvider::get_default().map(|p| p.signature_verification_algorithms.supported_schemes()).unwrap_or_default()
+    }
+}
+
+fn tls_config() -> rustls::ClientConfig {
+    rustls::ClientConfig::builder().dangerous().with_custom_certificate_verifier(std::sync::Arc::new(NoCertVerification)).with_no_client_auth()
+}
+
+#[tokio::main]
+async fn main() {
+    let args: Vec<String> = std::env::args().collect();
+    let [_, ip, access_code, serial] = args.as_slice() else {
+        eprintln!("usage: stop_print <ip> <access_code> <serial>");
+        std::process::exit(1);
+    };
+
+    let mut mqttoptions = MqttOptions::new(format!("spoolbook-stop-{}", std::process::id()), ip.clone(), 8883);
+    mqttoptions.set_credentials("bblp", access_code);
+    mqttoptions.set_keep_alive(Duration::from_secs(30));
+    mqttoptions.set_transport(Transport::tls_with_config(TlsConfiguration::Rustls(std::sync::Arc::new(tls_config()))));
+    mqttoptions.set_max_packet_size(1024 * 1024, 1024 * 1024);
+
+    let (client, mut eventloop) = AsyncClient::new(mqttoptions, 10);
+    let report_topic = format!("device/{serial}/report");
+    let request_topic = format!("device/{serial}/request");
+    client.subscribe(&report_topic, QoS::AtMostOnce).await.expect("subscribe");
+
+    let mut sent = false;
+    for _ in 0..40 {
+        match eventloop.poll().await {
+            Ok(Event::Incoming(Packet::ConnAck(_))) => {
+                let seq = format!("{}", std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_millis());
+                let payload = format!(r#"{{"print":{{"command":"stop","param":"","sequence_id":"{seq}"}}}}"#);
+                client.publish(&request_topic, QoS::AtMostOnce, false, payload).await.expect("publish stop");
+                println!("STOP sent (seq {seq})");
+                sent = true;
+            }
+            Ok(Event::Incoming(Packet::Publish(p))) => {
+                let body = String::from_utf8_lossy(&p.payload);
+                if let Some(i) = body.find("\"gcode_state\"") {
+                    println!("gcode_state {}", &body[i..(i + 30).min(body.len())]);
+                }
+            }
+            Ok(_) => {}
+            Err(e) => {
+                eprintln!("poll error: {e}");
+                if sent {
+                    return;
+                }
+            }
+        }
+    }
+    println!("done polling");
+}

spoolbook-rs/src/printer_mqtt.rs

@@ -182,7 +182,10 @@ async fn purge_stale_jobs_loop(pool: SqlitePool) {
 // polling the eventloop is simultaneously what maintains the connection and what delivers
 // messages. So unlike the .NET original's connect-then-watchdog-loop shape, this is one loop
 // that does both; behaviorally equivalent (connected, receiving, reconnecting on failure).
-async fn connect_and_subscribe_loop(
+// pub rather than private: driven directly by tests/printer_live.rs against a real printer on
+// the LAN (no mock -- the bugs this catches are all "the firmware's own wire format / packet
+// size", invisible to anything but real hardware). Same testing boundary as handle_message.
+pub async fn connect_and_subscribe_loop(
     printer_id: i64,
     ip_address: String,
     access_code: String,
@@ -202,23 +205,28 @@ async fn connect_and_subscribe_loop(
         mqttoptions.set_keep_alive(Duration::from_secs(30));
         mqttoptions.set_transport(Transport::tls_with_config(TlsConfiguration::Rustls(Arc::new(tls_config()))));
         // rumqttc's default incoming limit (10KB) is smaller than this printer's real device/report
-        // payload (~14KB, confirmed against a real P2S -- the connection ConnAcks and SubAcks fine,
-        // then errors out the instant the first real status report arrives, since nothing here ever
-        // subscribed to real hardware's own report size before). 128KB is a comfortable margin.
-        mqttoptions.set_max_packet_size(128 * 1024, 128 * 1024);
+        // payload (~14KB delta, confirmed against a real P2S -- the connection ConnAcks and SubAcks
+        // fine, then errors out the instant the first real status report arrives). The pushall
+        // *full* object (every AMS tray + the whole HMS array) is bigger again, so 128KB wasn't
+        // always enough -- an oversized-packet poll error right after ConnAck is exactly the
+        // "connects, flips to Live, then drops to No live job data" flap. 1MB is well clear.
+        mqttoptions.set_max_packet_size(1024 * 1024, 1024 * 1024);
 
         let (client, mut eventloop) = AsyncClient::new(mqttoptions, 10);
 
         let topic = format!("device/{serial_number}/report");
         if client.subscribe(&topic, QoS::AtMostOnce).await.is_err() {
+            eprintln!("[mqtt {printer_id}] subscribe to {topic} failed, retrying in 15s");
             tokio::time::sleep(Duration::from_secs(15)).await;
             continue;
         }
+        eprintln!("[mqtt {printer_id}] connecting to {ip_address}:8883");
         store.write().await.entry(printer_id).or_default().client = Some(client.clone());
 
         loop {
             match eventloop.poll().await {
                 Ok(Event::Incoming(Packet::ConnAck(_))) => {
+                    eprintln!("[mqtt {printer_id}] connected, sending pushall");
                     store.write().await.entry(printer_id).or_default().connected = true;
                     // Bambu's broker only emits a full status object (gcode_state, task_id, ams,
                     // the pause/HMS reason) on request or on its own slow (~minutes) schedule --
@@ -233,11 +241,21 @@ async fn connect_and_subscribe_loop(
                 }
                 Ok(Event::Incoming(Packet::Publish(publish))) => {
                     if let Ok(payload) = std::str::from_utf8(&publish.payload) {
+                        // Set SPOOLBOOK_MQTT_DEBUG=1 to dump every raw report to stderr -- the
+                        // only way to see the printer's own pause/HMS reason, gcode_state, and
+                        // error codes, none of which the parser keeps. Off by default (a real
+                        // report is ~14KB).
+                        if std::env::var("SPOOLBOOK_MQTT_DEBUG").is_ok() {
+                            eprintln!("[mqtt {printer_id}] report: {payload}");
+                        }
                         handle_message(printer_id, payload, &pool, &store, &mut active_task_id, &camera_registry).await;
                     }
                 }
                 Ok(_) => {}
-                Err(_) => break,
+                Err(e) => {
+                    eprintln!("[mqtt {printer_id}] disconnected: {e} -- reconnecting in 15s");
+                    break;
+                }
             }
         }
 
@@ -312,13 +330,17 @@ async fn snapshot_for_end_job(store: &LiveStatusStore, printer_id: i64) -> (Opti
 // than opening a new one per action — so a command fails outright (rather than queuing) if that
 // connection happens to be mid-reconnect.
 pub async fn publish_command(store: &LiveStatusStore, printer_id: i64, serial_number: &str, command: &str) -> Result<(), String> {
-    let payload = serde_json::json!({ "print": { "command": command, "sequence_id": uuid_v4() } }).to_string();
+    // sequence_id must be a stringified integer, not a uuid: the P2S firmware ignores pause/
+    // resume/stop commands whose sequence_id isn't numeric (the start-print "project_file" path
+    // already uses a numeric one via submission_id() and works). This is what makes the on-screen
+    // "select Resume to retry" prompt after an HMS actually respond to spoolbook's Resume button.
+    let payload = serde_json::json!({ "print": { "command": command, "param": "", "sequence_id": crate::send_print::submission_id() } }).to_string();
     publish_raw(store, printer_id, serial_number, payload).await
 }
 
 // Shared by publish_command above and send_print.rs's "project_file" start-print command —
 // both just publish an already-built payload on the same live connection telemetry holds open.
-pub(crate) async fn publish_raw(store: &LiveStatusStore, printer_id: i64, serial_number: &str, payload: String) -> Result<(), String> {
+pub async fn publish_raw(store: &LiveStatusStore, printer_id: i64, serial_number: &str, payload: String) -> Result<(), String> {
     let client = store.read().await.get(&printer_id).and_then(|s| s.client.clone());
     let Some(client) = client else {
         return Err("Printer isn't connected — telemetry link is down or still reconnecting.".to_string());

spoolbook-rs/src/printer_telemetry.rs

@@ -19,21 +19,36 @@ pub struct PrinterJob {
 
 const COLUMNS: &str = "id, printer_id, external_job_id, started_at, ended_at, print_id";
 
+// Called from the MQTT eventloop task (printer_mqtt::handle_message) for every active-state
+// reading. A DB error here must not unwind -- a panic in that spawned task kills the whole
+// telemetry+control connection for the printer (pause/resume/stop stop working mid-print) and
+// it doesn't come back until a restart. So this logs and drops the one reading instead.
 pub async fn record_reading(
     pool: &SqlitePool,
     printer_id: i64,
     external_job_id: &str,
     input: &ReadingInput,
     at: Option<&str>,
 ) {
+    if let Err(e) = try_record_reading(pool, printer_id, external_job_id, input, at).await {
+        eprintln!("[telemetry {printer_id}] dropped a reading for job {external_job_id}: {e}");
+    }
+}
+
+async fn try_record_reading(
+    pool: &SqlitePool,
+    printer_id: i64,
+    external_job_id: &str,
+    input: &ReadingInput,
+    at: Option<&str>,
+) -> Result<(), sqlx::Error> {
     let existing_job_id = sqlx::query_scalar::<_, i64>(
         "SELECT id FROM printer_jobs WHERE printer_id = ?1 AND external_job_id = ?2 AND ended_at IS NULL",
     )
     .bind(printer_id)
     .bind(external_job_id)
     .fetch_optional(pool)
-    .await
-    .expect("query failed");
+    .await?;
 
     let is_new_job = existing_job_id.is_none();
     let job_id = match existing_job_id {
@@ -47,8 +62,7 @@ pub async fn record_reading(
         .bind(external_job_id)
         .bind(at)
         .fetch_one(pool)
-        .await
-        .expect("insert failed"),
+        .await?,
     };
 
     sqlx::query(
@@ -66,8 +80,7 @@ pub async fn record_reading(
     .bind(input.layer_num)
     .bind(input.total_layer_num)
     .execute(pool)
-    .await
-    .expect("insert failed");
+    .await?;
 
     // Auto-create-on-send (docs/adr/0017's 2026-08-14 addendum): a brand-new Job attaches
     // straight to the printer's open (InProgress, not yet attached) Print instead of waiting for
@@ -84,18 +97,17 @@ pub async fn record_reading(
         )
         .bind(printer_id)
         .fetch_optional(pool)
-        .await
-        .expect("query failed");
+        .await?;
 
         if let Some(open_print_id) = open_print_id {
             sqlx::query("UPDATE printer_jobs SET print_id = ?1 WHERE id = ?2")
                 .bind(open_print_id)
                 .bind(job_id)
                 .execute(pool)
-                .await
-                .expect("update failed");
+                .await?;
         }
     }
+    Ok(())
 }
 
 #[derive(Serialize, serde::Deserialize, sqlx::FromRow, Clone)]

spoolbook-rs/src/reslicing.rs

@@ -81,7 +81,9 @@ async fn slice_and_save(pool: &SqlitePool, patched_path: &std::path::Path, displ
     result
 }
 
-async fn slice_via_service(patched_path: &std::path::Path) -> Result<Vec<u8>, String> {
+// pub for tests/printer_live.rs's non-sliced -> slice -> print flow against the real
+// slicer-service container. Takes any .3mf on disk; the caller patches settings first (or not).
+pub async fn slice_via_service(patched_path: &std::path::Path) -> Result<Vec<u8>, String> {
     let base_url = std::env::var("RESLICE_SERVICE_URL").unwrap_or_else(|_| "http://localhost:8100".to_string());
     let bytes = std::fs::read(patched_path).map_err(|e| e.to_string())?;
     let file_name = patched_path.file_name().and_then(|n| n.to_str()).unwrap_or("project.3mf").to_string();

spoolbook-rs/src/send_print.rs

@@ -48,16 +48,18 @@ pub fn build_project_file_payload(
     remote_file_name: &str,
     md5: &str,
     plate_gcode_file_name: &str,
-    use_ams: bool,
-    ams_slot: i64,
+    // Both kept for the wire contract with the frontend's AMS toggle + tray picker, but neither
+    // is sent: spoolbook's .3mf is re-sliced by the headless BambuStudio CLI, whose gcode carries
+    // no per-filament AMS-assignment metadata. With that missing, use_ams:true (with or without
+    // an ams_mapping) fails at HMS 07FF-8012 "Failed to get AMS mapping table" — confirmed
+    // against a real P2S: only use_ams:false prints. The printer then just feeds from whatever
+    // filament is threaded (the AMS tray's PTFE, in practice). Real AMS/multi-material support
+    // needs a file sliced *for* AMS and is a separate feature.
+    _use_ams: bool,
+    _ams_slot: i64,
     is_p2s: bool,
     submission_id: &str,
 ) -> String {
-    let flat_ams_mapping: Vec<i64> = if use_ams { vec![ams_slot] } else { vec![] };
-    // Global tray ID = ams_id*4 + slot_id (bambuddy's "regular AMS tray" case — spoolbook only
-    // targets a single onboard AMS unit, not AMS-HT/external-spool/multi-nozzle setups).
-    let ams_mapping2: Vec<serde_json::Value> =
-        flat_ams_mapping.iter().map(|t| json!({ "ams_id": t / 4, "slot_id": t % 4 })).collect();
     let subtask_name = std::path::Path::new(remote_file_name).file_stem().and_then(|s| s.to_str()).unwrap_or(remote_file_name);
 
     json!({
@@ -80,7 +82,9 @@ pub fn build_project_file_payload(
             // specifically for P2S/N7.
             "vibration_cali": !is_p2s,
             "layer_inspect": false,
-            "use_ams": use_ams,
+            // Forced false — see the doc comment on the _use_ams param. A re-sliced spoolbook
+            // .3mf has no AMS filament data, so use_ams:true fails at HMS 07FF-8012.
+            "use_ams": false,
             "cfg": "0",
             "extrude_cali_flag": 2,
             "extrude_cali_manual_mode": 0,
@@ -93,8 +97,16 @@ pub fn build_project_file_payload(
             "project_id": submission_id,
             "subtask_id": submission_id,
             "task_id": submission_id,
-            "ams_mapping": flat_ams_mapping,
-            "ams_mapping2": ams_mapping2,
+            // Deliberately empty, never a per-tray [tray_id]. spoolbook's .3mf is re-sliced by the
+            // headless BambuStudio CLI, whose gcode omits the per-filament AMS-assignment metadata
+            // the printer cross-references an incoming mapping against — supplying one (even a
+            // correct-looking [ams_id*4+slot_id]) fails at HMS 07FF-8012 "Failed to get AMS
+            // mapping table" (confirmed against a real P2S: empty prints fine, [1] does not).
+            // Empty tells the printer to feed from its currently-active AMS tray, matching Bambu
+            // Handy's own single-colour default. True per-tray selection needs a file sliced *for*
+            // AMS and isn't supported here yet.
+            "ams_mapping": [],
+            "ams_mapping2": [],
         }
     })
     .to_string()