From 3d42d1da821e47ba2ee9f8c0c3a29d1f81b4533a Mon Sep 17 00:00:00 2001 From: Brummel Date: Tue, 21 Apr 2026 12:56:02 +0200 Subject: [PATCH] fix: decouple oneliner-heal from request handler tasks MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit heal_orphans_if_idle called regenerate_missing_oneliners_for_user inline on the request task, blocking the browser for up to N×60s when orphans existed (observed with 16 cases after manual cleanup). Now: CAS on new OnelinerHealBusy flag + tokio::spawn + BusyGuard reset-on-drop. Handler returns immediately; the existing OnelinerUpdated SSE event delivers results via the usual reload. Startup recovery shares the flag so a concurrent page-load during boot cannot fire a parallel regen loop against Ollama. Regression test uses wiremock + timeout(150ms) + .expect(5) to verify both non-blocking return and dedup of parallel invocations. --- server/src/lib.rs | 16 +++ server/src/main.rs | 11 ++ server/src/routes/user_web.rs | 57 ++++++--- server/tests/magic_link_test.rs | 5 +- server/tests/oneliner_heal_decoupled_test.rs | 120 +++++++++++++++++++ 5 files changed, 193 insertions(+), 16 deletions(-) create mode 100644 server/tests/oneliner_heal_decoupled_test.rs diff --git a/server/src/lib.rs b/server/src/lib.rs index 1d60186..089bcbc 100644 --- a/server/src/lib.rs +++ b/server/src/lib.rs @@ -38,6 +38,12 @@ pub type WorkerBusy = Arc; pub struct AnalyzeBusy(pub WorkerBusy); #[derive(Clone)] pub struct TranscribeBusy(pub WorkerBusy); +/// Dedup flag for the detached oneliner-heal `tokio::spawn` fired from +/// `heal_orphans_if_idle`. The handler CAS's this to `true`; the spawned +/// task resets via [`BusyGuard`] on drop. Keeps concurrent page-loads +/// from launching parallel Ollama-call loops against the same data. +#[derive(Clone)] +pub struct OnelinerHealBusy(pub WorkerBusy); /// RAII guard that flips a [`WorkerBusy`] true on construction and false on /// drop — panic-safe. Shared across analyze + transcribe workers. @@ -68,6 +74,7 @@ pub struct PipelineState { pub analyze_tx: AnalyzeSender, pub transcribe_busy: TranscribeBusy, pub transcribe_tx: TranscribeSender, + pub oneliner_heal_busy: OnelinerHealBusy, } /// Shared application state. Clonable — all fields are cheap to clone @@ -81,6 +88,7 @@ pub struct AppState { pub magic_link_store: MagicLinkStore, pub analyze_busy: AnalyzeBusy, pub transcribe_busy: TranscribeBusy, + pub oneliner_heal_busy: OnelinerHealBusy, pub events_tx: events::EventSender, pub http_client: reqwest::Client, pub vocab: Arc, @@ -128,6 +136,12 @@ impl FromRef for TranscribeBusy { } } +impl FromRef for OnelinerHealBusy { + fn from_ref(state: &AppState) -> Self { + state.oneliner_heal_busy.clone() + } +} + impl FromRef for PipelineState { fn from_ref(state: &AppState) -> Self { PipelineState { @@ -135,6 +149,7 @@ impl FromRef for PipelineState { analyze_tx: state.analyze_tx.clone(), transcribe_busy: state.transcribe_busy.clone(), transcribe_tx: state.transcribe_tx.clone(), + oneliner_heal_busy: state.oneliner_heal_busy.clone(), } } } @@ -171,6 +186,7 @@ pub fn create_router(config: Arc) -> Router { magic_link_store: magic_link::new_store(), analyze_busy: AnalyzeBusy(Arc::new(AtomicBool::new(false))), transcribe_busy: TranscribeBusy(Arc::new(AtomicBool::new(false))), + oneliner_heal_busy: OnelinerHealBusy(Arc::new(AtomicBool::new(false))), events_tx: events::channel(), http_client: reqwest::Client::new(), vocab: Arc::new(Gazetteer::empty()), diff --git a/server/src/main.rs b/server/src/main.rs index 5e6008b..c23960b 100644 --- a/server/src/main.rs +++ b/server/src/main.rs @@ -116,6 +116,12 @@ async fn main() { // subscribed browsers. let events_tx = events::channel(); + // Dedup flag shared by the startup oneliner-regen task and every + // request-driven `heal_orphans_if_idle` spawn. One-at-a-time semantics + // keep Ollama from seeing overlapping regen loops. + let oneliner_heal_busy: doctate_server::WorkerBusy = + std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); + // Transcription pipeline: channel + worker + recovery scan. let (transcribe_tx, transcribe_rx) = transcribe::channel(); let transcribe_busy: doctate_server::WorkerBusy = @@ -135,10 +141,14 @@ async fn main() { let cfg = config.clone(); let vocab = vocab.clone(); let events = events_tx.clone(); + let heal_busy = oneliner_heal_busy.clone(); tokio::spawn(async move { transcribe::recovery::scan_and_enqueue(&data_path, &tx).await; // Then catch up oneliners for cases that crashed between // transcript-write and oneliner-write in a previous run. + // Hold the shared heal-busy flag while running so a concurrent + // page-load does not spawn a second parallel regen loop. + let _guard = doctate_server::BusyGuard::new(heal_busy); transcribe::recovery::regenerate_missing_oneliners( &data_path, &client, &cfg, &vocab, &events, ) @@ -174,6 +184,7 @@ async fn main() { magic_link_store: doctate_server::magic_link::new_store(), analyze_busy: doctate_server::AnalyzeBusy(analyze_busy), transcribe_busy: doctate_server::TranscribeBusy(transcribe_busy), + oneliner_heal_busy: doctate_server::OnelinerHealBusy(oneliner_heal_busy), events_tx, http_client, vocab, diff --git a/server/src/routes/user_web.rs b/server/src/routes/user_web.rs index 52f6089..373ef55 100644 --- a/server/src/routes/user_web.rs +++ b/server/src/routes/user_web.rs @@ -11,7 +11,6 @@ use tracing::{info, warn}; use time::{Date, OffsetDateTime, format_description::well_known::Rfc3339}; -use crate::PipelineState; use crate::analyze::{ ANALYSIS_INPUT_FILE, DOCUMENT_FILE, auto_trigger, recovery as analyze_recovery, }; @@ -25,6 +24,7 @@ use crate::paths; use crate::routes::case_actions::read_document; use crate::routes::web::{RecordingView, scan_recordings}; use crate::transcribe::recovery as transcribe_recovery; +use crate::{BusyGuard, PipelineState}; /// UI-layer oneliner status for the case list. Combines the persisted /// [`OnelinerState`] with derived transit states that only make sense @@ -299,8 +299,8 @@ impl PipelineState { user_root: &Path, slug: &str, http_client: &reqwest::Client, - config: &Config, - vocab: &Gazetteer, + config: &Arc, + vocab: &Arc, events_tx: &EventSender, ) { if !self.analyze_busy.0.load(Ordering::Acquire) { @@ -309,18 +309,45 @@ impl PipelineState { if !self.transcribe_busy.0.load(Ordering::Acquire) { transcribe_recovery::enqueue_pending_for_user(user_root, slug, &self.transcribe_tx) .await; - // Oneliner self-heal: if the transcribe worker is busy, it will - // write a fresh oneliner at batch-end anyway — skip to avoid a - // redundant LLM call. - transcribe_recovery::regenerate_missing_oneliners_for_user( - user_root, - slug, - http_client, - config, - vocab, - events_tx, - ) - .await; + // Oneliner self-heal: if the transcribe worker is busy, it + // will write a fresh oneliner at batch-end anyway — skip to + // avoid a redundant LLM call. Otherwise, fire the regen in a + // detached task so the request handler returns immediately; + // SSE `OnelinerUpdated` events deliver results to the browser. + // The CAS on `oneliner_heal_busy` guards against concurrent + // page-loads spawning their own parallel regen loops against + // the same cases. + if self + .oneliner_heal_busy + .0 + .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire) + .is_ok() + { + let user_root = user_root.to_path_buf(); + let slug_owned = slug.to_owned(); + let http_client = http_client.clone(); + let config = Arc::clone(config); + let vocab = Arc::clone(vocab); + let events_tx = events_tx.clone(); + let busy = Arc::clone(&self.oneliner_heal_busy.0); + info!(user = %slug, "oneliner heal spawned"); + tokio::spawn(async move { + // `BusyGuard::Drop` resets the flag to `false` — even + // on panic inside the regen loop. The redundant + // `store(true)` inside `BusyGuard::new` is harmless; + // the CAS above already owns the flag transition. + let _guard = BusyGuard::new(busy); + transcribe_recovery::regenerate_missing_oneliners_for_user( + &user_root, + &slug_owned, + &http_client, + &config, + &vocab, + &events_tx, + ) + .await; + }); + } } } } diff --git a/server/tests/magic_link_test.rs b/server/tests/magic_link_test.rs index 4b01f2b..dec78c5 100644 --- a/server/tests/magic_link_test.rs +++ b/server/tests/magic_link_test.rs @@ -11,7 +11,9 @@ use tower::util::ServiceExt; use doctate_common::API_KEY_HEADER; use doctate_server::config::{Config, User}; use doctate_server::magic_link::{MagicLinkStore, PendingMagicLink}; -use doctate_server::{AnalyzeBusy, AppState, TranscribeBusy, analyze, transcribe, web_session}; +use doctate_server::{ + AnalyzeBusy, AppState, OnelinerHealBusy, TranscribeBusy, analyze, transcribe, web_session, +}; const API_KEY: &str = "key-dr_a"; @@ -55,6 +57,7 @@ fn build_state() -> (AppState, Arc) { magic_link_store, analyze_busy: AnalyzeBusy(Arc::new(AtomicBool::new(false))), transcribe_busy: TranscribeBusy(Arc::new(AtomicBool::new(false))), + oneliner_heal_busy: OnelinerHealBusy(Arc::new(AtomicBool::new(false))), events_tx: doctate_server::events::channel(), http_client: reqwest::Client::new(), vocab: Arc::new(doctate_server::gazetteer::Gazetteer::empty()), diff --git a/server/tests/oneliner_heal_decoupled_test.rs b/server/tests/oneliner_heal_decoupled_test.rs new file mode 100644 index 0000000..90eefe1 --- /dev/null +++ b/server/tests/oneliner_heal_decoupled_test.rs @@ -0,0 +1,120 @@ +//! Regression test: `PipelineState::heal_orphans_if_idle` must not block +//! the request handler on Ollama. Before the fix, the function awaited a +//! sequential `regenerate_missing_oneliners_for_user` loop inline, so the +//! HTTP handler stayed pinned for N × LLM-latency. The fix detaches the +//! regeneration via `tokio::spawn`, guarded by a `compare_exchange` on +//! `OnelinerHealBusy` to dedup concurrent page-loads. + +use std::path::Path; +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::time::{Duration, Instant}; + +use doctate_server::config::Config; +use doctate_server::events; +use doctate_server::gazetteer::Gazetteer; +use doctate_server::{ + AnalyzeBusy, OnelinerHealBusy, PipelineState, TranscribeBusy, WorkerBusy, analyze, transcribe, +}; +use serde_json::json; +use tempfile::tempdir; +use tokio::time::timeout; +use wiremock::matchers::{method, path as wm_path}; +use wiremock::{Mock, MockServer, ResponseTemplate}; + +fn write_transcript(case_dir: &Path) { + let p = case_dir.join("2026-04-16T10-00-00Z.transcript.txt"); + std::fs::write(p, "Brustschmerz links, seit heute morgen.").unwrap(); +} + +#[tokio::test] +async fn heal_orphans_returns_fast_and_dedups_parallel_calls() { + // Five cases with a transcript but no oneliner.json. Without the fix, + // heal_orphans_if_idle would sequentially block on 5 × 200 ms = 1 s of + // mock delay. With the fix, it must return in well under that window. + let data = tempdir().unwrap(); + let slug = "dr_test"; + let user_root = data.path().join(slug); + std::fs::create_dir_all(&user_root).unwrap(); + for i in 0..5 { + let case_dir = user_root.join(format!("case-{i:02}")); + std::fs::create_dir_all(&case_dir).unwrap(); + write_transcript(&case_dir); + } + + // Ollama mock: 200 ms delay per chat call. `.expect(5)` asserts at + // MockServer drop that exactly five requests arrived — dedup would + // show up as 10 without the CAS guard. + let mock = MockServer::start().await; + Mock::given(method("POST")) + .and(wm_path("/api/chat")) + .respond_with( + ResponseTemplate::new(200) + .set_delay(Duration::from_millis(200)) + .set_body_json(json!({ + "message": { "content": "Brustschmerz links" } + })), + ) + .expect(5) + .mount(&mock) + .await; + + let mut cfg = Config::test_default(); + cfg.data_path = data.path().to_path_buf(); + cfg.ollama_url = mock.uri(); + let config = Arc::new(cfg); + + let vocab = Arc::new(Gazetteer::empty()); + let events_tx = events::channel(); + let http_client = reqwest::Client::new(); + + let (tx_a, _rx_a) = analyze::channel(); + let (tx_t, _rx_t) = transcribe::channel(); + + let heal_busy: WorkerBusy = Arc::new(AtomicBool::new(false)); + let pipeline = PipelineState { + analyze_busy: AnalyzeBusy(Arc::new(AtomicBool::new(false))), + analyze_tx: tx_a, + transcribe_busy: TranscribeBusy(Arc::new(AtomicBool::new(false))), + transcribe_tx: tx_t, + oneliner_heal_busy: OnelinerHealBusy(heal_busy.clone()), + }; + + // Two back-to-back heal invocations. The first wins the CAS and + // spawns. The second must no-op. Both heal futures themselves must + // finish in a fraction of the mock delay. + let before = Instant::now(); + timeout(Duration::from_millis(150), async { + pipeline + .heal_orphans_if_idle(&user_root, slug, &http_client, &config, &vocab, &events_tx) + .await; + pipeline + .heal_orphans_if_idle(&user_root, slug, &http_client, &config, &vocab, &events_tx) + .await; + }) + .await + .expect("heal_orphans_if_idle must return without awaiting the LLM loop"); + assert!( + before.elapsed() < Duration::from_millis(150), + "heal_orphans returned in {:?} — the inline regen is back", + before.elapsed() + ); + + // Spawn drops the busy flag on completion — poll until idle. + let poll_start = Instant::now(); + while heal_busy.load(Ordering::Acquire) { + if poll_start.elapsed() > Duration::from_secs(5) { + panic!("oneliner heal spawn did not finish within 5s"); + } + tokio::time::sleep(Duration::from_millis(25)).await; + } + + // Every case now has an oneliner.json. + for i in 0..5 { + let p = user_root.join(format!("case-{i:02}")).join("oneliner.json"); + assert!(p.exists(), "oneliner.json missing for case-{i:02}"); + } + + // Mock is dropped at end of scope → `.expect(5)` verifies no + // double-firing. A failing expectation panics with the actual count. +}