diff --git a/Cargo.lock b/Cargo.lock index 847a592..a56b7d7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1162,12 +1162,18 @@ dependencies = [ "directories", "doctate-common", "eframe", + "reqwest", "serde", + "serde_json", "tempfile", "thiserror 1.0.69", + "tokio", "toml", "tracing", "tracing-subscriber", + "uuid", + "which", + "wiremock", ] [[package]] @@ -1336,6 +1342,12 @@ dependencies = [ "winit", ] +[[package]] +name = "either" +version = "1.15.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "48c757948c5ede0e46177b7add2e67155f70e33c07fea8284df6576da70b3719" + [[package]] name = "emath" version = "0.28.1" @@ -1940,6 +1952,15 @@ version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "dfa686283ad6dd069f105e5ab091b04c62850d3e4cf5d67debad1933f55023df" +[[package]] +name = "home" +version = "0.5.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cc627f471c528ff0c4a49e1d5e60450c8f6461dd6d10ba9dcd3a61d3dff7728d" +dependencies = [ + "windows-sys 0.61.2", +] + [[package]] name = "http" version = "1.4.0" @@ -5035,6 +5056,18 @@ dependencies = [ "web-sys", ] +[[package]] +name = "which" +version = "6.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b4ee928febd44d98f2f459a4a79bd4d928591333a494a10a868418ac1b39cf1f" +dependencies = [ + "either", + "home", + "rustix 0.38.44", + "winsafe", +] + [[package]] name = "widestring" version = "1.2.1" @@ -5503,6 +5536,12 @@ dependencies = [ "memchr", ] +[[package]] +name = "winsafe" +version = "0.0.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d135d17ab770252ad95e9a872d365cf3090e3be864a34ab46f48555993efc904" + [[package]] name = "wiremock" version = "0.6.5" diff --git a/client-desktop/Cargo.toml b/client-desktop/Cargo.toml index 3ec84bf..aad56a2 100644 --- a/client-desktop/Cargo.toml +++ b/client-desktop/Cargo.toml @@ -7,12 +7,19 @@ description = "Desktop dictation client for the doctate medical dictation system [dependencies] doctate-common = { path = "../doctate-common" } eframe = "0.28" +reqwest = { workspace = true } serde = { workspace = true } +serde_json = { workspace = true } toml = { workspace = true } +tokio = { workspace = true } tracing = { workspace = true } tracing-subscriber = { version = "0.3", features = ["env-filter"] } +uuid = { workspace = true } directories = "5" thiserror = "1" +which = "6" [dev-dependencies] tempfile = "3" +tokio = { workspace = true, features = ["test-util"] } +wiremock = "0.6" diff --git a/client-desktop/src/main.rs b/client-desktop/src/main.rs index cb1dfc3..2fb9d93 100644 --- a/client-desktop/src/main.rs +++ b/client-desktop/src/main.rs @@ -1,5 +1,11 @@ #[allow(dead_code)] mod config; +#[allow(dead_code)] +mod paths; +#[allow(dead_code)] +mod recorder; +#[allow(dead_code)] +mod uploader; use eframe::egui; use tracing::info; diff --git a/client-desktop/src/paths.rs b/client-desktop/src/paths.rs new file mode 100644 index 0000000..b70dca4 --- /dev/null +++ b/client-desktop/src/paths.rs @@ -0,0 +1,23 @@ +//! OS-conventional paths for client-side storage. + +use std::path::PathBuf; + +use directories::ProjectDirs; +use thiserror::Error; + +#[derive(Debug, Error)] +pub enum PathError { + #[error("cannot determine OS data directory (is $HOME set?)")] + NoDataDir, +} + +/// Directory where recordings wait for upload. Persists across app +/// restarts and crashes. +/// Linux: `~/.local/share/doctate/pending/` +/// Windows: `%LOCALAPPDATA%\doctate\pending\` +/// macOS: `~/Library/Application Support/doctate/pending/` +pub fn pending_dir() -> Result { + ProjectDirs::from("", "", "doctate") + .map(|dirs| dirs.data_local_dir().join("pending")) + .ok_or(PathError::NoDataDir) +} diff --git a/client-desktop/src/recorder.rs b/client-desktop/src/recorder.rs new file mode 100644 index 0000000..f870fc6 --- /dev/null +++ b/client-desktop/src/recorder.rs @@ -0,0 +1,201 @@ +//! Audio recorder via an ffmpeg subprocess. Produces m4a/AAC files on disk. +//! +//! Portability note: this module is intentionally UI-agnostic and carries no +//! `egui`/`eframe` deps so it can be lifted into a shared client-core crate +//! once a second Rust client (Windows variant, etc.) arrives. +//! +//! Call `Recorder::start` inside a Tokio runtime context (entered with +//! `Runtime::enter()` or via `#[tokio::main]`). + +use std::path::{Path, PathBuf}; +use std::process::Stdio; +use std::time::Duration; + +use thiserror::Error; +use tokio::io::AsyncWriteExt; +use tokio::process::{Child, Command}; +use tokio::sync::{mpsc, oneshot}; +use tokio::time::timeout; +use tracing::warn; + +/// Max time to wait for ffmpeg to finalize the MOOV atom after stop; past +/// this we hard-kill and emit `Failed`. +const STOP_GRACEFUL_TIMEOUT: Duration = Duration::from_secs(5); + +#[derive(Debug)] +pub enum RecorderEvent { + /// ffmpeg spawned successfully; audio capture has begun. + Started, + /// ffmpeg exited cleanly; the m4a file is ready. + Finished { output: PathBuf }, + /// ffmpeg or I/O failed; output file may be missing or corrupt. + Failed { error: String }, +} + +#[derive(Debug, Error)] +pub enum RecorderError { + #[error("ffmpeg not found in PATH")] + FfmpegMissing, + #[error("spawn ffmpeg failed: {0}")] + Spawn(#[from] std::io::Error), +} + +/// Handle to a running recorder. Dropping it does NOT stop the recorder — +/// call `stop()` explicitly (consumes self). +pub struct Recorder { + stop_tx: oneshot::Sender<()>, +} + +impl Recorder { + /// Spawn a recorder task that captures the system default audio input + /// to `output_path`. Returns a handle + event receiver. The task + /// continues until `stop()` is called, then finalizes the m4a. + pub fn start( + output_path: PathBuf, + ) -> Result<(Self, mpsc::UnboundedReceiver), RecorderError> { + if which::which("ffmpeg").is_err() { + return Err(RecorderError::FfmpegMissing); + } + + let (events_tx, events_rx) = mpsc::unbounded_channel(); + let (stop_tx, stop_rx) = oneshot::channel(); + + let mut cmd = Command::new("ffmpeg"); + cmd.args(ffmpeg_args(&output_path)) + .stdin(Stdio::piped()) + .stdout(Stdio::null()) + .stderr(Stdio::null()); + + let child = cmd.spawn()?; + + tokio::spawn(run_recorder_task(child, output_path, stop_rx, events_tx)); + + Ok((Self { stop_tx }, events_rx)) + } + + /// Request the recorder to finalize and exit. Consumes the handle — + /// no double-stop. Completion arrives as `Finished` (or `Failed`) on + /// the event receiver. + pub fn stop(self) { + let _ = self.stop_tx.send(()); + } +} + +async fn run_recorder_task( + mut child: Child, + output_path: PathBuf, + stop_rx: oneshot::Receiver<()>, + events_tx: mpsc::UnboundedSender, +) { + let _ = events_tx.send(RecorderEvent::Started); + + tokio::select! { + exit_result = child.wait() => { + let event = match exit_result { + Ok(status) if status.success() => RecorderEvent::Finished { output: output_path }, + Ok(status) => RecorderEvent::Failed { + error: format!("ffmpeg exited unexpectedly: {status}"), + }, + Err(e) => RecorderEvent::Failed { + error: format!("wait ffmpeg: {e}"), + }, + }; + let _ = events_tx.send(event); + } + _ = stop_rx => { + finalize(&mut child, output_path, &events_tx).await; + } + } +} + +/// Send 'q' to ffmpeg's stdin, wait up to `STOP_GRACEFUL_TIMEOUT` for a +/// clean exit; hard-kill on timeout. +async fn finalize( + child: &mut Child, + output_path: PathBuf, + events_tx: &mpsc::UnboundedSender, +) { + if let Some(stdin) = child.stdin.as_mut() + && let Err(e) = stdin.write_all(b"q\n").await + { + warn!(error = %e, "failed to write stop to ffmpeg stdin"); + } + + let event = match timeout(STOP_GRACEFUL_TIMEOUT, child.wait()).await { + Ok(Ok(status)) if status.success() => RecorderEvent::Finished { output: output_path }, + Ok(Ok(status)) => RecorderEvent::Failed { + error: format!("ffmpeg exited with {status}"), + }, + Ok(Err(e)) => RecorderEvent::Failed { + error: format!("wait ffmpeg: {e}"), + }, + Err(_elapsed) => { + warn!("ffmpeg did not exit within timeout; killing"); + let _ = child.kill().await; + RecorderEvent::Failed { + error: "ffmpeg stop timeout — output may be corrupt".into(), + } + } + }; + let _ = events_tx.send(event); +} + +fn ffmpeg_args(output: &Path) -> Vec { + let mut args: Vec = vec!["-loglevel".into(), "error".into()]; + for s in audio_input_args() { + args.push(s.to_string()); + } + args.extend([ + "-c:a".into(), + "aac".into(), + "-b:a".into(), + "96k".into(), + "-ar".into(), + "48000".into(), + "-ac".into(), + "1".into(), + "-movflags".into(), + "+faststart".into(), + "-y".into(), + output.to_string_lossy().into_owned(), + ]); + args +} + +#[cfg(target_os = "linux")] +fn audio_input_args() -> Vec<&'static str> { + // PulseAudio interface (also works under PipeWire via pipewire-pulse). + vec!["-f", "pulse", "-i", "default"] +} + +#[cfg(target_os = "windows")] +fn audio_input_args() -> Vec<&'static str> { + // TODO(windows-client): enumerate dshow devices via + // `ffmpeg -list_devices true -f dshow -i dummy` and cache the first + // non-loopback capture device. "audio=default" below is a placeholder; + // most Windows systems do not have a device literally named "default", + // so the Windows client must resolve a real name before shipping. + vec!["-f", "dshow", "-i", "audio=default"] +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn ffmpeg_args_contain_output_path_and_encoder() { + let path = Path::new("/tmp/test.m4a"); + let args = ffmpeg_args(path); + assert!(args.contains(&"/tmp/test.m4a".to_string())); + assert!(args.contains(&"-y".to_string())); + assert!(args.contains(&"aac".to_string())); + assert!(args.contains(&"+faststart".to_string())); + } + + #[test] + fn ffmpeg_args_have_exactly_one_input() { + let args = ffmpeg_args(Path::new("/tmp/x.m4a")); + let input_flags = args.iter().filter(|s| *s == "-i").count(); + assert_eq!(input_flags, 1); + } +} diff --git a/client-desktop/src/uploader.rs b/client-desktop/src/uploader.rs new file mode 100644 index 0000000..75e48da --- /dev/null +++ b/client-desktop/src/uploader.rs @@ -0,0 +1,417 @@ +//! Async uploader with persistent on-disk queue and exponential-backoff +//! retry. Recordings live in the pending directory as `{stem}.m4a` plus +//! a sidecar `{stem}.meta.json`. The worker replays anything it finds +//! there at startup, then processes new submissions as they arrive. + +use std::path::{Path, PathBuf}; +use std::time::Duration; + +use doctate_common::ack::{AckResponse, AckStatus}; +use doctate_common::constants::{ + API_KEY_HEADER, CONTENT_TYPE_AUDIO_MP4, FIELD_AUDIO, FIELD_CASE_ID, FIELD_RECORDED_AT, + UPLOAD_PATH, +}; +use serde::{Deserialize, Serialize}; +use thiserror::Error; +use tokio::sync::mpsc; +use tracing::{info, warn}; +use uuid::Uuid; + +use crate::config::Config; + +const INITIAL_BACKOFF: Duration = Duration::from_secs(2); +const MAX_BACKOFF: Duration = Duration::from_secs(60); + +/// A recording waiting to be uploaded. The `file` path is derived from +/// the sidecar JSON's location, never from the JSON content itself — +/// that way the sidecar stays valid even if the pending directory is +/// renamed. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct PendingUpload { + pub case_id: Uuid, + pub recorded_at: String, + #[serde(skip, default)] + pub file: PathBuf, +} + +/// Lifecycle events emitted to the UI for a single upload attempt chain. +#[derive(Debug)] +pub enum UploadEvent { + Started(Uuid), + Succeeded { + case_id: Uuid, + status: AckStatus, + }, + Failed { + case_id: Uuid, + reason: String, + will_retry: bool, + }, +} + +#[derive(Debug, Error)] +pub enum UploaderError { + #[error("I/O error: {0}")] + Io(#[from] std::io::Error), + #[error("serialize sidecar: {0}")] + Serialize(#[from] serde_json::Error), +} + +/// Handle to the uploader's worker task. Drop does NOT shut down the +/// worker; it stays alive as long as the `submit_tx` is kept. +pub struct Uploader { + submit_tx: mpsc::UnboundedSender, +} + +impl Uploader { + /// Spawn the worker task. Must be called inside a Tokio runtime + /// context (`Runtime::enter()` guard or `#[tokio::main]`). + /// + /// Re-scans `pending_dir` at startup so leftover recordings from + /// a prior run are replayed automatically. + pub fn spawn( + config: Config, + pending_dir: PathBuf, + ) -> (Self, mpsc::UnboundedReceiver) { + let (submit_tx, submit_rx) = mpsc::unbounded_channel(); + let (events_tx, events_rx) = mpsc::unbounded_channel(); + tokio::spawn(run_worker(config, pending_dir, submit_rx, events_tx)); + (Self { submit_tx }, events_rx) + } + + /// Submit a pending upload for processing. Non-blocking. Fails only + /// if the worker has been dropped (should not happen in normal + /// operation). + pub fn submit( + &self, + upload: PendingUpload, + ) -> Result<(), mpsc::error::SendError> { + self.submit_tx.send(upload) + } +} + +/// Path of the sidecar JSON for a given m4a file — same directory, +/// `{stem}.meta.json`. +pub fn sidecar_path_for(m4a: &Path) -> PathBuf { + let stem = m4a + .file_stem() + .map(|s| s.to_string_lossy().into_owned()) + .unwrap_or_default(); + let dir = m4a.parent().unwrap_or_else(|| Path::new(".")); + dir.join(format!("{stem}.meta.json")) +} + +/// Persist a pending upload's metadata next to its m4a file. +pub fn write_sidecar(upload: &PendingUpload) -> Result<(), UploaderError> { + let sidecar = sidecar_path_for(&upload.file); + if let Some(parent) = sidecar.parent() { + std::fs::create_dir_all(parent)?; + } + let json = serde_json::to_string_pretty(upload)?; + std::fs::write(&sidecar, json)?; + Ok(()) +} + +/// Scan `pending_dir` for m4a files with matching sidecars. Orphans +/// (m4a without sidecar, or unreadable sidecar) are logged and skipped +/// rather than propagated as errors — one broken file should not take +/// down recovery of the rest. +pub fn scan_pending_dir(pending_dir: &Path) -> Vec { + let entries = match std::fs::read_dir(pending_dir) { + Ok(e) => e, + Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Vec::new(), + Err(e) => { + warn!(dir = ?pending_dir, error = %e, "scan pending dir failed"); + return Vec::new(); + } + }; + let mut out = Vec::new(); + for entry in entries.flatten() { + let path = entry.path(); + if path.extension().and_then(|s| s.to_str()) != Some("m4a") { + continue; + } + match read_pending(&path) { + Ok(upload) => out.push(upload), + Err(e) => warn!(path = ?path, error = %e, "skipping pending entry"), + } + } + out +} + +fn read_pending(m4a: &Path) -> Result { + let sidecar = sidecar_path_for(m4a); + let json = std::fs::read_to_string(&sidecar)?; + let mut upload: PendingUpload = serde_json::from_str(&json)?; + upload.file = m4a.to_path_buf(); + Ok(upload) +} + +async fn run_worker( + config: Config, + pending_dir: PathBuf, + mut submit_rx: mpsc::UnboundedReceiver, + events_tx: mpsc::UnboundedSender, +) { + let client = reqwest::Client::new(); + + let startup = scan_pending_dir(&pending_dir); + if !startup.is_empty() { + info!(count = startup.len(), "replaying pending uploads"); + } + for pending in startup { + upload_with_retry(&client, &config, pending, &events_tx).await; + } + + while let Some(pending) = submit_rx.recv().await { + upload_with_retry(&client, &config, pending, &events_tx).await; + } +} + +async fn upload_with_retry( + client: &reqwest::Client, + config: &Config, + upload: PendingUpload, + events_tx: &mpsc::UnboundedSender, +) { + let mut backoff = INITIAL_BACKOFF; + loop { + let _ = events_tx.send(UploadEvent::Started(upload.case_id)); + match attempt_upload(client, config, &upload).await { + Ok(ack) => { + let _ = tokio::fs::remove_file(&upload.file).await; + let _ = tokio::fs::remove_file(&sidecar_path_for(&upload.file)).await; + let _ = events_tx.send(UploadEvent::Succeeded { + case_id: upload.case_id, + status: ack.status, + }); + return; + } + Err(ErrorKind::Terminal(reason)) => { + let _ = events_tx.send(UploadEvent::Failed { + case_id: upload.case_id, + reason, + will_retry: false, + }); + return; + } + Err(ErrorKind::Transient(reason)) => { + let _ = events_tx.send(UploadEvent::Failed { + case_id: upload.case_id, + reason, + will_retry: true, + }); + tokio::time::sleep(backoff).await; + backoff = (backoff * 2).min(MAX_BACKOFF); + } + } + } +} + +enum ErrorKind { + Transient(String), + Terminal(String), +} + +async fn attempt_upload( + client: &reqwest::Client, + config: &Config, + upload: &PendingUpload, +) -> Result { + let audio = tokio::fs::read(&upload.file) + .await + .map_err(|e| ErrorKind::Terminal(format!("read audio file: {e}")))?; + + let file_name = upload + .file + .file_name() + .map(|s| s.to_string_lossy().into_owned()) + .unwrap_or_else(|| "recording.m4a".to_string()); + + let audio_part = reqwest::multipart::Part::bytes(audio) + .file_name(file_name) + .mime_str(CONTENT_TYPE_AUDIO_MP4) + .expect("static MIME string is valid"); + + let form = reqwest::multipart::Form::new() + .text(FIELD_CASE_ID, upload.case_id.to_string()) + .text(FIELD_RECORDED_AT, upload.recorded_at.clone()) + .part(FIELD_AUDIO, audio_part); + + let url = format!("{}{}", config.server_url.trim_end_matches('/'), UPLOAD_PATH); + + let resp = client + .post(&url) + .header(API_KEY_HEADER, &config.api_key) + .multipart(form) + .send() + .await + .map_err(|e| ErrorKind::Transient(format!("network: {e}")))?; + + let status = resp.status(); + if status.is_success() { + resp.json::() + .await + .map_err(|e| ErrorKind::Terminal(format!("parse ack: {e}"))) + } else if status.as_u16() == 408 || status.as_u16() == 429 || status.is_server_error() { + let body = resp.text().await.unwrap_or_default(); + Err(ErrorKind::Transient(format!("HTTP {status}: {body}"))) + } else { + let body = resp.text().await.unwrap_or_default(); + Err(ErrorKind::Terminal(format!("HTTP {status}: {body}"))) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use tempfile::TempDir; + use wiremock::matchers::{header, method, path as wpath}; + use wiremock::{Mock, MockServer, ResponseTemplate}; + + fn sample_pending(dir: &Path) -> PendingUpload { + let file = dir.join("sample.m4a"); + std::fs::write(&file, b"dummy audio").unwrap(); + PendingUpload { + case_id: Uuid::parse_str("550e8400-e29b-41d4-a716-446655440000").unwrap(), + recorded_at: "2026-04-15T10:32:00Z".into(), + file, + } + } + + fn ack_body(status: AckStatus) -> AckResponse { + AckResponse { + case_id: "550e8400-e29b-41d4-a716-446655440000".into(), + recorded_at: "2026-04-15T10:32:00Z".into(), + status, + } + } + + #[test] + fn sidecar_roundtrip_via_scan() { + let tmp = TempDir::new().unwrap(); + let upload = sample_pending(tmp.path()); + write_sidecar(&upload).unwrap(); + let found = scan_pending_dir(tmp.path()); + assert_eq!(found.len(), 1); + assert_eq!(found[0].case_id, upload.case_id); + assert_eq!(found[0].recorded_at, upload.recorded_at); + assert_eq!(found[0].file, upload.file); + } + + #[test] + fn scan_ignores_m4a_without_sidecar() { + let tmp = TempDir::new().unwrap(); + std::fs::write(tmp.path().join("orphan.m4a"), b"x").unwrap(); + let found = scan_pending_dir(tmp.path()); + assert!(found.is_empty()); + } + + #[test] + fn scan_returns_empty_for_missing_dir() { + let tmp = TempDir::new().unwrap(); + let missing = tmp.path().join("does-not-exist"); + assert!(scan_pending_dir(&missing).is_empty()); + } + + #[tokio::test] + async fn successful_upload_deletes_files_and_emits_events() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(wpath(UPLOAD_PATH)) + .and(header(API_KEY_HEADER, "test-key")) + .respond_with(ResponseTemplate::new(200).set_body_json(ack_body(AckStatus::Received))) + .mount(&server) + .await; + + let tmp = TempDir::new().unwrap(); + let upload = sample_pending(tmp.path()); + write_sidecar(&upload).unwrap(); + let config = Config { + server_url: server.uri(), + api_key: "test-key".into(), + }; + + let client = reqwest::Client::new(); + let (tx, mut rx) = mpsc::unbounded_channel(); + upload_with_retry(&client, &config, upload.clone(), &tx).await; + + assert!(!upload.file.exists(), "m4a should be deleted after ack"); + assert!( + !sidecar_path_for(&upload.file).exists(), + "sidecar should be deleted after ack" + ); + + assert!(matches!(rx.recv().await, Some(UploadEvent::Started(_)))); + let Some(UploadEvent::Succeeded { status, .. }) = rx.recv().await else { + panic!("expected Succeeded event"); + }; + assert_eq!(status, AckStatus::Received); + } + + #[tokio::test(start_paused = true)] + async fn transient_500_retries_then_succeeds() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(wpath(UPLOAD_PATH)) + .respond_with(ResponseTemplate::new(500)) + .up_to_n_times(1) + .mount(&server) + .await; + Mock::given(method("POST")) + .and(wpath(UPLOAD_PATH)) + .respond_with(ResponseTemplate::new(200).set_body_json(ack_body(AckStatus::Received))) + .mount(&server) + .await; + + let tmp = TempDir::new().unwrap(); + let upload = sample_pending(tmp.path()); + write_sidecar(&upload).unwrap(); + let config = Config { + server_url: server.uri(), + api_key: "test-key".into(), + }; + + let client = reqwest::Client::new(); + let (tx, mut rx) = mpsc::unbounded_channel(); + upload_with_retry(&client, &config, upload.clone(), &tx).await; + + assert!(matches!(rx.recv().await, Some(UploadEvent::Started(_)))); + let Some(UploadEvent::Failed { will_retry, .. }) = rx.recv().await else { + panic!("expected Failed event"); + }; + assert!(will_retry, "500 should be a transient failure"); + assert!(matches!(rx.recv().await, Some(UploadEvent::Started(_)))); + assert!(matches!(rx.recv().await, Some(UploadEvent::Succeeded { .. }))); + } + + #[tokio::test] + async fn terminal_401_keeps_files_and_does_not_retry() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(wpath(UPLOAD_PATH)) + .respond_with(ResponseTemplate::new(401).set_body_string("Unauthorized")) + .mount(&server) + .await; + + let tmp = TempDir::new().unwrap(); + let upload = sample_pending(tmp.path()); + write_sidecar(&upload).unwrap(); + let config = Config { + server_url: server.uri(), + api_key: "bad-key".into(), + }; + + let client = reqwest::Client::new(); + let (tx, mut rx) = mpsc::unbounded_channel(); + upload_with_retry(&client, &config, upload.clone(), &tx).await; + + assert!(upload.file.exists(), "m4a must be preserved on terminal failure"); + + assert!(matches!(rx.recv().await, Some(UploadEvent::Started(_)))); + let Some(UploadEvent::Failed { will_retry, .. }) = rx.recv().await else { + panic!("expected Failed event"); + }; + assert!(!will_retry, "401 is terminal; must not retry"); + } +}