diff --git a/server/src/analyze/auto_trigger.rs b/server/src/analyze/auto_trigger.rs index 2692518..d4a8cab 100644 --- a/server/src/analyze/auto_trigger.rs +++ b/server/src/analyze/auto_trigger.rs @@ -17,7 +17,7 @@ use time::format_description::well_known::Rfc3339; use tokio::io::AsyncWriteExt; use tracing::warn; -use super::{ANALYSIS_INPUT_FILE, AnalysisInput, AnalyzeJob, AnalyzeSender, DOCUMENT_FILE}; +use super::{AnalyzeJob, AnalyzeSender, DOCUMENT_FILE, InFlight}; use crate::events::{self, CaseEventKind, EventSender}; use crate::paths; use crate::routes::case_actions::build_analysis_input; @@ -155,7 +155,7 @@ pub enum CaseAnalysisState { Current, } -/// Compute the full per-case lifecycle state. Pure file-I/O, no +/// Compute the full per-case lifecycle state. Pure observation, no /// mutations or channel sends — safe to call redundantly per render. /// /// `idle_threshold` is the time without new recordings after which a @@ -172,11 +172,16 @@ pub enum CaseAnalysisState { /// fresh server start does not instantly classify every old case as /// `Idle` (and trigger a thundering herd of LLM calls on the first /// `/cases` render). Pass `SystemTime::UNIX_EPOCH` to disable. +/// +/// `in_flight` is the process-local "queued or running" set. It replaces +/// the previous on-disk `analysis_input.json` marker, which survived +/// crashes and left cases stuck in `Queued` after a restart. pub async fn evaluate_state( case_dir: &Path, idle_threshold: std::time::Duration, now: SystemTime, boot_time: SystemTime, + in_flight: &InFlight, ) -> CaseAnalysisState { let scan = scan_m4as(case_dir).await; @@ -190,11 +195,9 @@ pub async fn evaluate_state( return CaseAnalysisState::NoRecordings; }; - // Any existing analysis_input.json means a job is queued or running. - if tokio::fs::try_exists(case_dir.join(ANALYSIS_INPUT_FILE)) - .await - .unwrap_or(false) - { + // A claimed slot means a job is queued or in flight in this very + // process. Empty on every server boot — no more phantom Queued. + if in_flight.contains(case_dir) { return CaseAnalysisState::Queued; } @@ -244,20 +247,24 @@ pub(crate) async fn read_document_meta(case_dir: &Path) -> Option)`. /// `Required` is also mapped to `Skip` here — but with /// `Duration::ZERO` as the threshold, `Required` collapses into `Idle` /// in practice, so the eager pre-refactor semantics are preserved. +/// +/// Uses an empty in-flight set: tests that exercise the `Queued` branch +/// supply their own set via `evaluate_state` directly. pub async fn evaluate_case(case_dir: &Path) -> AutoDecision { let state = evaluate_state( case_dir, std::time::Duration::ZERO, SystemTime::now(), SystemTime::UNIX_EPOCH, + &InFlight::new(), ) .await; match state { @@ -290,18 +297,27 @@ pub async fn try_enqueue( idle_threshold: std::time::Duration, now: SystemTime, boot_time: SystemTime, + in_flight: &InFlight, tx: &AnalyzeSender, events_tx: &EventSender, ) -> bool { - match evaluate_state(case_dir, idle_threshold, now, boot_time).await { + match evaluate_state(case_dir, idle_threshold, now, boot_time, in_flight).await { CaseAnalysisState::Idle { .. } => {} _ => return false, } + // Atomic claim → race-guard against concurrent triggers. If another + // caller (manual button, second tab) just grabbed the slot, skip + // silently — they will send the job. + if !in_flight.try_claim(case_dir) { + return false; + } + let input = match build_analysis_input(case_dir, "").await { Ok(i) => i, Err(e) => { warn!(case = %case_dir.display(), error = ?e, "auto-analyze: build input failed"); + in_flight.release(case_dir); return false; } }; @@ -310,19 +326,15 @@ pub async fn try_enqueue( // the UI keeps rendering the last successful analysis as "letztes // Dokument" with a Stale-banner. The worker overwrites the file // atomically once the new analysis completes. - let input_path = case_dir.join(ANALYSIS_INPUT_FILE); - if let Err(e) = write_input_overwrite(&input_path, &input).await { - warn!(case = %case_dir.display(), error = %e, "auto-analyze: write input failed"); - return false; - } - if tx .send(AnalyzeJob { case_dir: case_dir.to_path_buf(), + input, }) .is_err() { warn!(case = %case_dir.display(), "auto-analyze: send failed (worker gone)"); + in_flight.release(case_dir); return false; } @@ -343,6 +355,7 @@ pub async fn try_enqueue_all_for_user( idle_threshold: std::time::Duration, now: SystemTime, boot_time: SystemTime, + in_flight: &InFlight, tx: &AnalyzeSender, events_tx: &EventSender, ) { @@ -363,7 +376,16 @@ pub async fn try_enqueue_all_for_user( if crate::paths::is_closed(&case_path).await { continue; } - try_enqueue(&case_path, idle_threshold, now, boot_time, tx, events_tx).await; + try_enqueue( + &case_path, + idle_threshold, + now, + boot_time, + in_flight, + tx, + events_tx, + ) + .await; } } @@ -511,24 +533,6 @@ pub(crate) async fn read_failure_marker(case_dir: &Path) -> Option std::io::Result<()> { - let bytes = serde_json::to_vec_pretty(input) - .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?; - let tmp = path.with_extension("json.tmp"); - let mut f = tokio::fs::File::create(&tmp).await?; - f.write_all(&bytes).await?; - f.sync_all().await?; - drop(f); - tokio::fs::rename(&tmp, path).await -} - /// Format an mtime in the same RFC3339 shape that /// `case_actions::build_analysis_input` writes, so equality comparison /// against a stored marker is string-exact. @@ -697,16 +701,45 @@ mod tests { ); } + /// `Queued` is now derived from the process-local `InFlight` set, + /// not from a file on disk. This test exercises that path directly + /// via `evaluate_state` (the `evaluate_case` adapter passes an empty + /// set, which is exactly the point — orphan files cannot fake + /// `Queued` any more). #[tokio::test] - async fn analysis_input_present_skips() { + async fn in_flight_claim_yields_queued() { let dir = TempDir::new().unwrap(); touch(&dir.path().join("2026-01-01T10-00-00Z.m4a")).await; seed_meta(dir.path(), "2026-01-01T10-00-00Z").await; - touch(&dir.path().join(ANALYSIS_INPUT_FILE)).await; - assert_eq!( - evaluate_case(dir.path()).await, - AutoDecision::Skip("job queued") - ); + + let in_flight = InFlight::new(); + assert!(in_flight.try_claim(dir.path())); + + let state = evaluate_state( + dir.path(), + std::time::Duration::ZERO, + SystemTime::now(), + SystemTime::UNIX_EPOCH, + &in_flight, + ) + .await; + assert_eq!(state, CaseAnalysisState::Queued); + } + + /// Orphan `analysis_input.json` left over from a pre-refactor server + /// run must NOT register as `Queued` any more — that was the whole + /// reason for the refactor. + #[tokio::test] + async fn orphan_input_file_does_not_count_as_queued() { + let dir = TempDir::new().unwrap(); + touch(&dir.path().join("2026-01-01T10-00-00Z.m4a")).await; + seed_meta_with(dir.path(), "2026-01-01T10-00-00Z", "patient mit fieber").await; + // Pre-refactor leftover: a stale file the server used to write. + fs::write(dir.path().join("analysis_input.json"), b"{}") + .await + .unwrap(); + + assert_eq!(evaluate_case(dir.path()).await, AutoDecision::Enqueue); } #[tokio::test] @@ -795,6 +828,7 @@ mod tests { Duration::from_secs(900), base + Duration::from_secs(100), SystemTime::UNIX_EPOCH, + &InFlight::new(), ) .await; @@ -823,6 +857,7 @@ mod tests { Duration::from_secs(900), base + Duration::from_secs(1000), SystemTime::UNIX_EPOCH, + &InFlight::new(), ) .await; @@ -855,7 +890,14 @@ mod tests { // Required, not Idle. let now = base + Duration::from_secs(30 * 24 * 3600); let boot = now - Duration::from_secs(100); - let state = evaluate_state(dir.path(), Duration::from_secs(900), now, boot).await; + let state = evaluate_state( + dir.path(), + Duration::from_secs(900), + now, + boot, + &InFlight::new(), + ) + .await; assert_eq!( state, @@ -867,7 +909,14 @@ mod tests { // Once the grace window has elapsed (boot was 1000 s ago) the // case promotes to Idle. let boot_old = now - Duration::from_secs(1000); - let state_old = evaluate_state(dir.path(), Duration::from_secs(900), now, boot_old).await; + let state_old = evaluate_state( + dir.path(), + Duration::from_secs(900), + now, + boot_old, + &InFlight::new(), + ) + .await; assert_eq!( state_old, CaseAnalysisState::Idle { @@ -891,6 +940,7 @@ mod tests { let (tx, mut rx) = analyze_channel(); let events_tx = events_channel(); + let in_flight = InFlight::new(); // Idle threshold not reached → must NOT enqueue. let sent_required = try_enqueue( @@ -898,6 +948,7 @@ mod tests { Duration::from_secs(900), base + Duration::from_secs(100), SystemTime::UNIX_EPOCH, + &in_flight, &tx, &events_tx, ) @@ -907,6 +958,10 @@ mod tests { rx.try_recv().is_err(), "channel must be empty after Required" ); + assert!( + !in_flight.contains(dir.path()), + "no claim should exist after Required" + ); // Same case, idle threshold passed → must enqueue exactly one job. let sent_idle = try_enqueue( @@ -914,6 +969,7 @@ mod tests { Duration::from_secs(900), base + Duration::from_secs(1000), SystemTime::UNIX_EPOCH, + &in_flight, &tx, &events_tx, ) @@ -924,6 +980,55 @@ mod tests { "channel must hold one job after Idle" ); assert!(rx.try_recv().is_err(), "channel must hold exactly one job"); + assert!( + in_flight.contains(dir.path()), + "claim must persist until the worker drops the slot guard" + ); + } + + /// Concurrent triggers (auto-trigger + manual click in another tab) + /// must produce **exactly one** job, not two. The atomic + /// `try_claim` is the race-guard. + #[tokio::test] + async fn try_enqueue_is_idempotent_under_concurrent_triggers() { + use crate::analyze::channel as analyze_channel; + use crate::events::channel as events_channel; + + let dir = TempDir::new().unwrap(); + let m4a = dir.path().join("2026-01-01T10-00-00Z.m4a"); + touch(&m4a).await; + seed_meta_with(dir.path(), "2026-01-01T10-00-00Z", "patient mit fieber").await; + let base = SystemTime::UNIX_EPOCH + Duration::from_secs(1_700_000_000); + set_mtime(&m4a, base); + + let (tx, mut rx) = analyze_channel(); + let events_tx = events_channel(); + let in_flight = InFlight::new(); + + let a = try_enqueue( + dir.path(), + Duration::from_secs(900), + base + Duration::from_secs(1000), + SystemTime::UNIX_EPOCH, + &in_flight, + &tx, + &events_tx, + ) + .await; + let b = try_enqueue( + dir.path(), + Duration::from_secs(900), + base + Duration::from_secs(1000), + SystemTime::UNIX_EPOCH, + &in_flight, + &tx, + &events_tx, + ) + .await; + assert!(a, "first call must enqueue"); + assert!(!b, "second call must skip — slot is already claimed"); + assert!(rx.try_recv().is_ok()); + assert!(rx.try_recv().is_err(), "exactly one job in the channel"); } #[tokio::test] @@ -968,12 +1073,14 @@ mod tests { let (tx, mut rx) = analyze_channel(); let events_tx = events_channel(); + let in_flight = InFlight::new(); try_enqueue_all_for_user( user_root.path(), std::time::Duration::ZERO, SystemTime::now(), SystemTime::UNIX_EPOCH, + &in_flight, &tx, &events_tx, ) diff --git a/server/src/analyze/in_flight.rs b/server/src/analyze/in_flight.rs new file mode 100644 index 0000000..23f0f8e --- /dev/null +++ b/server/src/analyze/in_flight.rs @@ -0,0 +1,144 @@ +//! Process-local "this case is queued or in flight" tracking. +//! +//! Replaces the previous design where the existence of `analysis_input.json` +//! on disk doubled as the lifecycle marker. That coupled disk state +//! (lives forever) to process state (lives one boot session), so a crash +//! mid-job left the case stuck in `Queued` after the next restart. +//! +//! By moving the marker into an [`Arc>>`] held in +//! `AppState`, the marker is reset to empty on every server boot. The +//! payload that used to live in `analysis_input.json` now travels with +//! the [`AnalyzeJob`] over the mpsc channel. +//! +//! [`AnalyzeJob`]: super::AnalyzeJob + +use std::collections::HashSet; +use std::path::{Path, PathBuf}; +use std::sync::{Arc, RwLock}; + +/// Shared, process-local set of cases currently queued or being analyzed. +/// Cheap to clone (`Arc`). Operations are non-blocking and short — the +/// inner `HashSet` is only ever held under the lock for an `insert`, +/// `remove`, or `contains` call. +#[derive(Clone, Default)] +pub struct InFlight(Arc>>); + +impl InFlight { + pub fn new() -> Self { + Self::default() + } + + /// Atomically claim a case. Returns `true` if the caller now owns the + /// slot (proceed to send a job), `false` if another caller already + /// holds it (skip — duplicate trigger). Replaces the `O_EXCL` create + /// race-guard the manual-trigger handlers used to rely on. + pub fn try_claim(&self, case_dir: &Path) -> bool { + self.0 + .write() + .expect("inflight lock poisoned") + .insert(case_dir.to_path_buf()) + } + + /// Release a slot. Idempotent — safe to call from cleanup paths that + /// do not know whether the slot was actually held. + pub fn release(&self, case_dir: &Path) { + self.0 + .write() + .expect("inflight lock poisoned") + .remove(case_dir); + } + + /// Read-only check used by `evaluate_state` to render the + /// `Queued` UI state. + pub fn contains(&self, case_dir: &Path) -> bool { + self.0 + .read() + .expect("inflight lock poisoned") + .contains(case_dir) + } +} + +/// RAII slot-release guard. Construct after a successful [`InFlight::try_claim`]; +/// the slot is released when the guard drops. Covers success, every +/// failure path, and worker panics — analogous to [`crate::BusyGuard`]. +pub struct InFlightGuard { + set: InFlight, + case_dir: PathBuf, +} + +impl InFlightGuard { + pub fn new(set: InFlight, case_dir: PathBuf) -> Self { + Self { set, case_dir } + } +} + +impl Drop for InFlightGuard { + fn drop(&mut self) { + self.set.release(&self.case_dir); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn p(s: &str) -> PathBuf { + PathBuf::from(s) + } + + #[test] + fn try_claim_returns_true_first_then_false() { + let set = InFlight::new(); + assert!(set.try_claim(&p("/case/a"))); + assert!(!set.try_claim(&p("/case/a"))); + assert!(set.try_claim(&p("/case/b"))); + } + + #[test] + fn release_makes_slot_reclaimable() { + let set = InFlight::new(); + assert!(set.try_claim(&p("/case/a"))); + set.release(&p("/case/a")); + assert!(set.try_claim(&p("/case/a"))); + } + + #[test] + fn release_unknown_slot_is_noop() { + let set = InFlight::new(); + set.release(&p("/case/never-claimed")); + } + + #[test] + fn contains_reflects_claim_state() { + let set = InFlight::new(); + assert!(!set.contains(&p("/case/a"))); + set.try_claim(&p("/case/a")); + assert!(set.contains(&p("/case/a"))); + set.release(&p("/case/a")); + assert!(!set.contains(&p("/case/a"))); + } + + #[test] + fn guard_releases_on_drop() { + let set = InFlight::new(); + assert!(set.try_claim(&p("/case/a"))); + { + let _g = InFlightGuard::new(set.clone(), p("/case/a")); + assert!(set.contains(&p("/case/a"))); + } + assert!(!set.contains(&p("/case/a"))); + } + + #[test] + fn guard_releases_on_panic_unwind() { + let set = InFlight::new(); + set.try_claim(&p("/case/a")); + let set_clone = set.clone(); + let result = std::panic::catch_unwind(move || { + let _g = InFlightGuard::new(set_clone, p("/case/a")); + panic!("simulated worker panic"); + }); + assert!(result.is_err()); + assert!(!set.contains(&p("/case/a"))); + } +} diff --git a/server/src/analyze/mod.rs b/server/src/analyze/mod.rs index f7469d9..aa08f96 100644 --- a/server/src/analyze/mod.rs +++ b/server/src/analyze/mod.rs @@ -5,18 +5,20 @@ use tokio::sync::mpsc; pub mod auto_trigger; pub mod backend; +pub mod in_flight; pub mod llm; pub mod prompt; pub mod recovery; pub mod render; pub mod worker; -/// Single-document-per-case filenames. A re-analysis overwrites the +pub use in_flight::{InFlight, InFlightGuard}; + +/// Single-document-per-case filename. A re-analysis overwrites the /// document in place; the file is a typed JSON envelope so the /// `covered_mtime` snapshot stays glued to the markdown body it was /// produced from. pub const DOCUMENT_FILE: &str = "document.json"; -pub const ANALYSIS_INPUT_FILE: &str = "analysis_input.json"; /// Persisted analysis output for a case. The `covered_mtime` field is /// the snapshot mtime of the recordings the LLM saw — *not* the @@ -33,13 +35,15 @@ pub struct Document { pub content_md: String, } -/// A single request to analyze a case. Payload is tiny (a path), so the -/// channel is unbounded — persistence lives in `analysis_input.json` on -/// disk, not in the queue. Recovery re-enqueues any inputs the server may -/// have held at crash time. +/// A single request to analyze a case. Carries the full [`AnalysisInput`] +/// payload over the channel so the worker does not need to read it back +/// from disk. Lifecycle is process-local — see [`InFlight`] — so a server +/// restart between send and worker pickup just drops the job; the next +/// `/cases` render re-triggers via the normal `Required → Idle` path. #[derive(Debug, Clone)] pub struct AnalyzeJob { pub case_dir: PathBuf, + pub input: AnalysisInput, } pub type AnalyzeSender = mpsc::UnboundedSender; @@ -49,10 +53,9 @@ pub fn channel() -> (AnalyzeSender, AnalyzeReceiver) { mpsc::unbounded_channel() } -/// Persistent input for a single analysis run. Written by the analyze -/// handler, consumed by the analyze worker, removed after a successful -/// document write. -#[derive(Debug, Serialize, Deserialize)] +/// Input payload for a single analysis run. Built by the trigger +/// (handler or auto-trigger), shipped to the worker via [`AnalyzeJob`]. +#[derive(Debug, Clone, Serialize, Deserialize)] pub struct AnalysisInput { /// Latest filesystem mtime (server clock) of any `.m4a` in the case at /// the moment the analyze button was clicked. Carried through for @@ -65,7 +68,7 @@ pub struct AnalysisInput { pub backend_id: String, } -#[derive(Debug, Serialize, Deserialize)] +#[derive(Debug, Clone, Serialize, Deserialize)] pub struct RecordingInput { /// Recording timestamp from the watch (filename-derived, ISO-8601). pub recorded_at: String, diff --git a/server/src/analyze/recovery.rs b/server/src/analyze/recovery.rs index 9b6e3ee..2f7c15a 100644 --- a/server/src/analyze/recovery.rs +++ b/server/src/analyze/recovery.rs @@ -13,6 +13,7 @@ use std::time::{Duration, SystemTime}; use tracing::info; +use super::InFlight; use super::auto_trigger::{CaseAnalysisState, evaluate_state}; /// Aggregate counts over every UUID-named, non-closed case under a user @@ -64,14 +65,21 @@ pub async fn boot_scan_state( idle_threshold: Duration, now: SystemTime, boot_time: SystemTime, + in_flight: &InFlight, ) -> BootStateSummary { let mut total = BootStateSummary::default(); let Ok(mut users) = tokio::fs::read_dir(data_path).await else { return total; }; while let Ok(Some(user_entry)) = users.next_entry().await { - let per_user = - compute_state_for_user(&user_entry.path(), idle_threshold, now, boot_time).await; + let per_user = compute_state_for_user( + &user_entry.path(), + idle_threshold, + now, + boot_time, + in_flight, + ) + .await; total.add(per_user); } info!( @@ -92,6 +100,7 @@ pub async fn compute_state_for_user( idle_threshold: Duration, now: SystemTime, boot_time: SystemTime, + in_flight: &InFlight, ) -> BootStateSummary { let mut summary = BootStateSummary::default(); let Ok(mut cases) = tokio::fs::read_dir(user_root).await else { @@ -108,7 +117,7 @@ pub async fn compute_state_for_user( if uuid::Uuid::parse_str(&name).is_err() { continue; } - let state = evaluate_state(&case_dir, idle_threshold, now, boot_time).await; + let state = evaluate_state(&case_dir, idle_threshold, now, boot_time, in_flight).await; summary.record(&state); } summary @@ -120,7 +129,7 @@ mod tests { use crate::analyze::auto_trigger::{ FAILURE_MARKER_FILE, FailureMarker, rfc3339_of as auto_rfc3339, }; - use crate::analyze::{ANALYSIS_INPUT_FILE, AnalysisInput, DOCUMENT_FILE, Document}; + use crate::analyze::{DOCUMENT_FILE, Document}; use filetime::{FileTime, set_file_mtime}; use tempfile::TempDir; use tokio::fs; @@ -174,20 +183,6 @@ mod tests { .unwrap(); } - async fn write_input(case_dir: &Path, mtime: SystemTime) { - let input = AnalysisInput { - last_recording_mtime: auto_rfc3339(mtime), - recordings: Vec::new(), - backend_id: String::new(), - }; - fs::write( - case_dir.join(ANALYSIS_INPUT_FILE), - serde_json::to_vec(&input).unwrap(), - ) - .await - .unwrap(); - } - async fn write_marker(case_dir: &Path, mtime: SystemTime) { let marker = FailureMarker { last_recording_mtime: auto_rfc3339(mtime), @@ -208,17 +203,23 @@ mod tests { SystemTime::UNIX_EPOCH + Duration::from_secs(1_700_000_000) } + /// `Queued` is now driven by the in-process [`InFlight`] set, not + /// by a file on disk. A claimed slot for a transcribed case must + /// surface as `queued`. #[tokio::test] - async fn pending_input_counts_as_queued() { + async fn in_flight_claim_counts_as_queued() { let user_root = TempDir::new().unwrap(); let case = seed_transcribed_case(user_root.path(), 1, base_time()).await; - write_input(&case, base_time()).await; + + let in_flight = InFlight::new(); + assert!(in_flight.try_claim(&case)); let summary = compute_state_for_user( user_root.path(), Duration::ZERO, base_time(), SystemTime::UNIX_EPOCH, + &in_flight, ) .await; @@ -227,6 +228,32 @@ mod tests { assert_eq!(summary.required, 0); } + /// Pre-refactor leftover: a stale `analysis_input.json` on disk (no + /// claim in [`InFlight`]) must NOT register as `Queued` any more — + /// that was the whole reason for the refactor. The case still + /// produces a transcribed-but-no-document state, so it counts as + /// `idle` (zero idle threshold collapses Required→Idle). + #[tokio::test] + async fn orphan_input_file_does_not_count_as_queued() { + let user_root = TempDir::new().unwrap(); + let case = seed_transcribed_case(user_root.path(), 7, base_time()).await; + fs::write(case.join("analysis_input.json"), b"{}") + .await + .unwrap(); + + let summary = compute_state_for_user( + user_root.path(), + Duration::ZERO, + base_time(), + SystemTime::UNIX_EPOCH, + &InFlight::new(), + ) + .await; + + assert_eq!(summary.queued, 0); + assert_eq!(summary.idle, 1); + } + #[tokio::test] async fn document_covering_latest_counts_as_current() { let user_root = TempDir::new().unwrap(); @@ -238,6 +265,7 @@ mod tests { Duration::ZERO, base_time(), SystemTime::UNIX_EPOCH, + &InFlight::new(), ) .await; @@ -258,6 +286,7 @@ mod tests { Duration::ZERO, base_time(), SystemTime::UNIX_EPOCH, + &InFlight::new(), ) .await; @@ -277,6 +306,7 @@ mod tests { Duration::ZERO, base_time(), SystemTime::UNIX_EPOCH, + &InFlight::new(), ) .await; @@ -298,6 +328,7 @@ mod tests { Duration::ZERO, base_time(), SystemTime::UNIX_EPOCH, + &InFlight::new(), ) .await; @@ -316,6 +347,7 @@ mod tests { Duration::from_secs(900), base_time() + Duration::from_secs(100), SystemTime::UNIX_EPOCH, + &InFlight::new(), ) .await; diff --git a/server/src/analyze/worker.rs b/server/src/analyze/worker.rs index 17074ad..14b2227 100644 --- a/server/src/analyze/worker.rs +++ b/server/src/analyze/worker.rs @@ -7,7 +7,8 @@ use tracing::{error, info, warn}; use super::auto_trigger::FailureDetails; use super::backend::{default_backend, find_backend}; use super::{ - ANALYSIS_INPUT_FILE, AnalysisInput, AnalyzeReceiver, DOCUMENT_FILE, auto_trigger, llm, prompt, + AnalysisInput, AnalyzeReceiver, DOCUMENT_FILE, InFlight, InFlightGuard, auto_trigger, llm, + prompt, }; use crate::events::{self, CaseEventKind, EventSender}; use crate::gazetteer::Gazetteer; @@ -20,14 +21,18 @@ pub async fn run( mut rx: AnalyzeReceiver, client: reqwest::Client, worker_busy: WorkerBusy, + in_flight: InFlight, vocab: Arc, events_tx: EventSender, ) { info!(vocab_entries = vocab.len(), "Analyze worker started"); while let Some(job) = rx.recv().await { - let _guard = BusyGuard::new(worker_busy.clone()); - process(&job.case_dir, &client, &vocab, &events_tx).await; + let _busy = BusyGuard::new(worker_busy.clone()); + // RAII guard: releases the in-flight slot on every exit path — + // success, all failure branches, and panics during the LLM call. + let _slot = InFlightGuard::new(in_flight.clone(), job.case_dir.clone()); + process(&job.case_dir, job.input, &client, &vocab, &events_tx).await; } warn!("Analyze worker stopped (channel closed)"); @@ -35,33 +40,18 @@ pub async fn run( async fn process( case_dir: &Path, + input: AnalysisInput, client: &reqwest::Client, vocab: &Gazetteer, events_tx: &EventSender, ) { - let input_path = case_dir.join(ANALYSIS_INPUT_FILE); let document_path = case_dir.join(DOCUMENT_FILE); // The previous "document exists → skip" guard has been removed: // re-analysis is a normal path now (auto-trigger Idle, manual // Aktualisieren-Bulk, Neu analysieren-Button), and the previous // document.json must be overwritten in place to keep "letztes - // Dokument" visible until the new run lands. Recovery still - // pre-filters cases with an existing document at recovery.rs:37. - - let input: AnalysisInput = match tokio::fs::read(&input_path).await { - Ok(bytes) => match serde_json::from_slice(&bytes) { - Ok(v) => v, - Err(e) => { - error!(path = %input_path.display(), error = %e, "analysis input parse failed"); - return; - } - }, - Err(e) => { - error!(path = %input_path.display(), error = %e, "analysis input read failed"); - return; - } - }; + // Dokument" visible until the new run lands. // Silent-only case: no usable transcripts. Skip LLM, write a stub. if input.recordings.is_empty() { @@ -86,9 +76,6 @@ async fn process( emit_analysis_failed(events_tx, case_dir); return; } - if let Err(e) = tokio::fs::remove_file(&input_path).await { - warn!(path = %input_path.display(), error = %e, "removing analysis input failed"); - } auto_trigger::remove_failure_marker(case_dir).await; info!(case = %case_dir.display(), "analysis done (stub, no recordings)"); events::emit( @@ -112,10 +99,6 @@ async fn process( "sending to llm" ); - // NOTE: If the server crashes between a successful LLM response and the - // document write below, the recovery scan will re-enqueue this job and - // the call will be made again. Accepted cost — rare event, small - // per-call price. Response-caching in a `.tmp` file would prevent it. let document = match llm::chat_once(client, backend, &user_content).await { Ok(text) => text, Err(e) => { @@ -205,12 +188,6 @@ 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"); - } auto_trigger::remove_failure_marker(case_dir).await; info!( diff --git a/server/src/lib.rs b/server/src/lib.rs index b09d326..9b86889 100644 --- a/server/src/lib.rs +++ b/server/src/lib.rs @@ -26,7 +26,7 @@ use axum::extract::FromRef; use axum::http::{HeaderName, HeaderValue}; use tower_http::set_header::SetResponseHeaderLayer; -use analyze::AnalyzeSender; +use analyze::{AnalyzeSender, InFlight}; use config::Config; use gazetteer::Gazetteer; use magic_link::MagicLinkStore; @@ -114,6 +114,10 @@ pub struct AppState { pub http_client: reqwest::Client, pub vocab: Arc, pub boot_time: BootTime, + /// Process-local set of cases currently queued or being analyzed. + /// Replaces the old "does `analysis_input.json` exist on disk?" + /// marker so a server restart never reports stale `Queued` states. + pub analyze_in_flight: InFlight, } impl FromRef for Arc { @@ -213,6 +217,12 @@ impl FromRef for BootTime { } } +impl FromRef for InFlight { + fn from_ref(state: &AppState) -> Self { + state.analyze_in_flight.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. @@ -241,9 +251,29 @@ pub fn create_router_and_session_store_with_settings( config: Arc, settings: Arc, ) -> (Router, SessionStore) { + let (router, session_store, _, _) = create_router_with_full_handles(config, settings); + (router, session_store) +} + +/// Test-friendly variant: returns the router, session store, the +/// process-local [`InFlight`] set, and the receiver half of the +/// analyze channel. Tests that need to inspect the [`AnalyzeJob`] +/// payload (recordings, backend_id) consume the receiver to read the +/// job a handler just sent. Tests that only care about the HTTP +/// response keep using [`create_router_and_session_store_with_settings`]. +pub fn create_router_with_full_handles( + config: Arc, + settings: Arc, +) -> ( + Router, + SessionStore, + analyze::InFlight, + analyze::AnalyzeReceiver, +) { let (transcribe_tx, _transcribe_rx) = transcribe::channel(); - let (analyze_tx, _analyze_rx) = analyze::channel(); + let (analyze_tx, analyze_rx) = analyze::channel(); let session_store = web_session::new_store(); + let in_flight = InFlight::new(); let router = create_router_with_state(AppState { config, settings, @@ -263,8 +293,9 @@ pub fn create_router_and_session_store_with_settings( // anyway. Tests that exercise the grace-window itself construct // an `AppState` with an explicit recent `BootTime`. boot_time: BootTime(std::time::SystemTime::UNIX_EPOCH), + analyze_in_flight: in_flight.clone(), }); - (router, session_store) + (router, session_store, in_flight, analyze_rx) } pub fn create_router_with_state(state: AppState) -> Router { diff --git a/server/src/main.rs b/server/src/main.rs index 5bc05e0..5ec90f2 100644 --- a/server/src/main.rs +++ b/server/src/main.rs @@ -190,10 +190,12 @@ async fn main() { let (analyze_tx, analyze_rx) = analyze::channel(); let analyze_busy: doctate_server::WorkerBusy = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); + let analyze_in_flight = analyze::InFlight::new(); tokio::spawn(analyze::worker::run( analyze_rx, http_client.clone(), analyze_busy.clone(), + analyze_in_flight.clone(), vocab.clone(), events_tx.clone(), )); @@ -201,12 +203,14 @@ async fn main() { { let data_path = config.data_path.clone(); let idle_threshold = std::time::Duration::from_secs(config.analysis_idle_threshold_secs); + let in_flight = analyze_in_flight.clone(); tokio::spawn(async move { analyze::recovery::boot_scan_state( &data_path, idle_threshold, std::time::SystemTime::now(), server_boot_time, + &in_flight, ) .await; }); @@ -227,6 +231,7 @@ async fn main() { http_client, vocab, boot_time: doctate_server::BootTime(server_boot_time), + analyze_in_flight, }; let addr = format!("0.0.0.0:{}", config.server_port); diff --git a/server/src/routes/bulk.rs b/server/src/routes/bulk.rs index 748078a..4aab93f 100644 --- a/server/src/routes/bulk.rs +++ b/server/src/routes/bulk.rs @@ -8,9 +8,9 @@ use doctate_common::timestamp::now_rfc3339; use serde::Deserialize; use tracing::{info, warn}; -use crate::analyze::auto_trigger::{self, CaseAnalysisState, write_input_overwrite}; +use crate::analyze::auto_trigger::{self, CaseAnalysisState}; use crate::analyze::backend::any_backend_available; -use crate::analyze::{ANALYSIS_INPUT_FILE, AnalyzeJob, AnalyzeSender}; +use crate::analyze::{AnalyzeJob, AnalyzeSender, InFlight}; use crate::auth::AuthenticatedWebUser; use crate::config::Config; use crate::csrf::{CsrfForm, HasCsrfToken}; @@ -18,9 +18,7 @@ use crate::error::AppError; use crate::events::{self, CaseEventKind, EventSender}; use crate::oneliner_locks::OnelinerLocks; use crate::paths::{CloseMarker, write_close_marker}; -use crate::routes::case_actions::{ - build_analysis_input, reset_case_artefacts, write_input_create_new, -}; +use crate::routes::case_actions::{build_analysis_input, reset_case_artefacts}; use crate::routes::user_web::locate_case; #[derive(Debug, Deserialize)] @@ -47,6 +45,7 @@ pub async fn handle_bulk_action( user: AuthenticatedWebUser, State(config): State>, State(analyze_tx): State, + State(in_flight): State, State(events_tx): State, State(locks): State, CsrfForm(form): CsrfForm, @@ -73,6 +72,7 @@ pub async fn handle_bulk_action( BulkAction::Analyze => { bulk_analyze( &analyze_tx, + &in_flight, &events_tx, &user.slug, &user_root, @@ -82,7 +82,15 @@ pub async fn handle_bulk_action( } BulkAction::Close => bulk_close(&events_tx, &user.slug, &user_root, &form.case_ids).await, BulkAction::Reset => { - bulk_reset(&events_tx, &locks, &user.slug, &user_root, &form.case_ids).await + bulk_reset( + &events_tx, + &locks, + &in_flight, + &user.slug, + &user_root, + &form.case_ids, + ) + .await } } @@ -91,6 +99,7 @@ pub async fn handle_bulk_action( async fn bulk_analyze( analyze_tx: &AnalyzeSender, + in_flight: &InFlight, events_tx: &EventSender, slug: &str, user_root: &std::path::Path, @@ -113,30 +122,31 @@ async fn bulk_analyze( skipped += 1; continue; }; - // The previous document.json is left in place: the UI keeps - // rendering it as "letztes Dokument" with a stale-banner until - // the worker overwrites it atomically on the next successful run. + if !in_flight.try_claim(&case_dir) { + // A job for this case is already queued or running. + skipped += 1; + continue; + } let input = match build_analysis_input(&case_dir, "").await { Ok(i) => i, Err(_) => { warn!(slug = %slug, case_id = %case_id, "bulk-analyze: build_analysis_input failed"); + in_flight.release(&case_dir); skipped += 1; continue; } }; - let input_path = case_dir.join(ANALYSIS_INPUT_FILE); - if write_input_create_new(&input_path, &input).await.is_err() { - warn!(slug = %slug, case_id = %case_id, "bulk-analyze: write input failed"); - skipped += 1; - continue; - } if analyze_tx .send(AnalyzeJob { case_dir: case_dir.clone(), + input, }) .is_err() { warn!(slug = %slug, case_id = %case_id, "bulk-analyze: send failed (worker gone)"); + in_flight.release(&case_dir); + skipped += 1; + continue; } events::emit(events_tx, slug, case_id, CaseEventKind::AnalysisQueued); ok += 1; @@ -181,6 +191,7 @@ async fn bulk_close( async fn bulk_reset( events_tx: &EventSender, locks: &OnelinerLocks, + in_flight: &InFlight, slug: &str, user_root: &std::path::Path, case_ids: &[String], @@ -198,6 +209,7 @@ async fn bulk_reset( skipped += 1; continue; }; + in_flight.release(&case_dir); if let Err(e) = reset_case_artefacts(&case_dir, locks).await { warn!(slug = %slug, case_id = %case_id, error = %e, "bulk-reset: failed"); skipped += 1; @@ -234,6 +246,7 @@ pub async fn handle_analyze_required( user: AuthenticatedWebUser, State(config): State>, State(analyze_tx): State, + State(in_flight): State, State(events_tx): State, State(boot_time): State, CsrfForm(_form): CsrfForm, @@ -276,7 +289,9 @@ pub async fn handle_analyze_required( // are intentionally left alone (Failed → manual per-backend retry // path on the case page; Queued → already in flight; Current → // nothing to do). - match auto_trigger::evaluate_state(&case_dir, idle_threshold, now, boot_time.0).await { + match auto_trigger::evaluate_state(&case_dir, idle_threshold, now, boot_time.0, &in_flight) + .await + { CaseAnalysisState::Required { .. } | CaseAnalysisState::Idle { .. } => {} _ => { skipped += 1; @@ -284,27 +299,30 @@ pub async fn handle_analyze_required( } } + if !in_flight.try_claim(&case_dir) { + // A concurrent trigger raced us — they will send the job. + skipped += 1; + continue; + } + let input = match build_analysis_input(&case_dir, "").await { Ok(i) => i, Err(_) => { warn!(slug = %user.slug, case = %case_dir.display(), "analyze-required: build_analysis_input failed"); + in_flight.release(&case_dir); skipped += 1; continue; } }; - let input_path = case_dir.join(ANALYSIS_INPUT_FILE); - if let Err(e) = write_input_overwrite(&input_path, &input).await { - warn!(slug = %user.slug, case = %case_dir.display(), error = %e, "analyze-required: write input failed"); - skipped += 1; - continue; - } if analyze_tx .send(AnalyzeJob { case_dir: case_dir.clone(), + input, }) .is_err() { warn!(slug = %user.slug, case = %case_dir.display(), "analyze-required: send failed (worker gone)"); + in_flight.release(&case_dir); skipped += 1; continue; } diff --git a/server/src/routes/case_actions.rs b/server/src/routes/case_actions.rs index dc0e2ab..197c287 100644 --- a/server/src/routes/case_actions.rs +++ b/server/src/routes/case_actions.rs @@ -8,7 +8,6 @@ use axum::response::Redirect; use serde::Deserialize; use time::OffsetDateTime; use time::format_description::well_known::Rfc3339; -use tokio::io::AsyncWriteExt; use tracing::{info, warn}; use doctate_common::timestamp::{filename_stem_to_recorded_at, now_rfc3339}; @@ -17,7 +16,7 @@ use doctate_common::{RECORDING_META_SUFFIX, TranscriptState}; use crate::analyze::auto_trigger::FAILURE_MARKER_FILE; use crate::analyze::backend::{any_backend_available, find_backend}; use crate::analyze::{ - ANALYSIS_INPUT_FILE, AnalysisInput, AnalyzeJob, AnalyzeSender, DOCUMENT_FILE, RecordingInput, + AnalysisInput, AnalyzeJob, AnalyzeSender, DOCUMENT_FILE, InFlight, RecordingInput, }; use crate::auth::AuthenticatedWebUser; use crate::case_id::CaseIdPath; @@ -54,16 +53,15 @@ impl HasCsrfToken for AnalyzeForm { /// POST /cases/{case_id}/analyze /// -/// Trigger an LLM analysis for the case. If no document exists yet, writes -/// `analysis_input_v1.json`; otherwise writes `analysis_input_v{N+1}.json` -/// where N is the current latest document version. The worker picks up the -/// new input and produces `document_v{version}.md`. Available to all logged-in -/// users; the optional `backend` form field is admin-only. Race protection -/// via `create_new` on the input file. +/// Trigger an LLM analysis for the case. Race protection via the +/// process-local [`InFlight`] set: a duplicate trigger while a job is +/// already queued or running returns 409. Available to all logged-in +/// users; the optional `backend` form field is admin-only. pub async fn handle_analyze_case( user: AuthenticatedWebUser, State(config): State>, State(analyze_tx): State, + State(in_flight): State, State(events_tx): State, CaseIdPath(case_id): CaseIdPath, CsrfForm(form): CsrfForm, @@ -96,47 +94,56 @@ pub async fn handle_analyze_case( let user_root = config.data_path.join(&user.slug); let case_dir = locate_case_or_404(&user_root, &case_id, &user.slug, "analyze").await?; + // Atomic claim. Concurrent triggers (manual + auto-trigger, + // duplicate clicks, second tab) hit this and the loser bails with + // 409 — the winner sends a single job. + if !in_flight.try_claim(&case_dir) { + return Err(AppError::Conflict( + "Analyse läuft bereits für diesen Fall".into(), + )); + } + // Re-analyze: delete the existing document first so the UI stops showing // the stale "ausgewertet" state while the worker re-computes. Ignore // NotFound (first-time analysis path). let document_path = case_dir.join(DOCUMENT_FILE); - match tokio::fs::remove_file(&document_path).await { - Ok(_) => {} - Err(e) if e.kind() == std::io::ErrorKind::NotFound => {} - Err(e) => return Err(AppError::Internal(format!("remove old document: {e}"))), + if let Err(e) = tokio::fs::remove_file(&document_path).await + && e.kind() != std::io::ErrorKind::NotFound + { + in_flight.release(&case_dir); + return Err(AppError::Internal(format!("remove old document: {e}"))); } - let input = build_analysis_input(&case_dir, &backend_id).await?; - let input_path = case_dir.join(ANALYSIS_INPUT_FILE); + let input = match build_analysis_input(&case_dir, &backend_id).await { + Ok(i) => i, + Err(e) => { + in_flight.release(&case_dir); + return Err(e); + } + }; - // Manual retry path: a failure marker means the worker has explicitly - // given up on the prior input. Drop the stale sentinel so create_new - // does not reject the legitimate retry as "Analyse läuft bereits". - // The marker itself is left in place — the worker overwrites it on - // re-failure and removes it on success, so the UI's "analyzing" flag - // (driven by input file presence) takes over from the next render. - if case_dir.join(FAILURE_MARKER_FILE).exists() { - remove_if_exists(&input_path).await?; - } + let recordings_count = input.recordings.len(); - write_input_create_new(&input_path, &input).await?; - - // Unbounded send: only fails if the worker has gone away (programmer error - // or shutdown race). Log but don't fail the request — the file is already - // on disk and recovery will pick it up on next startup. + // Unbounded send: only fails if the worker is gone (shutdown race). + // Release the slot so a future request can re-trigger; surface the + // failure as a log line rather than 500 to keep the UI honest — + // the user lands on the case page and the missing document is + // self-evident. if analyze_tx .send(AnalyzeJob { case_dir: case_dir.clone(), + input, }) .is_err() { warn!(case_id = %case_id, "analyze send failed (worker gone)"); + in_flight.release(&case_dir); } info!( slug = %user.slug, case_id = %case_id, - recordings = input.recordings.len(), + recordings = recordings_count, "analyze requested; analysis enqueued" ); @@ -343,40 +350,6 @@ async fn collect_m4as(case_dir: &Path) -> Result, App Ok(out) } -/// Serialize `input` and write it to `path` using `create_new` to atomically -/// reserve the destination. Returns `Conflict` if another close already won -/// the race. -/// -/// Trade-off: if the process crashes between `open` and `write_all`/`sync_all`, -/// a zero-byte or partially-written JSON remains. The worker will fail to -/// deserialize it and log an error — manual cleanup required. For MVP the -/// simpler check-and-create is acceptable. -pub(crate) async fn write_input_create_new( - path: &Path, - input: &AnalysisInput, -) -> Result<(), AppError> { - let bytes = serde_json::to_vec_pretty(input) - .map_err(|e| AppError::Internal(format!("serialize input: {e}")))?; - - let mut f = match tokio::fs::OpenOptions::new() - .write(true) - .create_new(true) - .open(path) - .await - { - Ok(f) => f, - Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => { - return Err(AppError::Conflict( - "Analyse läuft bereits für diesen Fall".into(), - )); - } - Err(e) => return Err(AppError::Internal(e.to_string())), - }; - f.write_all(&bytes).await?; - f.sync_all().await?; - Ok(()) -} - /// Read `case_dir/document.json` and return the embedded markdown /// body. Returns `None` if no document exists or the file is not valid /// JSON for the [`Document`] envelope. A parse error is logged so a @@ -394,9 +367,12 @@ pub(crate) async fn read_document(case_dir: &Path) -> Option { } /// Reset a case to its raw audio: delete derived artefacts -/// (transcripts, analysis input, document, non-Manual oneliner) and rename every -/// `*.m4a.failed` back to `*.m4a` so the transcribe self-heal picks it -/// up again. Idempotent: missing files are no-ops. +/// (transcripts, document, non-Manual oneliner, failure marker) and +/// rename every `*.m4a.failed` back to `*.m4a` so the transcribe +/// self-heal picks it up again. Idempotent: missing files are no-ops. +/// The caller is responsible for releasing any in-flight slot for this +/// case via [`InFlight::release`] — `reset_case_artefacts` itself is +/// pure file I/O. /// /// `oneliner.json` is special: a [`OnelinerState::Manual`] override is /// preserved (per the sticky-Manual rule). Routed through @@ -406,7 +382,11 @@ pub(crate) async fn reset_case_artefacts( case_dir: &Path, locks: &OnelinerLocks, ) -> std::io::Result<()> { - for name in [DOCUMENT_FILE, ANALYSIS_INPUT_FILE] { + // Failure marker is part of the lifecycle state — without removing + // it, a stale ``-pinned marker would keep classifying + // the case as `Failed` until a new recording arrives, even though + // the user has explicitly chosen to start over. + for name in [DOCUMENT_FILE, FAILURE_MARKER_FILE] { match tokio::fs::remove_file(case_dir.join(name)).await { Ok(_) => {} Err(e) if e.kind() == std::io::ErrorKind::NotFound => {} @@ -552,6 +532,7 @@ pub async fn handle_reset_case( State(config): State>, State(events_tx): State, State(locks): State, + State(in_flight): State, CaseIdPath(case_id): CaseIdPath, CsrfForm(form): CsrfForm, ) -> Result { @@ -562,6 +543,14 @@ pub async fn handle_reset_case( let user_root = config.data_path.join(&user.slug); let case_dir = locate_case_or_404(&user_root, &case_id, &user.slug, "reset").await?; + // A worker job may still be running against the old artefacts; the + // RAII `InFlightGuard` will release the slot when it completes. We + // also clear the slot here so the UI immediately stops showing + // "wird analysiert" — the worker's fresh-write path is harmless, + // it will just produce a document the user is about to overwrite + // anyway. + in_flight.release(&case_dir); + reset_case_artefacts(&case_dir, &locks) .await .map_err(|e| AppError::Internal(format!("reset: {e}")))?; @@ -696,6 +685,7 @@ pub async fn handle_delete_recording( State(config): State>, State(events_tx): State, State(locks): State, + State(in_flight): State, CaseIdPath(case_id): CaseIdPath, CsrfForm(form): CsrfForm, ) -> Result { @@ -709,15 +699,18 @@ pub async fn handle_delete_recording( .trim_end_matches(".failed") .trim_end_matches(".m4a"); + // The recording set is changing — any pending analysis is now stale. + // Release the in-flight slot so the next render auto-triggers fresh. + in_flight.release(&case_dir); + // Audio and its single metadata sidecar. NotFound on any of these // is fine: the JSON sidecar may never have been written (the // worker hadn't reached the atomic write yet, or the file got // deleted twice). - let targets: [PathBuf; 4] = [ + let targets: [PathBuf; 3] = [ case_dir.join(&form.filename), case_dir.join(format!("{stem}.{RECORDING_META_SUFFIX}")), case_dir.join(DOCUMENT_FILE), - case_dir.join(ANALYSIS_INPUT_FILE), ]; for p in &targets { remove_if_exists(p).await?; diff --git a/server/src/routes/user_web.rs b/server/src/routes/user_web.rs index 6c3e186..e53ac44 100644 --- a/server/src/routes/user_web.rs +++ b/server/src/routes/user_web.rs @@ -13,7 +13,7 @@ use tracing::{info, warn}; use time::{Date, OffsetDateTime, format_description::well_known::Rfc3339}; -use crate::analyze::{ANALYSIS_INPUT_FILE, DOCUMENT_FILE, auto_trigger}; +use crate::analyze::{DOCUMENT_FILE, InFlight, auto_trigger}; use crate::auth::AuthenticatedWebUser; use crate::case_id::{CaseId, CaseIdPath}; use crate::config::Config; @@ -479,12 +479,13 @@ async fn compute_flags( idle_threshold: std::time::Duration, now_systime: std::time::SystemTime, boot_time: std::time::SystemTime, + in_flight: &InFlight, ) -> CaseFlags { let has_document = any_document_exists(case_dir).await; - // 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; + // Honest status: a claimed in-flight slot is the only authoritative + // signal that a job is queued or running. The set is process-local + // so it cannot survive a crash — no orphan markers any more. + let analyzing = !has_document && worker_busy && in_flight.contains(case_dir); let non_failed: Vec<&RecordingView> = recordings.iter().filter(|r| !r.failed).collect(); let all_transcribed = @@ -521,7 +522,8 @@ async fn compute_flags( }; let state = - auto_trigger::evaluate_state(case_dir, idle_threshold, now_systime, boot_time).await; + auto_trigger::evaluate_state(case_dir, idle_threshold, now_systime, boot_time, in_flight) + .await; let analysis_stale_with_doc = matches!( state, auto_trigger::CaseAnalysisState::Required { has_document: true } @@ -538,14 +540,6 @@ async fn compute_flags( } } -/// True iff `analysis_input.json` exists. The worker removes it after a -/// successful document write, so existence implies a pending or in-flight job. -pub(crate) async fn any_analysis_input_exists(case_dir: &Path) -> bool { - tokio::fs::try_exists(case_dir.join(ANALYSIS_INPUT_FILE)) - .await - .unwrap_or(false) -} - /// True iff `document.md` exists. pub(crate) async fn any_document_exists(case_dir: &Path) -> bool { tokio::fs::try_exists(case_dir.join(DOCUMENT_FILE)) @@ -629,6 +623,7 @@ pub async fn handle_my_cases( State(http_client): State, State(vocab): State>, State(boot_time): State, + State(in_flight): State, Query(query): Query, ) -> Result, AppError> { let user_root = config.data_path.join(&user.slug); @@ -653,6 +648,7 @@ pub async fn handle_my_cases( idle_threshold, now, boot_time.0, + &in_flight, &pipeline.analyze_tx, &events_tx, ) @@ -679,6 +675,7 @@ pub async fn handle_my_cases( retention.auto_delete_days, crate::analyze::backend::any_backend_available(), boot_time.0, + &in_flight, ) .await; let total = cases.len(); @@ -737,6 +734,7 @@ pub async fn handle_case_page( State(http_client): State, State(vocab): State>, State(boot_time): State, + State(in_flight): State, CaseIdPath(case_id): CaseIdPath, Query(query): Query, ) -> Result, AppError> { @@ -780,6 +778,7 @@ pub async fn handle_case_page( idle_threshold, now, boot_time.0, + &in_flight, &pipeline.analyze_tx, &events_tx, ) @@ -809,6 +808,7 @@ pub async fn handle_case_page( idle_threshold, now, boot_time.0, + &in_flight, ) .await; @@ -1015,6 +1015,7 @@ async fn scan_user_cases( auto_delete_days: u32, llm_configured: bool, boot_time: std::time::SystemTime, + in_flight: &InFlight, ) -> Vec { let user_root = config.data_path.join(slug); let mut cases = Vec::new(); @@ -1040,6 +1041,7 @@ async fn scan_user_cases( idle_threshold, now_systime, boot_time, + in_flight, ) .await { @@ -1086,6 +1088,7 @@ async fn compute_case_view( idle_threshold: std::time::Duration, now_systime: std::time::SystemTime, boot_time: std::time::SystemTime, + in_flight: &InFlight, ) -> Option { let recordings = scan_recordings(case_path).await; if recordings.is_empty() { @@ -1111,11 +1114,12 @@ async fn compute_case_view( let (is_closed, days_until_purge) = closed_state(case_path, auto_delete_days, now).await; // The typed lifecycle is the single source of truth for both the - // badge and the "is the Analysieren-button enabled?" - // decision: a case in `Queued` (i.e. `analysis_input.json` on disk) - // must not let the user enqueue a duplicate job. + // badge and the "is the Analysieren-button enabled?" decision: a + // case in `Queued` (claim held in [`InFlight`]) must not let the + // user enqueue a duplicate job. let state = - auto_trigger::evaluate_state(case_path, idle_threshold, now_systime, boot_time).await; + auto_trigger::evaluate_state(case_path, idle_threshold, now_systime, boot_time, in_flight) + .await; let analysis_in_flight = matches!(state, auto_trigger::CaseAnalysisState::Queued); // Closed cases must never advertise as "Analyse erforderlich" — the // whole point of closing a case is to freeze it. Suppress the badge diff --git a/server/tests/analyze_test.rs b/server/tests/analyze_test.rs index 9ecbcb6..73c8b7d 100644 --- a/server/tests/analyze_test.rs +++ b/server/tests/analyze_test.rs @@ -188,7 +188,7 @@ async fn analyze_case_without_llm_returns_503() { // --------------------------------------------------------------------- #[tokio::test] -async fn analyze_case_happy_path_writes_input_and_redirects() { +async fn analyze_case_happy_path_sends_job_and_redirects() { let (cfg, settings) = cfg_llm("g"); let case_id = "11111111-1111-1111-1111-111111111111"; let case_dir = seed_case(&cfg.data_path, "dr_a", case_id); @@ -203,7 +203,8 @@ async fn analyze_case_happy_path_writes_input_and_redirects() { Some("Korrektur: links, nicht rechts."), ); - let (app, store) = doctate_server::create_router_and_session_store_with_settings(cfg, settings); + let (app, store, in_flight, mut analyze_rx) = + doctate_server::create_router_with_full_handles(cfg, settings); let (cookie, csrf) = login_with_csrf(&app, &store, "dr_a", "s").await; let resp = app @@ -225,17 +226,26 @@ async fn analyze_case_happy_path_writes_input_and_redirects() { format!("/cases/{case_id}") ); - let input_path = case_dir.join(ANALYSIS_INPUT_FILE); - assert!(input_path.exists(), "analysis_input.json missing"); + // Slot is claimed for the duration of the worker's run — proxied + // here by the receiver still holding the un-consumed job. + assert!( + in_flight.contains(&case_dir), + "in-flight slot must be held until the worker drops it" + ); - let raw = std::fs::read_to_string(&input_path).unwrap(); - let parsed: Value = serde_json::from_str(&raw).unwrap(); - let recs = parsed["recordings"].as_array().unwrap(); - assert_eq!(recs.len(), 2); - assert_eq!(recs[0]["recorded_at"], "2026-04-15T10:00:00Z"); - assert_eq!(recs[0]["text"], "Patient klagt über Knie."); - assert_eq!(recs[1]["recorded_at"], "2026-04-15T10:05:00Z"); - assert!(parsed["last_recording_mtime"].is_string()); + let job = analyze_rx + .try_recv() + .expect("handler must enqueue exactly one job"); + assert_eq!(job.case_dir, case_dir); + assert_eq!(job.input.recordings.len(), 2); + assert_eq!(job.input.recordings[0].recorded_at, "2026-04-15T10:00:00Z"); + assert_eq!(job.input.recordings[0].text, "Patient klagt über Knie."); + assert_eq!(job.input.recordings[1].recorded_at, "2026-04-15T10:05:00Z"); + assert!(!job.input.last_recording_mtime.is_empty()); + assert!( + analyze_rx.try_recv().is_err(), + "exactly one job must be enqueued" + ); } #[tokio::test] @@ -245,7 +255,12 @@ async fn analyze_case_second_time_returns_409() { let case_dir = seed_case(&cfg.data_path, "dr_a", case_id); seed_recording(&case_dir, "2026-04-15T10-00-00Z", Some("text")); - let (app, store) = doctate_server::create_router_and_session_store_with_settings(cfg, settings); + // Hold the receiver alive across both requests — otherwise the + // first send fails, the slot is released, and the second click + // would be accepted as a fresh trigger. In production the worker + // keeps the slot held via `InFlightGuard` until it finishes. + let (app, store, _in_flight, _analyze_rx) = + doctate_server::create_router_with_full_handles(cfg, settings); let (cookie, csrf) = login_with_csrf(&app, &store, "dr_a", "s").await; let first = app @@ -273,32 +288,29 @@ async fn analyze_case_second_time_returns_409() { } /// Bug regression (case c414cf52, 2026-05-03): a prior LLM run failed, -/// leaving `analysis_input.json` AND `.analysis_failed.json` on disk. -/// The user clicks "Erneut versuchen" — the handler must accept the -/// retry instead of rejecting it as 409 Conflict ("Analyse läuft -/// bereits"). The failure marker is the explicit "worker has given up" -/// signal that legitimises overwriting the stale sentinel. +/// leaving `.analysis_failed.json` on disk. The user clicks "Erneut +/// versuchen" — the handler must accept the retry. With the in-flight +/// marker now process-local, no on-disk sentinel can falsely block the +/// retry; the explicit user click claims a fresh slot and sends a job +/// with the current transcripts. #[tokio::test] -async fn analyze_retry_after_failure_marker_overwrites_stale_input() { +async fn analyze_retry_after_failure_marker_succeeds() { let (cfg, settings) = cfg_llm("retry-after-fail"); let case_id = "11111111-1111-1111-1111-111111111111"; let case_dir = seed_case(&cfg.data_path, "dr_a", case_id); seed_recording(&case_dir, "2026-04-15T10-00-00Z", Some("frischer Text")); - // Simulate the post-failure on-disk state: stale input from the - // attempt the worker gave up on, plus the matching failure marker. - std::fs::write( - case_dir.join(ANALYSIS_INPUT_FILE), - r#"{"last_recording_mtime":"2026-04-15T10:00:00Z","recordings":[{"recorded_at":"2026-04-15T10:00:00Z","text":"alter Text"}],"backend_id":""}"#, - ) - .unwrap(); + // Failure marker from the prior aborted attempt remains on disk — + // the new model leaves it in place; the worker overwrites it on + // re-failure or removes it on success. std::fs::write( case_dir.join(".analysis_failed.json"), r#"{"last_recording_mtime":"2026-04-15T10:00:00Z","reason":"llm: connect timeout","failed_at":"2026-04-15T10:05:00Z"}"#, ) .unwrap(); - let (app, store) = doctate_server::create_router_and_session_store_with_settings(cfg, settings); + let (app, store, _in_flight, mut analyze_rx) = + doctate_server::create_router_with_full_handles(cfg, settings); let (cookie, csrf) = login_with_csrf(&app, &store, "dr_a", "s").await; let resp = app @@ -316,13 +328,13 @@ async fn analyze_retry_after_failure_marker_overwrites_stale_input() { "retry with failure marker present must succeed, not 409" ); - // Input was rewritten with the fresh transcript — the stale "alter Text" - // must be gone. - let raw = std::fs::read_to_string(case_dir.join(ANALYSIS_INPUT_FILE)).unwrap(); - let parsed: Value = serde_json::from_str(&raw).unwrap(); - let recs = parsed["recordings"].as_array().unwrap(); - assert_eq!(recs.len(), 1); - assert_eq!(recs[0]["text"], "frischer Text"); + // The job carries the fresh transcript — there is no on-disk state + // to validate any more, the channel is the truth. + let job = analyze_rx + .try_recv() + .expect("retry must enqueue a fresh job"); + assert_eq!(job.input.recordings.len(), 1); + assert_eq!(job.input.recordings[0].text, "frischer Text"); } #[tokio::test] @@ -334,7 +346,8 @@ async fn analyze_case_skips_blank_transcript_from_recordings() { seed_recording(&case_dir, "2026-04-15T10-05-00Z", Some(" ")); // blank seed_recording(&case_dir, "2026-04-15T10-10-00Z", Some("auch echt")); - let (app, store) = doctate_server::create_router_and_session_store_with_settings(cfg, settings); + let (app, store, _in_flight, mut analyze_rx) = + doctate_server::create_router_with_full_handles(cfg, settings); let (cookie, csrf) = login_with_csrf(&app, &store, "dr_a", "s").await; let resp = app @@ -348,10 +361,13 @@ async fn analyze_case_skips_blank_transcript_from_recordings() { .unwrap(); assert_eq!(resp.status(), StatusCode::SEE_OTHER); - let raw = std::fs::read_to_string(case_dir.join(ANALYSIS_INPUT_FILE)).unwrap(); - let parsed: Value = serde_json::from_str(&raw).unwrap(); - let recs = parsed["recordings"].as_array().unwrap(); - assert_eq!(recs.len(), 2, "blank transcript must be filtered out"); + let job = analyze_rx.try_recv().expect("job must be enqueued"); + assert_eq!( + job.input.recordings.len(), + 2, + "blank transcript must be filtered out" + ); + let _ = case_dir; // name kept for grep symmetry with the seed calls } // --------------------------------------------------------------------- @@ -376,17 +392,14 @@ async fn analyze_worker_writes_document_via_wiremock() { let case_dir = tmp.join("dr_a/11111111-1111-1111-1111-111111111111"); std::fs::create_dir_all(&case_dir).unwrap(); - let input = json!({ - "last_recording_mtime": "2026-04-15T10:00:00Z", - "recordings": [ - { "recorded_at": "2026-04-15T10:00:00Z", "text": "Patient mit Knie." } - ] - }); - std::fs::write( - case_dir.join(ANALYSIS_INPUT_FILE), - serde_json::to_vec_pretty(&input).unwrap(), - ) - .unwrap(); + let input = analyze::AnalysisInput { + last_recording_mtime: "2026-04-15T10:00:00Z".to_owned(), + recordings: vec![analyze::RecordingInput { + recorded_at: "2026-04-15T10:00:00Z".to_owned(), + text: "Patient mit Knie.".to_owned(), + }], + backend_id: String::new(), + }; let mock = MockServer::start().await; Mock::given(method("POST")) @@ -410,16 +423,19 @@ async fn analyze_worker_writes_document_via_wiremock() { let client = reqwest::Client::new(); let busy = Arc::new(std::sync::atomic::AtomicBool::new(false)); let vocab = Arc::new(doctate_server::gazetteer::Gazetteer::empty()); + let in_flight = analyze::InFlight::new(); let handle = tokio::spawn(analyze::worker::run( rx, client, busy, + in_flight, vocab, doctate_server::events::channel(), )); tx.send(analyze::AnalyzeJob { case_dir: case_dir.clone(), + input, }) .unwrap(); @@ -455,17 +471,14 @@ async fn analyze_worker_normalizes_llm_output() { let case_dir = tmp.join("dr_a/22222222-2222-2222-2222-222222222222"); std::fs::create_dir_all(&case_dir).unwrap(); - let input = json!({ - "last_recording_mtime": "2026-04-15T10:00:00Z", - "recordings": [ - { "recorded_at": "2026-04-15T10:00:00Z", "text": "Patient mit Blutung im Zerebrum." } - ] - }); - std::fs::write( - case_dir.join(ANALYSIS_INPUT_FILE), - serde_json::to_vec_pretty(&input).unwrap(), - ) - .unwrap(); + let input = analyze::AnalysisInput { + last_recording_mtime: "2026-04-15T10:00:00Z".to_owned(), + recordings: vec![analyze::RecordingInput { + recorded_at: "2026-04-15T10:00:00Z".to_owned(), + text: "Patient mit Blutung im Zerebrum.".to_owned(), + }], + backend_id: String::new(), + }; // Vocab directory with a single anatomy entry. let vocab_dir = unique_tmpdir("analyze-vocab"); @@ -501,16 +514,19 @@ async fn analyze_worker_normalizes_llm_output() { let (tx, rx) = analyze::channel(); let client = reqwest::Client::new(); let busy = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let in_flight = analyze::InFlight::new(); let handle = tokio::spawn(analyze::worker::run( rx, client, busy, + in_flight, vocab, doctate_server::events::channel(), )); tx.send(analyze::AnalyzeJob { case_dir: case_dir.clone(), + input, }) .unwrap(); @@ -613,23 +629,27 @@ async fn replace_real_case_dry() { // Recovery // --------------------------------------------------------------------- -/// Boot-scan is observation-only after the typed-state refactor: even a -/// case sitting in the `Queued` state on disk must NOT be re-enqueued -/// at startup. The per-page-render `try_enqueue_all_for_user` is the -/// only auto-trigger. +/// Boot-scan is observation-only after the typed-state refactor: it +/// must never enqueue jobs. With the in-flight set replacing on-disk +/// markers, an orphan `analysis_input.json` left over from a +/// pre-refactor server run is a no-op: the empty `InFlight` set means +/// the case is not classified as `Queued`. #[tokio::test] async fn recovery_does_not_enqueue_at_boot() { let tmp = unique_tmpdir("analyze-r1"); let case_dir = tmp.join("dr_a/11111111-1111-1111-1111-111111111111"); std::fs::create_dir_all(&case_dir).unwrap(); + // Pre-refactor leftover; ignored by the new model. std::fs::write(case_dir.join(ANALYSIS_INPUT_FILE), "{}").unwrap(); let (_tx, mut rx) = analyze::channel(); + let in_flight = analyze::InFlight::new(); analyze::recovery::boot_scan_state( &tmp, std::time::Duration::ZERO, std::time::SystemTime::now(), std::time::SystemTime::UNIX_EPOCH, + &in_flight, ) .await; @@ -644,7 +664,7 @@ async fn recovery_does_not_enqueue_at_boot() { // --------------------------------------------------------------------- #[tokio::test] -async fn analyze_deletes_old_document_and_writes_fresh_input() { +async fn analyze_deletes_old_document_and_sends_fresh_job() { let (cfg, settings) = cfg_llm("rn-5"); let case_id = "11111111-1111-1111-1111-111111111111"; let case_dir = seed_case(&cfg.data_path, "dr_a", case_id); @@ -652,7 +672,8 @@ async fn analyze_deletes_old_document_and_writes_fresh_input() { seed_recording(&case_dir, "2026-04-15T10-00-00Z", Some("erste Aufnahme")); seed_recording(&case_dir, "2026-04-15T10-05-00Z", Some("zweite Aufnahme")); - let (app, store) = doctate_server::create_router_and_session_store_with_settings(cfg, settings); + let (app, store, _in_flight, mut analyze_rx) = + doctate_server::create_router_with_full_handles(cfg, settings); let (cookie, csrf) = login_with_csrf(&app, &store, "dr_a", "s").await; let resp = app @@ -670,11 +691,8 @@ async fn analyze_deletes_old_document_and_writes_fresh_input() { !case_dir.join(DOCUMENT_FILE).exists(), "old document must be deleted before re-analysis" ); - let input_path = case_dir.join(ANALYSIS_INPUT_FILE); - assert!(input_path.exists(), "analysis_input.json missing"); - let parsed: Value = - serde_json::from_str(&std::fs::read_to_string(&input_path).unwrap()).unwrap(); - assert_eq!(parsed["recordings"].as_array().unwrap().len(), 2); + let job = analyze_rx.try_recv().expect("re-analysis must enqueue"); + assert_eq!(job.input.recordings.len(), 2); } // --------------------------------------------------------------------- @@ -1211,7 +1229,8 @@ async fn bulk_analyze_processes_each_selected_case() { seed_recording(&dir_a, "2026-04-15T10-00-00Z", Some("a")); seed_recording(&dir_b, "2026-04-15T10-05-00Z", Some("b")); - let (app, store) = doctate_server::create_router_and_session_store_with_settings(cfg, settings); + let (app, store, in_flight, mut analyze_rx) = + doctate_server::create_router_with_full_handles(cfg, settings); let (cookie, csrf) = login_with_csrf(&app, &store, "dr_a", "s").await; let body = format!( @@ -1224,8 +1243,15 @@ async fn bulk_analyze_processes_each_selected_case() { .unwrap(); assert_eq!(resp.status(), StatusCode::SEE_OTHER); - assert!(dir_a.join(ANALYSIS_INPUT_FILE).exists()); - assert!(dir_b.join(ANALYSIS_INPUT_FILE).exists()); + assert!(in_flight.contains(&dir_a)); + assert!(in_flight.contains(&dir_b)); + let mut received = Vec::new(); + while let Ok(job) = analyze_rx.try_recv() { + received.push(job.case_dir); + } + assert_eq!(received.len(), 2, "both selected cases must be enqueued"); + assert!(received.iter().any(|p| p == &dir_a)); + assert!(received.iter().any(|p| p == &dir_b)); } #[tokio::test] @@ -1233,18 +1259,20 @@ async fn recovery_skips_closed_cases() { let tmp = unique_tmpdir("analyze-rec-closed"); let case_dir = tmp.join("dr_a/11111111-1111-1111-1111-111111111111"); std::fs::create_dir_all(&case_dir).unwrap(); - std::fs::write(case_dir.join(ANALYSIS_INPUT_FILE), "{}").unwrap(); std::fs::write( case_dir.join(CLOSE_MARKER), r#"{"closed_at":"2026-04-15T10:00:00Z"}"#, ) .unwrap(); + let in_flight = analyze::InFlight::new(); + in_flight.try_claim(&case_dir); let summary = analyze::recovery::boot_scan_state( &tmp, std::time::Duration::ZERO, std::time::SystemTime::now(), std::time::SystemTime::UNIX_EPOCH, + &in_flight, ) .await; @@ -1276,7 +1304,15 @@ async fn reset_case_clears_derived_and_unfails_audio() { ) .unwrap(); write_document_for_test(&dir, "# Doc\n", "1970-01-01T00:00:00Z"); - std::fs::write(dir.join(ANALYSIS_INPUT_FILE), "{}").unwrap(); + // Failure marker from a prior aborted run — the bonus-fix says + // reset must wipe this too, otherwise the case stays classified + // as `Failed` forever after the user explicitly requested a fresh + // start. + std::fs::write( + dir.join(".analysis_failed.json"), + r#"{"last_recording_mtime":"2026-04-15T09:00:00Z","reason":"x","failed_at":"2026-04-15T09:01:00Z"}"#, + ) + .unwrap(); let (app, store) = doctate_server::create_router_and_session_store_with_settings(cfg, settings); let (cookie, csrf) = login_with_csrf(&app, &store, "dr_a", "s").await; @@ -1304,7 +1340,10 @@ async fn reset_case_clears_derived_and_unfails_audio() { assert!(!dir.join("2026-04-15T09-00-00Z.json").exists()); assert!(!dir.join(ONELINER_FILENAME).exists()); assert!(!dir.join(DOCUMENT_FILE).exists()); - assert!(!dir.join(ANALYSIS_INPUT_FILE).exists()); + assert!( + !dir.join(".analysis_failed.json").exists(), + "reset must clear the failure marker so the case is no longer Failed" + ); } #[tokio::test] @@ -1440,17 +1479,14 @@ async fn analyze_works_against_ollama_style_endpoint_without_api_key() { let case_dir = tmp.join("dr_a/22222222-2222-2222-2222-222222222222"); std::fs::create_dir_all(&case_dir).unwrap(); - let input = json!({ - "last_recording_mtime": "2026-04-16T10:00:00Z", - "recordings": [ - { "recorded_at": "2026-04-16T10:00:00Z", "text": "Test." } - ] - }); - std::fs::write( - case_dir.join(ANALYSIS_INPUT_FILE), - serde_json::to_vec_pretty(&input).unwrap(), - ) - .unwrap(); + let input = analyze::AnalysisInput { + last_recording_mtime: "2026-04-16T10:00:00Z".to_owned(), + recordings: vec![analyze::RecordingInput { + recorded_at: "2026-04-16T10:00:00Z".to_owned(), + text: "Test.".to_owned(), + }], + backend_id: String::new(), + }; let mock = MockServer::start().await; Mock::given(method("POST")) @@ -1483,16 +1519,19 @@ async fn analyze_works_against_ollama_style_endpoint_without_api_key() { let client = reqwest::Client::new(); let busy = Arc::new(std::sync::atomic::AtomicBool::new(false)); let vocab = Arc::new(doctate_server::gazetteer::Gazetteer::empty()); + let in_flight = analyze::InFlight::new(); let handle = tokio::spawn(analyze::worker::run( rx, client, busy, + in_flight, vocab, doctate_server::events::channel(), )); tx.send(analyze::AnalyzeJob { case_dir: case_dir.clone(), + input, }) .unwrap(); @@ -1518,9 +1557,8 @@ async fn analyze_works_against_ollama_style_endpoint_without_api_key() { /// `POST /cases/analyze-required` walks every UUID-named case and force- /// enqueues those in `Required`/`Idle`. `Failed` and `Current` cases must -/// stay untouched. We probe the resulting state via the on-disk -/// `analysis_input.json` marker — written before the channel send, so its -/// presence is the canonical "would have been enqueued" signal. +/// stay untouched. The InFlight set is the canonical "would have been +/// enqueued" signal; the channel receiver lets us count delivered jobs. #[tokio::test] async fn analyze_required_endpoint_enqueues_only_required_cases() { use doctate_server::analyze::auto_trigger::FailureMarker; @@ -1570,7 +1608,8 @@ async fn analyze_required_endpoint_enqueues_only_required_cases() { ) .unwrap(); - let (app, store) = doctate_server::create_router_and_session_store_with_settings(cfg, settings); + let (app, store, in_flight, mut analyze_rx) = + doctate_server::create_router_with_full_handles(cfg, settings); let (cookie, csrf) = login_with_csrf(&app, &store, "dr_a", "s").await; let resp = app @@ -1584,31 +1623,42 @@ async fn analyze_required_endpoint_enqueues_only_required_cases() { "must redirect back to the case list" ); - // Required → analysis_input.json was written, the case is now Queued. + let req_path = user_root.join(req_id); + let cur_path = user_root.join(cur_id); + let fail_path = user_root.join(fail_id); + + // Required → claim held + job in channel. assert!( - user_root.join(req_id).join(ANALYSIS_INPUT_FILE).exists(), - "Required case must have been enqueued" + in_flight.contains(&req_path), + "Required case must have a claim" ); - // Current → no enqueue, document.json untouched. + // Current → no claim, document.json untouched. assert!( - !user_root.join(cur_id).join(ANALYSIS_INPUT_FILE).exists(), - "Current case must not be enqueued" + !in_flight.contains(&cur_path), + "Current case must not be claimed" ); assert!( - user_root.join(cur_id).join(DOCUMENT_FILE).exists(), + cur_path.join(DOCUMENT_FILE).exists(), "Current case's document must remain in place" ); - // Failed → no enqueue. Marker stays so the user sees the failure + // Failed → no claim. Marker stays so the user sees the failure // banner on the case page and can retry per-backend manually. assert!( - !user_root.join(fail_id).join(ANALYSIS_INPUT_FILE).exists(), - "Failed case must not be enqueued" + !in_flight.contains(&fail_path), + "Failed case must not be claimed" ); assert!( - user_root - .join(fail_id) - .join(".analysis_failed.json") - .exists(), + fail_path.join(".analysis_failed.json").exists(), "failure marker must persist" ); + + let mut received = Vec::new(); + while let Ok(job) = analyze_rx.try_recv() { + received.push(job.case_dir); + } + assert_eq!( + received, + vec![req_path], + "only the Required case must be enqueued" + ); } diff --git a/server/tests/common/artefacts.rs b/server/tests/common/artefacts.rs index aa9260c..1a9a92b 100644 --- a/server/tests/common/artefacts.rs +++ b/server/tests/common/artefacts.rs @@ -11,9 +11,16 @@ use std::path::Path; pub use doctate_common::ONELINER_FILENAME; -pub use doctate_server::analyze::{ANALYSIS_INPUT_FILE, DOCUMENT_FILE, Document}; +pub use doctate_server::analyze::{DOCUMENT_FILE, Document}; pub use doctate_server::paths::CLOSE_MARKER; +/// Legacy filename of the per-case analysis input. The server no +/// longer writes or reads it — the in-flight marker lives in `InFlight` +/// and the payload travels with the `AnalyzeJob`. This constant is +/// retained only for tests that simulate an orphan file from a +/// pre-refactor server. New tests should not seed this file. +pub const ANALYSIS_INPUT_FILE: &str = "analysis_input.json"; + /// Seed a `document.json` for tests. `covered_mtime` is the RFC3339 /// snapshot mtime the document is supposed to "cover" — for tests that /// don't care about the auto-trigger state machine, pass any RFC3339 diff --git a/server/tests/delete_recording_test.rs b/server/tests/delete_recording_test.rs index 51d7b20..c259116 100644 --- a/server/tests/delete_recording_test.rs +++ b/server/tests/delete_recording_test.rs @@ -14,9 +14,9 @@ use doctate_common::oneliners::OnelinerState; use tower::util::ServiceExt; use common::{ - ANALYSIS_INPUT_FILE, DOCUMENT_FILE, ONELINER_FILENAME, TestConfig, csrf_form_post, - login_with_csrf, paths, seed_case, seed_oneliner_ready, seed_oneliner_state, - seed_recording_with_sidecars, test_user, write_document_for_test, + DOCUMENT_FILE, ONELINER_FILENAME, TestConfig, csrf_form_post, login_with_csrf, paths, + seed_case, seed_oneliner_ready, seed_oneliner_state, seed_recording_with_sidecars, test_user, + write_document_for_test, }; fn delete_body(filename: &str) -> String { @@ -38,7 +38,6 @@ async fn delete_recording_removes_file_and_sidecars_and_invalidates_derived() { // delete — keeps the original contract „auto state must be wiped". seed_oneliner_ready(&case_dir, "knee, left, pain"); write_document_for_test(&case_dir, "# stale", "1970-01-01T00:00:00Z"); - std::fs::write(case_dir.join(ANALYSIS_INPUT_FILE), b"{}").unwrap(); let (app, store) = doctate_server::create_router_and_session_store(cfg); let (cookie, csrf) = login_with_csrf(&app, &store, "dr_a", "s").await; @@ -91,10 +90,6 @@ async fn delete_recording_removes_file_and_sidecars_and_invalidates_derived() { !case_dir.join(DOCUMENT_FILE).exists(), "document.md must be cleared" ); - assert!( - !case_dir.join(ANALYSIS_INPUT_FILE).exists(), - "analysis_input.json must be cleared" - ); } #[tokio::test] diff --git a/server/tests/magic_link_test.rs b/server/tests/magic_link_test.rs index d90b77d..c8b5d0a 100644 --- a/server/tests/magic_link_test.rs +++ b/server/tests/magic_link_test.rs @@ -72,6 +72,7 @@ fn build_state_with_stores() -> AppState { http_client: reqwest::Client::new(), vocab: Arc::new(doctate_server::gazetteer::Gazetteer::empty()), boot_time: doctate_server::BootTime(std::time::SystemTime::UNIX_EPOCH), + analyze_in_flight: doctate_server::analyze::InFlight::new(), } } diff --git a/server/tests/oneliner_override_test.rs b/server/tests/oneliner_override_test.rs index 1d24a4d..cf9f260 100644 --- a/server/tests/oneliner_override_test.rs +++ b/server/tests/oneliner_override_test.rs @@ -85,6 +85,7 @@ fn build_app_with_events_tx( http_client: reqwest::Client::new(), vocab: Arc::new(doctate_server::gazetteer::Gazetteer::empty()), boot_time: doctate_server::BootTime(std::time::SystemTime::UNIX_EPOCH), + analyze_in_flight: doctate_server::analyze::InFlight::new(), } }