fix: decouple oneliner-heal from request handler tasks
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.
This commit is contained in:
@@ -38,6 +38,12 @@ pub type WorkerBusy = Arc<AtomicBool>;
|
||||
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<Gazetteer>,
|
||||
@@ -128,6 +136,12 @@ impl FromRef<AppState> for TranscribeBusy {
|
||||
}
|
||||
}
|
||||
|
||||
impl FromRef<AppState> for OnelinerHealBusy {
|
||||
fn from_ref(state: &AppState) -> Self {
|
||||
state.oneliner_heal_busy.clone()
|
||||
}
|
||||
}
|
||||
|
||||
impl FromRef<AppState> for PipelineState {
|
||||
fn from_ref(state: &AppState) -> Self {
|
||||
PipelineState {
|
||||
@@ -135,6 +149,7 @@ impl FromRef<AppState> 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<Config>) -> 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()),
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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<Config>,
|
||||
vocab: &Arc<Gazetteer>,
|
||||
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;
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<Config>) {
|
||||
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()),
|
||||
|
||||
@@ -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.
|
||||
}
|
||||
Reference in New Issue
Block a user