diff --git a/server/src/analyze/worker.rs b/server/src/analyze/worker.rs index d66bac7..0980e9a 100644 --- a/server/src/analyze/worker.rs +++ b/server/src/analyze/worker.rs @@ -1,5 +1,4 @@ use std::path::Path; -use std::sync::atomic::Ordering; use std::sync::Arc; use std::time::Duration; @@ -8,25 +7,7 @@ use tracing::{error, info, warn}; use super::{llm, prompt, AnalysisInput, AnalyzeReceiver}; use crate::config::Config; -use crate::WorkerBusy; - -/// RAII guard that flips `WorkerBusy` to `true` on construction and back to -/// `false` on drop. Panic-safe: even if `process()` unwinds, the flag is -/// cleared. -struct BusyGuard(WorkerBusy); - -impl BusyGuard { - fn new(busy: WorkerBusy) -> Self { - busy.store(true, Ordering::Release); - Self(busy) - } -} - -impl Drop for BusyGuard { - fn drop(&mut self) { - self.0.store(false, Ordering::Release); - } -} +use crate::{BusyGuard, WorkerBusy}; /// Consume analyze jobs sequentially. The external LLM tolerates parallelism, /// but sequential processing keeps the architecture simple and shields the diff --git a/server/src/lib.rs b/server/src/lib.rs index 49004f1..6183457 100644 --- a/server/src/lib.rs +++ b/server/src/lib.rs @@ -8,7 +8,7 @@ pub mod routes; pub mod transcribe; pub mod web_session; -use std::sync::atomic::AtomicBool; +use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; use axum::extract::FromRef; @@ -19,12 +19,36 @@ use config::Config; use transcribe::TranscribeSender; use web_session::SessionStore; -/// Live "is the analyze worker currently processing a job?" flag. -/// Set true before each `process()` call, false after. UI uses it to -/// distinguish a real in-flight analysis from a stale `analysis_input_v*.json` -/// orphan left over from a crash. +/// Live "is this worker currently processing a job?" flag. +/// Set true before each job, false after. UI uses it to distinguish a real +/// in-flight job from a stale on-disk marker (orphan input, missing +/// transcript) left over from a crash. pub type WorkerBusy = Arc; +/// Newtype wrappers so `FromRef` can target the two busy-flags +/// separately even though both are `Arc` underneath. +#[derive(Clone)] +pub struct AnalyzeBusy(pub WorkerBusy); +#[derive(Clone)] +pub struct TranscribeBusy(pub WorkerBusy); + +/// RAII guard that flips a [`WorkerBusy`] true on construction and false on +/// drop — panic-safe. Shared across analyze + transcribe workers. +pub struct BusyGuard(WorkerBusy); + +impl BusyGuard { + pub fn new(busy: WorkerBusy) -> Self { + busy.store(true, Ordering::Release); + Self(busy) + } +} + +impl Drop for BusyGuard { + fn drop(&mut self) { + self.0.store(false, Ordering::Release); + } +} + /// Shared application state. Clonable — all fields are cheap to clone /// (`Arc` and `mpsc::Sender`). #[derive(Clone)] @@ -33,7 +57,8 @@ pub struct AppState { pub transcribe_tx: TranscribeSender, pub analyze_tx: AnalyzeSender, pub session_store: SessionStore, - pub worker_busy: WorkerBusy, + pub analyze_busy: AnalyzeBusy, + pub transcribe_busy: TranscribeBusy, } impl FromRef for Arc { @@ -60,9 +85,15 @@ impl FromRef for SessionStore { } } -impl FromRef for WorkerBusy { +impl FromRef for AnalyzeBusy { fn from_ref(state: &AppState) -> Self { - state.worker_busy.clone() + state.analyze_busy.clone() + } +} + +impl FromRef for TranscribeBusy { + fn from_ref(state: &AppState) -> Self { + state.transcribe_busy.clone() } } @@ -77,7 +108,8 @@ pub fn create_router(config: Arc) -> Router { transcribe_tx, analyze_tx, session_store: web_session::new_store(), - worker_busy: Arc::new(AtomicBool::new(false)), + analyze_busy: AnalyzeBusy(Arc::new(AtomicBool::new(false))), + transcribe_busy: TranscribeBusy(Arc::new(AtomicBool::new(false))), }) } diff --git a/server/src/main.rs b/server/src/main.rs index fe5a52a..2be7f28 100644 --- a/server/src/main.rs +++ b/server/src/main.rs @@ -66,10 +66,13 @@ async fn main() { // Transcription pipeline: channel + worker + recovery scan. let (transcribe_tx, transcribe_rx) = transcribe::channel(); + let transcribe_busy: doctate_server::WorkerBusy = + std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); tokio::spawn(transcribe::worker::run( transcribe_rx, config.clone(), http_client.clone(), + transcribe_busy.clone(), )); { let tx = transcribe_tx.clone(); @@ -81,13 +84,13 @@ async fn main() { // Analyze pipeline: channel + worker + recovery scan. let (analyze_tx, analyze_rx) = analyze::channel(); - let worker_busy: doctate_server::WorkerBusy = + let analyze_busy: doctate_server::WorkerBusy = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); tokio::spawn(analyze::worker::run( analyze_rx, config.clone(), http_client.clone(), - worker_busy.clone(), + analyze_busy.clone(), )); { let tx = analyze_tx.clone(); @@ -102,7 +105,8 @@ async fn main() { transcribe_tx, analyze_tx, session_store: doctate_server::web_session::new_store(), - worker_busy, + analyze_busy: doctate_server::AnalyzeBusy(analyze_busy), + transcribe_busy: doctate_server::TranscribeBusy(transcribe_busy), }; let addr = format!("0.0.0.0:{}", config.server_port); diff --git a/server/src/routes/user_web.rs b/server/src/routes/user_web.rs index b5269e9..799a4a3 100644 --- a/server/src/routes/user_web.rs +++ b/server/src/routes/user_web.rs @@ -12,7 +12,8 @@ use crate::auth::AuthenticatedWebUser; use crate::config::Config; use crate::error::AppError; use crate::routes::web::{scan_recordings, RecordingView}; -use crate::WorkerBusy; +use crate::transcribe::{recovery as transcribe_recovery, TranscribeSender}; +use crate::{AnalyzeBusy, TranscribeBusy}; struct UserCaseView { case_id: String, @@ -67,6 +68,10 @@ struct CaseDetailTemplate { llm_missing: bool, analyzing: bool, has_document: bool, + /// True iff the transcribe worker is currently running. Gate for the + /// per-recording "Transkription läuft…" label — suppresses the lie when + /// a transcript is missing but no worker is active. + transcribe_busy: bool, } /// Flags derived from filesystem state + config. @@ -142,30 +147,45 @@ pub(crate) async fn any_document_exists(case_dir: &Path) -> bool { false } -/// Self-heal: if the analyze worker is idle but orphan inputs (an input file -/// without its document) exist for this user, re-enqueue them. Page-load is -/// the trigger; no cron, no periodic task. +/// Self-heal: if a worker is idle but orphans exist on disk for this user, +/// re-enqueue them. Page-load is the trigger; no cron, no periodic task. +/// Runs for both pipelines so a page-load on either view cleans both. async fn heal_orphans_if_idle( user_root: &Path, - worker_busy: &WorkerBusy, + slug: &str, + analyze_busy: &AnalyzeBusy, analyze_tx: &AnalyzeSender, + transcribe_busy: &TranscribeBusy, + transcribe_tx: &TranscribeSender, ) { - if worker_busy.load(Ordering::Acquire) { - return; + if !analyze_busy.0.load(Ordering::Acquire) { + analyze_recovery::enqueue_pending_for_user(user_root, analyze_tx).await; + } + if !transcribe_busy.0.load(Ordering::Acquire) { + transcribe_recovery::enqueue_pending_for_user(user_root, slug, transcribe_tx).await; } - analyze_recovery::enqueue_pending_for_user(user_root, analyze_tx).await; } pub async fn handle_my_cases( user: AuthenticatedWebUser, State(config): State>, - State(worker_busy): State, + State(analyze_busy): State, State(analyze_tx): State, + State(transcribe_busy): State, + State(transcribe_tx): State, ) -> Result, AppError> { let user_root = config.data_path.join(&user.slug); - heal_orphans_if_idle(&user_root, &worker_busy, &analyze_tx).await; + heal_orphans_if_idle( + &user_root, + &user.slug, + &analyze_busy, + &analyze_tx, + &transcribe_busy, + &transcribe_tx, + ) + .await; - let busy = worker_busy.load(Ordering::Acquire); + let busy = analyze_busy.0.load(Ordering::Acquire); let cases = scan_user_cases(&config, &user.slug, busy).await; let undo_count = crate::routes::case_actions::summarize_latest_batch(&user_root) .await @@ -186,8 +206,10 @@ pub async fn handle_my_cases( pub async fn handle_case_detail( user: AuthenticatedWebUser, State(config): State>, - State(worker_busy): State, + State(analyze_busy): State, State(analyze_tx): State, + State(transcribe_busy): State, + State(transcribe_tx): State, AxumPath(case_id): AxumPath, ) -> Result, AppError> { uuid::Uuid::parse_str(&case_id) @@ -195,7 +217,15 @@ pub async fn handle_case_detail( // IDOR guard: case_dir must live under the session's user_slug. let user_root = config.data_path.join(&user.slug); - heal_orphans_if_idle(&user_root, &worker_busy, &analyze_tx).await; + heal_orphans_if_idle( + &user_root, + &user.slug, + &analyze_busy, + &analyze_tx, + &transcribe_busy, + &transcribe_tx, + ) + .await; let case_dir = match locate_case(&user_root, &case_id).await { Some(p) => p, @@ -213,8 +243,9 @@ pub async fn handle_case_detail( .map(|s| s.trim().to_string()) .filter(|s| !s.is_empty()); - let busy = worker_busy.load(Ordering::Acquire); - let flags = compute_flags(&case_dir, &recordings, config.llm_configured(), busy).await; + let a_busy = analyze_busy.0.load(Ordering::Acquire); + let t_busy = transcribe_busy.0.load(Ordering::Acquire); + let flags = compute_flags(&case_dir, &recordings, config.llm_configured(), a_busy).await; let case_id_short = case_id.chars().take(8).collect(); CaseDetailTemplate { @@ -227,6 +258,7 @@ pub async fn handle_case_detail( llm_missing: flags.llm_missing, analyzing: flags.analyzing, has_document: flags.has_document, + transcribe_busy: t_busy, } .render() .map(Html) diff --git a/server/src/transcribe/recovery.rs b/server/src/transcribe/recovery.rs index f837a2f..83ed0cb 100644 --- a/server/src/transcribe/recovery.rs +++ b/server/src/transcribe/recovery.rs @@ -1,85 +1,81 @@ -use std::path::{Path, PathBuf}; +use std::path::Path; use tracing::{info, warn}; use super::{TranscribeJob, TranscribeSender}; -/// Walk `data_path/*/*/*.m4a` and enqueue every recording that does not -/// yet have a `.transcript.txt` sibling. -/// -/// Intended to run once at startup (spawn as a task — sends may back-pressure -/// if many recordings are pending and the queue fills). +/// Walk `data_path/*/` and enqueue every recording pending transcription for +/// every user. Intended for startup; the same primitive backs the per-user +/// self-heal triggered by web handlers. pub async fn scan_and_enqueue(data_path: &Path, tx: &TranscribeSender) { - let pending = collect_pending(data_path).await; - let total = pending.len(); - info!(count = total, "Recovery scan found pending recordings"); + let Ok(mut users) = tokio::fs::read_dir(data_path).await else { + return; + }; + // Collect + sort by slug so enqueue order is deterministic across + // users (within a user, the primitive sorts by timestamp). + let mut user_dirs: Vec<(String, std::path::PathBuf)> = Vec::new(); + while let Ok(Some(user_entry)) = users.next_entry().await { + if let Some(slug) = user_entry.file_name().to_str().map(str::to_owned) { + user_dirs.push((slug, user_entry.path())); + } + } + user_dirs.sort_by(|a, b| a.0.cmp(&b.0)); - for (user_slug, audio_path) in pending { + let mut total = 0; + for (slug, path) in user_dirs { + total += enqueue_pending_for_user(&path, &slug, tx).await; + } + info!(count = total, "Transcribe recovery scan finished"); +} + +/// Scan a single `//*.m4a` set and enqueue every recording +/// whose `.transcript.txt` is missing. Returns the number sent. Channel send +/// is awaited (bounded channel); only fails if the worker is gone. +pub async fn enqueue_pending_for_user( + user_root: &Path, + slug: &str, + tx: &TranscribeSender, +) -> usize { + let Ok(mut cases) = tokio::fs::read_dir(user_root).await else { + return 0; + }; + let mut pending: Vec = Vec::new(); + while let Ok(Some(case_entry)) = cases.next_entry().await { + let case_dir = case_entry.path(); + if !case_dir.is_dir() || crate::paths::is_deleted(&case_dir).await { + continue; + } + let Ok(mut files) = tokio::fs::read_dir(&case_dir).await else { + continue; + }; + while let Ok(Some(file)) = files.next_entry().await { + let path = file.path(); + if path.extension().and_then(|s| s.to_str()) != Some("m4a") { + continue; + } + if path.with_extension("transcript.txt").exists() { + continue; + } + pending.push(path); + } + } + // Oldest first — filenames are ISO timestamps so path order = chronological. + pending.sort(); + + let mut sent = 0; + for path in pending { if tx .send(TranscribeJob { - audio_path: audio_path.clone(), - user_slug: user_slug.clone(), + audio_path: path.clone(), + user_slug: slug.to_owned(), }) .await .is_err() { - warn!(file = %audio_path.display(), "Recovery send failed (worker gone)"); - break; + warn!(file = %path.display(), "recovery send failed (worker gone)"); + return sent; } + sent += 1; } -} - -/// Walks `$data_path///*.m4a` and returns `(slug, audio_path)` -/// tuples for every recording without a `.transcript.txt` sibling. The slug -/// comes from the top-level user directory name — that's the only ownership -/// signal after a restart (auth context is gone). -async fn collect_pending(data_path: &Path) -> Vec<(String, PathBuf)> { - let mut out: Vec<(String, PathBuf)> = Vec::new(); - - let mut users = match tokio::fs::read_dir(data_path).await { - Ok(r) => r, - Err(_) => return out, - }; - - while let Ok(Some(user_entry)) = users.next_entry().await { - let Some(slug) = user_entry.file_name().to_str().map(str::to_owned) else { - continue; - }; - let mut cases = match tokio::fs::read_dir(user_entry.path()).await { - Ok(r) => r, - Err(_) => continue, - }; - - while let Ok(Some(case_entry)) = cases.next_entry().await { - let case_dir = case_entry.path(); - if !case_dir.is_dir() { - continue; - } - if crate::paths::is_deleted(&case_dir).await { - continue; - } - - let mut files = match tokio::fs::read_dir(&case_dir).await { - Ok(r) => r, - Err(_) => continue, - }; - - while let Ok(Some(file)) = files.next_entry().await { - let path = file.path(); - if path.extension().and_then(|s| s.to_str()) != Some("m4a") { - continue; - } - let transcript = path.with_extension("transcript.txt"); - if !transcript.exists() { - out.push((slug.clone(), path)); - } - } - } - } - - // Oldest first (filenames are ISO timestamps). Path ordering puts same-user - // entries together, which is fine — we still process strictly by timestamp - // within a user directory. - out.sort_by(|a, b| a.1.cmp(&b.1)); - out + sent } diff --git a/server/src/transcribe/worker.rs b/server/src/transcribe/worker.rs index a08fd78..815ec31 100644 --- a/server/src/transcribe/worker.rs +++ b/server/src/transcribe/worker.rs @@ -6,16 +6,23 @@ use tracing::{error, info, warn}; use super::{ffmpeg, ollama, whisper, TranscribeReceiver}; use crate::config::{Config, WhisperUserSettings}; +use crate::{BusyGuard, WorkerBusy}; const ONELINER_TIMEOUT: Duration = Duration::from_secs(60); /// Consume transcription jobs sequentially. Whisper holds the GPU, so parallelism /// here would only cause contention. One job in flight at a time. -pub async fn run(mut rx: TranscribeReceiver, config: Arc, client: reqwest::Client) { +pub async fn run( + mut rx: TranscribeReceiver, + config: Arc, + client: reqwest::Client, + worker_busy: WorkerBusy, +) { info!("Transcription worker started"); let timeout = Duration::from_secs(config.whisper_timeout_seconds); while let Some(job) = rx.recv().await { + let _guard = BusyGuard::new(worker_busy.clone()); let audio_path = job.audio_path; let transcript_path = audio_path.with_extension("transcript.txt"); diff --git a/server/templates/case_detail.html b/server/templates/case_detail.html index 46d6e82..464dee6 100644 --- a/server/templates/case_detail.html +++ b/server/templates/case_detail.html @@ -69,7 +69,11 @@ header form { margin: 0; } {% when Some with (t) %}
{{ t }}
{% when None %} +{% if transcribe_busy %}
Transkription läuft…
+{% else %} +
Transkription ausstehend.
+{% endif %} {% endmatch %} {% endif %} diff --git a/server/tests/transcribe_test.rs b/server/tests/transcribe_test.rs index 7c488e5..b3f9a5c 100644 --- a/server/tests/transcribe_test.rs +++ b/server/tests/transcribe_test.rs @@ -286,7 +286,8 @@ async fn worker_renames_audio_to_failed_on_whisper_error() { drop(tx); // close channel so the worker loop exits after processing the job let client = reqwest::Client::new(); - transcribe::worker::run(rx, config, client).await; + let busy = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); + transcribe::worker::run(rx, config, client, busy).await; // Audio must have been renamed so recovery skips it next time. assert!(!audio.exists(), "original .m4a still present");