From a8994fd84d51ffd0cc842f2512dba3966421ebb3 Mon Sep 17 00:00:00 2001 From: Brummel Date: Tue, 5 May 2026 09:52:29 +0200 Subject: [PATCH] fix(wear): keep recordings on hard upload errors, reap ghost markers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Treat every non-Success upload response as Transient so audio survives expired sessions, 4xx replies, and unparseable 2xx bodies — the watch keeps retrying instead of silently dropping the recording. Drop the UploadResult.Terminal variant entirely and the SyncWorkerLoop branch that consumed it. Also reap orphaned unsynced markers on app start: a syncedToServer=false marker without a matching pending pair (audio gone) and older than a 5-minute grace window is inconsistent — the user couldn't discard it, and no other path could clean it up. Eliminates the 2026-05-03 21:08 ghost-marker class. Test fixture (InMemoryCaseStore), SyncWorkerLoopTest, UploadClientTest, and StartupCleanupTest updated to reflect the new contract. --- .../com/doctate/watch/net/UploadClientTest.kt | 17 +++- .../main/java/com/doctate/watch/DoctateApp.kt | 1 + .../com/doctate/watch/domain/CaseStore.kt | 10 +++ .../com/doctate/watch/domain/DiskCaseStore.kt | 23 +++++ .../doctate/watch/domain/StartupCleanup.kt | 30 ++++++- .../com/doctate/watch/net/UploadClient.kt | 10 ++- .../com/doctate/watch/net/UploadResult.kt | 8 +- .../com/doctate/watch/sync/SyncWorkerLoop.kt | 6 -- .../watch/domain/StartupCleanupTest.kt | 87 ++++++++++++++++--- .../doctate/watch/sync/SyncWorkerLoopTest.kt | 27 +++--- .../watch/testfixtures/InMemoryCaseStore.kt | 20 +++++ 11 files changed, 199 insertions(+), 40 deletions(-) diff --git a/clients/wearos/app/src/androidTest/java/com/doctate/watch/net/UploadClientTest.kt b/clients/wearos/app/src/androidTest/java/com/doctate/watch/net/UploadClientTest.kt index 94a7753..c70fd57 100644 --- a/clients/wearos/app/src/androidTest/java/com/doctate/watch/net/UploadClientTest.kt +++ b/clients/wearos/app/src/androidTest/java/com/doctate/watch/net/UploadClientTest.kt @@ -68,10 +68,23 @@ class UploadClientTest { assertThat(result).isInstanceOf(UploadResult.Transient::class.java) } - @Test fun upload_classifies_401_as_terminal() = runTest { + @Test fun upload_classifies_401_as_transient() = runTest { + // Per the canonical client sync spec, anything other than a 2xx + // Success keeps the audio. A 401 typically means an expired session + // — retrying after re-auth should succeed, so we MUST NOT drop the + // audio. Same applies to all other former Terminal codes (4xx). server.enqueue(MockResponse().setResponseCode(401).setBody("nope")) val result = client.upload("uuid", "2026-04-23T10:30:00Z", audioFile) - assertThat(result).isInstanceOf(UploadResult.Terminal::class.java) + assertThat(result).isInstanceOf(UploadResult.Transient::class.java) + } + + @Test fun upload_classifies_2xx_with_unparseable_body_as_transient() = runTest { + // Server contract drift (200 OK but JSON has wrong shape) is a + // server-side issue. Keep the audio so the upload can succeed once + // the server is fixed. + server.enqueue(MockResponse().setResponseCode(200).setBody("not json")) + val result = client.upload("uuid", "2026-04-23T10:30:00Z", audioFile) + assertThat(result).isInstanceOf(UploadResult.Transient::class.java) } @Test fun upload_classifies_io_error_as_transient() = runTest { diff --git a/clients/wearos/app/src/main/java/com/doctate/watch/DoctateApp.kt b/clients/wearos/app/src/main/java/com/doctate/watch/DoctateApp.kt index 1dc3b92..b3c3b28 100644 --- a/clients/wearos/app/src/main/java/com/doctate/watch/DoctateApp.kt +++ b/clients/wearos/app/src/main/java/com/doctate/watch/DoctateApp.kt @@ -90,6 +90,7 @@ class DoctateApp : Application() { applicationScope.launch { StartupCleanup.run( store = caseStore, + pendingStore = pendingStore, pendingDir = File(filesDir, "recordings/unsynced"), ) Log.i(TAG, "startup: kicking sync service (queueLen=${pendingStore.scan().size})") diff --git a/clients/wearos/app/src/main/java/com/doctate/watch/domain/CaseStore.kt b/clients/wearos/app/src/main/java/com/doctate/watch/domain/CaseStore.kt index 5c489f4..927238f 100644 --- a/clients/wearos/app/src/main/java/com/doctate/watch/domain/CaseStore.kt +++ b/clients/wearos/app/src/main/java/com/doctate/watch/domain/CaseStore.kt @@ -50,6 +50,16 @@ interface CaseStore { /** Drop synced markers older than `cutoffMs`. Returns count removed. */ suspend fun cleanupStale(cutoffMs: Long): Int + /** + * Drop unsynced markers whose audio is gone *and* the server doesn't know + * about them — neither retry-able nor reconcilable. A marker counts as + * orphaned when its caseId is not in [pendingCaseIds] (no audio left to + * upload) and its `lastActivityAt` is older than [cutoffMs] (the grace + * window that protects a fresh marker whose pair is still being written). + * Returns count removed. + */ + suspend fun cleanupOrphanedUnsynced(pendingCaseIds: Set, cutoffMs: Long): Int + /** Wipe everything (config change, account switch). */ suspend fun clearAll() } diff --git a/clients/wearos/app/src/main/java/com/doctate/watch/domain/DiskCaseStore.kt b/clients/wearos/app/src/main/java/com/doctate/watch/domain/DiskCaseStore.kt index 5f7dae8..7aaacfb 100644 --- a/clients/wearos/app/src/main/java/com/doctate/watch/domain/DiskCaseStore.kt +++ b/clients/wearos/app/src/main/java/com/doctate/watch/domain/DiskCaseStore.kt @@ -191,6 +191,29 @@ class DiskCaseStore(private val dir: File) : CaseStore { return removed } + override suspend fun cleanupOrphanedUnsynced( + pendingCaseIds: Set, + cutoffMs: Long, + ): Int { + val cutoffRfc = epochMsToRfc(cutoffMs) + var removed = 0 + mutex.withLock { + val toRemove = state.values.filter { m -> + !m.syncedToServer && + m.caseId !in pendingCaseIds && + m.lastActivityAtRfc < cutoffRfc + }.map { it.caseId } + for (id in toRemove) { + deleteMarkerFile(id) + state.remove(id) + Log.i(TAG, "orphaned unsynced marker removed id=${id.take(8)} (audio gone, server unaware)") + removed++ + } + if (removed > 0) broadcast() + } + return removed + } + override suspend fun clearAll() { mutex.withLock { val files = dir.listFiles { f -> f.name.endsWith(".json") } ?: emptyArray() diff --git a/clients/wearos/app/src/main/java/com/doctate/watch/domain/StartupCleanup.kt b/clients/wearos/app/src/main/java/com/doctate/watch/domain/StartupCleanup.kt index ee0fc13..8ee534b 100644 --- a/clients/wearos/app/src/main/java/com/doctate/watch/domain/StartupCleanup.kt +++ b/clients/wearos/app/src/main/java/com/doctate/watch/domain/StartupCleanup.kt @@ -2,14 +2,20 @@ package com.doctate.watch.domain import android.util.Log import com.doctate.watch.audio.PendingCleanup +import com.doctate.watch.audio.PendingStore import java.io.File /** * One-shot startup cleanup. Mirrors `doctate-desktop::startup`. * - * Two retention horizons: 72 h for synced markers (non-sensitive UUIDs + - * oneliners), 24 h for orphan audio in the pending queue (sensitive). The - * watch is a "sophisticated microphone" — it must not hoard recordings + * Three retention horizons: + * - 72 h for synced markers (non-sensitive UUIDs + oneliners), + * - 24 h for orphan audio in the pending queue (sensitive), + * - 5 min grace for orphaned unsynced markers (audio is gone *and* the + * server doesn't know about them — the inconsistent state that an + * earlier code path could leave behind). + * + * The watch is a "sophisticated microphone" — it must not hoard recordings * that never reached the server past the dictation-day horizon. * * Best-effort: failures are logged, never thrown. A failing sweep @@ -24,12 +30,21 @@ object StartupCleanup { /** 24 hours — orphan recordings older than yesterday are presumed lost. */ const val DEFAULT_ORPHAN_AUDIO_RETENTION_MS: Long = 24L * 3600L * 1000L + /** + * 5 minutes — grace window before an unsynced marker without a pending + * pair is considered orphaned. Long enough to survive any plausible + * race between marker write and pending-pair write at recording stop. + */ + const val DEFAULT_ORPHAN_MARKER_GRACE_MS: Long = 5L * 60L * 1000L + suspend fun run( store: CaseStore, + pendingStore: PendingStore, pendingDir: File, nowMs: Long = System.currentTimeMillis(), markerRetentionMs: Long = DEFAULT_MARKER_RETENTION_MS, orphanAudioRetentionMs: Long = DEFAULT_ORPHAN_AUDIO_RETENTION_MS, + orphanMarkerGraceMs: Long = DEFAULT_ORPHAN_MARKER_GRACE_MS, ) { try { val markerCutoff = nowMs - markerRetentionMs @@ -46,5 +61,14 @@ object StartupCleanup { } catch (e: Exception) { Log.w(TAG, "startup: orphan-audio sweep failed: ${e.message}") } + + try { + val pendingCaseIds = pendingStore.scan().map { it.caseId }.toSet() + val orphanCutoff = nowMs - orphanMarkerGraceMs + val n = store.cleanupOrphanedUnsynced(pendingCaseIds, orphanCutoff) + if (n > 0) Log.i(TAG, "startup: orphaned unsynced markers removed: $n") + } catch (e: Exception) { + Log.w(TAG, "startup: orphan-marker sweep failed: ${e.message}") + } } } diff --git a/clients/wearos/app/src/main/java/com/doctate/watch/net/UploadClient.kt b/clients/wearos/app/src/main/java/com/doctate/watch/net/UploadClient.kt index 389e2df..3d79f79 100644 --- a/clients/wearos/app/src/main/java/com/doctate/watch/net/UploadClient.kt +++ b/clients/wearos/app/src/main/java/com/doctate/watch/net/UploadClient.kt @@ -48,10 +48,12 @@ class UploadClient( } } + // Anything other than a confirmed 2xx Success is Transient. Per the + // canonical client sync spec, a hard server error (401, 4xx, etc.) is + // semantically the same as "no contact": keep the audio, keep retrying. private fun classify(code: Int, body: String): UploadResult = when { code in 200..299 -> parseAck(body) - code == 408 || code == 429 || code in 500..599 -> UploadResult.Transient("HTTP $code: $body") - else -> UploadResult.Terminal("HTTP $code: $body") + else -> UploadResult.Transient("HTTP $code: $body") } private fun parseAck(body: String): UploadResult = try { @@ -62,7 +64,9 @@ class UploadClient( status = json.getString("status"), ) } catch (e: Exception) { - UploadResult.Terminal("parse ack: ${e.message}") + // 2xx with an unparseable body — server contract drift. Treat as + // transient so the audio survives until the server is fixed. + UploadResult.Transient("parse ack: ${e.message}") } companion object { diff --git a/clients/wearos/app/src/main/java/com/doctate/watch/net/UploadResult.kt b/clients/wearos/app/src/main/java/com/doctate/watch/net/UploadResult.kt index 2fbd35c..ae8f679 100644 --- a/clients/wearos/app/src/main/java/com/doctate/watch/net/UploadResult.kt +++ b/clients/wearos/app/src/main/java/com/doctate/watch/net/UploadResult.kt @@ -1,11 +1,13 @@ package com.doctate.watch.net /** - * Outcome of [UploadClient.upload]. Mirrors UploadOutcome in - * clients/desktop/src/server_sync.rs (Succeeded / Transient / Terminal). + * Outcome of [UploadClient.upload]. Only Success or Transient: per the + * canonical client sync spec, anything other than a confirmed Success is + * treated like "no contact" — the audio stays local and retry continues + * indefinitely. The desktop client mirror in clients/desktop/src/server_sync.rs + * still has a separate Terminal variant and must be reconciled separately. */ sealed interface UploadResult { data class Success(val caseId: String, val recordedAt: String, val status: String) : UploadResult data class Transient(val reason: String) : UploadResult - data class Terminal(val reason: String) : UploadResult } diff --git a/clients/wearos/app/src/main/java/com/doctate/watch/sync/SyncWorkerLoop.kt b/clients/wearos/app/src/main/java/com/doctate/watch/sync/SyncWorkerLoop.kt index 28f86d4..4c21df2 100644 --- a/clients/wearos/app/src/main/java/com/doctate/watch/sync/SyncWorkerLoop.kt +++ b/clients/wearos/app/src/main/java/com/doctate/watch/sync/SyncWorkerLoop.kt @@ -178,12 +178,6 @@ class SyncWorkerLoop( delay(backoffMs) backoffMs = (backoffMs * 2).coerceAtMost(maxBackoffMs) } - is UploadResult.Terminal -> { - Log.w(TAG, "terminal: ${result.reason} → dropping caseId=${next.caseId.take(8)}") - pendingStore.deletePair(next.file) - state.recordFailure(nowMs(), result.reason) - backoffMs = initialBackoffMs - } } } diff --git a/clients/wearos/app/src/test/java/com/doctate/watch/domain/StartupCleanupTest.kt b/clients/wearos/app/src/test/java/com/doctate/watch/domain/StartupCleanupTest.kt index 063f30d..d3de9b1 100644 --- a/clients/wearos/app/src/test/java/com/doctate/watch/domain/StartupCleanupTest.kt +++ b/clients/wearos/app/src/test/java/com/doctate/watch/domain/StartupCleanupTest.kt @@ -1,5 +1,7 @@ package com.doctate.watch.domain +import com.doctate.watch.audio.PendingStore +import com.doctate.watch.audio.PendingUpload import com.doctate.watch.testfixtures.InMemoryCaseStore import com.google.common.truth.Truth.assertThat import java.io.File @@ -11,8 +13,18 @@ import org.junit.rules.TemporaryFolder class StartupCleanupTest { @get:Rule val tempDir = TemporaryFolder() + private fun newPendingStore(root: File): PendingStore = PendingStore(File(root, "pending")) + + private fun PendingStore.seed(caseId: String, recordedAt: String): File { + val audio = audioPathFor(caseId, recordedAt) + audio.writeBytes(byteArrayOf(0x66, 0x74, 0x79, 0x70)) + writeSidecar(PendingUpload(caseId, recordedAt, audio)) + return audio + } + @Test fun sweeps_stale_synced_markers() = runTest { val store = InMemoryCaseStore() + val pendingStore = newPendingStore(tempDir.root) val now = 10_000_000_000L // Old + synced → eligible. val oldId = "case-old" @@ -25,7 +37,8 @@ class StartupCleanupTest { StartupCleanup.run( store = store, - pendingDir = tempDir.newFolder("pending"), + pendingStore = pendingStore, + pendingDir = File(tempDir.root, "pending"), nowMs = now, markerRetentionMs = 72L * 3600L * 1000L, ) @@ -33,42 +46,90 @@ class StartupCleanupTest { assertThat(ids).containsExactly(recentId) } - @Test fun preserves_pending_markers_regardless_of_age() = runTest { + @Test fun preserves_pending_marker_with_pending_pair_regardless_of_age() = runTest { + // A marker with syncedToServer=false AND a matching pending pair + // is a real waiting upload — must survive any age threshold. val store = InMemoryCaseStore() + val pendingStore = newPendingStore(tempDir.root) val now = 10_000_000_000L - val oldPendingId = "case-old-pending" - store.markActivity(oldPendingId, now = now - 200L * 3600L * 1000L) - // No markSynced — pending must be preserved. + val pendingId = "case-old-pending" + store.markActivity(pendingId, now = now - 200L * 3600L * 1000L) + pendingStore.seed(pendingId, "2026-04-26T10:00:00Z") StartupCleanup.run( store = store, - pendingDir = tempDir.newFolder("pending"), + pendingStore = pendingStore, + pendingDir = File(tempDir.root, "pending"), nowMs = now, markerRetentionMs = 72L * 3600L * 1000L, ) val ids = store.casesFlow.value.map { it.caseId } - assertThat(ids).containsExactly(oldPendingId) + assertThat(ids).containsExactly(pendingId) + } + + @Test fun reaps_orphaned_unsynced_marker_when_audio_is_gone() = runTest { + // syncedToServer=false + NO pending pair + older than the grace + // window → orphan ("ghost" marker the user can't otherwise remove). + // This is the exact regression fix for the 2026-05-03 21:08 ghost. + val store = InMemoryCaseStore() + val pendingStore = newPendingStore(tempDir.root) + val now = 10_000_000_000L + val ghostId = "case-ghost" + store.markActivity(ghostId, now = now - 60L * 60L * 1000L) // 1h ago + // No pendingStore.seed() — audio was already discarded. + + StartupCleanup.run( + store = store, + pendingStore = pendingStore, + pendingDir = File(tempDir.root, "pending"), + nowMs = now, + orphanMarkerGraceMs = 5L * 60L * 1000L, // 5 min + ) + assertThat(store.casesFlow.value).isEmpty() + } + + @Test fun preserves_fresh_unsynced_marker_within_grace_window() = runTest { + // A marker just written (within the grace window) might still have + // its pending pair on the way — must not be reaped. + val store = InMemoryCaseStore() + val pendingStore = newPendingStore(tempDir.root) + val now = 10_000_000_000L + val freshId = "case-fresh" + store.markActivity(freshId, now = now - 1_000L) // 1 s ago + // No pending pair yet — but grace window protects it. + + StartupCleanup.run( + store = store, + pendingStore = pendingStore, + pendingDir = File(tempDir.root, "pending"), + nowMs = now, + orphanMarkerGraceMs = 5L * 60L * 1000L, + ) + val ids = store.casesFlow.value.map { it.caseId } + assertThat(ids).containsExactly(freshId) } @Test fun reaps_old_orphan_audio_in_pending_dir() = runTest { val store = InMemoryCaseStore() - val pending = tempDir.newFolder("pending") + val pendingDir = File(tempDir.root, "pending").apply { mkdirs() } + val pendingStore = PendingStore(pendingDir) val now = 10_000_000_000L // Old orphan m4a (no sidecar) — should die. - File(pending, "old-orphan.m4a").apply { + File(pendingDir, "old-orphan.m4a").apply { writeText("x"); setLastModified(now - 30L * 3600L * 1000L) // 30h ago } // Young orphan m4a — should survive (within 24h). - File(pending, "fresh-orphan.m4a").apply { + File(pendingDir, "fresh-orphan.m4a").apply { writeText("x"); setLastModified(now - 1L * 3600L * 1000L) // 1h ago } StartupCleanup.run( store = store, - pendingDir = pending, + pendingStore = pendingStore, + pendingDir = pendingDir, nowMs = now, orphanAudioRetentionMs = 24L * 3600L * 1000L, ) - assertThat(File(pending, "old-orphan.m4a").exists()).isFalse() - assertThat(File(pending, "fresh-orphan.m4a").exists()).isTrue() + assertThat(File(pendingDir, "old-orphan.m4a").exists()).isFalse() + assertThat(File(pendingDir, "fresh-orphan.m4a").exists()).isTrue() } } diff --git a/clients/wearos/app/src/test/java/com/doctate/watch/sync/SyncWorkerLoopTest.kt b/clients/wearos/app/src/test/java/com/doctate/watch/sync/SyncWorkerLoopTest.kt index 4296579..66d758b 100644 --- a/clients/wearos/app/src/test/java/com/doctate/watch/sync/SyncWorkerLoopTest.kt +++ b/clients/wearos/app/src/test/java/com/doctate/watch/sync/SyncWorkerLoopTest.kt @@ -87,26 +87,33 @@ class SyncWorkerLoopTest { assertThat(state.state.value.lastFailureReason).contains("HTTP 503") } - @Test fun terminal_drops_pair_no_retry() = runTest { + @Test fun former_terminal_code_keeps_pair_and_retries() = runTest { + // Per the canonical client sync spec, a 4xx-class server response + // (e.g. 401 expired session) must not destroy the audio. The + // UploadClient now classifies it as Transient, so the sync worker + // sees the same retry path as a 5xx — the pair stays on disk and + // the worker keeps trying with backoff. val store = InMemoryCaseStore() val pending = newPendingStore() - val audio = pending.seed("case-terminal", "2026-04-26T10:00:00Z") + val audio = pending.seed("case-formerly-terminal", "2026-04-26T10:00:00Z") var calls = 0 val state = SyncStateProvider() val uploader = SyncWorkerLoop.Uploader { _, _, _ -> calls++ - UploadResult.Terminal("HTTP 401") + UploadResult.Transient("HTTP 401: nope") } - val loop = SyncWorkerLoop(pending, uploader, store, state, Channel(Channel.UNLIMITED)) + val loop = SyncWorkerLoop( + pending, uploader, store, state, Channel(Channel.UNLIMITED), + initialBackoffMs = 100L, maxBackoffMs = 800L, + ) loop.runOnce() - // Second runOnce sees an empty queue — terminal must have dropped the pair. - val nextStep = loop.runOnce() - assertThat(nextStep).isEqualTo(SyncWorkerLoop.Step.Idle) - assertThat(audio.exists()).isFalse() - assertThat(pending.sidecarPathFor(audio).exists()).isFalse() - assertThat(calls).isEqualTo(1) + loop.runOnce() + assertThat(audio.exists()).isTrue() + assertThat(pending.sidecarPathFor(audio).exists()).isTrue() + assertThat(calls).isEqualTo(2) + assertThat(state.state.value.lastFailureReason).contains("HTTP 401") } @Test fun idle_branch_publishes_zero_queue() = runTest { diff --git a/clients/wearos/app/src/test/java/com/doctate/watch/testfixtures/InMemoryCaseStore.kt b/clients/wearos/app/src/test/java/com/doctate/watch/testfixtures/InMemoryCaseStore.kt index 3db8fb1..b1921f7 100644 --- a/clients/wearos/app/src/test/java/com/doctate/watch/testfixtures/InMemoryCaseStore.kt +++ b/clients/wearos/app/src/test/java/com/doctate/watch/testfixtures/InMemoryCaseStore.kt @@ -130,6 +130,26 @@ class InMemoryCaseStore : CaseStore { return removed } + override suspend fun cleanupOrphanedUnsynced( + pendingCaseIds: Set, + cutoffMs: Long, + ): Int { + var removed = 0 + mutex.withLock { + val toRemove = state.values.filter { e -> + !e.syncedToServer && + e.caseId !in pendingCaseIds && + e.lastActivityAt < cutoffMs + }.map { it.caseId } + for (id in toRemove) { + state.remove(id) + removed++ + } + if (removed > 0) broadcast() + } + return removed + } + override suspend fun clearAll() { mutex.withLock { state.clear()