Refactor auto-trigger and recovery logic
This commit introduces a significant refactor to the auto-trigger and recovery mechanisms for the analysis pipeline. The core change is the introduction of a more granular `CaseAnalysisState` enum, replacing the simpler `AutoDecision`. The `evaluate_state` function now computes the full lifecycle state of a case, considering factors like the presence of recordings, pending transcripts, queued jobs, failure markers, document staleness, and an idle threshold. This provides a richer understanding of the case's status. The `evaluate_case` function is retained as a backwards-compatible adapter for existing callsites and tests. It maps the new `CaseAnalysisState` to the old `AutoDecision` enum, simplifying the decision to "enqueue" or "skip." The `recovery` module is also refactored. The `scan_and_enqueue` function is replaced by `boot_scan_state`, which is now purely observational. It aggregates statistics about cases in different states at boot time and logs them, but it no longer attempts to re-enqueue jobs. The responsibility for triggering analysis now lies with the per-render `try_enqueue_all_for_user` logic, which utilizes the new typed state machine. This refactoring aims to provide more clarity and control over the analysis pipeline's state management, particularly in handling cases that have previously failed or are waiting for further input.
This commit is contained in:
@@ -112,56 +112,146 @@ impl FailureDetails {
|
||||
|
||||
/// Outcome of the pre-condition check. `Skip` carries a static string so
|
||||
/// the common "nothing to do" path allocates nothing.
|
||||
///
|
||||
/// Adapter over [`CaseAnalysisState`]: the auto-trigger reduces the rich
|
||||
/// state machine to a binary "fire or skip" decision. UI layers that need
|
||||
/// the full state should call [`evaluate_state`] directly.
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub enum AutoDecision {
|
||||
Enqueue,
|
||||
Skip(&'static str),
|
||||
}
|
||||
|
||||
/// Pure pre-condition check: decide whether this case is due for an
|
||||
/// automatic analysis. Does file I/O but no channel sends and no
|
||||
/// state mutations — safe to call redundantly.
|
||||
pub async fn evaluate_case(case_dir: &Path) -> AutoDecision {
|
||||
/// Per-case lifecycle state, computed from the artefacts on disk plus the
|
||||
/// idle threshold. The variants below are precedence-ordered: the first
|
||||
/// matching condition wins, so a `Failed` case with a stale failure
|
||||
/// marker is reported as `Required`, not `Failed`.
|
||||
///
|
||||
/// `has_document: bool` on `Required`/`Idle` lets the UI decide whether
|
||||
/// to render the previously-generated document (= „letztes Dokument")
|
||||
/// alongside the „Analyse erforderlich"-banner.
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub enum CaseAnalysisState {
|
||||
/// No `.m4a` files in the directory at all (e.g. fresh case before
|
||||
/// the first upload completes).
|
||||
NoRecordings,
|
||||
/// At least one recording is still missing its transcript sidecar.
|
||||
Pending,
|
||||
/// `analysis_input.json` exists — a job is queued or in flight.
|
||||
Queued,
|
||||
/// Failure marker matches the current `latest_mtime`. Stays `Failed`
|
||||
/// until either the input changes (→ `Required`) or the user wipes
|
||||
/// the marker via the manual retry path.
|
||||
Failed,
|
||||
/// Document does not yet cover the newest recording, but the
|
||||
/// idle-threshold has not elapsed yet — the user may still be
|
||||
/// dictating. Auto-trigger waits.
|
||||
Required { has_document: bool },
|
||||
/// Like `Required`, plus the idle window has elapsed. The auto-
|
||||
/// trigger fires.
|
||||
Idle { has_document: bool },
|
||||
/// Document on disk covers exactly the current recording set
|
||||
/// (`covered_mtime == latest_mtime`).
|
||||
Current,
|
||||
}
|
||||
|
||||
/// Compute the full per-case lifecycle state. Pure file-I/O, no
|
||||
/// mutations or channel sends — safe to call redundantly per render.
|
||||
///
|
||||
/// `idle_threshold` is the time without new recordings after which a
|
||||
/// stale-document case is auto-promoted from `Required` to `Idle`. Pass
|
||||
/// `Duration::ZERO` to disable the wait (every Required becomes Idle
|
||||
/// immediately — the historical eager behaviour, used by the
|
||||
/// [`evaluate_case`] adapter and most existing tests).
|
||||
///
|
||||
/// `now` is injected so tests can pin filesystem mtimes against a known
|
||||
/// reference point. Production callers pass `SystemTime::now()`.
|
||||
pub async fn evaluate_state(
|
||||
case_dir: &Path,
|
||||
idle_threshold: std::time::Duration,
|
||||
now: SystemTime,
|
||||
) -> CaseAnalysisState {
|
||||
let scan = scan_m4as(case_dir).await;
|
||||
|
||||
if !scan.saw_any {
|
||||
return AutoDecision::Skip("no recordings");
|
||||
return CaseAnalysisState::NoRecordings;
|
||||
}
|
||||
if !scan.all_transcribed {
|
||||
return AutoDecision::Skip("not all transcribed");
|
||||
return CaseAnalysisState::Pending;
|
||||
}
|
||||
let Some(latest_mtime) = scan.latest_mtime else {
|
||||
return AutoDecision::Skip("no recordings");
|
||||
return CaseAnalysisState::NoRecordings;
|
||||
};
|
||||
|
||||
// Any existing analysis_input.json means a job is queued or running.
|
||||
// Worker removes the file on success — next reload re-evaluates.
|
||||
if tokio::fs::try_exists(case_dir.join(ANALYSIS_INPUT_FILE))
|
||||
.await
|
||||
.unwrap_or(false)
|
||||
{
|
||||
return AutoDecision::Skip("job queued");
|
||||
return CaseAnalysisState::Queued;
|
||||
}
|
||||
|
||||
// A failure marker matching the current input signature means we
|
||||
// already tried and the input has not changed since.
|
||||
let latest_rfc = rfc3339_of(latest_mtime);
|
||||
|
||||
// Failure marker matching the current input signature → previous
|
||||
// attempt for *this* recording set failed, do not retry until the
|
||||
// input changes (or the user clears the marker manually).
|
||||
if let Some(marker) = read_failure_marker(case_dir).await
|
||||
&& marker.last_recording_mtime == rfc3339_of(latest_mtime)
|
||||
&& marker.last_recording_mtime == latest_rfc
|
||||
{
|
||||
return AutoDecision::Skip("previously failed");
|
||||
return CaseAnalysisState::Failed;
|
||||
}
|
||||
|
||||
// Fresh document older than the newest recording → stale, re-analyse.
|
||||
// Fresh document newer or equal → nothing to do.
|
||||
let doc_path = case_dir.join(DOCUMENT_FILE);
|
||||
if let Ok(meta) = tokio::fs::metadata(&doc_path).await
|
||||
&& let Ok(doc_mtime) = meta.modified()
|
||||
&& doc_mtime >= latest_mtime
|
||||
let doc_meta = read_document_meta(case_dir).await;
|
||||
let has_document = doc_meta.is_some();
|
||||
|
||||
if let Some(doc) = doc_meta
|
||||
&& doc.covered_mtime == latest_rfc
|
||||
{
|
||||
return AutoDecision::Skip("analysis current");
|
||||
return CaseAnalysisState::Current;
|
||||
}
|
||||
|
||||
AutoDecision::Enqueue
|
||||
// Stale document (or no document at all): differentiate between
|
||||
// "user might still be dictating" (Required) and "long enough quiet,
|
||||
// fire" (Idle).
|
||||
let elapsed = now
|
||||
.duration_since(latest_mtime)
|
||||
.unwrap_or(std::time::Duration::ZERO);
|
||||
if elapsed >= idle_threshold {
|
||||
CaseAnalysisState::Idle { has_document }
|
||||
} else {
|
||||
CaseAnalysisState::Required { has_document }
|
||||
}
|
||||
}
|
||||
|
||||
/// Read and parse `document.json` if present. Returns `None` for
|
||||
/// missing files or parse failures (a corrupt envelope makes the case
|
||||
/// appear as „has no document", which downgrades it to `Required` and
|
||||
/// triggers a fresh analysis on the next idle window).
|
||||
pub(crate) async fn read_document_meta(case_dir: &Path) -> Option<crate::analyze::Document> {
|
||||
let bytes = tokio::fs::read(case_dir.join(DOCUMENT_FILE)).await.ok()?;
|
||||
serde_json::from_slice(&bytes).ok()
|
||||
}
|
||||
|
||||
/// Backwards-compatible binary decision: kept for the existing
|
||||
/// `try_enqueue` callsites and unit tests that predate the typed state
|
||||
/// machine. New callers should consume [`evaluate_state`] directly.
|
||||
///
|
||||
/// Reduction: `Idle` → `Enqueue`, everything else → `Skip(<tag>)`.
|
||||
/// `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.
|
||||
pub async fn evaluate_case(case_dir: &Path) -> AutoDecision {
|
||||
let state = evaluate_state(case_dir, std::time::Duration::ZERO, SystemTime::now()).await;
|
||||
match state {
|
||||
CaseAnalysisState::NoRecordings => AutoDecision::Skip("no recordings"),
|
||||
CaseAnalysisState::Pending => AutoDecision::Skip("not all transcribed"),
|
||||
CaseAnalysisState::Queued => AutoDecision::Skip("job queued"),
|
||||
CaseAnalysisState::Failed => AutoDecision::Skip("previously failed"),
|
||||
CaseAnalysisState::Current => AutoDecision::Skip("analysis current"),
|
||||
CaseAnalysisState::Idle { .. } => AutoDecision::Enqueue,
|
||||
CaseAnalysisState::Required { .. } => AutoDecision::Skip("waiting for idle"),
|
||||
}
|
||||
}
|
||||
|
||||
/// Evaluate and, on `Enqueue`, prepare and send the job. Returns `true`
|
||||
@@ -187,15 +277,10 @@ pub async fn try_enqueue(case_dir: &Path, tx: &AnalyzeSender, events_tx: &EventS
|
||||
}
|
||||
};
|
||||
|
||||
// Re-analyse path: delete stale document.md up front so the worker's
|
||||
// "document exists → skip" guard does not short-circuit, and the UI
|
||||
// stops showing the outdated text during the render race.
|
||||
let doc_path = case_dir.join(DOCUMENT_FILE);
|
||||
if let Err(e) = remove_if_exists(&doc_path).await {
|
||||
warn!(case = %case_dir.display(), error = %e, "auto-analyze: remove old doc failed");
|
||||
return false;
|
||||
}
|
||||
|
||||
// The previous document.json stays in place: while the worker runs,
|
||||
// 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");
|
||||
@@ -394,14 +479,6 @@ pub(crate) async fn read_failure_marker(case_dir: &Path) -> Option<FailureMarker
|
||||
serde_json::from_slice(&bytes).ok()
|
||||
}
|
||||
|
||||
async fn remove_if_exists(path: &Path) -> std::io::Result<()> {
|
||||
match tokio::fs::remove_file(path).await {
|
||||
Ok(_) => Ok(()),
|
||||
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
|
||||
Err(e) => Err(e),
|
||||
}
|
||||
}
|
||||
|
||||
/// Overwrite-capable sibling of `case_actions::write_input_create_new`.
|
||||
/// Auto-trigger re-analyses may legitimately replace an existing input
|
||||
/// (e.g. a new recording arrived between two reloads while the previous
|
||||
@@ -420,13 +497,14 @@ async fn write_input_overwrite(path: &Path, input: &AnalysisInput) -> std::io::R
|
||||
/// 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.
|
||||
fn rfc3339_of(t: SystemTime) -> String {
|
||||
pub(crate) fn rfc3339_of(t: SystemTime) -> String {
|
||||
OffsetDateTime::from(t).format(&Rfc3339).unwrap_or_default()
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::analyze::Document;
|
||||
use filetime::{FileTime, set_file_mtime};
|
||||
use std::time::Duration;
|
||||
use tempfile::TempDir;
|
||||
@@ -436,6 +514,25 @@ mod tests {
|
||||
fs::write(path, b"").await.unwrap();
|
||||
}
|
||||
|
||||
/// Seed a `document.json` whose `covered_mtime` matches the supplied
|
||||
/// `SystemTime`. Use `set_mtime` on `.m4a` files separately to drive
|
||||
/// the latest-mtime → covered-mtime comparison in [`evaluate_state`].
|
||||
async fn write_document_covering(case_dir: &std::path::Path, covered_mtime: SystemTime) {
|
||||
let covered_rfc = rfc3339_of(covered_mtime);
|
||||
let doc = Document {
|
||||
covered_mtime: covered_rfc.clone(),
|
||||
written_at: covered_rfc,
|
||||
backend_id: "test".into(),
|
||||
content_md: "doc".into(),
|
||||
};
|
||||
fs::write(
|
||||
case_dir.join(DOCUMENT_FILE),
|
||||
serde_json::to_vec(&doc).unwrap(),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
/// Seed a `<stem>.json` next to a (presumed already-touched)
|
||||
/// `<stem>.m4a`. Variant is `Transcript::Silent` — these tests
|
||||
/// only care that the recording is in *some* terminal state
|
||||
@@ -495,17 +592,15 @@ mod tests {
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn document_newer_than_recordings_skips() {
|
||||
async fn document_covered_equal_latest_returns_current() {
|
||||
let dir = TempDir::new().unwrap();
|
||||
let m4a = dir.path().join("2026-01-01T10-00-00Z.m4a");
|
||||
touch(&m4a).await;
|
||||
seed_meta(dir.path(), "2026-01-01T10-00-00Z").await;
|
||||
let doc = dir.path().join(DOCUMENT_FILE);
|
||||
touch(&doc).await;
|
||||
|
||||
let base = SystemTime::UNIX_EPOCH + Duration::from_secs(1_700_000_000);
|
||||
set_mtime(&m4a, base);
|
||||
set_mtime(&doc, base + Duration::from_secs(60));
|
||||
write_document_covering(dir.path(), base).await;
|
||||
|
||||
assert_eq!(
|
||||
evaluate_case(dir.path()).await,
|
||||
@@ -514,32 +609,28 @@ mod tests {
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn document_older_than_recordings_enqueues() {
|
||||
async fn document_covered_older_than_recordings_enqueues() {
|
||||
let dir = TempDir::new().unwrap();
|
||||
let m4a = dir.path().join("2026-01-01T10-00-00Z.m4a");
|
||||
touch(&m4a).await;
|
||||
seed_meta(dir.path(), "2026-01-01T10-00-00Z").await;
|
||||
let doc = dir.path().join(DOCUMENT_FILE);
|
||||
touch(&doc).await;
|
||||
|
||||
let base = SystemTime::UNIX_EPOCH + Duration::from_secs(1_700_000_000);
|
||||
set_mtime(&doc, base);
|
||||
set_mtime(&m4a, base + Duration::from_secs(60));
|
||||
// Document covers an earlier snapshot — a recording arrived since.
|
||||
write_document_covering(dir.path(), base).await;
|
||||
|
||||
assert_eq!(evaluate_case(dir.path()).await, AutoDecision::Enqueue);
|
||||
}
|
||||
|
||||
/// Burst race: when a recording arrives DURING the LLM call for an
|
||||
/// earlier recording, the worker writes a `document.md` whose mtime
|
||||
/// is later than the new recording's mtime — even though the
|
||||
/// document content cannot include that recording (its
|
||||
/// `analysis_input.json` snapshot was frozen before the recording
|
||||
/// existed). The current `doc_mtime >= latest_mtime` check would
|
||||
/// then incorrectly report "analysis current" and the new recording
|
||||
/// would never make it into a document.
|
||||
///
|
||||
/// This test pins the desired behaviour: `evaluate_case` must return
|
||||
/// `Enqueue` so the late recording is folded into a fresh document.
|
||||
/// earlier recording, the worker writes a `document.json` whose
|
||||
/// `covered_mtime` snapshot only includes the recordings that were
|
||||
/// on disk when the analyze job started. A later-arriving recording
|
||||
/// (B) is reflected by a strictly larger `latest_mtime` on disk than
|
||||
/// the document's `covered_mtime`, so [`evaluate_state`] correctly
|
||||
/// returns `Required` (and `evaluate_case` enqueues, given a zero
|
||||
/// idle threshold).
|
||||
#[tokio::test]
|
||||
async fn recording_arrived_during_llm_call_re_enqueues() {
|
||||
let dir = TempDir::new().unwrap();
|
||||
@@ -558,13 +649,11 @@ mod tests {
|
||||
seed_meta_with(dir.path(), "2026-01-01T10-00-05Z", "bravo").await;
|
||||
set_mtime(&b_m4a, base + Duration::from_secs(5));
|
||||
|
||||
// document.md: written when the LLM responded for A's snapshot.
|
||||
// Its mtime is newer than both recordings — but its content only
|
||||
// covers A. analysis_input.json has already been removed by the
|
||||
// worker on success, and there is no failure marker.
|
||||
let doc = dir.path().join(DOCUMENT_FILE);
|
||||
touch(&doc).await;
|
||||
set_mtime(&doc, base + Duration::from_secs(30));
|
||||
// document.json: covers only A's snapshot. Wallclock-mtime of
|
||||
// the file itself would be misleading (later than B), but the
|
||||
// typed `covered_mtime` field pins the truth — the contained
|
||||
// markdown describes A only.
|
||||
write_document_covering(dir.path(), base).await;
|
||||
|
||||
assert_eq!(
|
||||
evaluate_case(dir.path()).await,
|
||||
@@ -675,18 +764,16 @@ mod tests {
|
||||
)
|
||||
.await;
|
||||
|
||||
// Not eligible: has a fresh document.
|
||||
// Not eligible: document already covers the latest recording.
|
||||
let current_id = "22222222-2222-2222-2222-222222222222";
|
||||
let current = user_root.path().join(current_id);
|
||||
fs::create_dir_all(¤t).await.unwrap();
|
||||
let m4a = current.join("2026-01-01T10-00-00Z.m4a");
|
||||
fs::write(&m4a, b"fake").await.unwrap();
|
||||
seed_meta_with(¤t, "2026-01-01T10-00-00Z", "stable").await;
|
||||
let doc = current.join(DOCUMENT_FILE);
|
||||
fs::write(&doc, b"doc").await.unwrap();
|
||||
let base = SystemTime::UNIX_EPOCH + Duration::from_secs(1_700_000_000);
|
||||
set_mtime(&m4a, base);
|
||||
set_mtime(&doc, base + Duration::from_secs(60));
|
||||
write_document_covering(¤t, base).await;
|
||||
|
||||
// Ignored: non-UUID directory name.
|
||||
let junk = user_root.path().join("not-a-uuid");
|
||||
|
||||
+201
-142
@@ -1,97 +1,159 @@
|
||||
//! Boot-time state observation for the analyze pipeline.
|
||||
//!
|
||||
//! Pre-refactor this module re-enqueued every case that looked pending
|
||||
//! at startup. With the typed [`CaseAnalysisState`] machine the per-page
|
||||
//! `try_enqueue_all_for_user` covers that responsibility lazily — boot
|
||||
//! is now reduced to a telemetry pass that logs how many cases sit in
|
||||
//! each observable state, without any worker channel sends.
|
||||
//!
|
||||
//! [`CaseAnalysisState`]: super::auto_trigger::CaseAnalysisState
|
||||
|
||||
use std::path::Path;
|
||||
use std::time::{Duration, SystemTime};
|
||||
|
||||
use tracing::{info, warn};
|
||||
use tracing::info;
|
||||
|
||||
use super::auto_trigger::read_failure_marker;
|
||||
use super::{ANALYSIS_INPUT_FILE, AnalysisInput, AnalyzeJob, AnalyzeSender, DOCUMENT_FILE};
|
||||
use super::auto_trigger::{CaseAnalysisState, evaluate_state};
|
||||
|
||||
/// 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) {
|
||||
/// Aggregate counts over every UUID-named, non-closed case under a user
|
||||
/// root. `NoRecordings` and `Pending` are deliberately not represented:
|
||||
/// they are transient pre-states that need no operator visibility.
|
||||
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
|
||||
pub struct BootStateSummary {
|
||||
/// `analysis_input.json` present → a job is queued or in flight.
|
||||
pub queued: usize,
|
||||
/// Failure marker matches `latest_mtime` → manual retry needed.
|
||||
pub failed: usize,
|
||||
/// Idle threshold elapsed → next page render will auto-enqueue.
|
||||
pub idle: usize,
|
||||
/// Document not covering latest, idle threshold not yet reached.
|
||||
pub required: usize,
|
||||
/// Document covers `latest_mtime` → nothing to do.
|
||||
pub current: usize,
|
||||
}
|
||||
|
||||
impl BootStateSummary {
|
||||
fn add(&mut self, other: BootStateSummary) {
|
||||
self.queued += other.queued;
|
||||
self.failed += other.failed;
|
||||
self.idle += other.idle;
|
||||
self.required += other.required;
|
||||
self.current += other.current;
|
||||
}
|
||||
|
||||
fn record(&mut self, state: &CaseAnalysisState) {
|
||||
match state {
|
||||
CaseAnalysisState::Queued => self.queued += 1,
|
||||
CaseAnalysisState::Failed => self.failed += 1,
|
||||
CaseAnalysisState::Idle { .. } => self.idle += 1,
|
||||
CaseAnalysisState::Required { .. } => self.required += 1,
|
||||
CaseAnalysisState::Current => self.current += 1,
|
||||
CaseAnalysisState::NoRecordings | CaseAnalysisState::Pending => {}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Walk `data_path/*/` and emit a single info-level summary line. The
|
||||
/// returned summary is mainly for tests; production callers ignore it.
|
||||
pub async fn boot_scan_state(
|
||||
data_path: &Path,
|
||||
idle_threshold: Duration,
|
||||
now: SystemTime,
|
||||
) -> BootStateSummary {
|
||||
let mut total = BootStateSummary::default();
|
||||
let Ok(mut users) = tokio::fs::read_dir(data_path).await else {
|
||||
return;
|
||||
return total;
|
||||
};
|
||||
let mut total = 0;
|
||||
while let Ok(Some(user_entry)) = users.next_entry().await {
|
||||
total += enqueue_pending_for_user(&user_entry.path(), tx).await;
|
||||
let per_user = compute_state_for_user(&user_entry.path(), idle_threshold, now).await;
|
||||
total.add(per_user);
|
||||
}
|
||||
info!(count = total, "Analyze recovery scan finished");
|
||||
info!(
|
||||
queued = total.queued,
|
||||
failed = total.failed,
|
||||
idle = total.idle,
|
||||
required = total.required,
|
||||
current = total.current,
|
||||
"analyze boot state scan finished"
|
||||
);
|
||||
total
|
||||
}
|
||||
|
||||
/// Scan a user's case dirs and enqueue any case that has an
|
||||
/// `analysis_input.json` but no `document.md`. Returns the number sent.
|
||||
pub async fn enqueue_pending_for_user(user_root: &Path, tx: &AnalyzeSender) -> usize {
|
||||
/// Aggregate the per-case state for one user root. Skips closed cases
|
||||
/// and non-UUID directory names. Pure observation — no side effects.
|
||||
pub async fn compute_state_for_user(
|
||||
user_root: &Path,
|
||||
idle_threshold: Duration,
|
||||
now: SystemTime,
|
||||
) -> BootStateSummary {
|
||||
let mut summary = BootStateSummary::default();
|
||||
let Ok(mut cases) = tokio::fs::read_dir(user_root).await else {
|
||||
return 0;
|
||||
return summary;
|
||||
};
|
||||
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_closed(&case_dir).await {
|
||||
continue;
|
||||
}
|
||||
if !case_dir.join(ANALYSIS_INPUT_FILE).exists() {
|
||||
let Ok(name) = case_entry.file_name().into_string() else {
|
||||
continue;
|
||||
};
|
||||
if uuid::Uuid::parse_str(&name).is_err() {
|
||||
continue;
|
||||
}
|
||||
if case_dir.join(DOCUMENT_FILE).exists() {
|
||||
continue;
|
||||
let state = evaluate_state(&case_dir, idle_threshold, now).await;
|
||||
summary.record(&state);
|
||||
}
|
||||
// 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(),
|
||||
})
|
||||
.is_err()
|
||||
{
|
||||
warn!(case = %case_dir.display(), "recovery send failed (worker gone)");
|
||||
return sent;
|
||||
}
|
||||
sent += 1;
|
||||
}
|
||||
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::<AnalysisInput>(&bytes) else {
|
||||
return false;
|
||||
};
|
||||
marker.last_recording_mtime == input.last_recording_mtime
|
||||
summary
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::analyze::auto_trigger::{FAILURE_MARKER_FILE, FailureMarker};
|
||||
use crate::analyze::{Document, RecordingInput, channel as analyze_channel};
|
||||
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 filetime::{FileTime, set_file_mtime};
|
||||
use tempfile::TempDir;
|
||||
use tokio::fs;
|
||||
|
||||
async fn write_document(case_dir: &Path, content: &str) {
|
||||
fn case_id(n: u8) -> String {
|
||||
let h = format!("{n:08x}");
|
||||
format!("{h}-1111-1111-1111-111111111111")
|
||||
}
|
||||
|
||||
fn set_mtime(path: &Path, t: SystemTime) {
|
||||
set_file_mtime(path, FileTime::from_system_time(t)).unwrap();
|
||||
}
|
||||
|
||||
/// Seed a case with a single transcribed recording at `mtime`.
|
||||
/// Returns the case_dir for further fixture operations.
|
||||
async fn seed_transcribed_case(user_root: &Path, n: u8, mtime: SystemTime) -> std::path::PathBuf {
|
||||
let case = user_root.join(case_id(n));
|
||||
fs::create_dir_all(&case).await.unwrap();
|
||||
let stem = "2026-01-01T10-00-00Z";
|
||||
let m4a = case.join(format!("{stem}.m4a"));
|
||||
fs::write(&m4a, b"fake").await.unwrap();
|
||||
crate::paths::write_recording_meta_sync(
|
||||
&case,
|
||||
stem,
|
||||
doctate_common::Transcript::Content {
|
||||
text: "transcript".into(),
|
||||
},
|
||||
None,
|
||||
);
|
||||
set_mtime(&m4a, mtime);
|
||||
case
|
||||
}
|
||||
|
||||
async fn write_document_covering(case_dir: &Path, covered_mtime: SystemTime) {
|
||||
let rfc = auto_rfc3339(covered_mtime);
|
||||
let doc = Document {
|
||||
covered_mtime: "1970-01-01T00:00:00Z".into(),
|
||||
written_at: "1970-01-01T00:00:00Z".into(),
|
||||
covered_mtime: rfc.clone(),
|
||||
written_at: rfc,
|
||||
backend_id: "test".into(),
|
||||
content_md: content.into(),
|
||||
content_md: "doc".into(),
|
||||
};
|
||||
fs::write(
|
||||
case_dir.join(DOCUMENT_FILE),
|
||||
@@ -101,29 +163,23 @@ mod tests {
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
fn make_input(mtime: &str) -> AnalysisInput {
|
||||
AnalysisInput {
|
||||
last_recording_mtime: mtime.to_owned(),
|
||||
recordings: vec![RecordingInput {
|
||||
recorded_at: mtime.to_owned(),
|
||||
text: "ignored".into(),
|
||||
}],
|
||||
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(),
|
||||
}
|
||||
}
|
||||
|
||||
async fn write_input(case_dir: &Path, input: &AnalysisInput) {
|
||||
};
|
||||
fs::write(
|
||||
case_dir.join(ANALYSIS_INPUT_FILE),
|
||||
serde_json::to_vec(input).unwrap(),
|
||||
serde_json::to_vec(&input).unwrap(),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
async fn write_marker(case_dir: &Path, mtime: &str) {
|
||||
async fn write_marker(case_dir: &Path, mtime: SystemTime) {
|
||||
let marker = FailureMarker {
|
||||
last_recording_mtime: mtime.to_owned(),
|
||||
last_recording_mtime: auto_rfc3339(mtime),
|
||||
reason: "test".into(),
|
||||
failed_at: "2026-05-03T00:00:00Z".into(),
|
||||
backend_id: None,
|
||||
@@ -137,93 +193,96 @@ mod tests {
|
||||
.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<std::path::PathBuf> {
|
||||
let mut out = Vec::new();
|
||||
while let Some(job) = rx.recv().await {
|
||||
out.push(job.case_dir);
|
||||
}
|
||||
out
|
||||
fn base_time() -> SystemTime {
|
||||
SystemTime::UNIX_EPOCH + Duration::from_secs(1_700_000_000)
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn enqueues_pending_input_without_doc_or_marker() {
|
||||
async fn pending_input_counts_as_queued() {
|
||||
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 case = seed_transcribed_case(user_root.path(), 1, base_time()).await;
|
||||
write_input(&case, base_time()).await;
|
||||
|
||||
let (tx, rx) = analyze_channel();
|
||||
let sent = enqueue_pending_for_user(user_root.path(), &tx).await;
|
||||
drop(tx);
|
||||
let summary =
|
||||
compute_state_for_user(user_root.path(), Duration::ZERO, base_time()).await;
|
||||
|
||||
assert_eq!(sent, 1);
|
||||
let received = drain(rx).await;
|
||||
assert_eq!(received.len(), 1);
|
||||
assert!(received[0].ends_with(case_id));
|
||||
assert_eq!(summary.queued, 1);
|
||||
assert_eq!(summary.idle, 0);
|
||||
assert_eq!(summary.required, 0);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn skips_when_document_already_exists() {
|
||||
async fn document_covering_latest_counts_as_current() {
|
||||
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;
|
||||
write_document(&case, "already analysed").await;
|
||||
let case = seed_transcribed_case(user_root.path(), 2, base_time()).await;
|
||||
write_document_covering(&case, base_time()).await;
|
||||
|
||||
let (tx, rx) = analyze_channel();
|
||||
let sent = enqueue_pending_for_user(user_root.path(), &tx).await;
|
||||
drop(tx);
|
||||
let summary =
|
||||
compute_state_for_user(user_root.path(), Duration::ZERO, base_time()).await;
|
||||
|
||||
assert_eq!(sent, 0);
|
||||
assert!(drain(rx).await.is_empty());
|
||||
assert_eq!(summary.current, 1);
|
||||
assert_eq!(summary.queued, 0);
|
||||
}
|
||||
|
||||
/// 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).
|
||||
/// Failure marker matching the input → operator must intervene; not
|
||||
/// retried automatically.
|
||||
#[tokio::test]
|
||||
async fn skips_when_failure_marker_matches_input_mtime() {
|
||||
async fn failure_marker_matching_mtime_counts_as_failed() {
|
||||
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 case = seed_transcribed_case(user_root.path(), 3, base_time()).await;
|
||||
write_marker(&case, base_time()).await;
|
||||
|
||||
let (tx, rx) = analyze_channel();
|
||||
let sent = enqueue_pending_for_user(user_root.path(), &tx).await;
|
||||
drop(tx);
|
||||
let summary =
|
||||
compute_state_for_user(user_root.path(), Duration::ZERO, base_time()).await;
|
||||
|
||||
assert_eq!(sent, 0, "must not re-enqueue an already-failed input");
|
||||
assert!(drain(rx).await.is_empty());
|
||||
assert_eq!(summary.failed, 1);
|
||||
}
|
||||
|
||||
/// A new recording bumps `last_recording_mtime`, so the old marker
|
||||
/// no longer applies and recovery should re-enqueue.
|
||||
/// Stale failure marker (older than current input) → recoverable on
|
||||
/// next idle tick.
|
||||
#[tokio::test]
|
||||
async fn enqueues_when_failure_marker_is_stale() {
|
||||
async fn stale_failure_marker_falls_through_to_idle() {
|
||||
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 case = seed_transcribed_case(user_root.path(), 4, base_time()).await;
|
||||
write_marker(&case, base_time() - Duration::from_secs(3600)).await;
|
||||
|
||||
let (tx, rx) = analyze_channel();
|
||||
let sent = enqueue_pending_for_user(user_root.path(), &tx).await;
|
||||
drop(tx);
|
||||
let summary =
|
||||
compute_state_for_user(user_root.path(), Duration::ZERO, base_time()).await;
|
||||
|
||||
assert_eq!(sent, 1);
|
||||
let received = drain(rx).await;
|
||||
assert_eq!(received.len(), 1);
|
||||
assert!(received[0].ends_with(case_id));
|
||||
assert_eq!(summary.idle, 1);
|
||||
assert_eq!(summary.failed, 0);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn closed_cases_are_skipped() {
|
||||
let user_root = TempDir::new().unwrap();
|
||||
let case = seed_transcribed_case(user_root.path(), 5, base_time()).await;
|
||||
// Mark closed; the closed-cases scan must skip this entirely.
|
||||
fs::write(case.join(crate::paths::CLOSE_MARKER), "{}")
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let summary =
|
||||
compute_state_for_user(user_root.path(), Duration::ZERO, base_time()).await;
|
||||
|
||||
assert_eq!(summary, BootStateSummary::default());
|
||||
}
|
||||
|
||||
/// Idle threshold not yet reached → state is `Required`, not `Idle`.
|
||||
#[tokio::test]
|
||||
async fn idle_threshold_not_reached_counts_as_required() {
|
||||
let user_root = TempDir::new().unwrap();
|
||||
let _ = seed_transcribed_case(user_root.path(), 6, base_time()).await;
|
||||
|
||||
// `now` is only 100s after the recording — below a 900s threshold.
|
||||
let summary = compute_state_for_user(
|
||||
user_root.path(),
|
||||
Duration::from_secs(900),
|
||||
base_time() + Duration::from_secs(100),
|
||||
)
|
||||
.await;
|
||||
|
||||
assert_eq!(summary.required, 1);
|
||||
assert_eq!(summary.idle, 0);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -42,11 +42,12 @@ async fn process(
|
||||
let input_path = case_dir.join(ANALYSIS_INPUT_FILE);
|
||||
let document_path = case_dir.join(DOCUMENT_FILE);
|
||||
|
||||
// If the document already exists, the job is obsolete (e.g. recovery
|
||||
// re-enqueued a finished case during a race). Skip silently.
|
||||
if document_path.exists() {
|
||||
return;
|
||||
}
|
||||
// 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) {
|
||||
|
||||
+11
-3
@@ -184,7 +184,9 @@ async fn main() {
|
||||
});
|
||||
}
|
||||
|
||||
// Analyze pipeline: channel + worker + recovery scan.
|
||||
// Analyze pipeline: channel + worker. Boot-scan is observation-only —
|
||||
// the per-render `try_enqueue_all_for_user` covers auto-trigger via
|
||||
// the typed `CaseAnalysisState` machine.
|
||||
let (analyze_tx, analyze_rx) = analyze::channel();
|
||||
let analyze_busy: doctate_server::WorkerBusy =
|
||||
std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
|
||||
@@ -196,10 +198,16 @@ async fn main() {
|
||||
events_tx.clone(),
|
||||
));
|
||||
{
|
||||
let tx = analyze_tx.clone();
|
||||
let data_path = config.data_path.clone();
|
||||
let idle_threshold =
|
||||
std::time::Duration::from_secs(config.analysis_idle_threshold_secs);
|
||||
tokio::spawn(async move {
|
||||
analyze::recovery::scan_and_enqueue(&data_path, &tx).await;
|
||||
analyze::recovery::boot_scan_state(
|
||||
&data_path,
|
||||
idle_threshold,
|
||||
std::time::SystemTime::now(),
|
||||
)
|
||||
.await;
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
@@ -13,9 +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, recovery as analyze_recovery,
|
||||
};
|
||||
use crate::analyze::{ANALYSIS_INPUT_FILE, DOCUMENT_FILE, auto_trigger};
|
||||
use crate::auth::AuthenticatedWebUser;
|
||||
use crate::case_id::{CaseId, CaseIdPath};
|
||||
use crate::config::Config;
|
||||
@@ -519,9 +517,6 @@ impl PipelineState {
|
||||
vocab: &Arc<Gazetteer>,
|
||||
events_tx: &EventSender,
|
||||
) {
|
||||
if !self.analyze_busy.0.load(Ordering::Acquire) {
|
||||
analyze_recovery::enqueue_pending_for_user(user_root, &self.analyze_tx).await;
|
||||
}
|
||||
if !self.transcribe_busy.0.load(Ordering::Acquire) {
|
||||
transcribe_recovery::enqueue_pending_for_user(user_root, slug, &self.transcribe_tx)
|
||||
.await;
|
||||
|
||||
@@ -613,35 +613,28 @@ 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.
|
||||
#[tokio::test]
|
||||
async fn recovery_enqueues_pending_analysis() {
|
||||
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();
|
||||
std::fs::write(case_dir.join(ANALYSIS_INPUT_FILE), "{}").unwrap();
|
||||
|
||||
let (tx, mut rx) = analyze::channel();
|
||||
analyze::recovery::scan_and_enqueue(&tmp, &tx).await;
|
||||
|
||||
let job = rx.try_recv().expect("expected one job");
|
||||
assert_eq!(job.case_dir, case_dir);
|
||||
assert!(rx.try_recv().is_err(), "no further jobs expected");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn recovery_skips_completed_analysis() {
|
||||
let tmp = unique_tmpdir("analyze-r2");
|
||||
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();
|
||||
write_document_for_test(&case_dir, "done", "1970-01-01T00:00:00Z");
|
||||
|
||||
let (tx, mut rx) = analyze::channel();
|
||||
analyze::recovery::scan_and_enqueue(&tmp, &tx).await;
|
||||
let (_tx, mut rx) = analyze::channel();
|
||||
analyze::recovery::boot_scan_state(
|
||||
&tmp,
|
||||
std::time::Duration::ZERO,
|
||||
std::time::SystemTime::now(),
|
||||
)
|
||||
.await;
|
||||
|
||||
assert!(
|
||||
rx.try_recv().is_err(),
|
||||
"completed analysis must not re-enqueue"
|
||||
"boot scan must not enqueue any jobs in the new state model"
|
||||
);
|
||||
}
|
||||
|
||||
@@ -1246,10 +1239,18 @@ async fn recovery_skips_closed_cases() {
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
let (tx, mut rx) = analyze::channel();
|
||||
analyze::recovery::scan_and_enqueue(&tmp, &tx).await;
|
||||
let summary = analyze::recovery::boot_scan_state(
|
||||
&tmp,
|
||||
std::time::Duration::ZERO,
|
||||
std::time::SystemTime::now(),
|
||||
)
|
||||
.await;
|
||||
|
||||
assert!(rx.try_recv().is_err(), "closed case must not be enqueued");
|
||||
assert_eq!(
|
||||
summary,
|
||||
analyze::recovery::BootStateSummary::default(),
|
||||
"closed case must not appear in any state bucket"
|
||||
);
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------
|
||||
|
||||
Reference in New Issue
Block a user