diff --git a/server/src/analyze/recovery.rs b/server/src/analyze/recovery.rs index d0b262b..7ba2166 100644 --- a/server/src/analyze/recovery.rs +++ b/server/src/analyze/recovery.rs @@ -2,7 +2,8 @@ use std::path::Path; use tracing::{info, warn}; -use super::{ANALYSIS_INPUT_FILE, AnalyzeJob, AnalyzeSender, DOCUMENT_FILE}; +use super::auto_trigger::read_failure_marker; +use super::{ANALYSIS_INPUT_FILE, AnalysisInput, AnalyzeJob, AnalyzeSender, DOCUMENT_FILE}; /// Walk `data_path/*/` and enqueue every pending analysis for every user. /// Intended to run once at startup; the same primitive backs the per-user @@ -36,6 +37,15 @@ pub async fn enqueue_pending_for_user(user_root: &Path, tx: &AnalyzeSender) -> u if case_dir.join(DOCUMENT_FILE).exists() { continue; } + // A failure marker matching the input's `last_recording_mtime` + // means the worker has already tried this exact input and failed + // (e.g. LLM timeout). Re-enqueueing here would block the worker + // for another full timeout window and starve fresh user clicks. + // `auto_trigger::evaluate_case` performs the same skip; recovery + // must mirror it. + if input_already_failed(&case_dir).await { + continue; + } if tx .send(AnalyzeJob { case_dir: case_dir.clone(), @@ -49,3 +59,157 @@ pub async fn enqueue_pending_for_user(user_root: &Path, tx: &AnalyzeSender) -> u } sent } + +/// Returns `true` iff a failure marker exists whose +/// `last_recording_mtime` equals the one in `analysis_input.json`. Any +/// I/O or parse error is treated as "no usable marker" — recovery then +/// proceeds with the enqueue, matching the optimistic intent of the +/// outer scan. +async fn input_already_failed(case_dir: &Path) -> bool { + let Some(marker) = read_failure_marker(case_dir).await else { + return false; + }; + let Ok(bytes) = tokio::fs::read(case_dir.join(ANALYSIS_INPUT_FILE)).await else { + return false; + }; + let Ok(input) = serde_json::from_slice::(&bytes) else { + return false; + }; + marker.last_recording_mtime == input.last_recording_mtime +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::analyze::auto_trigger::{FAILURE_MARKER_FILE, FailureMarker}; + use crate::analyze::{RecordingInput, channel as analyze_channel}; + use tempfile::TempDir; + use tokio::fs; + + fn make_input(mtime: &str) -> AnalysisInput { + AnalysisInput { + last_recording_mtime: mtime.to_owned(), + recordings: vec![RecordingInput { + recorded_at: mtime.to_owned(), + text: "ignored".into(), + }], + backend_id: String::new(), + } + } + + async fn write_input(case_dir: &Path, input: &AnalysisInput) { + fs::write( + case_dir.join(ANALYSIS_INPUT_FILE), + serde_json::to_vec(input).unwrap(), + ) + .await + .unwrap(); + } + + async fn write_marker(case_dir: &Path, mtime: &str) { + let marker = FailureMarker { + last_recording_mtime: mtime.to_owned(), + reason: "test".into(), + failed_at: "2026-05-03T00:00:00Z".into(), + backend_id: None, + }; + fs::write( + case_dir.join(FAILURE_MARKER_FILE), + serde_json::to_vec(&marker).unwrap(), + ) + .await + .unwrap(); + } + + /// Drain the receiver after sender drop — same pattern as + /// `auto_trigger::tests::try_enqueue_all_sends_only_eligible_cases`. + async fn drain(mut rx: crate::analyze::AnalyzeReceiver) -> Vec { + let mut out = Vec::new(); + while let Some(job) = rx.recv().await { + out.push(job.case_dir); + } + out + } + + #[tokio::test] + async fn enqueues_pending_input_without_doc_or_marker() { + let user_root = TempDir::new().unwrap(); + let case_id = "11111111-1111-1111-1111-111111111111"; + let case = user_root.path().join(case_id); + fs::create_dir_all(&case).await.unwrap(); + write_input(&case, &make_input("2026-05-03T10:00:00Z")).await; + + let (tx, rx) = analyze_channel(); + let sent = enqueue_pending_for_user(user_root.path(), &tx).await; + drop(tx); + + assert_eq!(sent, 1); + let received = drain(rx).await; + assert_eq!(received.len(), 1); + assert!(received[0].ends_with(case_id)); + } + + #[tokio::test] + async fn skips_when_document_already_exists() { + let user_root = TempDir::new().unwrap(); + let case = user_root + .path() + .join("22222222-2222-2222-2222-222222222222"); + fs::create_dir_all(&case).await.unwrap(); + write_input(&case, &make_input("2026-05-03T10:00:00Z")).await; + fs::write(case.join(DOCUMENT_FILE), b"already analysed") + .await + .unwrap(); + + let (tx, rx) = analyze_channel(); + let sent = enqueue_pending_for_user(user_root.path(), &tx).await; + drop(tx); + + assert_eq!(sent, 0); + assert!(drain(rx).await.is_empty()); + } + + /// Regression: a stuck Llama job times out, leaves `analysis_input.json` + /// behind, and writes a failure marker. Without this skip the next + /// page reload re-enqueues the same job and blocks the worker for + /// another full timeout — see Bug 1 (case c414cf52, 2026-05-03). + #[tokio::test] + async fn skips_when_failure_marker_matches_input_mtime() { + let user_root = TempDir::new().unwrap(); + let case = user_root + .path() + .join("33333333-3333-3333-3333-333333333333"); + fs::create_dir_all(&case).await.unwrap(); + let mtime = "2026-05-03T10:00:00Z"; + write_input(&case, &make_input(mtime)).await; + write_marker(&case, mtime).await; + + let (tx, rx) = analyze_channel(); + let sent = enqueue_pending_for_user(user_root.path(), &tx).await; + drop(tx); + + assert_eq!(sent, 0, "must not re-enqueue an already-failed input"); + assert!(drain(rx).await.is_empty()); + } + + /// A new recording bumps `last_recording_mtime`, so the old marker + /// no longer applies and recovery should re-enqueue. + #[tokio::test] + async fn enqueues_when_failure_marker_is_stale() { + let user_root = TempDir::new().unwrap(); + let case_id = "44444444-4444-4444-4444-444444444444"; + let case = user_root.path().join(case_id); + fs::create_dir_all(&case).await.unwrap(); + write_input(&case, &make_input("2026-05-03T11:00:00Z")).await; + write_marker(&case, "2026-05-03T10:00:00Z").await; + + let (tx, rx) = analyze_channel(); + let sent = enqueue_pending_for_user(user_root.path(), &tx).await; + drop(tx); + + assert_eq!(sent, 1); + let received = drain(rx).await; + assert_eq!(received.len(), 1); + assert!(received[0].ends_with(case_id)); + } +}