diff --git a/Cargo.lock b/Cargo.lock index 490ea7d..d64b3c1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1150,6 +1150,7 @@ name = "doctate-client-core" version = "0.1.0" dependencies = [ "doctate-common", + "filetime", "reqwest", "serde", "serde_json", diff --git a/client-desktop/src/app.rs b/client-desktop/src/app.rs index 0c75119..3ca0da6 100644 --- a/client-desktop/src/app.rs +++ b/client-desktop/src/app.rs @@ -6,7 +6,7 @@ use std::sync::Arc; use std::time::{Duration, Instant}; use doctate_client_core::{ - CaseMarker, CaseStore, OnelinerPoller, PollerConfig, SnapshotCache, + cleanup_orphan_audio, CaseMarker, CaseStore, OnelinerPoller, PollerConfig, SnapshotCache, }; use doctate_common::timestamp::{now_rfc3339, recorded_at_to_filename_stem}; use eframe::egui; @@ -86,6 +86,32 @@ impl DoctateApp { let snapshot_cache = Arc::new(SnapshotCache::new(snapshot_cache_path)); let http_client = Arc::new(reqwest::Client::new()); + // Startup cleanup — runs once, before the poller and uploader + // spawn. Keeps the client a lean ephemeral microphone: + // - markers older than 72 h (and server-confirmed) drop out, + // - orphan audio older than 24 h is reaped (patient-data + // minimization; a file that never found its sidecar will + // never be uploaded). + { + let store = case_store.clone(); + let pending = pending_dir.clone(); + runtime.block_on(async move { + let marker_cutoff = + time::OffsetDateTime::now_utc() - time::Duration::hours(72); + match store.cleanup_stale(marker_cutoff).await { + Ok(n) if n > 0 => info!(count = n, "startup cleanup: stale markers removed"), + Ok(_) => {} + Err(e) => warn!(error = %e, "startup cleanup: marker sweep failed"), + } + let audio_cutoff = std::time::SystemTime::now() + - std::time::Duration::from_secs(24 * 3600); + let n = cleanup_orphan_audio(&pending, audio_cutoff).await; + if n > 0 { + info!(count = n, "startup cleanup: orphan audio removed"); + } + }); + } + let (state, uploader, upload_events, poller) = if let Some(cfg) = &config { let (up, rx, pol) = spawn_background( &runtime, diff --git a/client-desktop/src/uploader.rs b/client-desktop/src/uploader.rs index 4712d90..941080a 100644 --- a/client-desktop/src/uploader.rs +++ b/client-desktop/src/uploader.rs @@ -194,6 +194,14 @@ async fn upload_with_retry( return; } Err(ErrorKind::Terminal(reason)) => { + // Terminal = configuration or operator error (wrong API + // key, payload too big, etc.). No amount of retries will + // help, and keeping sensitive audio on the client "just + // in case" would violate the data-minimization rule + // (clients are ephemeral microphones, not archives). + // Delete both artefacts and surface the failure to the UI. + let _ = tokio::fs::remove_file(&upload.file).await; + let _ = tokio::fs::remove_file(&sidecar_path_for(&upload.file)).await; let _ = events_tx.send(UploadEvent::Failed { case_id: upload.case_id, reason, @@ -411,8 +419,14 @@ mod tests { assert!(matches!(rx.recv().await, Some(UploadEvent::Succeeded { .. }))); } + /// Terminal failures (401, 413, ...) indicate a config or operator + /// problem that retries cannot fix. The sensitive audio file must + /// be deleted immediately — keeping it locally "in case the user + /// fixes the key" would leak patient data indefinitely for a + /// situation that in practice is rarely recoverable without manual + /// intervention anyway. #[tokio::test] - async fn terminal_401_keeps_files_and_does_not_retry() { + async fn terminal_401_deletes_files_and_does_not_retry() { let server = MockServer::start().await; Mock::given(method("POST")) .and(wpath(UPLOAD_PATH)) @@ -434,7 +448,14 @@ mod tests { 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!( + !upload.file.exists(), + "m4a must be deleted on terminal failure" + ); + assert!( + !sidecar_path_for(&upload.file).exists(), + "sidecar must be deleted on terminal failure" + ); assert!(matches!(rx.recv().await, Some(UploadEvent::Started(_)))); let Some(UploadEvent::Failed { will_retry, .. }) = rx.recv().await else { diff --git a/doctate-client-core/Cargo.toml b/doctate-client-core/Cargo.toml index 8875c19..09057d8 100644 --- a/doctate-client-core/Cargo.toml +++ b/doctate-client-core/Cargo.toml @@ -19,3 +19,4 @@ thiserror = "1" tempfile = "3" tokio = { workspace = true, features = ["test-util"] } wiremock = "0.6" +filetime = "0.2" diff --git a/doctate-client-core/src/case_store.rs b/doctate-client-core/src/case_store.rs index be205e2..8abfd37 100644 --- a/doctate-client-core/src/case_store.rs +++ b/doctate-client-core/src/case_store.rs @@ -194,6 +194,41 @@ impl CaseStore { Ok(()) } + /// Drop markers whose last activity is older than `cutoff` **and** + /// that the server has already confirmed (`synced_to_server = true`). + /// Pending/unsynced markers are preserved — they guard uploads that + /// have not yet reached the server, and their loss would silently + /// erase patient recordings from the client's awareness. + /// + /// Returns the number of markers removed. Intended as an + /// app-startup pass so the data-minimization rule holds: metadata + /// ages out after a few days even without an explicit UI action. + pub async fn cleanup_stale( + &self, + cutoff: OffsetDateTime, + ) -> Result { + let cutoff_str = format_utc(cutoff)?; + let mut state = self.state.lock().await; + let to_remove: Vec = state + .iter() + .filter(|(_, m)| m.synced_to_server && m.last_activity_at < cutoff_str) + .map(|(id, _)| *id) + .collect(); + let count = to_remove.len(); + for id in &to_remove { + let path = self.dir.join(format!("{id}.json")); + match tokio::fs::remove_file(&path).await { + Ok(_) => tracing::info!(case_id = %id, path = %path.display(), "stale marker removed"), + Err(e) => warn!(case_id = %id, error = %e, "failed to remove stale marker"), + } + state.remove(id); + } + if count > 0 { + self.broadcast(&state); + } + Ok(count) + } + /// Remove every marker from memory and disk. Used on config change /// (new API key = potentially different user). pub async fn clear_all(&self) -> Result<(), CaseStoreError> { @@ -546,6 +581,72 @@ mod tests { assert_eq!(list[0].last_activity_at, "2026-04-18T10:00:00Z"); } + #[tokio::test] + async fn cleanup_stale_removes_old_synced_markers() { + let tmp = TempDir::new().unwrap(); + let (store, _rx) = CaseStore::open(tmp.path().to_owned()).await.unwrap(); + let old_synced = Uuid::new_v4(); + let young_synced = Uuid::new_v4(); + let old_unsynced = Uuid::new_v4(); + + // old_synced: created 4 days ago, then merged in → synced=true, old + store.create_local(old_synced, t("2026-04-14T10:00:00Z")).await.unwrap(); + store + .merge_server_snapshot(&server_snapshot(vec![OnelinerEntry { + case_id: old_synced.to_string(), + oneliner: Some("old".into()), + created_at: "2026-04-14T10:00:00Z".into(), + last_recording_at: Some("2026-04-14T10:00:00Z".into()), + updated_at: Some("2026-04-14T10:00:00Z".into()), + }])) + .await + .unwrap(); + + // young_synced: created today, synced + store.create_local(young_synced, t("2026-04-18T10:00:00Z")).await.unwrap(); + store + .merge_server_snapshot(&server_snapshot(vec![OnelinerEntry { + case_id: young_synced.to_string(), + oneliner: Some("young".into()), + created_at: "2026-04-18T10:00:00Z".into(), + last_recording_at: Some("2026-04-18T10:00:00Z".into()), + updated_at: Some("2026-04-18T10:00:00Z".into()), + }])) + .await + .unwrap(); + + // old_unsynced: created 4 days ago, NEVER merged in → pending upload + store.create_local(old_unsynced, t("2026-04-14T10:00:00Z")).await.unwrap(); + + // Cutoff: 72h before the "today" markers, i.e. 2026-04-15. + let cutoff = t("2026-04-15T10:00:00Z"); + let removed = store.cleanup_stale(cutoff).await.unwrap(); + + assert_eq!(removed, 1, "only old+synced must be removed"); + let list = store.list().await; + let ids: Vec = list.iter().map(|m| m.case_id).collect(); + assert!(ids.contains(&young_synced), "young synced must stay"); + assert!( + ids.contains(&old_unsynced), + "old unsynced must stay (pending upload)" + ); + assert!(!ids.contains(&old_synced), "old synced must be gone"); + + // Disk: old_synced's file deleted. + let old_path = tmp.path().join(format!("{old_synced}.json")); + assert!(!old_path.exists(), "marker file for old_synced must be deleted"); + } + + #[tokio::test] + async fn cleanup_stale_is_noop_when_nothing_old() { + let tmp = TempDir::new().unwrap(); + let (store, _rx) = CaseStore::open(tmp.path().to_owned()).await.unwrap(); + store.create_local(Uuid::new_v4(), t("2026-04-18T10:00:00Z")).await.unwrap(); + let removed = store.cleanup_stale(t("2026-01-01T00:00:00Z")).await.unwrap(); + assert_eq!(removed, 0); + assert_eq!(store.list().await.len(), 1); + } + #[tokio::test] async fn sorted_newest_first() { let tmp = TempDir::new().unwrap(); diff --git a/doctate-client-core/src/lib.rs b/doctate-client-core/src/lib.rs index f808221..2d9fcb5 100644 --- a/doctate-client-core/src/lib.rs +++ b/doctate-client-core/src/lib.rs @@ -13,8 +13,10 @@ pub mod case_store; pub mod oneliner_poller; +pub mod pending_cleanup; pub mod snapshot_cache; pub use case_store::{CaseMarker, CaseStore, CaseStoreError}; pub use oneliner_poller::{OnelinerPoller, PollerConfig}; +pub use pending_cleanup::cleanup_orphan_audio; pub use snapshot_cache::{CachedSnapshot, SnapshotCache}; diff --git a/doctate-client-core/src/pending_cleanup.rs b/doctate-client-core/src/pending_cleanup.rs new file mode 100644 index 0000000..5f239d5 --- /dev/null +++ b/doctate-client-core/src/pending_cleanup.rs @@ -0,0 +1,219 @@ +//! Orphan reaping for the client's pending-upload directory. +//! +//! The uploader happy-path deletes `.m4a` + `.meta.json` pairs after +//! ACK. A crash between recorder-stop and sidecar-write, or a crash +//! between ACK and the two `remove_file` calls, can leave **orphans** +//! behind: +//! +//! - `*.m4a` without `*.meta.json` (or vice versa): the uploader's +//! `scan_pending_dir` silently skips these on next launch, so they +//! would accumulate indefinitely without explicit cleanup. +//! - `*.json.tmp` / `*.m4a.tmp` leftovers from interrupted atomic +//! writes: pure garbage, always safe to remove. +//! +//! Orphan audio is sensitive — clients must not hoard patient +//! recordings that never made it to the server. This module reaps them +//! on a caller-supplied age threshold. The caller passes an `age_cutoff: +//! SystemTime`; everything whose mtime is strictly older is deleted. +//! `.tmp` leftovers bypass the cutoff (they are never legitimate work +//! in progress at app startup — the lock file guarantees exclusivity). + +use std::path::Path; +use std::time::SystemTime; + +use tracing::{info, warn}; + +/// Reap orphan artefacts in `pending_dir`. Returns the number of files +/// deleted. Missing directory is treated as "nothing to do". +pub async fn cleanup_orphan_audio(pending_dir: &Path, age_cutoff: SystemTime) -> usize { + let Ok(mut entries) = tokio::fs::read_dir(pending_dir).await else { + return 0; + }; + + // First pass: collect filenames so we can answer "does the sibling + // exist?" without doing N separate stat calls. + let mut names: Vec = Vec::new(); + while let Ok(Some(e)) = entries.next_entry().await { + if let Some(n) = e.file_name().to_str() { + names.push(n.to_owned()); + } + } + let has_name = |n: &str| names.iter().any(|x| x == n); + + let mut deleted = 0usize; + for name in &names { + let path = pending_dir.join(name); + + // Atomic-write leftovers: always delete, no age check. A running + // process would hold the single-instance lock, so `.tmp` at + // startup is by definition stale. + if name.ends_with(".m4a.tmp") || name.ends_with(".meta.json.tmp") { + if let Err(e) = tokio::fs::remove_file(&path).await { + warn!(path = %path.display(), error = %e, "failed to remove tmp leftover"); + } else { + info!(path = %path.display(), "tmp leftover removed"); + deleted += 1; + } + continue; + } + + // Candidate: orphan `.m4a` — only delete if older than cutoff. + if let Some(stem) = name.strip_suffix(".m4a") { + let sibling = format!("{stem}.meta.json"); + if has_name(&sibling) { + continue; // paired, leave for the uploader + } + if mtime_older_than(&path, age_cutoff).await { + if let Err(e) = tokio::fs::remove_file(&path).await { + warn!(path = %path.display(), error = %e, "failed to remove orphan m4a"); + } else { + info!(path = %path.display(), "orphan m4a removed (no sidecar)"); + deleted += 1; + } + } + continue; + } + + // Candidate: orphan sidecar — only delete if older than cutoff. + if let Some(stem) = name.strip_suffix(".meta.json") { + let sibling = format!("{stem}.m4a"); + if has_name(&sibling) { + continue; + } + if mtime_older_than(&path, age_cutoff).await { + if let Err(e) = tokio::fs::remove_file(&path).await { + warn!(path = %path.display(), error = %e, "failed to remove orphan sidecar"); + } else { + info!(path = %path.display(), "orphan sidecar removed (no m4a)"); + deleted += 1; + } + } + continue; + } + } + + deleted +} + +async fn mtime_older_than(path: &Path, cutoff: SystemTime) -> bool { + match tokio::fs::metadata(path).await { + Ok(meta) => match meta.modified() { + Ok(mtime) => mtime < cutoff, + Err(_) => false, + }, + Err(_) => false, + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::time::Duration; + use tempfile::TempDir; + + async fn touch_with_mtime(path: &Path, mtime: SystemTime) { + tokio::fs::write(path, b"x").await.unwrap(); + filetime::set_file_mtime(path, filetime::FileTime::from_system_time(mtime)).unwrap(); + } + + fn now() -> SystemTime { + SystemTime::now() + } + + #[tokio::test] + async fn missing_directory_returns_zero() { + let tmp = TempDir::new().unwrap(); + let n = cleanup_orphan_audio(&tmp.path().join("nope"), now()).await; + assert_eq!(n, 0); + } + + #[tokio::test] + async fn paired_files_are_preserved() { + let tmp = TempDir::new().unwrap(); + let stem = "uuid_2026-04-18T08-00-00Z"; + touch_with_mtime(&tmp.path().join(format!("{stem}.m4a")), now()).await; + touch_with_mtime(&tmp.path().join(format!("{stem}.meta.json")), now()).await; + // Cutoff in the future → any orphan would be deleted — but + // these are paired, must survive. + let cutoff = now() + Duration::from_secs(3600); + let n = cleanup_orphan_audio(tmp.path(), cutoff).await; + assert_eq!(n, 0); + assert!(tmp.path().join(format!("{stem}.m4a")).exists()); + assert!(tmp.path().join(format!("{stem}.meta.json")).exists()); + } + + #[tokio::test] + async fn young_orphan_m4a_is_preserved() { + let tmp = TempDir::new().unwrap(); + let m4a = tmp.path().join("uuid_young.m4a"); + touch_with_mtime(&m4a, now()).await; + // cutoff = 1h ago → file (mtime=now) is younger → keep + let cutoff = now() - Duration::from_secs(3600); + assert_eq!(cleanup_orphan_audio(tmp.path(), cutoff).await, 0); + assert!(m4a.exists()); + } + + #[tokio::test] + async fn old_orphan_m4a_is_deleted() { + let tmp = TempDir::new().unwrap(); + let m4a = tmp.path().join("uuid_old.m4a"); + touch_with_mtime(&m4a, now() - Duration::from_secs(48 * 3600)).await; + let cutoff = now() - Duration::from_secs(24 * 3600); + let n = cleanup_orphan_audio(tmp.path(), cutoff).await; + assert_eq!(n, 1); + assert!(!m4a.exists(), "old orphan m4a must be gone"); + } + + #[tokio::test] + async fn old_orphan_sidecar_is_deleted() { + let tmp = TempDir::new().unwrap(); + let sc = tmp.path().join("uuid_orphan.meta.json"); + touch_with_mtime(&sc, now() - Duration::from_secs(48 * 3600)).await; + let cutoff = now() - Duration::from_secs(24 * 3600); + let n = cleanup_orphan_audio(tmp.path(), cutoff).await; + assert_eq!(n, 1); + assert!(!sc.exists()); + } + + #[tokio::test] + async fn tmp_leftovers_always_removed_regardless_of_age() { + let tmp = TempDir::new().unwrap(); + let young_tmp = tmp.path().join("fresh.meta.json.tmp"); + let m4a_tmp = tmp.path().join("interrupted.m4a.tmp"); + touch_with_mtime(&young_tmp, now()).await; + touch_with_mtime(&m4a_tmp, now()).await; + // Cutoff in the far past → age check would preserve everything. + // But tmp leftovers bypass age check. + let n = cleanup_orphan_audio(tmp.path(), now() - Duration::from_secs(10 * 86400)).await; + assert_eq!(n, 2); + assert!(!young_tmp.exists()); + assert!(!m4a_tmp.exists()); + } + + #[tokio::test] + async fn mix_of_pairs_orphans_and_tmp() { + let tmp = TempDir::new().unwrap(); + let keep_stem = "uuid_keep_2026-04-18"; + let orphan_stem = "uuid_orphan_old"; + + // Paired (keep) + touch_with_mtime(&tmp.path().join(format!("{keep_stem}.m4a")), now()).await; + touch_with_mtime(&tmp.path().join(format!("{keep_stem}.meta.json")), now()).await; + // Old orphan m4a (delete) + touch_with_mtime( + &tmp.path().join(format!("{orphan_stem}.m4a")), + now() - Duration::from_secs(48 * 3600), + ) + .await; + // tmp leftover (delete) + touch_with_mtime(&tmp.path().join("crash.m4a.tmp"), now()).await; + + let cutoff = now() - Duration::from_secs(24 * 3600); + let n = cleanup_orphan_audio(tmp.path(), cutoff).await; + assert_eq!(n, 2, "orphan + tmp must be removed, pair must stay"); + assert!(tmp.path().join(format!("{keep_stem}.m4a")).exists()); + assert!(tmp.path().join(format!("{keep_stem}.meta.json")).exists()); + assert!(!tmp.path().join(format!("{orphan_stem}.m4a")).exists()); + assert!(!tmp.path().join("crash.m4a.tmp").exists()); + } +}