diff --git a/backend/Cargo.lock b/backend/Cargo.lock index 60991c0..6e509cc 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -1558,7 +1558,7 @@ checksum = "c41e0c4fef86961ac6d6f8a82609f55f31b05e4fce149ac5710e439df7619ba4" [[package]] name = "mangalord" -version = "0.93.5" +version = "0.93.6" dependencies = [ "anyhow", "argon2", diff --git a/backend/Cargo.toml b/backend/Cargo.toml index bdf5927..1535031 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "mangalord" -version = "0.93.5" +version = "0.93.6" edition = "2021" default-run = "mangalord" diff --git a/backend/src/analysis/ocr.rs b/backend/src/analysis/ocr.rs index f840fc8..a5f8aff 100644 --- a/backend/src/analysis/ocr.rs +++ b/backend/src/analysis/ocr.rs @@ -155,6 +155,37 @@ pub fn ocr_concurrency_limit(workers: usize, cores: usize) -> usize { workers.min(cores.max(1)).max(1) } +/// Run a CPU-bound OCR closure on the blocking pool while holding an +/// **owned** permit for the entire duration of the work. +/// +/// The permit is acquired with `acquire_owned` and moved *into* the +/// blocking task rather than being held by the caller's future. This +/// matters on cancellation: if the dispatcher future is dropped (graceful +/// shutdown), a borrowed permit would be released the instant the future +/// unwinds — but `spawn_blocking` work is not cancellable and keeps +/// running detached, so the concurrency bound (ANALYSIS_WORKERS) would be +/// briefly exceeded. Moving the permit into the task ties the slot's +/// lifetime to the actual CPU work. +async fn run_ocr_blocking( + permits: Arc, + f: F, +) -> anyhow::Result +where + F: FnOnce() -> T + Send + 'static, + T: Send + 'static, +{ + let permit = permits + .acquire_owned() + .await + .map_err(|e| anyhow::anyhow!("OCR semaphore closed: {e}"))?; + tokio::task::spawn_blocking(move || { + let _permit = permit; // held until f() returns, even if the caller is cancelled + f() + }) + .await + .map_err(|e| anyhow::anyhow!("OCR task join error: {e}")) +} + /// Production dispatcher for the OCR backend: load the page, read its image /// from storage, run OCR on the blocking pool, and persist the lines. Mirrors /// [`crate::analysis::daemon::RealAnalyzeDispatcher`] but with no network I/O. @@ -192,15 +223,10 @@ impl AnalyzeDispatcher for OcrAnalyzeDispatcher { // OCR inference is CPU-bound and synchronous — keep it off the async // worker's runtime thread, and gate it behind the shared permit pool so // ANALYSIS_WORKERS > cores can't oversubscribe the blocking pool. - let _permit = self - .ocr_permits - .acquire() - .await - .map_err(|e| anyhow::anyhow!("OCR semaphore closed: {e}"))?; let engine = Arc::clone(&self.engine); - let lines = tokio::task::spawn_blocking(move || engine.recognize(&bytes)) - .await - .map_err(|e| anyhow::anyhow!("OCR task join error: {e}"))??; + let lines = + run_ocr_blocking(Arc::clone(&self.ocr_permits), move || engine.recognize(&bytes)) + .await??; let analysis = lines_to_analysis(lines); repo::page_analysis::persist_analysis(&self.db, page_id, &analysis, OCR_MODEL_LABEL).await?; Ok(()) @@ -235,6 +261,55 @@ pub mod test_support { mod tests { use super::*; + #[tokio::test] + async fn run_ocr_blocking_holds_permit_until_work_completes_despite_cancellation() { + use std::sync::mpsc; + use tokio::sync::{Notify, Semaphore}; + + let permits = Arc::new(Semaphore::new(1)); + let (release_tx, release_rx) = mpsc::channel::<()>(); + let started = Arc::new(Notify::new()); + + // Model the dispatch path: acquire an owned permit and run blocking + // work that we hold open via a channel. + let sem = Arc::clone(&permits); + let started2 = Arc::clone(&started); + let caller = tokio::spawn(async move { + run_ocr_blocking(sem, move || { + // Signal that the blocking task now holds the permit, then + // block until the test releases us. + started2.notify_one(); + release_rx.recv().ok(); + }) + .await + .unwrap(); + }); + + // Wait until the blocking task is running and owns the permit. + started.notified().await; + assert_eq!(permits.available_permits(), 0, "permit taken by blocking work"); + + // Cancel the caller future (simulates graceful shutdown). The + // blocking task is not cancellable and keeps running detached; with + // an *owned* permit the slot must stay occupied. A borrowed permit + // would have been released here, regressing the concurrency bound. + caller.abort(); + let _ = caller.await; + assert_eq!( + permits.available_permits(), + 0, + "owned permit must remain held by the still-running blocking task" + ); + + // Let the blocking work finish; the permit is then returned. + release_tx.send(()).unwrap(); + let permit = tokio::time::timeout(std::time::Duration::from_secs(5), permits.acquire()) + .await + .expect("permit should be released once blocking work completes") + .unwrap(); + drop(permit); + } + #[test] fn lines_to_analysis_maps_lines_in_order_and_leaves_rest_empty() { let v = lines_to_analysis(vec!["Hello".to_string(), "world!".to_string()]); diff --git a/frontend/package.json b/frontend/package.json index 61cbe69..9c8582a 100644 --- a/frontend/package.json +++ b/frontend/package.json @@ -1,6 +1,6 @@ { "name": "mangalord-frontend", - "version": "0.93.5", + "version": "0.93.6", "private": true, "type": "module", "scripts": {