printer telemetry fix

spoolbook-rs/src/lib.rs

@@ -41,6 +41,15 @@ pub fn app_with_live_status(pool: SqlitePool, live_status: printer_mqtt::LiveSta
 }
 
 pub fn app_with_camera(pool: SqlitePool, live_status: printer_mqtt::LiveStatusStore, camera_registry: printer_camera::CameraRegistry) -> Router {
+    app_with_camera_supervised(pool, live_status, camera_registry, printer_mqtt::new_supervisor())
+}
+
+pub fn app_with_camera_supervised(
+    pool: SqlitePool,
+    live_status: printer_mqtt::LiveStatusStore,
+    camera_registry: printer_camera::CameraRegistry,
+    conn_supervisor: printer_mqtt::ConnSupervisor,
+) -> Router {
     filaments::router()
         .merge(colors::router())
         .merge(spools::router())
@@ -62,6 +71,7 @@ pub fn app_with_camera(pool: SqlitePool, live_status: printer_mqtt::LiveStatusSt
         .route("/api/version", get(|| async { Json(serde_json::json!({ "version": env!("CARGO_PKG_VERSION") })) }))
         .layer(Extension(live_status))
         .layer(Extension(camera_registry))
+        .layer(Extension(conn_supervisor))
         // axum's own default (2MB) is well below a real sliced .3mf's size (embedded gcode +
         // thumbnails) -- match project_upload's own MAX_BYTES, the limit it already validates
         // import-url downloads against but this layer is what actually enforces it for uploads.

spoolbook-rs/src/main.rs

@@ -30,7 +30,8 @@ async fn main() {
 
     let live_status = spoolbook_rs::printer_mqtt::new_store();
     let camera_registry = spoolbook_rs::printer_camera::new_registry();
-    spoolbook_rs::printer_mqtt::spawn_all(pool.clone(), live_status.clone(), camera_registry.clone()).await;
+    let conn_supervisor = spoolbook_rs::printer_mqtt::new_supervisor();
+    spoolbook_rs::printer_mqtt::spawn_all(pool.clone(), live_status.clone(), camera_registry.clone(), conn_supervisor.clone()).await;
 
     // Throttled to once/24h via app_settings.last_filament_sync_at, same as the .NET app's
     // Program.cs startup block — silent on failure, the Filaments page's manual sync button
@@ -58,7 +59,7 @@ async fn main() {
     let index_path = std::path::Path::new(&static_root).join("index.html");
     let serve_static = ServeDir::new(&static_root).fallback(ServeFile::new(index_path));
 
-    let app = spoolbook_rs::app_with_camera(pool, live_status, camera_registry).fallback_service(serve_static);
+    let app = spoolbook_rs::app_with_camera_supervised(pool, live_status, camera_registry, conn_supervisor).fallback_service(serve_static);
 
     // 0.0.0.0, not 127.0.0.1: matches .NET's ASPNETCORE_URLS=http://0.0.0.0:5070 (self-hosted,
     // LAN-accessible per CLAUDE.md) and is required for Docker port publishing to reach it at all.

spoolbook-rs/src/printer_mqtt.rs

@@ -29,6 +29,55 @@ pub fn new_store() -> LiveStatusStore {
     Arc::new(RwLock::new(HashMap::new()))
 }
 
+// Abort handles for the per-printer connect_and_subscribe_loop tasks, keyed by printer id.
+// spawn_all fills this at startup; printers.rs create/update/delete call respawn_one/stop_one so a
+// printer configured while the process is already running gets a live connection without a
+// restart. That's the normal case under Docker (fresh data volume, printer added through the UI,
+// container never bounced): before this, the one-shot Test Connection worked but the persistent
+// loop never started, so the card stayed "Not connected" and Print failed at publish_raw with
+// "Printer isn't connected".
+pub type ConnSupervisor = Arc<tokio::sync::Mutex<HashMap<i64, tokio::task::AbortHandle>>>;
+
+pub fn new_supervisor() -> ConnSupervisor {
+    Arc::new(tokio::sync::Mutex::new(HashMap::new()))
+}
+
+// Spawn (or replace) the live-telemetry loop for one printer. Aborts any existing loop for this id
+// first, so an edit that changes the IP doesn't leave the old loop reconnecting to the old
+// address forever with both loops fighting over store[id]. Collapses to just the abort when the
+// printer has no connection details.
+#[allow(clippy::too_many_arguments)]
+pub async fn respawn_one(
+    supervisor: &ConnSupervisor,
+    id: i64,
+    ip_address: Option<String>,
+    access_code: Option<String>,
+    serial_number: Option<String>,
+    pool: SqlitePool,
+    store: LiveStatusStore,
+    camera_registry: crate::printer_camera::CameraRegistry,
+) {
+    let mut sup = supervisor.lock().await;
+    if let Some(handle) = sup.remove(&id) {
+        handle.abort();
+    }
+    store.write().await.remove(&id);
+    if let (Some(ip_address), Some(access_code), Some(serial_number)) = (ip_address, access_code, serial_number) {
+        let task = tokio::spawn(connect_and_subscribe_loop(
+            id, ip_address, access_code, serial_number, pool, store, camera_registry,
+        ));
+        sup.insert(id, task.abort_handle());
+    }
+}
+
+// Drop the live-telemetry loop for a deleted printer.
+pub async fn stop_one(supervisor: &ConnSupervisor, store: &LiveStatusStore, id: i64) {
+    if let Some(handle) = supervisor.lock().await.remove(&id) {
+        handle.abort();
+    }
+    store.write().await.remove(&id);
+}
+
 pub async fn snapshot(store: &LiveStatusStore, printer_id: i64) -> PrinterLiveStatus {
     store.read().await.get(&printer_id).cloned().unwrap_or_default()
 }
@@ -103,7 +152,7 @@ pub(crate) fn tls12_no_verify_config() -> rustls::ClientConfig {
 // default 10KB max packet size is smaller than this printer's real device/report payload
 // (~14KB) — connect_and_subscribe_loop's own set_max_packet_size call below is the fix. No
 // reference test suite to port forward (the .NET original has none either).
-pub async fn spawn_all(pool: SqlitePool, store: LiveStatusStore, camera_registry: crate::printer_camera::CameraRegistry) {
+pub async fn spawn_all(pool: SqlitePool, store: LiveStatusStore, camera_registry: crate::printer_camera::CameraRegistry, supervisor: ConnSupervisor) {
     tokio::spawn(purge_stale_jobs_loop(pool.clone()));
 
     let printers = sqlx::query_as::<_, (i64, Option<String>, Option<String>, Option<String>)>(
@@ -114,9 +163,7 @@ pub async fn spawn_all(pool: SqlitePool, store: LiveStatusStore, camera_registry
     .unwrap_or_default();
 
     for (id, ip_address, access_code, serial_number) in printers {
-        if let (Some(ip_address), Some(access_code), Some(serial_number)) = (ip_address, access_code, serial_number) {
-            tokio::spawn(connect_and_subscribe_loop(id, ip_address, access_code, serial_number, pool.clone(), store.clone(), camera_registry.clone()));
-        }
+        respawn_one(&supervisor, id, ip_address, access_code, serial_number, pool.clone(), store.clone(), camera_registry.clone()).await;
     }
 }
 

spoolbook-rs/src/printers.rs

@@ -1,4 +1,5 @@
-use crate::printer_mqtt::{self, LiveStatusStore};
+use crate::printer_camera::CameraRegistry;
+use crate::printer_mqtt::{self, ConnSupervisor, LiveStatusStore};
 use axum::http::StatusCode;
 use axum::response::sse::{Event as SseEvent, KeepAlive, Sse};
 use axum::{
@@ -186,6 +187,9 @@ async fn is_duplicate(pool: &SqlitePool, name: &str, exclude_id: Option<i64>) ->
 async fn create(
     _editor: crate::auth::Editor,
     State(pool): State<SqlitePool>,
+    Extension(store): Extension<LiveStatusStore>,
+    Extension(camera_registry): Extension<CameraRegistry>,
+    Extension(supervisor): Extension<ConnSupervisor>,
     Json(input): Json<PrinterInput>,
 ) -> (StatusCode, Json<PrinterResult>) {
     if input.name.trim().is_empty() {
@@ -208,12 +212,29 @@ async fn create(
     .await
     .expect("insert failed");
 
+    // Start its live-telemetry loop now, not just at the next restart's spawn_all — a printer
+    // added while the process is already running is the norm under Docker.
+    printer_mqtt::respawn_one(
+        &supervisor,
+        printer.id,
+        printer.ip_address.clone(),
+        printer.access_code.clone(),
+        printer.serial_number.clone(),
+        pool.clone(),
+        store,
+        camera_registry,
+    )
+    .await;
+
     (StatusCode::OK, Json(PrinterResult { ok: true, error: None, printer: Some(printer) }))
 }
 
 async fn update(
     _editor: crate::auth::Editor,
     State(pool): State<SqlitePool>,
+    Extension(store): Extension<LiveStatusStore>,
+    Extension(camera_registry): Extension<CameraRegistry>,
+    Extension(supervisor): Extension<ConnSupervisor>,
     Path(id): Path<i64>,
     Json(input): Json<PrinterInput>,
 ) -> (StatusCode, Json<PrinterResult>) {
@@ -240,12 +261,33 @@ async fn update(
     .expect("update failed");
 
     match printer {
-        Some(printer) => (StatusCode::OK, Json(PrinterResult { ok: true, error: None, printer: Some(printer) })),
+        Some(printer) => {
+            // Connection details may have changed — restart the loop against the new ones (and
+            // drop it entirely if they were cleared). respawn_one aborts the stale loop first.
+            printer_mqtt::respawn_one(
+                &supervisor,
+                printer.id,
+                printer.ip_address.clone(),
+                printer.access_code.clone(),
+                printer.serial_number.clone(),
+                pool.clone(),
+                store,
+                camera_registry,
+            )
+            .await;
+            (StatusCode::OK, Json(PrinterResult { ok: true, error: None, printer: Some(printer) }))
+        }
         None => err(StatusCode::NOT_FOUND, "not_found"),
     }
 }
 
-async fn delete(_editor: crate::auth::Editor, State(pool): State<SqlitePool>, Path(id): Path<i64>) -> (StatusCode, Json<PrinterResult>) {
+async fn delete(
+    _editor: crate::auth::Editor,
+    State(pool): State<SqlitePool>,
+    Extension(store): Extension<LiveStatusStore>,
+    Extension(supervisor): Extension<ConnSupervisor>,
+    Path(id): Path<i64>,
+) -> (StatusCode, Json<PrinterResult>) {
     let has_prints = sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM prints WHERE printer_id = ?1")
         .bind(id)
         .fetch_one(&pool)
@@ -266,5 +308,7 @@ async fn delete(_editor: crate::auth::Editor, State(pool): State<SqlitePool>, Pa
         return err(StatusCode::NOT_FOUND, "not_found");
     }
 
+    printer_mqtt::stop_one(&supervisor, &store, id).await;
+
     (StatusCode::OK, Json(PrinterResult { ok: true, error: None, printer: None }))
 }

spoolbook-rs/tests/printer_respawn.rs

@@ -0,0 +1,89 @@
+// A printer configured while the process is already running (the normal case under Docker: fresh
+// data volume, printer added through the UI, container never bounced) must get its live-telemetry
+// loop started right then -- not only at the next startup's spawn_all. Without it, Test Connection
+// works but the card stays "Not connected" and Print fails with "Printer isn't connected".
+mod common;
+
+use axum::body::Body;
+use axum::http::{Request, StatusCode};
+use serde_json::{Value, json};
+use sqlx::sqlite::SqlitePoolOptions;
+use tower::ServiceExt;
+
+async fn test_pool() -> sqlx::SqlitePool {
+    let pool = SqlitePoolOptions::new()
+        .max_connections(1)
+        .connect("sqlite::memory:")
+        .await
+        .expect("failed to open in-memory db");
+    sqlx::migrate!().run(&pool).await.expect("migration failed");
+    pool
+}
+
+async fn send(
+    pool: &sqlx::SqlitePool,
+    supervisor: &spoolbook_rs::printer_mqtt::ConnSupervisor,
+    method: &str,
+    uri: &str,
+    body: Option<Value>,
+) -> (StatusCode, Value) {
+    let app = spoolbook_rs::app_with_camera_supervised(
+        pool.clone(),
+        spoolbook_rs::printer_mqtt::new_store(),
+        spoolbook_rs::printer_camera::new_registry(),
+        supervisor.clone(),
+    );
+    let body = body.map(|b| b.to_string()).unwrap_or_default();
+    let response = app
+        .oneshot(
+            Request::builder()
+                .method(method)
+                .uri(uri)
+                .header("content-type", "application/json")
+                .header("cookie", common::auth_cookie_header(pool).await)
+                .body(Body::from(body))
+                .unwrap(),
+        )
+        .await
+        .unwrap();
+    let status = response.status();
+    let bytes = axum::body::to_bytes(response.into_body(), usize::MAX).await.unwrap();
+    let json: Value = if bytes.is_empty() { Value::Null } else { serde_json::from_slice(&bytes).unwrap() };
+    (status, json)
+}
+
+#[tokio::test]
+async fn create_then_delete_starts_and_stops_the_telemetry_loop() {
+    let pool = test_pool().await;
+    let supervisor = spoolbook_rs::printer_mqtt::new_supervisor();
+
+    let body = json!({ "name": "P2S", "model": "P2S", "ipAddress": "192.168.1.50", "accessCode": "12345678", "serialNumber": "ABC123" });
+    let (status, created) = send(&pool, &supervisor, "POST", "/api/printers", Some(body)).await;
+    assert_eq!(status, StatusCode::OK);
+    let id = created["printer"]["id"].as_i64().unwrap();
+
+    assert!(
+        supervisor.lock().await.contains_key(&id),
+        "creating a printer with connection details should start its telemetry loop"
+    );
+
+    let (status, _) = send(&pool, &supervisor, "DELETE", &format!("/api/printers/{id}"), None).await;
+    assert_eq!(status, StatusCode::OK);
+    assert!(
+        !supervisor.lock().await.contains_key(&id),
+        "deleting a printer should stop its telemetry loop"
+    );
+}
+
+#[tokio::test]
+async fn create_without_connection_details_registers_no_loop() {
+    let pool = test_pool().await;
+    let supervisor = spoolbook_rs::printer_mqtt::new_supervisor();
+
+    let body = json!({ "name": "bare", "model": "P2S" });
+    let (status, created) = send(&pool, &supervisor, "POST", "/api/printers", Some(body)).await;
+    assert_eq!(status, StatusCode::OK);
+    let id = created["printer"]["id"].as_i64().unwrap();
+
+    assert!(!supervisor.lock().await.contains_key(&id));
+}