feat: Add recovery scan for pending recordings

This commit introduces a new module `transcribe::recovery` and a
function `scan_and_enqueue`.
This function is called at startup to find audio files that have not yet
been transcribed
and re-enqueues them for processing. This ensures that recordings are
not lost if the
server restarts before transcription is complete.

A new integration test `recovery_enqueues_only_pending_recordings` has
been added to
verify the functionality.
This commit is contained in:
2026-04-13 16:47:38 +02:00
parent bcae4e5466
commit 13375b8297
4 changed files with 122 additions and 0 deletions
+9
View File
@@ -50,6 +50,15 @@ async fn main() {
http_client,
));
// Re-enqueue pending recordings left over from a previous run.
{
let tx = transcribe_tx.clone();
let data_path = config.data_path.clone();
tokio::spawn(async move {
transcribe::recovery::scan_and_enqueue(&data_path, &tx).await;
});
}
let state = AppState {
config: config.clone(),
transcribe_tx,
+1
View File
@@ -3,6 +3,7 @@ use std::path::PathBuf;
use tokio::sync::mpsc;
pub mod ffmpeg;
pub mod recovery;
pub mod whisper;
pub mod worker;
+73
View File
@@ -0,0 +1,73 @@
use std::path::{Path, PathBuf};
use tracing::{info, warn};
use super::{TranscribeJob, TranscribeSender};
/// Walk `data_path/*/open/*/*.m4a` and enqueue every recording that does not
/// yet have a `.transcript.txt` sibling.
///
/// Intended to run once at startup (spawn as a task — sends may back-pressure
/// if many recordings are pending and the queue fills).
pub async fn scan_and_enqueue(data_path: &Path, tx: &TranscribeSender) {
let pending = collect_pending(data_path).await;
let total = pending.len();
info!(count = total, "Recovery scan found pending recordings");
for audio_path in pending {
if tx
.send(TranscribeJob {
audio_path: audio_path.clone(),
})
.await
.is_err()
{
warn!(file = %audio_path.display(), "Recovery send failed (worker gone)");
break;
}
}
}
async fn collect_pending(data_path: &Path) -> Vec<PathBuf> {
let mut out = Vec::new();
let mut users = match tokio::fs::read_dir(data_path).await {
Ok(r) => r,
Err(_) => return out,
};
while let Ok(Some(user_entry)) = users.next_entry().await {
let open_dir = user_entry.path().join("open");
let mut cases = match tokio::fs::read_dir(&open_dir).await {
Ok(r) => r,
Err(_) => continue,
};
while let Ok(Some(case_entry)) = cases.next_entry().await {
let case_dir = case_entry.path();
if !case_dir.is_dir() {
continue;
}
let mut files = match tokio::fs::read_dir(&case_dir).await {
Ok(r) => r,
Err(_) => continue,
};
while let Ok(Some(file)) = files.next_entry().await {
let path = file.path();
if path.extension().and_then(|s| s.to_str()) != Some("m4a") {
continue;
}
let transcript = path.with_extension("transcript.txt");
if !transcript.exists() {
out.push(path);
}
}
}
}
// Oldest first (filenames are ISO timestamps).
out.sort();
out
}