diff --git a/spoolbook-rs/src/lib.rs b/spoolbook-rs/src/lib.rs index e3e74f5..6e69229 100644 --- a/spoolbook-rs/src/lib.rs +++ b/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. diff --git a/spoolbook-rs/src/main.rs b/spoolbook-rs/src/main.rs index 6618505..6cef45d 100644 --- a/spoolbook-rs/src/main.rs +++ b/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. diff --git a/spoolbook-rs/src/printer_mqtt.rs b/spoolbook-rs/src/printer_mqtt.rs index b2c993d..392c3d3 100644 --- a/spoolbook-rs/src/printer_mqtt.rs +++ b/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>>; + +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, + access_code: Option, + serial_number: Option, + 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, Option, Option)>( @@ -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; } } diff --git a/spoolbook-rs/src/printers.rs b/spoolbook-rs/src/printers.rs index f8f924c..db16399 100644 --- a/spoolbook-rs/src/printers.rs +++ b/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) -> async fn create( _editor: crate::auth::Editor, State(pool): State, + Extension(store): Extension, + Extension(camera_registry): Extension, + Extension(supervisor): Extension, Json(input): Json, ) -> (StatusCode, Json) { 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, + Extension(store): Extension, + Extension(camera_registry): Extension, + Extension(supervisor): Extension, Path(id): Path, Json(input): Json, ) -> (StatusCode, Json) { @@ -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, Path(id): Path) -> (StatusCode, Json) { +async fn delete( + _editor: crate::auth::Editor, + State(pool): State, + Extension(store): Extension, + Extension(supervisor): Extension, + Path(id): Path, +) -> (StatusCode, Json) { 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, 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 })) } diff --git a/spoolbook-rs/tests/printer_respawn.rs b/spoolbook-rs/tests/printer_respawn.rs new file mode 100644 index 0000000..c65bf29 --- /dev/null +++ b/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, +) -> (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)); +}