feat: Add per-case dir oneliner locks
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.
This commit is contained in:
+10
-1
@@ -122,6 +122,12 @@ async fn main() {
|
||||
let oneliner_heal_busy: doctate_server::WorkerBusy =
|
||||
std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
|
||||
|
||||
// Per-`case_dir` mutex registry. Shared by the PUT handler (manual
|
||||
// override write) and the transcribe worker (post-LLM re-check)
|
||||
// so a doctor's manual edit always wins over a concurrent
|
||||
// auto-regen, no matter the timing.
|
||||
let oneliner_locks = doctate_server::oneliner_locks::OnelinerLocks::new();
|
||||
|
||||
// Transcription pipeline: channel + worker + recovery scan.
|
||||
let (transcribe_tx, transcribe_rx) = transcribe::channel();
|
||||
let transcribe_busy: doctate_server::WorkerBusy =
|
||||
@@ -133,6 +139,7 @@ async fn main() {
|
||||
transcribe_busy.clone(),
|
||||
vocab.clone(),
|
||||
events_tx.clone(),
|
||||
oneliner_locks.clone(),
|
||||
));
|
||||
{
|
||||
let tx = transcribe_tx.clone();
|
||||
@@ -142,6 +149,7 @@ async fn main() {
|
||||
let vocab = vocab.clone();
|
||||
let events = events_tx.clone();
|
||||
let heal_busy = oneliner_heal_busy.clone();
|
||||
let locks = oneliner_locks.clone();
|
||||
tokio::spawn(async move {
|
||||
transcribe::recovery::scan_and_enqueue(&data_path, &tx).await;
|
||||
// Then catch up oneliners for cases that crashed between
|
||||
@@ -150,7 +158,7 @@ async fn main() {
|
||||
// page-load does not spawn a second parallel regen loop.
|
||||
let _guard = doctate_server::BusyGuard::new(heal_busy);
|
||||
transcribe::recovery::regenerate_missing_oneliners(
|
||||
&data_path, &client, &cfg, &vocab, &events,
|
||||
&data_path, &client, &cfg, &vocab, &events, &locks,
|
||||
)
|
||||
.await;
|
||||
});
|
||||
@@ -185,6 +193,7 @@ async fn main() {
|
||||
analyze_busy: doctate_server::AnalyzeBusy(analyze_busy),
|
||||
transcribe_busy: doctate_server::TranscribeBusy(transcribe_busy),
|
||||
oneliner_heal_busy: doctate_server::OnelinerHealBusy(oneliner_heal_busy),
|
||||
oneliner_locks,
|
||||
events_tx,
|
||||
http_client,
|
||||
vocab,
|
||||
|
||||
Reference in New Issue
Block a user