fix(desktop): keep recordings on hard upload errors, reap ghost markers
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.
This commit is contained in:
@@ -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;
|
||||
});
|
||||
|
||||
@@ -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<Uuid>,
|
||||
cutoff: OffsetDateTime,
|
||||
) -> Result<usize, CaseStoreError> {
|
||||
let cutoff_str = format_utc(cutoff)?;
|
||||
let mut state = self.state.lock().await;
|
||||
let to_remove: Vec<Uuid> = 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<Uuid> = 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<Uuid> = 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<Uuid> = 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<Uuid> = 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);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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::<AckResponse>().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;
|
||||
}
|
||||
}
|
||||
assert!(saw_terminal, "expected terminal failure event");
|
||||
// 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;
|
||||
|
||||
// Files gone.
|
||||
assert!(!upload.file.exists());
|
||||
assert!(!sidecar_path_for(&upload.file).exists());
|
||||
// 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_retryable_failure,
|
||||
"expected at least one Failed{{will_retry:true}} event"
|
||||
);
|
||||
|
||||
// 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
|
||||
|
||||
@@ -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<Uuid> = 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;
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user