Add caller-thread progress for native sheet processing

Co-authored-by: TKIPisalegacycipher <git@chir.us>

crates/print/README.md

@@ -54,6 +54,14 @@ is closed and requests cancellation. Calls into a driver and interactive output-
 are not covered by this notification timeout. Windows answers when the job is in
 the queue ("spooled 3 sheet(s)"): a paused queue or an offline printer keeps it there. CUPS jobs
 never use a PDF temporary file. Printing is synchronous; UI callers should submit on a worker.
+`submit_with_progress` reports `(processed, sheets)` before the first Windows sheet and after
+each sheet is successfully added to the native print job. These are source PDF sheets, excluding
+copies and duplex; the events describe sheet processing, and do not confirm physical output.
+The callback runs on the thread calling submission, can capture non-`Send` state, and should
+return promptly. It is never retained. Keep waiting for the final `Result` even at full progress:
+the job can still fail. The API does not provide cancellation. CUPS reports no progress.
+UI callers must forward events to their own progress state; adopting the API alone does not
+change an application's UI.
 Receipts are backend messages, not portable job IDs. `Job::duplex = None` preserves the queue's
 duplex default; `Some(Duplex::Off)` forces simplex. `Job::scaling` defaults to `Scaling::Exact`;
 callers exporting arbitrary document sizes can choose `Scaling::ShrinkToFit`. Exact mode reports

crates/print/src/lib.rs

@@ -8,6 +8,8 @@
 
 #![deny(clippy::unwrap_used, clippy::expect_used, clippy::panic, clippy::unimplemented, clippy::todo, clippy::unreachable)]
 
+#[cfg(any(test, windows))]
+mod progress;
 #[cfg(any(test, windows))]
 mod sheets;
 #[cfg(test)]
@@ -437,17 +439,30 @@ pub fn parse_windows_printers(out: &str) -> Vec<Printer> {
 /// Send a print-ready PDF to the spooler. Returns the spooler's message: CUPS includes a job id;
 /// Windows confirms submission only ("spooled 3 sheet(s)").
 pub fn submit(pdf: &[u8], job: &Job) -> Result<String, Error> {
+    submit_with_progress(pdf, job, &mut |_, _| {})
+}
+
+/// [`submit`] with Windows sheet-processing progress: `(processed, sheets)` before the first
+/// sheet and after each sheet is successfully added to the native print job. Counts refer to
+/// source PDF sheets, regardless of copies or duplex. CUPS reports no progress.
+///
+/// The callback runs synchronously on the thread calling this function, need not be `Send`,
+/// and should return promptly. It is never retained. All sheets processed is not terminal
+/// success: only the returned result confirms submission, and a later failure may follow the
+/// last progress event. This does not report physical printing or provide cancellation.
+pub fn submit_with_progress(pdf: &[u8], job: &Job, progress: &mut dyn FnMut(usize, usize)) -> Result<String, Error> {
     #[cfg(all(unix, not(target_arch = "wasm32")))]
     {
+        let _ = progress;
         submit_via(std::process::Command::new("lp"), pdf, job)
     }
     #[cfg(windows)]
     {
-        windows::submit(pdf, job)
+        windows::submit_with_progress(pdf, job, progress)
     }
     #[cfg(not(any(windows, all(unix, not(target_arch = "wasm32")))))]
     {
-        let _ = (pdf, job);
+        let _ = (pdf, job, progress);
         Err(Error::Spool("printing to a printer isn't available on this platform yet; save the print-ready PDF instead".into()))
     }
 }

crates/print/src/progress.rs

@@ -0,0 +1,110 @@
+//! Bridge native worker events to callbacks on the submission thread.
+
+use std::sync::mpsc::{SyncSender, sync_channel};
+
+use crate::Error;
+
+pub(crate) fn run<T: Send>(
+    progress: &mut dyn FnMut(usize, usize),
+    work: impl FnOnce(SyncSender<(usize, usize)>) -> Result<T, Error> + Send,
+) -> Result<T, Error> {
+    // A slow UI callback must not accumulate one queued event per PDF sheet. The sender must
+    // ignore disconnection: dropping a progress receiver does not cancel an already-started job.
+    let (sender, receiver) = sync_channel(1);
+    std::thread::scope(move |scope| {
+        let worker = std::thread::Builder::new()
+            .name("print job".into())
+            .spawn_scoped(scope, move || work(sender))
+            .map_err(|e| Error::Spool(format!("the print job could not be started: {e}")))?;
+        // Own the receiver here so unwinding a callback closes it before scope joins the
+        // worker; otherwise a bounded sender could block forever during that join.
+        for (processed, sheets) in receiver {
+            progress(processed, sheets);
+        }
+        worker.join().map_err(|_| Error::Spool("the print job stopped unexpectedly; check the printer's queue before printing again".into()))?
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::cell::RefCell;
+    use std::rc::Rc;
+
+    #[test]
+    fn progress_arrives_on_the_calling_thread_before_submission_finishes() {
+        let caller = std::thread::current().id();
+        let seen = Rc::new(RefCell::new(Vec::new()));
+        let observations = Rc::clone(&seen);
+        let (observed, waiting) = sync_channel(0);
+        let result = run(
+            &mut move |processed, sheets| {
+                assert_eq!(std::thread::current().id(), caller);
+                observations.borrow_mut().push((processed, sheets));
+                observed.send(()).unwrap();
+            },
+            move |events| {
+                assert_ne!(std::thread::current().id(), caller);
+                for completed in 0..=3 {
+                    events.send((completed, 3)).unwrap();
+                    // No terminal result until the caller processes this intermediate event.
+                    waiting.recv().unwrap();
+                }
+                Ok("accepted")
+            },
+        );
+        assert_eq!(result, Ok("accepted"));
+        assert_eq!(*seen.borrow(), [(0, 3), (1, 3), (2, 3), (3, 3)]);
+    }
+
+    #[test]
+    fn complete_progress_does_not_hide_a_terminal_failure() {
+        let mut seen = Vec::new();
+        let error = Error::Spool("the native print package failed".into());
+        let result = run(&mut |done, total| seen.push((done, total)), |events| {
+            events.send((0, 1)).unwrap();
+            events.send((1, 1)).unwrap();
+            Err::<(), _>(error.clone())
+        });
+        assert_eq!(seen, [(0, 1), (1, 1)]);
+        assert_eq!(result, Err(error));
+    }
+
+    #[test]
+    fn a_partial_job_keeps_its_failure_and_last_processed_sheet() {
+        let mut seen = Vec::new();
+        let result = run(&mut |done, total| seen.push((done, total)), |events| {
+            events.send((0, 3)).unwrap();
+            events.send((1, 3)).unwrap();
+            Err::<(), _>(Error::Spool("the second sheet could not be rendered".into()))
+        });
+        assert_eq!(seen, [(0, 3), (1, 3)]);
+        assert_eq!(result, Err(Error::Spool("the second sheet could not be rendered".into())));
+    }
+
+    #[test]
+    fn worker_failure_is_reported_without_inventing_progress() {
+        let mut seen = Vec::new();
+        let result = run(&mut |done, total| seen.push((done, total)), |_| -> Result<(), Error> {
+            panic!("synthetic native worker failure");
+        });
+        assert!(seen.is_empty());
+        assert!(result.is_err_and(|e| e.to_string().contains("check the printer's queue")));
+    }
+
+    #[test]
+    fn callback_unwind_closes_progress_and_reaps_the_worker() {
+        let finished = std::sync::atomic::AtomicBool::new(false);
+        let caught = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
+            run(&mut |_, _| panic!("synthetic caller callback failure"), |events| {
+                for processed in 0..=20 {
+                    let _ = events.send((processed, 20));
+                }
+                finished.store(true, std::sync::atomic::Ordering::SeqCst);
+                Ok(())
+            })
+        }));
+        assert!(caught.is_err());
+        assert!(finished.load(std::sync::atomic::Ordering::SeqCst));
+    }
+}

crates/print/src/tests/windows_native.rs

@@ -368,11 +368,22 @@ fn windows_jobs_reach_the_spooler() {
     let listed = spool::printers().into_iter().find(|p| p.name == name);
     let mixed = test_pdf(&[LETTER, LETTER_LANDSCAPE, LETTER], true);
     let job = Job { printer: Some(name.into()), copies: 2, duplex: Some(Duplex::LongEdge), title: "mixed.pdf".into(), ..Job::default() };
-    let sent = spool::submit(&mixed, &job);
+    let caller = std::thread::current().id();
+    // Rc proves callers are not required to make their callback Send just because native
+    // printing uses a COM worker. Counts describe PDF sheets, despite copies and duplex.
+    let processed = std::rc::Rc::new(std::cell::RefCell::new(Vec::new()));
+    let observed = std::rc::Rc::clone(&processed);
+    let sent = spool::submit_with_progress(&mixed, &job, &mut move |done, total| {
+        assert_eq!(std::thread::current().id(), caller);
+        observed.borrow_mut().push((done, total));
+    });
     let tabloid = test_pdf(&[TABLOID, LETTER], false);
     let shrink = Job { printer: Some(name.into()), scaling: Scaling::ShrinkToFit, title: "shrink".into(), ..Job::default() };
     let shrunk = spool::submit(&tabloid, &shrink);
-    let exact_refused = spool::submit(&tabloid, &Job { printer: Some(name.into()), ..Job::default() });
+    let mut refused_progress = Vec::new();
+    let exact_refused = spool::submit_with_progress(&tabloid, &Job { printer: Some(name.into()), ..Job::default() }, &mut |done, total| {
+        refused_progress.push((done, total));
+    });
     let queued = windows_powershell(
         "@(Get-PrintJob -PrinterName $name | Sort-Object Id | ForEach-Object { $_.DocumentName + ' ' + $_.TotalPages }) -join '|'",
         name,
@@ -385,8 +396,10 @@ fn windows_jobs_reach_the_spooler() {
     );
     assert!(listed.as_ref().is_some_and(|p| p.two_sided.is_some()), "the new printer is listed, its duplex support known: {listed:?}");
     assert_eq!(sent.as_deref(), Ok("spooled 3 sheet(s)"), "{sent:?}");
+    assert_eq!(*processed.borrow(), [(0, 3), (1, 3), (2, 3), (3, 3)]);
     assert_eq!(shrunk.as_deref(), Ok("spooled 2 sheet(s)"), "{shrunk:?}");
     assert!(matches!(exact_refused, Err(Error::Spool(ref m)) if m.contains("one paper size")), "{exact_refused:?}");
+    assert!(refused_progress.is_empty(), "rejection before a job starts has no progress: {refused_progress:?}");
     assert_eq!(queued.as_deref().map(str::trim), Ok("mixed.pdf 3|shrink.pdf 2"), "one job each, all sheets, in the paused queue: {queued:?}");
     assert!(cleaned.is_ok(), "the test printer is removed again: {cleaned:?}");
     // Asking for a printer that isn't there is an error, never a job on another one.