From a142725a8a421c32689f8b063a0714735e27705c Mon Sep 17 00:00:00 2001 From: Brummel Date: Tue, 5 May 2026 10:21:33 +0200 Subject: [PATCH] fix(desktop): keep recordings on hard upload errors, reap ghost markers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Mirror of the 2026-05-05 Wear-OS fix: port the canonical client sync spec (project_client_sync_semantics.md) into the Rust desktop client. Before this change, server_sync::run_loop had a `UploadOutcome::Terminal` branch that deleted the m4a + sidecar pair on any non-2xx + non-5xx response (401, 413, 415, parse drift, audio read failures) and never called mark_synced — so the marker stayed `synced_to_server=false` as a non-removable ghost in the desktop UI. Same shape as the Wear-OS bug. Changes: - `UploadOutcome` reduced to two variants (`Succeeded` | `Transient`). The Terminal arm in `run_loop` is gone; `delete_pair` only fires from the success path. - `post_upload` now classifies every failure mode as `Transient`: 4xx incl. 401, parse-ack drift on 200, file-read errors, multipart build errors, and network errors. Backoff retries forever. - New `CaseStore::cleanup_orphaned_unsynced(pending_case_ids, cutoff)`: removes markers that are unsynced AND have no waiting upload pair AND are older than a small grace window. Self-heals pre-spec ghosts. - `run_startup_cleanup` gains a third sweep that calls `cleanup_orphaned_unsynced` with `DEFAULT_ORPHAN_MARKER_GRACE = 5 min`, matching the Wear-OS StartupCleanup contract. - Tests inverted/added: `post_upload_transient_on_401`, `post_upload_transient_on_2xx_unparseable_body`, `former_terminal_401_keeps_pair_and_retries`, four new `cleanup_orphaned_unsynced_*` cases, plus `sweeps_orphan_unsynced_marker` and `keeps_unsynced_marker_with_pending_pair` integration tests. Followup (out of scope): `UploadEvent::Failed.will_retry` can now only be `true` and is vestigial. Removing it touches every Failed-event consumer in app.rs/UI; left for a separate cleanup commit. --- clients/desktop/src/app.rs | 4 +- clients/desktop/src/case_store.rs | 157 +++++++++++++++++++++++++++++ clients/desktop/src/server_sync.rs | 139 ++++++++++++++++--------- clients/desktop/src/startup.rs | 140 +++++++++++++++++++++++-- 4 files changed, 387 insertions(+), 53 deletions(-) diff --git a/clients/desktop/src/app.rs b/clients/desktop/src/app.rs index fb33b8c..879a2b9 100644 --- a/clients/desktop/src/app.rs +++ b/clients/desktop/src/app.rs @@ -19,7 +19,8 @@ use crate::footer_status::{FooterStatus, pick_footer_status}; use crate::server_sync::{ServerSync, SyncConfig, WorkerSnapshot, WorkerState}; use crate::snapshot_cache::SnapshotCache; use crate::startup::{ - DEFAULT_MARKER_RETENTION, DEFAULT_ORPHAN_AUDIO_RETENTION, run_startup_cleanup, + DEFAULT_MARKER_RETENTION, DEFAULT_ORPHAN_AUDIO_RETENTION, DEFAULT_ORPHAN_MARKER_GRACE, + run_startup_cleanup, }; use crate::upload::{PendingUpload, UploadEvent, write_sidecar}; use doctate_common::join_url; @@ -156,6 +157,7 @@ impl DoctateApp { &pending, DEFAULT_MARKER_RETENTION, DEFAULT_ORPHAN_AUDIO_RETENTION, + DEFAULT_ORPHAN_MARKER_GRACE, ) .await; }); diff --git a/clients/desktop/src/case_store.rs b/clients/desktop/src/case_store.rs index de622ec..bd789e2 100644 --- a/clients/desktop/src/case_store.rs +++ b/clients/desktop/src/case_store.rs @@ -326,6 +326,59 @@ impl CaseStore { Ok(count) } + /// Sweep "ghost" markers: unsynced markers whose recording is no + /// longer waiting on disk and whose `last_activity_at` is older than + /// `cutoff` are removed. A marker qualifies when ALL of these hold: + /// 1. `synced_to_server == false` (server doesn't know it), + /// 2. `case_id` is **not** in `pending_case_ids` (no m4a/sidecar + /// pair is currently waiting to be uploaded), and + /// 3. `last_activity_at < cutoff` (a small grace window — usually + /// 5 minutes — protects freshly-written markers whose pair is + /// still being committed to disk). + /// + /// Without this sweep, a `Transient`-classified upload that the + /// server later refuses with a non-2xx wouldn't actually leave a + /// ghost (the spec keeps the pair retrying); but markers can also + /// orphan from older, pre-spec runs whose Terminal-branch erased + /// the audio without finalizing the marker. Running the sweep on + /// every startup self-heals those. + pub async fn cleanup_orphaned_unsynced( + &self, + pending_case_ids: &HashSet, + cutoff: OffsetDateTime, + ) -> Result { + let cutoff_str = format_utc(cutoff)?; + let mut state = self.state.lock().await; + let to_remove: Vec = state + .iter() + .filter(|(id, m)| { + !m.synced_to_server + && !pending_case_ids.contains(id) + && 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(_) => info!( + case_id = %id, + "orphaned unsynced marker removed (audio gone, server unaware)" + ), + Err(e) => warn!( + case_id = %id, error = %e, + "failed to remove orphaned unsynced marker" + ), + } + state.remove(id); + } + if count > 0 { + self.broadcast(&state); + } + Ok(count) + } + /// 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 @@ -1122,4 +1175,108 @@ mod tests { assert!(result.is_ok(), "no-op must not error"); assert!(store.list().await.is_empty(), "no marker should be created"); } + + /// The exact regression contract for the canonical client sync spec: + /// a marker with `synced_to_server == false`, no waiting upload pair, + /// and `last_activity_at` older than the grace cutoff is a "ghost". + /// Same shape as the Wear-OS `reaps_orphaned_unsynced_marker_when_audio_is_gone` + /// test. + #[tokio::test] + async fn cleanup_orphaned_unsynced_reaps_unsynced_without_pending_pair() { + let tmp = TempDir::new().unwrap(); + let (store, _rx) = CaseStore::open(tmp.path().to_owned()).await.unwrap(); + let ghost = Uuid::new_v4(); + // Marker created an hour ago, never synced, no pair on disk. + store + .create_local(ghost, t("2026-04-18T09:00:00Z")) + .await + .unwrap(); + + let pending: HashSet = HashSet::new(); + // Cutoff: 5 min before "now" (use a fixed point past the marker). + let cutoff = t("2026-04-18T09:55:00Z"); + let removed = store + .cleanup_orphaned_unsynced(&pending, cutoff) + .await + .unwrap(); + + assert_eq!(removed, 1, "ghost must be removed"); + assert!(store.list().await.is_empty()); + let path = tmp.path().join(format!("{ghost}.json")); + assert!(!path.exists(), "marker file must be deleted"); + } + + /// A marker WITH a waiting upload pair is a real pending upload — + /// must survive any age threshold. The `pending_case_ids` set is the + /// authoritative signal for "this case is mid-flight". + #[tokio::test] + async fn cleanup_orphaned_unsynced_keeps_unsynced_with_pending_pair() { + let tmp = TempDir::new().unwrap(); + let (store, _rx) = CaseStore::open(tmp.path().to_owned()).await.unwrap(); + let id = Uuid::new_v4(); + store + .create_local(id, t("2026-04-18T09:00:00Z")) + .await + .unwrap(); + + let mut pending: HashSet = HashSet::new(); + pending.insert(id); + let cutoff = t("2026-04-18T09:55:00Z"); + let removed = store + .cleanup_orphaned_unsynced(&pending, cutoff) + .await + .unwrap(); + + assert_eq!(removed, 0); + assert_eq!(store.list().await.len(), 1); + } + + /// A freshly-written marker (younger than the grace cutoff) might + /// still be in the middle of having its pair flushed — must survive. + #[tokio::test] + async fn cleanup_orphaned_unsynced_keeps_fresh_unsynced_within_grace() { + let tmp = TempDir::new().unwrap(); + let (store, _rx) = CaseStore::open(tmp.path().to_owned()).await.unwrap(); + let id = Uuid::new_v4(); + store + .create_local(id, t("2026-04-18T10:00:00Z")) + .await + .unwrap(); + + let pending: HashSet = HashSet::new(); + // Cutoff = same instant as the marker → marker is NOT strictly older. + let cutoff = t("2026-04-18T10:00:00Z"); + let removed = store + .cleanup_orphaned_unsynced(&pending, cutoff) + .await + .unwrap(); + + assert_eq!(removed, 0); + assert_eq!(store.list().await.len(), 1); + } + + /// Synced markers are not in scope for this sweep — they are the + /// `cleanup_stale` sweep's territory. Even an old synced marker + /// without a pending pair must survive `cleanup_orphaned_unsynced`. + #[tokio::test] + async fn cleanup_orphaned_unsynced_keeps_synced_markers() { + let tmp = TempDir::new().unwrap(); + let (store, _rx) = CaseStore::open(tmp.path().to_owned()).await.unwrap(); + let id = Uuid::new_v4(); + store + .create_local(id, t("2026-04-18T09:00:00Z")) + .await + .unwrap(); + store.mark_synced(id).await.unwrap(); + + let pending: HashSet = HashSet::new(); + let cutoff = t("2026-04-18T09:55:00Z"); + let removed = store + .cleanup_orphaned_unsynced(&pending, cutoff) + .await + .unwrap(); + + assert_eq!(removed, 0, "synced markers belong to cleanup_stale"); + assert_eq!(store.list().await.len(), 1); + } } diff --git a/clients/desktop/src/server_sync.rs b/clients/desktop/src/server_sync.rs index 9f9ab1d..0fa26f7 100644 --- a/clients/desktop/src/server_sync.rs +++ b/clients/desktop/src/server_sync.rs @@ -13,9 +13,12 @@ //! disk. //! - **Backoff is a local `Duration`.** Exponential, resets on success, //! capped at `MAX_BACKOFF`. Not persisted across restarts. -//! - **Terminal upload failures delete the file.** 401/413/etc. will -//! never succeed; keeping the audio around would violate the -//! data-minimization rule. +//! - **Only `Success` deletes the file.** Per the canonical client sync +//! spec, every non-2xx outcome (401, 413, 5xx, network failure, parse +//! drift) is `Transient`: the worker retries forever with backoff and +//! never destroys locally-recorded audio. A persistently bad config +//! shows up as "stuck pending" in the UI rather than as silent data +//! loss. use std::collections::HashSet; use std::path::{Path, PathBuf}; @@ -163,11 +166,14 @@ impl Drop for ServerSync { } } -/// Outcome classification for a single upload attempt. +/// Outcome classification for a single upload attempt. Per the canonical +/// client sync spec there are only two states: a 2xx ACK (success) or a +/// retry-worthy hiccup (transient). There is **no** "terminal" outcome — +/// a 401, 413, parse drift, or read failure all keep the audio on disk +/// and the worker keeps trying. pub(crate) enum UploadOutcome { Succeeded(AckStatus), Transient(String), - Terminal(String), } /// Outcome classification for a single poll cycle. Symmetric to the old @@ -281,17 +287,6 @@ async fn run_loop( tokio::time::sleep(backoff).await; backoff = (backoff * 2).min(MAX_BACKOFF); } - UploadOutcome::Terminal(reason) => { - warn!(case_id = %upload.case_id, reason = %reason, "upload failed terminally — deleting files"); - delete_pair(&upload.file).await; - last_failure = Some(Instant::now()); - backoff = INITIAL_BACKOFF; - let _ = events_tx.send(UploadEvent::Failed { - case_id: upload.case_id, - reason, - will_retry: false, - }); - } } } else { publish( @@ -334,13 +329,19 @@ async fn run_loop( } } -/// Perform one upload attempt. Error classification follows the rules: -/// - I/O errors reading the audio file → Terminal (file is broken, retry -/// won't help). +/// Perform one upload attempt. Per the canonical client sync spec the +/// only success path is a 2xx with parseable ACK; everything else is +/// `Transient` and gets retried with backoff. Concretely: +/// - I/O reading the audio file → Transient (temporary lock, disk hiccup; +/// if the file is truly gone the orphan-cleanup sweeps catch it on the +/// next startup). +/// - multipart build failure → Transient (in practice unreachable with a +/// constant MIME, kept for spec consistency). /// - `reqwest::send` errors → Transient (network). /// - 2xx with parseable ACK → Succeeded. -/// - 408/429/5xx → Transient (server will likely recover). -/// - Other 4xx → Terminal (401, 413, 415, …; operator must fix). +/// - 2xx with unparseable body → Transient (server contract drift, fix +/// server-side; don't punish the client by destroying audio). +/// - Any non-2xx (4xx incl. 401/413/415, 5xx) → Transient. pub(crate) async fn post_upload( http: &reqwest::Client, config: &SyncConfig, @@ -348,7 +349,7 @@ pub(crate) async fn post_upload( ) -> UploadOutcome { let audio = match tokio::fs::read(&upload.file).await { Ok(bytes) => bytes, - Err(e) => return UploadOutcome::Terminal(format!("read audio file: {e}")), + Err(e) => return UploadOutcome::Transient(format!("read audio file: {e}")), }; let file_name = upload @@ -362,7 +363,7 @@ pub(crate) async fn post_upload( .mime_str(CONTENT_TYPE_AUDIO_MP4) { Ok(p) => p, - Err(e) => return UploadOutcome::Terminal(format!("build multipart: {e}")), + Err(e) => return UploadOutcome::Transient(format!("build multipart: {e}")), }; let form = reqwest::multipart::Form::new() @@ -387,14 +388,11 @@ pub(crate) async fn post_upload( if status.is_success() { match resp.json::().await { Ok(ack) => UploadOutcome::Succeeded(ack.status), - Err(e) => UploadOutcome::Terminal(format!("parse ack: {e}")), + Err(e) => UploadOutcome::Transient(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(); - UploadOutcome::Transient(format!("HTTP {status}: {body}")) } else { let body = resp.text().await.unwrap_or_default(); - UploadOutcome::Terminal(format!("HTTP {status}: {body}")) + UploadOutcome::Transient(format!("HTTP {status}: {body}")) } } @@ -586,8 +584,12 @@ mod tests { )); } + /// Per the canonical client sync spec, a 401 (typically expired API + /// key) must NOT destroy the audio. Retrying after the operator fixes + /// the key is the intended path. Same applies to all other former + /// "terminal" 4xx codes (413, 415, …). #[tokio::test] - async fn post_upload_terminal_on_401() { + async fn post_upload_transient_on_401() { let server = MockServer::start().await; Mock::given(method("POST")) .respond_with(ResponseTemplate::new(401).set_body_string("nope")) @@ -600,7 +602,28 @@ mod tests { let http = reqwest::Client::new(); assert!(matches!( post_upload(&http, &cfg(&server), &upload).await, - UploadOutcome::Terminal(_) + UploadOutcome::Transient(_) + )); + } + + /// Server contract drift (200 OK but the JSON body has the wrong + /// shape) is a server-side issue. Keeping the audio gives the user + /// a chance to receive it once the server is fixed. + #[tokio::test] + async fn post_upload_transient_on_2xx_unparseable_body() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(200).set_body_string("not json")) + .mount(&server) + .await; + + let tmp = TempDir::new().unwrap(); + let upload = sample_pending(tmp.path(), Uuid::new_v4(), "2026-04-18T10-32-00Z"); + + let http = reqwest::Client::new(); + assert!(matches!( + post_upload(&http, &cfg(&server), &upload).await, + UploadOutcome::Transient(_) )); } @@ -805,14 +828,21 @@ mod tests { // Mock expectations are checked on `server` drop at end of test. } - /// Terminal 401: file + sidecar must be deleted, no retry loop. + /// Per the canonical client sync spec, a 4xx-class server response + /// (e.g. 401 expired key) must NOT destroy the audio. The classifier + /// returns `Transient`, so the worker behaves like it would on a 5xx: + /// the pair stays on disk, backoff kicks in, retries continue. This + /// is the same regression contract Wear OS captures with its own + /// `former_terminal_code_keeps_pair_and_retries` test. #[tokio::test] - async fn terminal_401_deletes_files_no_retry() { + async fn former_terminal_401_keeps_pair_and_retries() { let server = MockServer::start().await; Mock::given(method("POST")) .and(wpath(UPLOAD_PATH)) .respond_with(ResponseTemplate::new(401).set_body_string("nope")) - .expect(1) + // At least 2 POSTs proves the worker did not abandon the pair + // after the first 401. + .expect(2..) .mount(&server) .await; Mock::given(method("GET")) @@ -835,24 +865,41 @@ mod tests { pending_dir.clone(), ); - // Expect a Failed{will_retry:false} event. - let mut saw_terminal = false; - for _ in 0..10 { - if let Ok(Some(UploadEvent::Failed { - will_retry: false, .. - })) = tokio::time::timeout(Duration::from_millis(500), events.recv()).await - { - saw_terminal = true; - break; + // INITIAL_BACKOFF is 2 s and doubles after each failure, so 5 s + // of real time gives the worker time for two POSTs (t=0 and + // t=2s, plus a margin). We rely on wiremock's `.expect(2..)` + // for the actual count assertion at server drop. + tokio::time::sleep(Duration::from_secs(5)).await; + + // While we're at it, drain the event stream and verify the new + // contract: every Failed event must carry will_retry=true. + let mut saw_retryable_failure = false; + while let Ok(maybe_ev) = + tokio::time::timeout(Duration::from_millis(50), events.recv()).await + { + match maybe_ev { + Some(UploadEvent::Failed { will_retry, .. }) => { + assert!(will_retry, "must never emit will_retry:false anymore"); + saw_retryable_failure = true; + } + Some(UploadEvent::Succeeded { .. }) => panic!("401 must never report success"), + Some(UploadEvent::Started(_)) | None => {} } } - assert!(saw_terminal, "expected terminal failure event"); + assert!( + saw_retryable_failure, + "expected at least one Failed{{will_retry:true}} event" + ); - // Files gone. - assert!(!upload.file.exists()); - assert!(!sidecar_path_for(&upload.file).exists()); + // Files survived. + assert!(upload.file.exists(), "m4a must remain on disk after 401"); + assert!( + sidecar_path_for(&upload.file).exists(), + "sidecar must remain on disk after 401" + ); drop(sync); + // wiremock's `.expect(2..)` is verified on `server` drop. } /// Kick wakes the worker out of the idle sleep. With a long diff --git a/clients/desktop/src/startup.rs b/clients/desktop/src/startup.rs index 9d76943..e031a6e 100644 --- a/clients/desktop/src/startup.rs +++ b/clients/desktop/src/startup.rs @@ -5,21 +5,29 @@ //! must not hoard patient-sensitive audio or metadata past the //! dictation-day horizon. //! -//! Two retention horizons exist: +//! Three retention horizons exist: //! - **Audio** (sensitive): 24 h by default. Orphan m4a/sidecar files //! that never reached the server get deleted. -//! - **Markers** (non-sensitive: UUIDs + oneliners): 72 h by default. -//! Long enough to span a weekend; short enough that a stolen device -//! can't reconstruct a long case history. +//! - **Synced markers** (non-sensitive: UUIDs + oneliners): 72 h by +//! default. Long enough to span a weekend; short enough that a stolen +//! device can't reconstruct a long case history. +//! - **Orphan unsynced markers**: 5 min grace by default. An unsynced +//! marker without a corresponding pending pair is the symptom of a +//! pre-spec ghost (audio destroyed by the now-removed Terminal branch +//! without finalizing the marker). The short grace protects markers +//! whose pair is still being flushed to disk. +use std::collections::HashSet; use std::path::Path; use std::time::{Duration, SystemTime}; use time::OffsetDateTime; use tracing::{info, warn}; +use uuid::Uuid; use crate::case_store::CaseStore; use crate::pending_cleanup::cleanup_orphan_audio; +use crate::upload::scan_pending_dir; /// Default retention for case-markers on disk. Confirmed-by-server /// markers older than this are pruned; unsynced markers are kept @@ -31,17 +39,25 @@ pub const DEFAULT_MARKER_RETENTION: Duration = Duration::from_secs(72 * 3600); /// older than this horizon. pub const DEFAULT_ORPHAN_AUDIO_RETENTION: Duration = Duration::from_secs(24 * 3600); +/// Grace window before a marker that is `synced_to_server=false` AND +/// has no pending pair on disk is considered a ghost. Short by design: +/// the only legitimate reason to be in that state momentarily is the +/// brief instant between writing the marker and flushing the m4a + +/// sidecar pair. +pub const DEFAULT_ORPHAN_MARKER_GRACE: Duration = Duration::from_secs(5 * 60); + /// Run the one-shot startup cleanup. Intended to be called once at app /// launch, before the server-sync worker spawns. /// -/// Logs each outcome via `tracing`; no error bubbles up because both -/// sweeps are best-effort (a failing sweep shouldn't prevent the app +/// Logs each outcome via `tracing`; no error bubbles up because every +/// sweep is best-effort (a failing sweep shouldn't prevent the app /// from starting). pub async fn run_startup_cleanup( store: &CaseStore, pending_dir: &Path, marker_retention: Duration, orphan_audio_retention: Duration, + orphan_marker_grace: Duration, ) { let marker_cutoff = OffsetDateTime::now_utc() - time::Duration::seconds(marker_retention.as_secs() as i64); @@ -56,6 +72,27 @@ pub async fn run_startup_cleanup( if n > 0 { info!(count = n, "startup cleanup: orphan audio removed"); } + + // Third sweep: orphan unsynced markers whose pair is gone. The + // pending_case_ids set is the authoritative "this case still has + // work in flight" signal — same idea as Wear OS's StartupCleanup.run. + let pending_case_ids: HashSet = scan_pending_dir(pending_dir) + .into_iter() + .map(|u| u.case_id) + .collect(); + let orphan_marker_cutoff = + OffsetDateTime::now_utc() - time::Duration::seconds(orphan_marker_grace.as_secs() as i64); + match store + .cleanup_orphaned_unsynced(&pending_case_ids, orphan_marker_cutoff) + .await + { + Ok(n) if n > 0 => info!( + count = n, + "startup cleanup: orphan unsynced markers removed" + ), + Ok(_) => {} + Err(e) => warn!(error = %e, "startup cleanup: orphan marker sweep failed"), + } } #[cfg(test)] @@ -116,6 +153,7 @@ mod tests { &pending_dir, Duration::from_secs(3600), // markers older than 1h: nuke Duration::from_secs(3600), // orphans older than 1h: nuke + DEFAULT_ORPHAN_MARKER_GRACE, ) .await; @@ -130,6 +168,95 @@ mod tests { assert!(fresh_m4a.exists(), "fresh paired m4a must remain"); } + /// Regression contract: a `synced_to_server=false` marker without a + /// corresponding pair on disk is the exact symptom the Wear-OS ghost + /// fix targeted. The third sweep must catch it on the desktop too. + #[tokio::test] + async fn sweeps_orphan_unsynced_marker() { + let tmp = TempDir::new().unwrap(); + let cases_dir = tmp.path().join("cases"); + let pending_dir = tmp.path().join("pending"); + std::fs::create_dir_all(&pending_dir).unwrap(); + + // Seed an unsynced marker dated yesterday, no pair on disk. + let ghost_id = { + let (store, _rx) = CaseStore::open(cases_dir.clone()).await.unwrap(); + let id = Uuid::new_v4(); + let then = OffsetDateTime::parse( + "2020-01-01T00:00:00Z", + &time::format_description::well_known::Rfc3339, + ) + .unwrap(); + store.create_local(id, then).await.unwrap(); + // Note: no mark_synced — this is the ghost shape. + id + }; + + let (store, _rx) = CaseStore::open(cases_dir).await.unwrap(); + run_startup_cleanup( + &store, + &pending_dir, + DEFAULT_MARKER_RETENTION, + DEFAULT_ORPHAN_AUDIO_RETENTION, + Duration::from_secs(60), // 1-min grace, the marker is years old + ) + .await; + + assert!( + store.list().await.is_empty(), + "ghost marker must be swept by the orphan-unsynced sweep" + ); + let path = tmp.path().join(format!("cases/{ghost_id}.json")); + assert!(!path.exists(), "marker file must be deleted on disk"); + } + + /// Negative case for the third sweep: a fresh unsynced marker with + /// its pair on disk is a real pending upload — must survive. + #[tokio::test] + async fn keeps_unsynced_marker_with_pending_pair() { + let tmp = TempDir::new().unwrap(); + let cases_dir = tmp.path().join("cases"); + let pending_dir = tmp.path().join("pending"); + std::fs::create_dir_all(&pending_dir).unwrap(); + + let id = Uuid::new_v4(); + // Create marker. + { + let (store, _rx) = CaseStore::open(cases_dir.clone()).await.unwrap(); + store + .create_local(id, OffsetDateTime::now_utc()) + .await + .unwrap(); + } + // Create the matching pair. + let stem = format!("{id}_2026-04-18T10-00-00Z"); + let m4a = pending_dir.join(format!("{stem}.m4a")); + let upload = PendingUpload { + case_id: id, + recorded_at: "2026-04-18T10:00:00Z".into(), + file: m4a.clone(), + }; + std::fs::write(&m4a, b"audio").unwrap(); + write_sidecar(&upload).unwrap(); + + let (store, _rx) = CaseStore::open(cases_dir).await.unwrap(); + run_startup_cleanup( + &store, + &pending_dir, + DEFAULT_MARKER_RETENTION, + DEFAULT_ORPHAN_AUDIO_RETENTION, + Duration::from_secs(60), + ) + .await; + + assert_eq!( + store.list().await.len(), + 1, + "unsynced marker with a pending pair must survive" + ); + assert!(m4a.exists()); + } + #[tokio::test] async fn no_op_on_clean_directories() { let tmp = TempDir::new().unwrap(); @@ -144,6 +271,7 @@ mod tests { &pending_dir, DEFAULT_MARKER_RETENTION, DEFAULT_ORPHAN_AUDIO_RETENTION, + DEFAULT_ORPHAN_MARKER_GRACE, ) .await; }