c769abe3d5
Introduces `OnelinerLocks` to serialize read-modify-write operations on
`oneliner.json`. This prevents race conditions between the manual
override API (`PUT /api/cases/{case_id}/oneliner`) and the transcription
worker's auto-regeneration process.
The manual override now acquires a lock specific to the `case_dir`
before writing. The transcription worker's `update_oneliner` function
also acquires the same lock around its post-LLM re-read-and-write
sequence. This ensures that a doctor's manual edit always takes
precedence, even if it arrives while an auto-regeneration is in
progress.
This change also includes:
- A new `routes::oneliner_override` module for the PUT handler.
- Validation for the oneliner text length and emptiness.
- Integration tests for the override functionality, covering happy path,
error conditions, early latching, race conditions, and parallel PUTs.
136 lines
5.0 KiB
Rust
136 lines
5.0 KiB
Rust
//! Regression test: `PipelineState::heal_orphans_if_idle` must not block
|
||
//! the request handler on Ollama. Before the fix, the function awaited a
|
||
//! sequential `regenerate_missing_oneliners_for_user` loop inline, so the
|
||
//! HTTP handler stayed pinned for N × LLM-latency. The fix detaches the
|
||
//! regeneration via `tokio::spawn`, guarded by a `compare_exchange` on
|
||
//! `OnelinerHealBusy` to dedup concurrent page-loads.
|
||
|
||
mod common;
|
||
|
||
use std::path::Path;
|
||
use std::sync::Arc;
|
||
use std::sync::atomic::{AtomicBool, Ordering};
|
||
use std::time::{Duration, Instant};
|
||
|
||
use doctate_server::events;
|
||
use doctate_server::gazetteer::Gazetteer;
|
||
use doctate_server::{
|
||
AnalyzeBusy, OnelinerHealBusy, PipelineState, TranscribeBusy, WorkerBusy, analyze, transcribe,
|
||
};
|
||
use serde_json::json;
|
||
use tempfile::tempdir;
|
||
use tokio::time::timeout;
|
||
use wiremock::matchers::{method, path as wm_path};
|
||
use wiremock::{Mock, MockServer, ResponseTemplate};
|
||
|
||
use common::{ONELINER_FILENAME, TestConfig};
|
||
|
||
fn write_transcript(case_dir: &Path) {
|
||
// Seed an m4a + transcript pair, matching the worker's on-disk
|
||
// layout. The oneliner worker iterates `.m4a` files and resolves
|
||
// the sidecar from the stem; a stray transcript without an m4a
|
||
// would be invisible to it.
|
||
let stem = "2026-04-16T10-00-00Z";
|
||
std::fs::write(case_dir.join(format!("{stem}.m4a")), b"audio").unwrap();
|
||
std::fs::write(
|
||
case_dir.join(format!("{stem}.transcript.txt")),
|
||
"Brustschmerz links, seit heute morgen.",
|
||
)
|
||
.unwrap();
|
||
}
|
||
|
||
#[tokio::test]
|
||
async fn heal_orphans_returns_fast_and_dedups_parallel_calls() {
|
||
// Five cases with a transcript but no oneliner.json. Without the fix,
|
||
// heal_orphans_if_idle would sequentially block on 5 × 200 ms = 1 s of
|
||
// mock delay. With the fix, it must return in well under that window.
|
||
let data = tempdir().unwrap();
|
||
let slug = "dr_test";
|
||
let user_root = data.path().join(slug);
|
||
std::fs::create_dir_all(&user_root).unwrap();
|
||
for i in 0..5 {
|
||
let case_dir = user_root.join(format!("case-{i:02}"));
|
||
std::fs::create_dir_all(&case_dir).unwrap();
|
||
write_transcript(&case_dir);
|
||
}
|
||
|
||
// Ollama mock: 200 ms delay per chat call. `.expect(5)` asserts at
|
||
// MockServer drop that exactly five requests arrived — dedup would
|
||
// show up as 10 without the CAS guard.
|
||
let mock = MockServer::start().await;
|
||
Mock::given(method("POST"))
|
||
.and(wm_path("/api/chat"))
|
||
.respond_with(
|
||
ResponseTemplate::new(200)
|
||
.set_delay(Duration::from_millis(200))
|
||
.set_body_json(json!({
|
||
"message": { "content": "Brustschmerz links" }
|
||
})),
|
||
)
|
||
.expect(5)
|
||
.mount(&mock)
|
||
.await;
|
||
|
||
let config = TestConfig::new()
|
||
.with_data_path(data.path().to_path_buf())
|
||
.with_ollama(mock.uri())
|
||
.build();
|
||
|
||
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 heal_busy: WorkerBusy = Arc::new(AtomicBool::new(false));
|
||
let pipeline = PipelineState {
|
||
analyze_busy: AnalyzeBusy(Arc::new(AtomicBool::new(false))),
|
||
analyze_tx: tx_a,
|
||
transcribe_busy: TranscribeBusy(Arc::new(AtomicBool::new(false))),
|
||
transcribe_tx: tx_t,
|
||
oneliner_heal_busy: OnelinerHealBusy(heal_busy.clone()),
|
||
oneliner_locks: doctate_server::oneliner_locks::OnelinerLocks::new(),
|
||
};
|
||
|
||
// Two back-to-back heal invocations. The first wins the CAS and
|
||
// spawns. The second must no-op. Both heal futures themselves must
|
||
// finish in a fraction of the mock delay.
|
||
let before = Instant::now();
|
||
timeout(Duration::from_millis(150), async {
|
||
pipeline
|
||
.heal_orphans_if_idle(&user_root, slug, &http_client, &config, &vocab, &events_tx)
|
||
.await;
|
||
pipeline
|
||
.heal_orphans_if_idle(&user_root, slug, &http_client, &config, &vocab, &events_tx)
|
||
.await;
|
||
})
|
||
.await
|
||
.expect("heal_orphans_if_idle must return without awaiting the LLM loop");
|
||
assert!(
|
||
before.elapsed() < Duration::from_millis(150),
|
||
"heal_orphans returned in {:?} — the inline regen is back",
|
||
before.elapsed()
|
||
);
|
||
|
||
// Spawn drops the busy flag on completion — poll until idle.
|
||
let poll_start = Instant::now();
|
||
while heal_busy.load(Ordering::Acquire) {
|
||
if poll_start.elapsed() > Duration::from_secs(5) {
|
||
panic!("oneliner heal spawn did not finish within 5s");
|
||
}
|
||
tokio::time::sleep(Duration::from_millis(25)).await;
|
||
}
|
||
|
||
// Every case now has an oneliner.json.
|
||
for i in 0..5 {
|
||
let p = user_root
|
||
.join(format!("case-{i:02}"))
|
||
.join(ONELINER_FILENAME);
|
||
assert!(p.exists(), "oneliner.json missing for case-{i:02}");
|
||
}
|
||
|
||
// Mock is dropped at end of scope → `.expect(5)` verifies no
|
||
// double-firing. A failing expectation panics with the actual count.
|
||
}
|