fix(wear): keep recordings on hard upload errors, reap ghost markers

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.
This commit is contained in:
2026-05-05 09:52:29 +02:00
parent 2c6062a53e
commit a8994fd84d
11 changed files with 199 additions and 40 deletions
@@ -68,10 +68,23 @@ class UploadClientTest {
assertThat(result).isInstanceOf(UploadResult.Transient::class.java) 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")) server.enqueue(MockResponse().setResponseCode(401).setBody("nope"))
val result = client.upload("uuid", "2026-04-23T10:30:00Z", audioFile) 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 { @Test fun upload_classifies_io_error_as_transient() = runTest {
@@ -90,6 +90,7 @@ class DoctateApp : Application() {
applicationScope.launch { applicationScope.launch {
StartupCleanup.run( StartupCleanup.run(
store = caseStore, store = caseStore,
pendingStore = pendingStore,
pendingDir = File(filesDir, "recordings/unsynced"), pendingDir = File(filesDir, "recordings/unsynced"),
) )
Log.i(TAG, "startup: kicking sync service (queueLen=${pendingStore.scan().size})") Log.i(TAG, "startup: kicking sync service (queueLen=${pendingStore.scan().size})")
@@ -50,6 +50,16 @@ interface CaseStore {
/** Drop synced markers older than `cutoffMs`. Returns count removed. */ /** Drop synced markers older than `cutoffMs`. Returns count removed. */
suspend fun cleanupStale(cutoffMs: Long): Int 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<String>, cutoffMs: Long): Int
/** Wipe everything (config change, account switch). */ /** Wipe everything (config change, account switch). */
suspend fun clearAll() suspend fun clearAll()
} }
@@ -191,6 +191,29 @@ class DiskCaseStore(private val dir: File) : CaseStore {
return removed return removed
} }
override suspend fun cleanupOrphanedUnsynced(
pendingCaseIds: Set<String>,
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() { override suspend fun clearAll() {
mutex.withLock { mutex.withLock {
val files = dir.listFiles { f -> f.name.endsWith(".json") } ?: emptyArray() val files = dir.listFiles { f -> f.name.endsWith(".json") } ?: emptyArray()
@@ -2,14 +2,20 @@ package com.doctate.watch.domain
import android.util.Log import android.util.Log
import com.doctate.watch.audio.PendingCleanup import com.doctate.watch.audio.PendingCleanup
import com.doctate.watch.audio.PendingStore
import java.io.File import java.io.File
/** /**
* One-shot startup cleanup. Mirrors `doctate-desktop::startup`. * One-shot startup cleanup. Mirrors `doctate-desktop::startup`.
* *
* Two retention horizons: 72 h for synced markers (non-sensitive UUIDs + * Three retention horizons:
* oneliners), 24 h for orphan audio in the pending queue (sensitive). The * - 72 h for synced markers (non-sensitive UUIDs + oneliners),
* watch is a "sophisticated microphone" — it must not hoard recordings * - 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. * that never reached the server past the dictation-day horizon.
* *
* Best-effort: failures are logged, never thrown. A failing sweep * 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. */ /** 24 hours — orphan recordings older than yesterday are presumed lost. */
const val DEFAULT_ORPHAN_AUDIO_RETENTION_MS: Long = 24L * 3600L * 1000L 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( suspend fun run(
store: CaseStore, store: CaseStore,
pendingStore: PendingStore,
pendingDir: File, pendingDir: File,
nowMs: Long = System.currentTimeMillis(), nowMs: Long = System.currentTimeMillis(),
markerRetentionMs: Long = DEFAULT_MARKER_RETENTION_MS, markerRetentionMs: Long = DEFAULT_MARKER_RETENTION_MS,
orphanAudioRetentionMs: Long = DEFAULT_ORPHAN_AUDIO_RETENTION_MS, orphanAudioRetentionMs: Long = DEFAULT_ORPHAN_AUDIO_RETENTION_MS,
orphanMarkerGraceMs: Long = DEFAULT_ORPHAN_MARKER_GRACE_MS,
) { ) {
try { try {
val markerCutoff = nowMs - markerRetentionMs val markerCutoff = nowMs - markerRetentionMs
@@ -46,5 +61,14 @@ object StartupCleanup {
} catch (e: Exception) { } catch (e: Exception) {
Log.w(TAG, "startup: orphan-audio sweep failed: ${e.message}") 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}")
}
} }
} }
@@ -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 { private fun classify(code: Int, body: String): UploadResult = when {
code in 200..299 -> parseAck(body) code in 200..299 -> parseAck(body)
code == 408 || code == 429 || code in 500..599 -> UploadResult.Transient("HTTP $code: $body") else -> UploadResult.Transient("HTTP $code: $body")
else -> UploadResult.Terminal("HTTP $code: $body")
} }
private fun parseAck(body: String): UploadResult = try { private fun parseAck(body: String): UploadResult = try {
@@ -62,7 +64,9 @@ class UploadClient(
status = json.getString("status"), status = json.getString("status"),
) )
} catch (e: Exception) { } 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 { companion object {
@@ -1,11 +1,13 @@
package com.doctate.watch.net package com.doctate.watch.net
/** /**
* Outcome of [UploadClient.upload]. Mirrors UploadOutcome in * Outcome of [UploadClient.upload]. Only Success or Transient: per the
* clients/desktop/src/server_sync.rs (Succeeded / Transient / Terminal). * 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 { sealed interface UploadResult {
data class Success(val caseId: String, val recordedAt: String, val status: String) : UploadResult data class Success(val caseId: String, val recordedAt: String, val status: String) : UploadResult
data class Transient(val reason: String) : UploadResult data class Transient(val reason: String) : UploadResult
data class Terminal(val reason: String) : UploadResult
} }
@@ -178,12 +178,6 @@ class SyncWorkerLoop(
delay(backoffMs) delay(backoffMs)
backoffMs = (backoffMs * 2).coerceAtMost(maxBackoffMs) 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
}
} }
} }
@@ -1,5 +1,7 @@
package com.doctate.watch.domain 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.doctate.watch.testfixtures.InMemoryCaseStore
import com.google.common.truth.Truth.assertThat import com.google.common.truth.Truth.assertThat
import java.io.File import java.io.File
@@ -11,8 +13,18 @@ import org.junit.rules.TemporaryFolder
class StartupCleanupTest { class StartupCleanupTest {
@get:Rule val tempDir = TemporaryFolder() @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 { @Test fun sweeps_stale_synced_markers() = runTest {
val store = InMemoryCaseStore() val store = InMemoryCaseStore()
val pendingStore = newPendingStore(tempDir.root)
val now = 10_000_000_000L val now = 10_000_000_000L
// Old + synced → eligible. // Old + synced → eligible.
val oldId = "case-old" val oldId = "case-old"
@@ -25,7 +37,8 @@ class StartupCleanupTest {
StartupCleanup.run( StartupCleanup.run(
store = store, store = store,
pendingDir = tempDir.newFolder("pending"), pendingStore = pendingStore,
pendingDir = File(tempDir.root, "pending"),
nowMs = now, nowMs = now,
markerRetentionMs = 72L * 3600L * 1000L, markerRetentionMs = 72L * 3600L * 1000L,
) )
@@ -33,42 +46,90 @@ class StartupCleanupTest {
assertThat(ids).containsExactly(recentId) 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 store = InMemoryCaseStore()
val pendingStore = newPendingStore(tempDir.root)
val now = 10_000_000_000L val now = 10_000_000_000L
val oldPendingId = "case-old-pending" val pendingId = "case-old-pending"
store.markActivity(oldPendingId, now = now - 200L * 3600L * 1000L) store.markActivity(pendingId, now = now - 200L * 3600L * 1000L)
// No markSynced — pending must be preserved. pendingStore.seed(pendingId, "2026-04-26T10:00:00Z")
StartupCleanup.run( StartupCleanup.run(
store = store, store = store,
pendingDir = tempDir.newFolder("pending"), pendingStore = pendingStore,
pendingDir = File(tempDir.root, "pending"),
nowMs = now, nowMs = now,
markerRetentionMs = 72L * 3600L * 1000L, markerRetentionMs = 72L * 3600L * 1000L,
) )
val ids = store.casesFlow.value.map { it.caseId } 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 { @Test fun reaps_old_orphan_audio_in_pending_dir() = runTest {
val store = InMemoryCaseStore() 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 val now = 10_000_000_000L
// Old orphan m4a (no sidecar) — should die. // 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 writeText("x"); setLastModified(now - 30L * 3600L * 1000L) // 30h ago
} }
// Young orphan m4a — should survive (within 24h). // 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 writeText("x"); setLastModified(now - 1L * 3600L * 1000L) // 1h ago
} }
StartupCleanup.run( StartupCleanup.run(
store = store, store = store,
pendingDir = pending, pendingStore = pendingStore,
pendingDir = pendingDir,
nowMs = now, nowMs = now,
orphanAudioRetentionMs = 24L * 3600L * 1000L, orphanAudioRetentionMs = 24L * 3600L * 1000L,
) )
assertThat(File(pending, "old-orphan.m4a").exists()).isFalse() assertThat(File(pendingDir, "old-orphan.m4a").exists()).isFalse()
assertThat(File(pending, "fresh-orphan.m4a").exists()).isTrue() assertThat(File(pendingDir, "fresh-orphan.m4a").exists()).isTrue()
} }
} }
@@ -87,26 +87,33 @@ class SyncWorkerLoopTest {
assertThat(state.state.value.lastFailureReason).contains("HTTP 503") 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 store = InMemoryCaseStore()
val pending = newPendingStore() 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 var calls = 0
val state = SyncStateProvider() val state = SyncStateProvider()
val uploader = SyncWorkerLoop.Uploader { _, _, _ -> val uploader = SyncWorkerLoop.Uploader { _, _, _ ->
calls++ 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() loop.runOnce()
// Second runOnce sees an empty queue — terminal must have dropped the pair. loop.runOnce()
val nextStep = loop.runOnce() assertThat(audio.exists()).isTrue()
assertThat(nextStep).isEqualTo(SyncWorkerLoop.Step.Idle) assertThat(pending.sidecarPathFor(audio).exists()).isTrue()
assertThat(audio.exists()).isFalse() assertThat(calls).isEqualTo(2)
assertThat(pending.sidecarPathFor(audio).exists()).isFalse() assertThat(state.state.value.lastFailureReason).contains("HTTP 401")
assertThat(calls).isEqualTo(1)
} }
@Test fun idle_branch_publishes_zero_queue() = runTest { @Test fun idle_branch_publishes_zero_queue() = runTest {
@@ -130,6 +130,26 @@ class InMemoryCaseStore : CaseStore {
return removed return removed
} }
override suspend fun cleanupOrphanedUnsynced(
pendingCaseIds: Set<String>,
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() { override suspend fun clearAll() {
mutex.withLock { mutex.withLock {
state.clear() state.clear()