feat(watch): foreground SyncService with backoff worker loop
SyncWorkerLoop is the upload+poll core, mirrored from doctate-client-core::server_sync::run. Pending uploads have priority over polling (no point asking the server for state if we still hold the doctor's audio). Backoff is in-process state — after restart it drops back to 2s and gives the server one fresh chance, which beats 'wait 60s because we cancelled mid-backoff'. Architecturally split into runOnce (single iteration, suspends only on real I/O) and run (infinite loop, suspends on idle). The split makes the loop directly testable on JVM virtual time without dealing with advanceUntilIdle vs. infinite-while semantics — tests call runOnce N times and assert in between. Wired through: - SyncService (foregroundServiceType=dataSync, low-importance silent notification, ServiceCompat.startForeground for API 34+) - SyncServiceLauncher replaces NoopSyncTrigger in DoctateApp - DoctateApp.onCreate kicks the service after StartupCleanup if the pending queue is non-empty (recovery from crash/reboot) - AndroidManifest: FOREGROUND_SERVICE / FOREGROUND_SERVICE_DATA_SYNC / POST_NOTIFICATIONS perms; service declaration 6 SyncWorkerLoopTest cases (success/transient/terminal/idle/fifo plus backoff-resets-after-success). 58 tests green total.
This commit is contained in:
@@ -4,6 +4,9 @@
|
||||
<uses-permission android:name="android.permission.WAKE_LOCK" />
|
||||
<uses-permission android:name="android.permission.INTERNET" />
|
||||
<uses-permission android:name="android.permission.RECORD_AUDIO" />
|
||||
<uses-permission android:name="android.permission.FOREGROUND_SERVICE" />
|
||||
<uses-permission android:name="android.permission.FOREGROUND_SERVICE_DATA_SYNC" />
|
||||
<uses-permission android:name="android.permission.POST_NOTIFICATIONS" />
|
||||
|
||||
<uses-feature android:name="android.hardware.type.watch" />
|
||||
<uses-feature android:name="android.hardware.microphone" android:required="true" />
|
||||
@@ -59,6 +62,11 @@
|
||||
android:resource="@drawable/tile_preview" />
|
||||
</service>
|
||||
|
||||
<service
|
||||
android:name=".sync.SyncService"
|
||||
android:exported="false"
|
||||
android:foregroundServiceType="dataSync" />
|
||||
|
||||
<service
|
||||
android:name=".complication.DoctateComplicationService"
|
||||
android:exported="true"
|
||||
|
||||
@@ -11,7 +11,9 @@ import com.doctate.watch.net.HttpClientProvider
|
||||
import com.doctate.watch.net.UploadClient
|
||||
import com.doctate.watch.settings.BuildConfigSettings
|
||||
import com.doctate.watch.settings.Settings
|
||||
import com.doctate.watch.sync.NoopSyncTrigger
|
||||
import com.doctate.watch.sync.SyncKick
|
||||
import com.doctate.watch.sync.SyncServiceLauncher
|
||||
import com.doctate.watch.sync.SyncStateProvider
|
||||
import com.doctate.watch.sync.SyncTrigger
|
||||
import com.doctate.watch.tile.DoctateTileService
|
||||
import java.io.File
|
||||
@@ -45,8 +47,11 @@ class DoctateApp : Application() {
|
||||
PendingStore(File(filesDir, "recordings/unsynced"))
|
||||
}
|
||||
|
||||
/** Replaced by [com.doctate.watch.sync.SyncServiceLauncher] in Phase 4. */
|
||||
var syncTrigger: SyncTrigger = NoopSyncTrigger
|
||||
/** Routes sync kicks into [com.doctate.watch.sync.SyncService]. */
|
||||
val syncTrigger: SyncTrigger = SyncServiceLauncher
|
||||
|
||||
/** Single source of truth for the worker's status, observed by UI/Tile. */
|
||||
val syncStateProvider: SyncStateProvider by lazy { SyncStateProvider() }
|
||||
|
||||
/** Process-scoped coroutine scope for work that must outlive any single ViewModel
|
||||
* (e.g. the post-stop OneLiner-burst fires after the user swipes back to the list). */
|
||||
@@ -68,12 +73,17 @@ class DoctateApp : Application() {
|
||||
}
|
||||
.launchIn(applicationScope)
|
||||
|
||||
// One-shot data-minimization sweep before the sync worker starts.
|
||||
// One-shot data-minimization sweep before the sync worker starts,
|
||||
// then kick the service if the pending queue still has work.
|
||||
applicationScope.launch {
|
||||
StartupCleanup.run(
|
||||
store = caseStore,
|
||||
pendingDir = File(filesDir, "recordings/unsynced"),
|
||||
)
|
||||
if (pendingStore.scan().isNotEmpty()) {
|
||||
Log.i(TAG, "startup: pending queue non-empty → kicking sync service")
|
||||
syncTrigger.kick(this@DoctateApp, SyncKick.NewPending)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,163 @@
|
||||
package com.doctate.watch.sync
|
||||
|
||||
import android.app.Notification
|
||||
import android.app.NotificationChannel
|
||||
import android.app.NotificationManager
|
||||
import android.app.Service
|
||||
import android.content.Context
|
||||
import android.content.Intent
|
||||
import android.content.pm.ServiceInfo
|
||||
import android.os.Build
|
||||
import android.os.IBinder
|
||||
import android.util.Log
|
||||
import androidx.core.app.NotificationCompat
|
||||
import androidx.core.app.ServiceCompat
|
||||
import com.doctate.watch.DoctateApp
|
||||
import com.doctate.watch.R
|
||||
import kotlinx.coroutines.CoroutineScope
|
||||
import kotlinx.coroutines.Dispatchers
|
||||
import kotlinx.coroutines.Job
|
||||
import kotlinx.coroutines.SupervisorJob
|
||||
import kotlinx.coroutines.channels.Channel
|
||||
import kotlinx.coroutines.launch
|
||||
|
||||
/**
|
||||
* Foreground sync service. Runs the [SyncWorkerLoop] for the lifetime of
|
||||
* the process; survives Doze via `foregroundServiceType="dataSync"`.
|
||||
*
|
||||
* Lifecycle: started either from [SyncServiceLauncher.kick] (recording
|
||||
* just queued an upload) or from [DoctateApp.onCreate] when the pending
|
||||
* queue is non-empty at app launch (recovery after restart).
|
||||
*
|
||||
* Notification is intentionally low-importance and silent — the doctor
|
||||
* shouldn't be alerted every time a case syncs in the background. The
|
||||
* channel is registered once on first run.
|
||||
*/
|
||||
class SyncService : Service() {
|
||||
private val scope = CoroutineScope(SupervisorJob() + Dispatchers.IO)
|
||||
private var loopJob: Job? = null
|
||||
private val kickChannel = Channel<SyncKick>(Channel.UNLIMITED)
|
||||
|
||||
override fun onBind(intent: Intent?): IBinder? = null
|
||||
|
||||
override fun onCreate() {
|
||||
super.onCreate()
|
||||
registerNotificationChannel()
|
||||
}
|
||||
|
||||
override fun onStartCommand(intent: Intent?, flags: Int, startId: Int): Int {
|
||||
val notification = buildNotification()
|
||||
ServiceCompat.startForeground(
|
||||
this,
|
||||
NOTIFICATION_ID,
|
||||
notification,
|
||||
if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.UPSIDE_DOWN_CAKE) {
|
||||
ServiceInfo.FOREGROUND_SERVICE_TYPE_DATA_SYNC
|
||||
} else 0,
|
||||
)
|
||||
|
||||
// Forward the incoming kick (if any) to the loop. Multiple onStartCommand
|
||||
// calls just enqueue more kicks — the loop coalesces.
|
||||
intent?.let { i ->
|
||||
val kindOrdinal = i.getIntExtra(EXTRA_KICK_KIND, KICK_KIND_NEW_PENDING)
|
||||
val kick = when (kindOrdinal) {
|
||||
KICK_KIND_BURST_POLL -> {
|
||||
val caseId = i.getStringExtra(EXTRA_BURST_CASE_ID).orEmpty()
|
||||
val deadline = i.getLongExtra(EXTRA_BURST_DEADLINE_MS, 0L)
|
||||
SyncKick.BurstPoll(caseId, deadline)
|
||||
}
|
||||
else -> SyncKick.NewPending
|
||||
}
|
||||
kickChannel.trySend(kick)
|
||||
}
|
||||
|
||||
if (loopJob?.isActive != true) {
|
||||
loopJob = startLoop()
|
||||
}
|
||||
return START_STICKY
|
||||
}
|
||||
|
||||
private fun startLoop(): Job {
|
||||
val app = applicationContext as DoctateApp
|
||||
val loop = SyncWorkerLoop(
|
||||
pendingStore = app.pendingStore,
|
||||
uploader = SyncWorkerLoop.Uploader { caseId, recordedAt, audio ->
|
||||
app.uploadClient.upload(caseId, recordedAt, audio)
|
||||
},
|
||||
caseStore = app.caseStore,
|
||||
state = app.syncStateProvider,
|
||||
kickChannel = kickChannel,
|
||||
)
|
||||
return scope.launch {
|
||||
try {
|
||||
loop.run()
|
||||
} catch (e: Throwable) {
|
||||
Log.w(TAG, "loop crashed: ${e.message}", e)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
override fun onDestroy() {
|
||||
super.onDestroy()
|
||||
loopJob?.cancel()
|
||||
scope.cancel()
|
||||
kickChannel.close()
|
||||
}
|
||||
|
||||
private fun registerNotificationChannel() {
|
||||
if (Build.VERSION.SDK_INT < Build.VERSION_CODES.O) return
|
||||
val nm = getSystemService(NotificationManager::class.java) ?: return
|
||||
if (nm.getNotificationChannel(CHANNEL_ID) != null) return
|
||||
val channel = NotificationChannel(
|
||||
CHANNEL_ID,
|
||||
"Doctate sync",
|
||||
NotificationManager.IMPORTANCE_LOW,
|
||||
).apply {
|
||||
description = "Background uploads of recordings"
|
||||
setShowBadge(false)
|
||||
}
|
||||
nm.createNotificationChannel(channel)
|
||||
}
|
||||
|
||||
private fun buildNotification(): Notification =
|
||||
NotificationCompat.Builder(this, CHANNEL_ID)
|
||||
.setContentTitle("Doctate")
|
||||
.setContentText("Synchronisiere Aufnahmen")
|
||||
.setSmallIcon(R.drawable.ic_complication)
|
||||
.setOngoing(true)
|
||||
.setPriority(NotificationCompat.PRIORITY_LOW)
|
||||
.build()
|
||||
|
||||
companion object {
|
||||
private const val TAG = "SyncService"
|
||||
private const val NOTIFICATION_ID = 1
|
||||
private const val CHANNEL_ID = "doctate_sync"
|
||||
|
||||
const val ACTION_KICK = "com.doctate.watch.sync.KICK"
|
||||
const val EXTRA_KICK_KIND = "kick_kind"
|
||||
const val EXTRA_BURST_CASE_ID = "burst_case_id"
|
||||
const val EXTRA_BURST_DEADLINE_MS = "burst_deadline_ms"
|
||||
|
||||
const val KICK_KIND_NEW_PENDING = 0
|
||||
const val KICK_KIND_BURST_POLL = 1
|
||||
|
||||
fun startIntent(context: Context, kick: SyncKick): Intent {
|
||||
val intent = Intent(context, SyncService::class.java).apply {
|
||||
action = ACTION_KICK
|
||||
when (kick) {
|
||||
SyncKick.NewPending -> putExtra(EXTRA_KICK_KIND, KICK_KIND_NEW_PENDING)
|
||||
is SyncKick.BurstPoll -> {
|
||||
putExtra(EXTRA_KICK_KIND, KICK_KIND_BURST_POLL)
|
||||
putExtra(EXTRA_BURST_CASE_ID, kick.caseId)
|
||||
putExtra(EXTRA_BURST_DEADLINE_MS, kick.deadlineMs)
|
||||
}
|
||||
}
|
||||
}
|
||||
return intent
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private fun CoroutineScope.cancel() {
|
||||
(coroutineContext[Job] ?: return).cancel()
|
||||
}
|
||||
@@ -0,0 +1,20 @@
|
||||
package com.doctate.watch.sync
|
||||
|
||||
import android.content.Context
|
||||
import android.os.Build
|
||||
|
||||
/**
|
||||
* [SyncTrigger] that routes kicks into [SyncService]. Wired in
|
||||
* [com.doctate.watch.DoctateApp.onCreate]; replaces [NoopSyncTrigger]
|
||||
* once the service is registered in the manifest.
|
||||
*/
|
||||
object SyncServiceLauncher : SyncTrigger {
|
||||
override fun kick(context: Context, kick: SyncKick) {
|
||||
val intent = SyncService.startIntent(context, kick)
|
||||
if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.O) {
|
||||
context.startForegroundService(intent)
|
||||
} else {
|
||||
context.startService(intent)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,21 @@
|
||||
package com.doctate.watch.sync
|
||||
|
||||
/**
|
||||
* Snapshot of what the sync worker is currently doing, surfaced to the UI
|
||||
* (Tile / List / Detail) so it can render the cloud indicator. Mirrors
|
||||
* `WorkerSnapshot` in `doctate-client-core::server_sync`.
|
||||
*
|
||||
* This is the authoritative view — never trust the queue length you'd
|
||||
* compute by listing the directory yourself, because the worker can be
|
||||
* mid-delete-pair and the disk count is briefly off-by-one.
|
||||
*/
|
||||
enum class WorkerPhase { Idle, Uploading, Polling }
|
||||
|
||||
data class SyncState(
|
||||
val phase: WorkerPhase = WorkerPhase.Idle,
|
||||
val queueLen: Int = 0,
|
||||
val pendingCaseIds: Set<String> = emptySet(),
|
||||
val lastSuccessAt: Long? = null,
|
||||
val lastFailureAt: Long? = null,
|
||||
val lastFailureReason: String? = null,
|
||||
)
|
||||
@@ -0,0 +1,36 @@
|
||||
package com.doctate.watch.sync
|
||||
|
||||
import kotlinx.coroutines.flow.MutableStateFlow
|
||||
import kotlinx.coroutines.flow.StateFlow
|
||||
import kotlinx.coroutines.flow.asStateFlow
|
||||
import kotlinx.coroutines.flow.update
|
||||
|
||||
/**
|
||||
* Owns the [SyncState] StateFlow. Written by [SyncWorkerLoop], read by
|
||||
* the UI. Decoupled from the Service / Application class so JVM tests
|
||||
* can drive the worker against a fresh provider.
|
||||
*/
|
||||
class SyncStateProvider {
|
||||
private val _state = MutableStateFlow(SyncState())
|
||||
val state: StateFlow<SyncState> = _state.asStateFlow()
|
||||
|
||||
fun setIdle(queueLen: Int, pendingCaseIds: Set<String>) {
|
||||
_state.update { it.copy(phase = WorkerPhase.Idle, queueLen = queueLen, pendingCaseIds = pendingCaseIds) }
|
||||
}
|
||||
|
||||
fun setUploading(queueLen: Int, pendingCaseIds: Set<String>) {
|
||||
_state.update { it.copy(phase = WorkerPhase.Uploading, queueLen = queueLen, pendingCaseIds = pendingCaseIds) }
|
||||
}
|
||||
|
||||
fun setPolling(queueLen: Int, pendingCaseIds: Set<String>) {
|
||||
_state.update { it.copy(phase = WorkerPhase.Polling, queueLen = queueLen, pendingCaseIds = pendingCaseIds) }
|
||||
}
|
||||
|
||||
fun recordSuccess(at: Long) {
|
||||
_state.update { it.copy(lastSuccessAt = at, lastFailureReason = null) }
|
||||
}
|
||||
|
||||
fun recordFailure(at: Long, reason: String) {
|
||||
_state.update { it.copy(lastFailureAt = at, lastFailureReason = reason) }
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,147 @@
|
||||
package com.doctate.watch.sync
|
||||
|
||||
import android.util.Log
|
||||
import com.doctate.watch.audio.PendingStore
|
||||
import com.doctate.watch.audio.PendingUpload
|
||||
import com.doctate.watch.domain.CaseStore
|
||||
import com.doctate.watch.net.UploadResult
|
||||
import java.io.File
|
||||
import kotlinx.coroutines.ExperimentalCoroutinesApi
|
||||
import kotlinx.coroutines.channels.Channel
|
||||
import kotlinx.coroutines.currentCoroutineContext
|
||||
import kotlinx.coroutines.delay
|
||||
import kotlinx.coroutines.isActive
|
||||
import kotlinx.coroutines.selects.onTimeout
|
||||
import kotlinx.coroutines.selects.select
|
||||
|
||||
/**
|
||||
* Pure upload + poll worker, testable on the JVM. Mirrors the loop in
|
||||
* `doctate-client-core::server_sync::run` — uploads have priority over
|
||||
* polling (no point asking the server about state if we still have the
|
||||
* doctor's audio in our pocket).
|
||||
*
|
||||
* The poll branch is wired in Phase 6; for now the empty-queue arm just
|
||||
* idles on the kick channel + interval.
|
||||
*
|
||||
* Backoff is purely in-process state. After a process restart it drops
|
||||
* back to [initialBackoffMs] and tries again immediately — which is the
|
||||
* right behaviour: "the watch came back to life, give the server one
|
||||
* fresh chance" is friendlier than "wait 60 s because we left the loop
|
||||
* mid-backoff".
|
||||
*
|
||||
* Architecturally split into [runOnce] (single iteration, suspends only
|
||||
* on real I/O — testable directly) and [run] (infinite loop, suspends
|
||||
* indefinitely on the idle branch — used by the production service).
|
||||
*/
|
||||
class SyncWorkerLoop(
|
||||
private val pendingStore: PendingStore,
|
||||
private val uploader: Uploader,
|
||||
private val caseStore: CaseStore,
|
||||
private val state: SyncStateProvider,
|
||||
private val kickChannel: Channel<SyncKick>,
|
||||
private val initialBackoffMs: Long = INITIAL_BACKOFF_MS,
|
||||
private val maxBackoffMs: Long = MAX_BACKOFF_MS,
|
||||
private val pollIntervalMsProvider: () -> Long = { DEFAULT_POLL_INTERVAL_MS },
|
||||
private val nowMs: () -> Long = { System.currentTimeMillis() },
|
||||
) {
|
||||
/** Functional seam: takes the same args as `UploadClient.upload`. */
|
||||
fun interface Uploader {
|
||||
suspend fun upload(caseId: String, recordedAt: String, audio: File): UploadResult
|
||||
}
|
||||
|
||||
/** Outcome of a single iteration. */
|
||||
sealed interface Step {
|
||||
/** A pending upload was processed, regardless of HTTP outcome. */
|
||||
data class Uploaded(val result: UploadResult) : Step
|
||||
/** Queue was empty when this iteration started. */
|
||||
data object Idle : Step
|
||||
}
|
||||
|
||||
private var backoffMs: Long = initialBackoffMs
|
||||
|
||||
/**
|
||||
* Run a single iteration. Uploads have priority over polling.
|
||||
*
|
||||
* - Pending non-empty: do exactly one upload and apply its outcome
|
||||
* (delete pair + mark synced on success, delete on terminal, keep
|
||||
* for retry on transient with backoff already incremented).
|
||||
* - Pending empty: publish Idle and return — caller decides how to
|
||||
* wait for the next wake-up (kick channel + poll timeout in
|
||||
* production; immediate return in tests).
|
||||
*/
|
||||
suspend fun runOnce(): Step {
|
||||
val pending = pendingStore.scan()
|
||||
val next = pending.firstOrNull()
|
||||
if (next != null) {
|
||||
publish(WorkerPhase.Uploading, pending)
|
||||
val result = uploader.upload(next.caseId, next.recordedAt, next.file)
|
||||
applyResult(next, result)
|
||||
return Step.Uploaded(result)
|
||||
}
|
||||
publish(WorkerPhase.Idle, emptyList())
|
||||
return Step.Idle
|
||||
}
|
||||
|
||||
/** Infinite loop variant for production. Idle branch waits on kick or poll interval. */
|
||||
@OptIn(ExperimentalCoroutinesApi::class)
|
||||
suspend fun run() {
|
||||
while (currentCoroutineContext().isActive) {
|
||||
when (runOnce()) {
|
||||
is Step.Uploaded -> Unit // proceed to next iteration immediately
|
||||
Step.Idle -> waitForKickOrTimeout()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@OptIn(ExperimentalCoroutinesApi::class)
|
||||
private suspend fun waitForKickOrTimeout() {
|
||||
select<Unit> {
|
||||
onTimeout(pollIntervalMsProvider()) {}
|
||||
kickChannel.onReceiveCatching { /* drained, just wake up */ }
|
||||
}
|
||||
}
|
||||
|
||||
private suspend fun applyResult(next: PendingUpload, result: UploadResult) {
|
||||
when (result) {
|
||||
is UploadResult.Success -> {
|
||||
pendingStore.deletePair(next.file)
|
||||
caseStore.markSynced(next.caseId)
|
||||
state.recordSuccess(nowMs())
|
||||
backoffMs = initialBackoffMs
|
||||
Log.i(TAG, "upload ok caseId=${next.caseId.take(8)} status=${result.status}")
|
||||
}
|
||||
is UploadResult.Transient -> {
|
||||
state.recordFailure(nowMs(), result.reason)
|
||||
Log.w(TAG, "transient: ${result.reason} → backoff ${backoffMs}ms")
|
||||
delay(backoffMs)
|
||||
backoffMs = (backoffMs * 2).coerceAtMost(maxBackoffMs)
|
||||
}
|
||||
is UploadResult.Terminal -> {
|
||||
// Server says "do not retry" (4xx) — drop the pair so the queue can drain.
|
||||
Log.w(TAG, "terminal: ${result.reason} → dropping caseId=${next.caseId.take(8)}")
|
||||
pendingStore.deletePair(next.file)
|
||||
state.recordFailure(nowMs(), result.reason)
|
||||
backoffMs = initialBackoffMs
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private fun publish(phase: WorkerPhase, pending: List<PendingUpload>) {
|
||||
val ids = pending.map { it.caseId }.toSet()
|
||||
when (phase) {
|
||||
WorkerPhase.Idle -> state.setIdle(pending.size, ids)
|
||||
WorkerPhase.Uploading -> state.setUploading(pending.size, ids)
|
||||
WorkerPhase.Polling -> state.setPolling(pending.size, ids)
|
||||
}
|
||||
}
|
||||
|
||||
/** Test-only accessor. */
|
||||
internal val currentBackoffMs: Long get() = backoffMs
|
||||
|
||||
companion object {
|
||||
private const val TAG = "SyncWorkerLoop"
|
||||
const val INITIAL_BACKOFF_MS: Long = 2_000L
|
||||
const val MAX_BACKOFF_MS: Long = 60_000L
|
||||
const val DEFAULT_POLL_INTERVAL_MS: Long = 30_000L
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,147 @@
|
||||
package com.doctate.watch.sync
|
||||
|
||||
import com.doctate.watch.audio.PendingStore
|
||||
import com.doctate.watch.audio.PendingUpload
|
||||
import com.doctate.watch.net.UploadResult
|
||||
import com.doctate.watch.testfixtures.InMemoryCaseStore
|
||||
import com.google.common.truth.Truth.assertThat
|
||||
import java.io.File
|
||||
import kotlinx.coroutines.ExperimentalCoroutinesApi
|
||||
import kotlinx.coroutines.channels.Channel
|
||||
import kotlinx.coroutines.test.runTest
|
||||
import org.junit.Rule
|
||||
import org.junit.Test
|
||||
import org.junit.rules.TemporaryFolder
|
||||
|
||||
@OptIn(ExperimentalCoroutinesApi::class)
|
||||
class SyncWorkerLoopTest {
|
||||
@get:Rule val tempDir = TemporaryFolder()
|
||||
|
||||
private fun newPendingStore(): PendingStore = PendingStore(File(tempDir.root, "unsynced"))
|
||||
|
||||
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 success_deletes_pair_marks_synced_and_resets_backoff() = runTest {
|
||||
val store = InMemoryCaseStore()
|
||||
val caseId = store.createLocal()
|
||||
val pending = newPendingStore()
|
||||
val audio = pending.seed(caseId, "2026-04-26T10:00:00Z")
|
||||
val state = SyncStateProvider()
|
||||
val uploader = SyncWorkerLoop.Uploader { c, t, _ -> UploadResult.Success(c, t, "received") }
|
||||
val loop = SyncWorkerLoop(pending, uploader, store, state, Channel(Channel.UNLIMITED))
|
||||
|
||||
val step = loop.runOnce()
|
||||
assertThat(step).isInstanceOf(SyncWorkerLoop.Step.Uploaded::class.java)
|
||||
assertThat(audio.exists()).isFalse()
|
||||
assertThat(pending.sidecarPathFor(audio).exists()).isFalse()
|
||||
assertThat(store.casesFlow.value.first { it.caseId == caseId }.syncedToServer).isTrue()
|
||||
assertThat(state.state.value.lastSuccessAt).isNotNull()
|
||||
}
|
||||
|
||||
@Test fun fifo_order_processes_oldest_recordedAt_first() = runTest {
|
||||
val store = InMemoryCaseStore()
|
||||
val pending = newPendingStore()
|
||||
pending.seed("case-newer", "2026-04-26T12:00:00Z")
|
||||
pending.seed("case-older", "2026-04-26T10:00:00Z")
|
||||
|
||||
val processed = mutableListOf<String>()
|
||||
val state = SyncStateProvider()
|
||||
val uploader = SyncWorkerLoop.Uploader { c, t, _ ->
|
||||
processed += c
|
||||
UploadResult.Success(c, t, "received")
|
||||
}
|
||||
val loop = SyncWorkerLoop(pending, uploader, store, state, Channel(Channel.UNLIMITED))
|
||||
|
||||
loop.runOnce() // first
|
||||
loop.runOnce() // second
|
||||
assertThat(processed).containsExactly("case-older", "case-newer").inOrder()
|
||||
}
|
||||
|
||||
@Test fun transient_keeps_pair_and_increments_backoff_capped() = runTest {
|
||||
val store = InMemoryCaseStore()
|
||||
val pending = newPendingStore()
|
||||
val audio = pending.seed("case-x", "2026-04-26T10:00:00Z")
|
||||
|
||||
val state = SyncStateProvider()
|
||||
val uploader = SyncWorkerLoop.Uploader { _, _, _ -> UploadResult.Transient("HTTP 503") }
|
||||
val loop = SyncWorkerLoop(
|
||||
pending, uploader, store, state, Channel(Channel.UNLIMITED),
|
||||
initialBackoffMs = 100L,
|
||||
maxBackoffMs = 800L,
|
||||
)
|
||||
|
||||
// Each runOnce in a transient state delays internally — runTest uses
|
||||
// virtual time, so this completes instantly under the test scheduler.
|
||||
loop.runOnce(); assertThat(loop.currentBackoffMs).isEqualTo(200L)
|
||||
loop.runOnce(); assertThat(loop.currentBackoffMs).isEqualTo(400L)
|
||||
loop.runOnce(); assertThat(loop.currentBackoffMs).isEqualTo(800L)
|
||||
loop.runOnce(); assertThat(loop.currentBackoffMs).isEqualTo(800L) // capped
|
||||
|
||||
assertThat(audio.exists()).isTrue()
|
||||
assertThat(pending.sidecarPathFor(audio).exists()).isTrue()
|
||||
assertThat(state.state.value.lastFailureReason).contains("HTTP 503")
|
||||
}
|
||||
|
||||
@Test fun terminal_drops_pair_no_retry() = runTest {
|
||||
val store = InMemoryCaseStore()
|
||||
val pending = newPendingStore()
|
||||
val audio = pending.seed("case-terminal", "2026-04-26T10:00:00Z")
|
||||
|
||||
var calls = 0
|
||||
val state = SyncStateProvider()
|
||||
val uploader = SyncWorkerLoop.Uploader { _, _, _ ->
|
||||
calls++
|
||||
UploadResult.Terminal("HTTP 401")
|
||||
}
|
||||
val loop = SyncWorkerLoop(pending, uploader, store, state, Channel(Channel.UNLIMITED))
|
||||
|
||||
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)
|
||||
}
|
||||
|
||||
@Test fun idle_branch_publishes_zero_queue() = runTest {
|
||||
val store = InMemoryCaseStore()
|
||||
val pending = newPendingStore()
|
||||
val state = SyncStateProvider()
|
||||
val uploader = SyncWorkerLoop.Uploader { _, _, _ -> error("must not be called") }
|
||||
val loop = SyncWorkerLoop(pending, uploader, store, state, Channel(Channel.UNLIMITED))
|
||||
|
||||
val step = loop.runOnce()
|
||||
assertThat(step).isEqualTo(SyncWorkerLoop.Step.Idle)
|
||||
assertThat(state.state.value.phase).isEqualTo(WorkerPhase.Idle)
|
||||
assertThat(state.state.value.queueLen).isEqualTo(0)
|
||||
}
|
||||
|
||||
@Test fun success_after_transient_resets_backoff() = runTest {
|
||||
val store = InMemoryCaseStore()
|
||||
val pending = newPendingStore()
|
||||
pending.seed("case-y", "2026-04-26T10:00:00Z")
|
||||
|
||||
var attempt = 0
|
||||
val state = SyncStateProvider()
|
||||
val uploader = SyncWorkerLoop.Uploader { c, t, _ ->
|
||||
attempt++
|
||||
if (attempt == 1) UploadResult.Transient("server warming up")
|
||||
else UploadResult.Success(c, t, "received")
|
||||
}
|
||||
val loop = SyncWorkerLoop(
|
||||
pending, uploader, store, state, Channel(Channel.UNLIMITED),
|
||||
initialBackoffMs = 100L, maxBackoffMs = 800L,
|
||||
)
|
||||
|
||||
loop.runOnce() // transient → backoff doubles to 200
|
||||
assertThat(loop.currentBackoffMs).isEqualTo(200L)
|
||||
loop.runOnce() // success → backoff resets
|
||||
assertThat(loop.currentBackoffMs).isEqualTo(100L)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user