doctate/server/tests/transient_failure_retries_test.rs

//! End-to-end regression test for the transient-transcribe-failure heal.
//!
//! Scenario: Minerva is briefly offline (HTTP 503) when the doctor's
//! recording arrives. The transcribe worker classifies the error as
//! transient and leaves the audio as plain `.m4a` — no `.failed`
//! suffix. On the next page-load, `heal_orphans_if_idle` re-enqueues
//! the pending `.m4a`. By then Minerva is back; Whisper returns the
//! transcript and the case advances normally.
//!
//! Pre-fix behaviour: any Whisper error renamed `.m4a` → `.m4a.failed`,
//! terminating progress permanently. Only a manual admin reset un-failed
//! the recording.

mod common;

use std::path::PathBuf;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;

use doctate_server::events;
use doctate_server::gazetteer::Gazetteer;
use doctate_server::{
    AnalyzeBusy, OnelinerHealBusy, PipelineState, TranscribeBusy, WorkerBusy, analyze, transcribe,
};
use tempfile::tempdir;
use wiremock::matchers::{method, path as wm_path};
use wiremock::{Mock, MockServer, ResponseTemplate};

use common::{TestConfig, test_user};

fn fixture(name: &str) -> PathBuf {
    PathBuf::from(env!("CARGO_MANIFEST_DIR"))
        .join("tests/fixtures")
        .join(name)
}

#[tokio::test]
async fn transient_whisper_failure_is_reenqueued_and_recovers() {
    // Arrange filesystem: one real m4a for user "dr_test".
    let data = tempdir().unwrap();
    let slug = "dr_test";
    let user_root = data.path().join(slug);
    let case_dir = user_root.join("case-transient");
    std::fs::create_dir_all(&case_dir).unwrap();
    let audio = case_dir.join("2026-04-15T20-30-00Z.m4a");
    std::fs::copy(fixture("sample.m4a"), &audio).unwrap();

    // Arrange Whisper mock: first call → 503 (Minerva down),
    // subsequent calls → 200 with a transcript. wiremock matches
    // mocks in registration order; `up_to_n_times(1)` retires the
    // first mock after one hit, so the second takes over for any
    // retry.
    let whisper = MockServer::start().await;
    Mock::given(method("POST"))
        .and(wm_path("/asr"))
        .respond_with(ResponseTemplate::new(503).set_body_string("service unavailable"))
        .up_to_n_times(1)
        .mount(&whisper)
        .await;
    Mock::given(method("POST"))
        .and(wm_path("/asr"))
        .respond_with(ResponseTemplate::new(200).set_body_string("recovered text"))
        .mount(&whisper)
        .await;

    // Config points at the mock Whisper, ollama intentionally unreachable
    // (the oneliner heal may fire in parallel; we only assert on
    // the transcript artefact, not on the oneliner).
    let (config, settings) = TestConfig::new()
        .with_data_path(data.path().to_path_buf())
        .with_user(test_user(slug))
        .with_whisper(whisper.uri(), 5)
        .build_pair();

    let vocab = Arc::new(Gazetteer::empty());
    let events_tx = events::channel();
    let http_client = reqwest::Client::new();

    let (tx_a, _rx_a) = analyze::channel();
    let (tx_t, rx_t) = transcribe::channel();
    let transcribe_busy: WorkerBusy = Arc::new(AtomicBool::new(false));
    let heal_busy: WorkerBusy = Arc::new(AtomicBool::new(false));

    // Spawn the transcribe worker in the background; it processes jobs
    // until the channel is closed at test teardown.
    let oneliner_locks = doctate_server::oneliner_locks::OnelinerLocks::new();
    let worker = tokio::spawn(transcribe::worker::run(
        rx_t,
        config.clone(),
        settings.clone(),
        http_client.clone(),
        transcribe_busy.clone(),
        vocab.clone(),
        events_tx.clone(),
        oneliner_locks.clone(),
    ));

    let pipeline = PipelineState {
        analyze_busy: AnalyzeBusy(Arc::new(AtomicBool::new(false))),
        analyze_tx: tx_a,
        transcribe_busy: TranscribeBusy(transcribe_busy.clone()),
        transcribe_tx: tx_t.clone(),
        oneliner_heal_busy: OnelinerHealBusy(heal_busy.clone()),
        oneliner_locks,
    };

    // Act 1: enqueue the initial job, wait for the 503 failure.
    tx_t.send(transcribe::TranscribeJob {
        audio_path: audio.clone(),
        user_slug: slug.into(),
    })
    .await
    .unwrap();

    // Wait until the worker has processed the first job (busy false
    // AND the 503 mock has recorded a hit). Polling both conditions
    // avoids a race where the worker hasn't started yet.
    let start = std::time::Instant::now();
    loop {
        let idle = !transcribe_busy.load(Ordering::Acquire);
        let hits = whisper.received_requests().await.unwrap().len();
        if idle && hits >= 1 {
            break;
        }
        if start.elapsed() > Duration::from_secs(10) {
            panic!("worker did not process first job within 10s (idle={idle}, hits={hits})");
        }
        tokio::time::sleep(Duration::from_millis(25)).await;
    }

    // Invariant after transient failure: .m4a still there, no .failed.
    assert!(audio.exists(), "transient error must not consume the .m4a");
    assert!(
        !case_dir.join("2026-04-15T20-30-00Z.m4a.failed").exists(),
        ".m4a.failed must not exist after transient error"
    );
    assert!(
        !case_dir.join("2026-04-15T20-30-00Z.json").exists(),
        "no metadata sidecar yet — the 503 produced nothing"
    );

    // Act 2: heal re-enqueues the pending .m4a (second mock → 200).
    pipeline
        .heal_orphans_if_idle(
            &user_root,
            slug,
            &http_client,
            &settings,
            &vocab,
            &events_tx,
        )
        .await;

    // Assert: the recording metadata sidecar lands. Poll because the
    // worker runs asynchronously after the heal hands off the job.
    let meta_path = case_dir.join("2026-04-15T20-30-00Z.json");
    let start = std::time::Instant::now();
    while !meta_path.exists() {
        if start.elapsed() > Duration::from_secs(10) {
            panic!(
                "meta sidecar did not appear within 10s (whisper hits: {})",
                whisper.received_requests().await.unwrap().len()
            );
        }
        tokio::time::sleep(Duration::from_millis(25)).await;
    }

    let bytes = std::fs::read(&meta_path).unwrap();
    let meta: doctate_common::RecordingMeta = serde_json::from_slice(&bytes).unwrap();
    match meta.transcript {
        doctate_common::Transcript::Content { text } => assert_eq!(text, "recovered text"),
        other => panic!("expected Transcript::Content, got {other:?}"),
    }

    // Teardown: close the worker channel so the background task exits.
    drop(tx_t);
    drop(pipeline);
    let _ = tokio::time::timeout(Duration::from_secs(5), worker).await;
}