From 9072d7a833b1d5149630235aee1e2b5647bdd3de Mon Sep 17 00:00:00 2001 From: Brummel Date: Thu, 16 Apr 2026 00:19:08 +0200 Subject: [PATCH] Refactor analyze recovery and worker busy status The analyze recovery scan has been refactored to iterate through user directories and enqueue pending analysis jobs more efficiently. The `WorkerBusy` status has been introduced as an `Arc` to track whether the analyze worker is currently processing a job. This `WorkerBusy` flag is used in the `analyze::worker::run` function and managed by a `BusyGuard` RAII struct, ensuring the flag is correctly set and unset even in case of panics. The `handle_my_cases` and `handle_case_detail` handlers in `user_web.rs` now utilize the `WorkerBusy` flag to accurately display the "analyzing" status and to trigger a self-heal of orphaned analysis inputs when the worker is idle. Additionally, the `WorkerBusy` type is now exported from `server/src/lib.rs` to be accessible by other modules. The test cases have been updated to include the `WorkerBusy` parameter when spawning the analyze worker. --- server/src/analyze/recovery.rs | 105 +++++++++++++++------------------ server/src/analyze/worker.rs | 35 ++++++++++- server/src/lib.rs | 15 +++++ server/src/main.rs | 4 ++ server/src/routes/user_web.rs | 91 +++++++++++++++++++++++----- server/templates/my_cases.html | 19 +++--- server/tests/analyze_test.rs | 6 +- 7 files changed, 191 insertions(+), 84 deletions(-) diff --git a/server/src/analyze/recovery.rs b/server/src/analyze/recovery.rs index d22c5d7..62a4903 100644 --- a/server/src/analyze/recovery.rs +++ b/server/src/analyze/recovery.rs @@ -1,78 +1,65 @@ -use std::path::{Path, PathBuf}; +use std::path::Path; use tracing::{info, warn}; use super::{AnalyzeJob, AnalyzeSender}; -/// Walk `data_path/*/*/analysis_input_v*.json` and enqueue every input -/// that does not yet have a matching `document_v*.md` sibling. -/// -/// Intended to run once at startup. Send is synchronous on the unbounded -/// channel — only fails if the worker is gone, which is a programming error. +/// Walk `data_path/*/` and enqueue every pending analysis for every user. +/// Intended to run once at startup; the same primitive backs the per-user +/// self-heal triggered by web handlers. pub async fn scan_and_enqueue(data_path: &Path, tx: &AnalyzeSender) { - let pending = collect_pending(data_path).await; - let total = pending.len(); - info!(count = total, "Analyze recovery scan found pending inputs"); - - for (case_dir, version) in pending { - if tx.send(AnalyzeJob { case_dir: case_dir.clone(), version }).is_err() { - warn!(case = %case_dir.display(), "Recovery send failed (worker gone)"); - break; - } + let Ok(mut users) = tokio::fs::read_dir(data_path).await else { + return; + }; + let mut total = 0; + while let Ok(Some(user_entry)) = users.next_entry().await { + total += enqueue_pending_for_user(&user_entry.path(), tx).await; } + info!(count = total, "Analyze recovery scan finished"); } -/// Returns `(case_dir, version)` tuples for every `analysis_input_v{N}.json` -/// whose `document_v{N}.md` is missing. -async fn collect_pending(data_path: &Path) -> Vec<(PathBuf, u32)> { - let mut out: Vec<(PathBuf, u32)> = Vec::new(); - - let mut users = match tokio::fs::read_dir(data_path).await { - Ok(r) => r, - Err(_) => return out, +/// Scan a single `//analysis_input_v*.json` set and enqueue +/// every input whose `document_v{N}.md` is missing. Returns the number sent. +/// Channel send is synchronous on the unbounded channel — only fails if the +/// worker is gone, which is a programming error. +pub async fn enqueue_pending_for_user(user_root: &Path, tx: &AnalyzeSender) -> usize { + let Ok(mut cases) = tokio::fs::read_dir(user_root).await else { + return 0; }; - - while let Ok(Some(user_entry)) = users.next_entry().await { - let mut cases = match tokio::fs::read_dir(user_entry.path()).await { - Ok(r) => r, - Err(_) => continue, + let mut sent = 0; + 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(case_entry)) = cases.next_entry().await { - let case_dir = case_entry.path(); - if !case_dir.is_dir() { + while let Ok(Some(file)) = files.next_entry().await { + let path = file.path(); + let Some(name) = path.file_name().and_then(|s| s.to_str()) else { 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(); - let Some(name) = path.file_name().and_then(|s| s.to_str()) else { - continue; - }; - let Some(version) = parse_input_version(name) else { - continue; - }; - let document = case_dir.join(format!("document_v{version}.md")); - if !document.exists() { - out.push((case_dir.clone(), version)); - } + let Some(version) = parse_input_version(name) else { + continue; + }; + if case_dir.join(format!("document_v{version}.md")).exists() { + continue; } + if tx + .send(AnalyzeJob { + case_dir: case_dir.clone(), + version, + }) + .is_err() + { + warn!(case = %case_dir.display(), "recovery send failed (worker gone)"); + return sent; + } + sent += 1; } } - - // Oldest case directories first — not strictly required for correctness, - // but gives deterministic recovery order. Cases are UUIDs so sorting is - // purely lexicographic, which is fine. - out.sort(); - out + sent } /// Parse `analysis_input_v{N}.json` and return `N` (as u32) if the pattern diff --git a/server/src/analyze/worker.rs b/server/src/analyze/worker.rs index 202c532..d66bac7 100644 --- a/server/src/analyze/worker.rs +++ b/server/src/analyze/worker.rs @@ -1,4 +1,5 @@ use std::path::Path; +use std::sync::atomic::Ordering; use std::sync::Arc; use std::time::Duration; @@ -7,15 +8,40 @@ 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); + } +} /// Consume analyze jobs sequentially. The external LLM tolerates parallelism, /// but sequential processing keeps the architecture simple and shields the /// provider from bursts. Trivial to lift later via `tokio::spawn(process(..))`. -pub async fn run(mut rx: AnalyzeReceiver, config: Arc, client: reqwest::Client) { +pub async fn run( + mut rx: AnalyzeReceiver, + config: Arc, + client: reqwest::Client, + worker_busy: WorkerBusy, +) { info!("Analyze worker started"); let timeout = Duration::from_secs(config.llm_timeout_seconds); while let Some(job) = rx.recv().await { + let _guard = BusyGuard::new(worker_busy.clone()); process(&job.case_dir, job.version, &config, &client, timeout).await; } @@ -100,6 +126,13 @@ async fn process( return; } + // Job done — drop the queue marker so the live "analyzing" flag + // becomes accurate immediately and recovery on next boot does not + // re-enqueue this version. + if let Err(e) = tokio::fs::remove_file(&input_path).await { + warn!(path = %input_path.display(), error = %e, "removing analysis input failed"); + } + info!( case = %case_dir.display(), version, diff --git a/server/src/lib.rs b/server/src/lib.rs index 7f8f37a..49004f1 100644 --- a/server/src/lib.rs +++ b/server/src/lib.rs @@ -8,6 +8,7 @@ pub mod routes; pub mod transcribe; pub mod web_session; +use std::sync::atomic::AtomicBool; use std::sync::Arc; use axum::extract::FromRef; @@ -18,6 +19,12 @@ 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. +pub type WorkerBusy = Arc; + /// Shared application state. Clonable — all fields are cheap to clone /// (`Arc` and `mpsc::Sender`). #[derive(Clone)] @@ -26,6 +33,7 @@ pub struct AppState { pub transcribe_tx: TranscribeSender, pub analyze_tx: AnalyzeSender, pub session_store: SessionStore, + pub worker_busy: WorkerBusy, } impl FromRef for Arc { @@ -52,6 +60,12 @@ impl FromRef for SessionStore { } } +impl FromRef for WorkerBusy { + fn from_ref(state: &AppState) -> Self { + state.worker_busy.clone() + } +} + /// Test/simple entrypoint: jobs pushed into either channel are dropped /// because the receivers are not retained. Use [`create_router_with_state`] /// from `main.rs` where real workers own the receivers. @@ -63,6 +77,7 @@ pub fn create_router(config: Arc) -> Router { transcribe_tx, analyze_tx, session_store: web_session::new_store(), + worker_busy: Arc::new(AtomicBool::new(false)), }) } diff --git a/server/src/main.rs b/server/src/main.rs index fc166ba..fe5a52a 100644 --- a/server/src/main.rs +++ b/server/src/main.rs @@ -81,10 +81,13 @@ async fn main() { // Analyze pipeline: channel + worker + recovery scan. let (analyze_tx, analyze_rx) = analyze::channel(); + let worker_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(), )); { let tx = analyze_tx.clone(); @@ -99,6 +102,7 @@ async fn main() { transcribe_tx, analyze_tx, session_store: doctate_server::web_session::new_store(), + worker_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 43d05fc..b5269e9 100644 --- a/server/src/routes/user_web.rs +++ b/server/src/routes/user_web.rs @@ -1,4 +1,5 @@ use std::path::Path; +use std::sync::atomic::Ordering; use std::sync::Arc; use askama::Template; @@ -6,15 +7,19 @@ use axum::extract::{Path as AxumPath, State}; use axum::response::Html; use tracing::{info, warn}; +use crate::analyze::{recovery as analyze_recovery, AnalyzeSender}; use crate::auth::AuthenticatedWebUser; use crate::config::Config; use crate::error::AppError; use crate::routes::web::{scan_recordings, RecordingView}; +use crate::WorkerBusy; struct UserCaseView { case_id: String, - case_id_short: String, most_recent: String, + /// HH:MM extracted from `most_recent` filename (e.g. "10:32"), + /// empty string if filename does not match the expected timestamp form. + time_hms: String, oneliner: Option, recordings_count: usize, analyzing: bool, @@ -25,6 +30,19 @@ struct UserCaseView { can_analyze: bool, } +/// Extract `HH:MM` from a recording filename of the form +/// `YYYY-MM-DDTHH-MM-SSZ.m4a[.failed]`. Returns `""` on any mismatch. +fn extract_hhmm(filename: &str) -> String { + let (Some(h), Some(m)) = (filename.get(11..13), filename.get(14..16)) else { + return String::new(); + }; + if h.bytes().all(|b| b.is_ascii_digit()) && m.bytes().all(|b| b.is_ascii_digit()) { + format!("{h}:{m}") + } else { + String::new() + } +} + #[derive(Template)] #[template(path = "my_cases.html")] struct MyCasesTemplate { @@ -33,6 +51,8 @@ struct MyCasesTemplate { /// Number of cases that "Undo last delete" would restore. 0 hides the /// undo button entirely. undo_count: usize, + /// True iff the session user is an admin — toggles full case_id display. + is_admin: bool, } #[derive(Template)] @@ -65,12 +85,13 @@ async fn compute_flags( case_dir: &Path, recordings: &[RecordingView], llm_configured: bool, + worker_busy: bool, ) -> CaseFlags { let has_document = any_document_exists(case_dir).await; - let analyzing = !has_document - && tokio::fs::try_exists(case_dir.join("analysis_input_v1.json")) - .await - .unwrap_or(false); + // Honest status: an input file alone does not mean a job is in flight. + // The worker may be idle and the file an orphan from a crash. Page-level + // self-heal (see `heal_orphans_if_idle`) re-enqueues those. + let analyzing = !has_document && worker_busy && any_analysis_input_exists(case_dir).await; let non_failed: Vec<&RecordingView> = recordings.iter().filter(|r| !r.failed).collect(); let all_transcribed = @@ -86,6 +107,24 @@ async fn compute_flags( } } +/// True iff the case directory contains at least one `analysis_input_v*.json`. +/// The worker removes the input after successful document write, so existence +/// implies a pending or in-flight analyze job. +pub(crate) async fn any_analysis_input_exists(case_dir: &Path) -> bool { + let mut entries = match tokio::fs::read_dir(case_dir).await { + Ok(r) => r, + Err(_) => return false, + }; + while let Ok(Some(entry)) = entries.next_entry().await { + let name = entry.file_name(); + let Some(name_str) = name.to_str() else { continue }; + if name_str.starts_with("analysis_input_v") && name_str.ends_with(".json") { + return true; + } + } + false +} + /// True iff the case directory contains at least one `document_v{N}.md`. /// We don't care about N here — any version means "Ausgewertet" for the UI. pub(crate) async fn any_document_exists(case_dir: &Path) -> bool { @@ -103,20 +142,41 @@ 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. +async fn heal_orphans_if_idle( + user_root: &Path, + worker_busy: &WorkerBusy, + analyze_tx: &AnalyzeSender, +) { + if worker_busy.load(Ordering::Acquire) { + return; + } + 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_tx): State, ) -> Result, AppError> { - let cases = scan_user_cases(&config, &user.slug).await; let user_root = config.data_path.join(&user.slug); + heal_orphans_if_idle(&user_root, &worker_busy, &analyze_tx).await; + + let busy = worker_busy.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 .map(|(_, n)| n) .unwrap_or(0); + let is_admin = user.is_admin(); MyCasesTemplate { slug: user.slug, cases, undo_count, + is_admin, } .render() .map(Html) @@ -126,6 +186,8 @@ pub async fn handle_my_cases( pub async fn handle_case_detail( user: AuthenticatedWebUser, State(config): State>, + State(worker_busy): State, + State(analyze_tx): State, AxumPath(case_id): AxumPath, ) -> Result, AppError> { uuid::Uuid::parse_str(&case_id) @@ -133,6 +195,8 @@ 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; + let case_dir = match locate_case(&user_root, &case_id).await { Some(p) => p, None => { @@ -149,7 +213,8 @@ pub async fn handle_case_detail( .map(|s| s.trim().to_string()) .filter(|s| !s.is_empty()); - let flags = compute_flags(&case_dir, &recordings, config.llm_configured()).await; + let busy = worker_busy.load(Ordering::Acquire); + let flags = compute_flags(&case_dir, &recordings, config.llm_configured(), busy).await; let case_id_short = case_id.chars().take(8).collect(); CaseDetailTemplate { @@ -186,7 +251,7 @@ pub(crate) async fn locate_case(user_root: &Path, case_id: &str) -> Option Vec { +async fn scan_user_cases(config: &Config, slug: &str, worker_busy: bool) -> Vec { let user_root = config.data_path.join(slug); let llm_configured = config.llm_configured(); let mut cases = Vec::new(); @@ -230,7 +295,6 @@ async fn scan_user_cases(config: &Config, slug: &str) -> Vec { .filter(|r| !r.failed) .all(|r| r.transcript.is_some()); - let case_id_short = case_id.chars().take(8).collect(); let oneliner = tokio::fs::read_to_string(case_path.join("oneliner.txt")) .await .ok() @@ -238,16 +302,15 @@ async fn scan_user_cases(config: &Config, slug: &str) -> Vec { .filter(|s| !s.is_empty()); let has_document = any_document_exists(&case_path).await; - let analyzing = !has_document - && tokio::fs::try_exists(case_path.join("analysis_input_v1.json")) - .await - .unwrap_or(false); + let analyzing = + !has_document && worker_busy && any_analysis_input_exists(&case_path).await; let can_analyze = !analyzing && all_transcribed && llm_configured; + let time_hms = extract_hhmm(&most_recent); cases.push(UserCaseView { case_id, - case_id_short, most_recent, + time_hms, oneliner, recordings_count, analyzing, diff --git a/server/templates/my_cases.html b/server/templates/my_cases.html index a6c5b4b..aead603 100644 --- a/server/templates/my_cases.html +++ b/server/templates/my_cases.html @@ -18,10 +18,17 @@ section h2 { font-size: 1.1em; color: #555; border-bottom: 1px solid #ddd; paddi .case-row .check { align-self: start; padding-top: 0.25em; } .case-row .body { min-width: 0; } .case-row .body a { color: inherit; text-decoration: none; } -.case-row .body a:hover h3 { text-decoration: underline; } +.case-row .body a:hover .line1 { text-decoration: underline; } .case-row .body h3 { margin: 0 0 0.2em 0; font-size: 1em; font-weight: 600; } .case-row .body small { color: #666; font-family: monospace; font-size: 0.85em; } .case-row .oneliner { color: #333; margin: 0.2em 0 0; } +.case-row .line1 { font-size: 1em; margin: 0; } +.case-row .line1 .time { font-family: monospace; font-weight: bold; color: #444; } +.case-row .line2 { margin: 0.2em 0 0; color: #555; font-size: 0.95em; display: flex; align-items: center; gap: 0.5em; } +.case-row .uuid { color: #999; font-family: monospace; font-size: 0.7em; margin-top: 0.2em; } +.status-badge { display: inline-block; padding: 0.05em 0.5em; border-radius: 999px; font-size: 0.75em; font-weight: bold; color: white; } +.status-badge.open { background: #e24a4a; } +.status-badge.done { background: #7ed321; } .case-row .actions { display: flex; gap: 0.4em; } .case-row .actions button { font-size: 0.85em; padding: 0.35em 0.8em; } @@ -60,13 +67,9 @@ section h2 { font-size: 1.1em; color: #555; border-bottom: 1px solid #ddd; paddi
{% if case.has_document %}{% else %}{% endif %} -

{{ case.case_id_short }} — {{ case.recordings_count }} Aufnahme(n){% if case.has_document %}ausgewertet{% else if case.analyzing %}wird analysiert{% endif %}

-{% match case.oneliner %} -{% when Some with (t) %} -
{{ t }}
-{% when None %} -{% endmatch %} -{{ case.most_recent }} +
{{ case.time_hms }}{% match case.oneliner %}{% when Some with (t) %} — {{ t }}{% when None %}{% endmatch %}
+
{{ case.recordings_count }} Aufnahmen{% if case.has_document %} — ausgewertet{% else %} — offen{% endif %}{% if case.analyzing %} wird analysiert{% endif %}
+{% if is_admin %}
{{ case.case_id }}
{% endif %}
diff --git a/server/tests/analyze_test.rs b/server/tests/analyze_test.rs index 71f8331..291235d 100644 --- a/server/tests/analyze_test.rs +++ b/server/tests/analyze_test.rs @@ -327,7 +327,8 @@ async fn analyze_worker_writes_document_via_wiremock() { let config = config_with_llm(tmp.clone(), mock.uri()); let (tx, rx) = analyze::channel(); let client = reqwest::Client::new(); - let handle = tokio::spawn(analyze::worker::run(rx, config, client)); + let busy = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); + let handle = tokio::spawn(analyze::worker::run(rx, config, client, busy)); tx.send(analyze::AnalyzeJob { case_dir: case_dir.clone(), @@ -480,7 +481,8 @@ async fn analyze_v2_worker_writes_document_v2_via_wiremock() { let config = config_with_llm(tmp.clone(), mock.uri()); let (tx, rx) = analyze::channel(); - let handle = tokio::spawn(analyze::worker::run(rx, config, reqwest::Client::new())); + let busy = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); + let handle = tokio::spawn(analyze::worker::run(rx, config, reqwest::Client::new(), busy)); tx.send(analyze::AnalyzeJob { case_dir: case_dir.clone(),