From d2a22e49e3e1f82db20380591f576aaa7f538825 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Thu, 11 Jun 2026 12:05:35 -0400 Subject: [PATCH] fix: harden offline/playback recording (contract-audit follow-ups) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Audit of every Android↔server connection point (2026-06-11) cleared the silent-contract class that caused the events `type` bug, but surfaced a cluster of offline/playback-robustness defects. Fixes: 1. Playback double-count on background. PlayEventsReporter no longer enqueues a partial play_offline + leaves the live row open on every screen-lock. closeCurrent() now routes by whether the server has an open row: close-by-id (durable PLAY_ENDED on failure) when it does, offline only when no row exists, and a no-op while a play_started is in flight (the server auto-closes that orphan). onStop only durably closes a *paused* play — a still-playing one is left to the live path under the foreground service. Adds the PLAY_ENDED mutation kind. 2. Replayer poison rows. MutationReplayer now classifies each replay as SENT / DROP / RETRY: permanent 4xx (and corrupt payloads) are dropped instead of retried forever; 408/429/5xx/transport still retry. 3. Offline-play / close-by-id idempotency (server). RecordOfflinePlay dedups on (user, track, started_at); RecordPlayEnded skips a second skip_events insert when re-closing an already-ended row. Makes the at-least-once replay safe against lost-response duplicates. 4. Like-toggle collapse. Replayer drops like-toggles superseded by a later toggle for the same entity, so partial-failure + differential retry can't invert the final like state. 5. Connectivity-return trigger. MutationReplayer + SyncController now also drain/sync when NetworkStatusController recovers to Healthy, so an offline→online transition mid-session doesn't wait for a cold start. SyncController.syncSafe gains a single-in-flight mutex. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../minstrel/cache/mutations/MutationQueue.kt | 31 ++ .../cache/mutations/MutationReplayer.kt | 324 +++++++++--------- .../minstrel/cache/sync/SyncController.kt | 26 +- .../minstrel/player/PlayEventsReporter.kt | 106 ++++-- internal/playevents/writer.go | 26 +- internal/playevents/writer_test.go | 49 +++ 6 files changed, 373 insertions(+), 189 deletions(-) diff --git a/android/app/src/main/java/com/fabledsword/minstrel/cache/mutations/MutationQueue.kt b/android/app/src/main/java/com/fabledsword/minstrel/cache/mutations/MutationQueue.kt index ec72f104..607f2839 100644 --- a/android/app/src/main/java/com/fabledsword/minstrel/cache/mutations/MutationQueue.kt +++ b/android/app/src/main/java/com/fabledsword/minstrel/cache/mutations/MutationQueue.kt @@ -28,6 +28,13 @@ object MutationKind { const val PLAY_OFFLINE: String = "play_offline" const val REQUEST_CANCEL: String = "request_cancel" const val PLAYBACK_ERROR_REPORT: String = "playback_error_report" + + // Durable close-by-id for a play whose server row is already open + // (play_started succeeded). Distinct from PLAY_OFFLINE, which records + // a *whole* play when no server row exists. Using close-by-id on + // background avoids the duplicate + orphan row the old offline-on-stop + // path produced (see 2026-06-11 contract audit). + const val PLAY_ENDED: String = "play_ended" } /** @@ -167,6 +174,17 @@ class MutationQueue @Inject constructor( ), ) + /** + * Durable close-by-id for an already-open server play row. Background + * path only (no user hint) — same as [enqueuePlayOffline]. + */ + suspend fun enqueuePlayEnded(payload: PlayEndedPayload): Long = dao.insert( + CachedMutationEntity( + kind = MutationKind.PLAY_ENDED, + payload = json.encodeToString(PlayEndedPayload.serializer(), payload), + ), + ) + private suspend fun insertUserDriven(kind: String, payload: String): Long { val id = dao.insert(CachedMutationEntity(kind = kind, payload = payload)) _userEnqueueHints.tryEmit(QUEUED_MESSAGE) @@ -223,6 +241,19 @@ data class PlayOfflinePayload( val source: String? = null, ) +/** + * Persisted payload for `MutationKind.PLAY_ENDED`. The replayer re-fires + * `POST /api/events` with type=play_ended, closing the already-open server + * row [playEventId] at [durationPlayedMs]. Server-side close-by-id is + * idempotent (re-ending updates the same row and skips a duplicate + * skip_events insert), so a lost-response retry can't duplicate history. + */ +@Serializable +data class PlayEndedPayload( + val playEventId: String, + val durationPlayedMs: Long, +) + /** * Persisted payload for `MutationKind.REQUEST_CANCEL` — the * `POST /api/requests/{id}/cancel` call lost during a connectivity diff --git a/android/app/src/main/java/com/fabledsword/minstrel/cache/mutations/MutationReplayer.kt b/android/app/src/main/java/com/fabledsword/minstrel/cache/mutations/MutationReplayer.kt index 4fd4c8fb..6d08a5f8 100644 --- a/android/app/src/main/java/com/fabledsword/minstrel/cache/mutations/MutationReplayer.kt +++ b/android/app/src/main/java/com/fabledsword/minstrel/cache/mutations/MutationReplayer.kt @@ -1,3 +1,5 @@ +@file:Suppress("TooManyFunctions") // one dispatcher per mutation kind + replay helpers + package com.fabledsword.minstrel.cache.mutations import com.fabledsword.minstrel.api.endpoints.AppendTracksRequest @@ -10,6 +12,9 @@ import com.fabledsword.minstrel.api.endpoints.PlaybackErrorsApi import com.fabledsword.minstrel.api.endpoints.PlaylistsApi import com.fabledsword.minstrel.api.endpoints.QuarantineApi import com.fabledsword.minstrel.api.endpoints.RequestsApi +import com.fabledsword.minstrel.connectivity.NetworkStatusController +import com.fabledsword.minstrel.connectivity.ServerHealth +import com.fabledsword.minstrel.models.wire.PlayEndedRequest import com.fabledsword.minstrel.models.wire.PlayOfflineRequest import com.fabledsword.minstrel.auth.AuthStore import com.fabledsword.minstrel.cache.db.dao.CachedMutationDao @@ -19,12 +24,16 @@ import com.fabledsword.minstrel.likes.data.LikesRepository import com.fabledsword.minstrel.models.wire.CreateRequestBody import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.flow.distinctUntilChanged +import kotlinx.coroutines.flow.filter import kotlinx.coroutines.flow.filterNotNull +import kotlinx.coroutines.flow.map import kotlinx.coroutines.launch import kotlinx.coroutines.sync.Mutex import kotlinx.coroutines.sync.withLock import kotlinx.datetime.Clock +import kotlinx.serialization.SerializationException import kotlinx.serialization.json.Json +import retrofit2.HttpException import retrofit2.Retrofit import retrofit2.create import javax.inject.Inject @@ -34,25 +43,24 @@ import javax.inject.Singleton * Drains the offline MutationQueue. Reads pending rows in FIFO order, * re-fires each call against the raw Retrofit API (NOT the high-level * Repository, which would re-enqueue on failure → infinite loop), and - * deletes the row on 2xx success. + * resolves each to an [Outcome]: + * - SENT — server accepted (2xx) → delete the row. + * - DROP — permanently un-sendable (4xx, unknown kind/variant, corrupt + * payload) → delete the row so it can't wedge the queue forever. + * - RETRY — transient (transport, 5xx, 408, 429) → leave queued, bump + * attempts, retry on the next trigger. * * Triggered by: - * - app start (MinstrelApplication @Inject forces the singleton's - * init block to subscribe) - * - every time AuthStore.sessionCookie transitions to a non-null - * value (sign-in, app launch with persisted cookie) + * - app start (MinstrelApplication @Inject forces the singleton's init) + * - AuthStore.sessionCookie transitioning to non-null (sign-in / cold + * start with a persisted cookie) + * - NetworkStatusController recovering to Healthy (server reachable again + * mid-session — covers offline→online without needing a cold start) * - * Single in-flight via the `mutex` — back-to-back triggers collapse - * into one drain pass. A proper connectivity-listener / WorkManager - * job that re-tries on Wi-Fi-came-back is a follow-up; for MVP the - * sign-in + app-open triggers cover the common offline → online - * transition (operator unlocks phone, opens app). - * - * Failed rows stay in the queue with their `attempts` and - * `lastAttemptAt` updated. They get another chance on the next - * trigger; no exponential backoff or max-attempts yet — the rows are - * small (low single-digit count for typical use) so retry-forever is - * cheap. + * Single in-flight via the [mutex]; back-to-back triggers collapse into one + * drain pass. No exponential backoff / max-attempts: the rows are few for + * typical use, and permanent failures are now dropped rather than retried + * forever. */ @Singleton class MutationReplayer @Inject constructor( @@ -60,6 +68,7 @@ class MutationReplayer @Inject constructor( private val authStore: AuthStore, private val json: Json, @ApplicationScope private val scope: CoroutineScope, + networkStatus: NetworkStatusController, retrofit: Retrofit, ) { private val likesApi: LikesApi = retrofit.create() @@ -72,6 +81,8 @@ class MutationReplayer @Inject constructor( private val mutex = Mutex() + private enum class Outcome { SENT, DROP, RETRY } + init { scope.launch { authStore.sessionCookie @@ -79,6 +90,16 @@ class MutationReplayer @Inject constructor( .distinctUntilChanged() .collect { drainSafe() } } + // Server-reachable-again: replay queued writes the moment the + // health signal recovers, so an offline→online transition mid- + // session doesn't wait for the next cold start / re-auth. + scope.launch { + networkStatus.state + .map { it == ServerHealth.Healthy } + .distinctUntilChanged() + .filter { it } + .collect { drainSafe() } + } } /** Public entry-point; coalesces concurrent triggers via a Mutex. */ @@ -93,193 +114,182 @@ class MutationReplayer @Inject constructor( private suspend fun drain() { val rows = dao.getAll() + // Collapse superseded like-toggles: only the latest desired state per + // (entity) is replayed; older toggles for the same entity are dropped + // unsent. Without this, partial-failure + differential retry could + // replay an older toggle last and invert the final like state. + val superseded = supersededLikeToggleIds(rows) for (row in rows) { - val sent = runCatching { dispatch(row) }.getOrDefault(false) - if (sent) { + if (row.id in superseded) { dao.delete(row.id) - } else { - dao.recordAttempt(row.id, Clock.System.now()) + continue + } + when (outcomeFor(row)) { + Outcome.SENT, Outcome.DROP -> dao.delete(row.id) + Outcome.RETRY -> dao.recordAttempt(row.id, Clock.System.now()) } } } + /** Row ids of like-toggles superseded by a later toggle for the same entity. */ + private fun supersededLikeToggleIds(rows: List): Set { + val latestByEntity = HashMap() + val superseded = HashSet() + for (row in rows) { + if (row.kind != MutationKind.LIKE_TOGGLE) continue + val decoded = runCatching { + json.decodeFromString(LikeTogglePayload.serializer(), row.payload) + }.getOrNull() ?: continue + val key = "${decoded.entityType}:${decoded.entityId}" + // `rows` is ascending by id, so a prior entry is always older. + latestByEntity.put(key, row.id)?.let(superseded::add) + } + return superseded + } + + private suspend fun outcomeFor(row: CachedMutationEntity): Outcome = try { + dispatch(row) + } catch (e: HttpException) { + if (isPermanent(e.code())) Outcome.DROP else Outcome.RETRY + } catch (@Suppress("SwallowedException") e: SerializationException) { + // Corrupt / schema-incompatible payload — permanent, drop it. + Outcome.DROP + } catch (@Suppress("TooGenericExceptionCaught", "SwallowedException") e: Throwable) { + // Transport / unknown — leave queued for the next pass. + Outcome.RETRY + } + + /** 4xx are permanent except the two "retry me" statuses. */ + private fun isPermanent(code: Int): Boolean = + code in HTTP_CLIENT_ERR_MIN..HTTP_CLIENT_ERR_MAX && + code != HTTP_TIMEOUT && code != HTTP_TOO_MANY + /** - * Returns true if the server accepted the call, false on any - * failure (transport, server error, decode). The replayer treats - * false as "leave the row for next pass" — it doesn't try to - * distinguish permanent rejections from transient ones; a row - * that consistently fails will accumulate attempt count and - * lastAttemptAt for diagnostics. + * Returns the outcome of replaying one row. SENT on a 2xx; DROP for an + * unknown kind/variant. Network / HTTP failures THROW and are classified + * by [outcomeFor] — dispatchers no longer swallow. */ - private suspend fun dispatch(row: CachedMutationEntity): Boolean = when (row.kind) { + private suspend fun dispatch(row: CachedMutationEntity): Outcome = when (row.kind) { MutationKind.LIKE_TOGGLE -> dispatchLikeToggle(row.payload) MutationKind.REQUEST_CREATE -> dispatchRequestCreate(row.payload) MutationKind.QUARANTINE_UNFLAG -> dispatchQuarantineUnflag(row.payload) MutationKind.PLAYLIST_APPEND -> dispatchPlaylistAppend(row.payload) MutationKind.QUARANTINE_FLAG -> dispatchQuarantineFlag(row.payload) MutationKind.PLAY_OFFLINE -> dispatchPlayOffline(row.payload) + MutationKind.PLAY_ENDED -> dispatchPlayEnded(row.payload) MutationKind.REQUEST_CANCEL -> dispatchRequestCancel(row.payload) MutationKind.PLAYBACK_ERROR_REPORT -> dispatchPlaybackErrorReport(row.payload) - else -> { - // Unknown kind — drop the row by claiming success so a - // stale schema entry can't wedge the queue forever. - true - } + // Unknown kind — drop so a stale schema entry can't wedge the queue. + else -> Outcome.DROP } - private suspend fun dispatchLikeToggle(payload: String): Boolean { + private suspend fun dispatchLikeToggle(payload: String): Outcome { val decoded = json.decodeFromString(LikeTogglePayload.serializer(), payload) val kindPath = when (decoded.entityType) { LikesRepository.ENTITY_ARTIST -> "artists" LikesRepository.ENTITY_ALBUM -> "albums" LikesRepository.ENTITY_TRACK -> "tracks" - else -> return true // unknown variant — drop. + else -> return Outcome.DROP } - return try { - if (decoded.desiredState) { - likesApi.like(kindPath, decoded.entityId) - } else { - likesApi.unlike(kindPath, decoded.entityId) - } - true - } catch ( - @Suppress("TooGenericExceptionCaught", "SwallowedException") e: Throwable, - ) { - // Intentional swallow: the row stays in the queue with - // attempts + lastAttemptAt incremented; next drain pass - // retries. Per-attempt diagnostic logging lands when we - // wire up Timber at the replayer level. - false + if (decoded.desiredState) { + likesApi.like(kindPath, decoded.entityId) + } else { + likesApi.unlike(kindPath, decoded.entityId) } + return Outcome.SENT } - private suspend fun dispatchRequestCreate(payload: String): Boolean { + private suspend fun dispatchRequestCreate(payload: String): Outcome { val decoded = json.decodeFromString(RequestCreatePayload.serializer(), payload) - val body = CreateRequestBody( - kind = decoded.kind, - artistMbid = decoded.artistMbid, - artistName = decoded.artistName, - albumMbid = decoded.albumMbid, - albumTitle = decoded.albumTitle, - trackMbid = decoded.trackMbid, - trackTitle = decoded.trackTitle, + discoverApi.createRequest( + CreateRequestBody( + kind = decoded.kind, + artistMbid = decoded.artistMbid, + artistName = decoded.artistName, + albumMbid = decoded.albumMbid, + albumTitle = decoded.albumTitle, + trackMbid = decoded.trackMbid, + trackTitle = decoded.trackTitle, + ), ) - return try { - discoverApi.createRequest(body) - true - } catch ( - @Suppress("TooGenericExceptionCaught", "SwallowedException") e: Throwable, - ) { - // Intentional swallow: the row stays in the queue with - // attempts + lastAttemptAt incremented; next drain pass - // retries. Per-attempt diagnostic logging lands when we - // wire up Timber at the replayer level. - false - } + return Outcome.SENT } - private suspend fun dispatchQuarantineUnflag(payload: String): Boolean { + private suspend fun dispatchQuarantineUnflag(payload: String): Outcome { val decoded = json.decodeFromString(QuarantineUnflagPayload.serializer(), payload) - return try { - quarantineApi.unflag(decoded.trackId) - true - } catch ( - @Suppress("TooGenericExceptionCaught", "SwallowedException") e: Throwable, - ) { - // Intentional swallow: the row stays in the queue with - // attempts + lastAttemptAt incremented; next drain pass - // retries. Per-attempt diagnostic logging lands when we - // wire up Timber at the replayer level. - false - } + quarantineApi.unflag(decoded.trackId) + return Outcome.SENT } - private suspend fun dispatchPlaylistAppend(payload: String): Boolean { + private suspend fun dispatchPlaylistAppend(payload: String): Outcome { val decoded = json.decodeFromString(PlaylistAppendPayload.serializer(), payload) - return try { - playlistsApi.appendTracks( - playlistId = decoded.playlistId, - body = AppendTracksRequest(trackIds = decoded.trackIds), - ) - true - } catch ( - @Suppress("TooGenericExceptionCaught", "SwallowedException") e: Throwable, - ) { - // Intentional swallow: row stays queued; next drain pass retries. - false - } + playlistsApi.appendTracks( + playlistId = decoded.playlistId, + body = AppendTracksRequest(trackIds = decoded.trackIds), + ) + return Outcome.SENT } - private suspend fun dispatchQuarantineFlag(payload: String): Boolean { + private suspend fun dispatchQuarantineFlag(payload: String): Outcome { val decoded = json.decodeFromString(QuarantineFlagPayload.serializer(), payload) - return try { - quarantineApi.flag( - FlagRequest( - trackId = decoded.trackId, - reason = decoded.reason, - notes = decoded.notes, - ), - ) - true - } catch ( - @Suppress("TooGenericExceptionCaught", "SwallowedException") e: Throwable, - ) { - // Intentional swallow: row stays queued; next drain pass retries. - false - } + quarantineApi.flag( + FlagRequest( + trackId = decoded.trackId, + reason = decoded.reason, + notes = decoded.notes, + ), + ) + return Outcome.SENT } - private suspend fun dispatchPlayOffline(payload: String): Boolean { + private suspend fun dispatchPlayOffline(payload: String): Outcome { val decoded = json.decodeFromString(PlayOfflinePayload.serializer(), payload) - return try { - eventsApi.playOffline( - PlayOfflineRequest( - trackId = decoded.trackId, - clientId = decoded.clientId, - at = decoded.atIso, - durationPlayedMs = decoded.durationPlayedMs, - source = decoded.source, - ), - ) - true - } catch ( - @Suppress("TooGenericExceptionCaught", "SwallowedException") e: Throwable, - ) { - // Intentional swallow: row stays queued; next drain pass retries. - false - } + eventsApi.playOffline( + PlayOfflineRequest( + trackId = decoded.trackId, + clientId = decoded.clientId, + at = decoded.atIso, + durationPlayedMs = decoded.durationPlayedMs, + source = decoded.source, + ), + ) + return Outcome.SENT } - private suspend fun dispatchRequestCancel(payload: String): Boolean { + private suspend fun dispatchPlayEnded(payload: String): Outcome { + val decoded = json.decodeFromString(PlayEndedPayload.serializer(), payload) + eventsApi.playEnded( + PlayEndedRequest( + playEventId = decoded.playEventId, + durationPlayedMs = decoded.durationPlayedMs, + ), + ) + return Outcome.SENT + } + + private suspend fun dispatchRequestCancel(payload: String): Outcome { val decoded = json.decodeFromString(RequestCancelPayload.serializer(), payload) - return try { - requestsApi.cancel(decoded.requestId) - true - } catch ( - @Suppress("TooGenericExceptionCaught", "SwallowedException") e: Throwable, - ) { - // Intentional swallow: row stays queued; next drain pass retries. - false - } + requestsApi.cancel(decoded.requestId) + return Outcome.SENT } - private suspend fun dispatchPlaybackErrorReport(payload: String): Boolean { + private suspend fun dispatchPlaybackErrorReport(payload: String): Outcome { val decoded = json.decodeFromString(PlaybackErrorReportPayload.serializer(), payload) - return try { - playbackErrorsApi.report( - PlaybackErrorReportRequest( - trackId = decoded.trackId, - kind = decoded.kind, - detail = decoded.detail, - clientId = decoded.clientId, - ), - ) - true - } catch ( - @Suppress("TooGenericExceptionCaught", "SwallowedException") e: Throwable, - ) { - // Intentional swallow: row stays queued; next drain pass retries. - false - } + playbackErrorsApi.report( + PlaybackErrorReportRequest( + trackId = decoded.trackId, + kind = decoded.kind, + detail = decoded.detail, + clientId = decoded.clientId, + ), + ) + return Outcome.SENT + } + + private companion object { + const val HTTP_CLIENT_ERR_MIN = 400 + const val HTTP_CLIENT_ERR_MAX = 499 + const val HTTP_TIMEOUT = 408 + const val HTTP_TOO_MANY = 429 } } diff --git a/android/app/src/main/java/com/fabledsword/minstrel/cache/sync/SyncController.kt b/android/app/src/main/java/com/fabledsword/minstrel/cache/sync/SyncController.kt index d50fac19..635f7290 100644 --- a/android/app/src/main/java/com/fabledsword/minstrel/cache/sync/SyncController.kt +++ b/android/app/src/main/java/com/fabledsword/minstrel/cache/sync/SyncController.kt @@ -2,6 +2,8 @@ package com.fabledsword.minstrel.cache.sync import com.fabledsword.minstrel.api.endpoints.SyncApi import com.fabledsword.minstrel.auth.AuthStore +import com.fabledsword.minstrel.connectivity.NetworkStatusController +import com.fabledsword.minstrel.connectivity.ServerHealth import com.fabledsword.minstrel.cache.db.dao.CachedAlbumDao import com.fabledsword.minstrel.cache.db.dao.CachedArtistDao import com.fabledsword.minstrel.cache.db.dao.CachedTrackDao @@ -16,8 +18,11 @@ import com.fabledsword.minstrel.models.wire.SyncArtistWire import com.fabledsword.minstrel.models.wire.SyncTrackWire import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.flow.distinctUntilChanged +import kotlinx.coroutines.flow.filter import kotlinx.coroutines.flow.filterNotNull +import kotlinx.coroutines.flow.map import kotlinx.coroutines.launch +import kotlinx.coroutines.sync.Mutex import kotlinx.datetime.Clock import retrofit2.Retrofit import retrofit2.create @@ -50,10 +55,15 @@ class SyncController @Inject constructor( private val trackDao: CachedTrackDao, private val authStore: AuthStore, @ApplicationScope private val scope: CoroutineScope, + networkStatus: NetworkStatusController, retrofit: Retrofit, ) { private val api: SyncApi = retrofit.create() + // Single in-flight guard: the cookie + health-recovery triggers (and a + // pull-to-refresh) can fire concurrently; coalesce them into one pass. + private val mutex = Mutex() + init { scope.launch { authStore.sessionCookie @@ -61,11 +71,25 @@ class SyncController @Inject constructor( .distinctUntilChanged() .collect { syncSafe() } } + // Re-sync when the server becomes reachable again mid-session, so a + // delta missed while offline lands without waiting for a cold start. + scope.launch { + networkStatus.state + .map { it == ServerHealth.Healthy } + .distinctUntilChanged() + .filter { it } + .collect { syncSafe() } + } } /** Public entry-point for "Sync now" affordances. Swallows errors. */ suspend fun syncSafe() { - runCatching { sync() } + if (!mutex.tryLock()) return + try { + runCatching { sync() } + } finally { + mutex.unlock() + } } private suspend fun sync() { diff --git a/android/app/src/main/java/com/fabledsword/minstrel/player/PlayEventsReporter.kt b/android/app/src/main/java/com/fabledsword/minstrel/player/PlayEventsReporter.kt index a8fe713c..0325fba9 100644 --- a/android/app/src/main/java/com/fabledsword/minstrel/player/PlayEventsReporter.kt +++ b/android/app/src/main/java/com/fabledsword/minstrel/player/PlayEventsReporter.kt @@ -6,6 +6,7 @@ import androidx.lifecycle.ProcessLifecycleOwner import com.fabledsword.minstrel.api.endpoints.EventsApi import com.fabledsword.minstrel.auth.AuthStore import com.fabledsword.minstrel.cache.mutations.MutationQueue +import com.fabledsword.minstrel.cache.mutations.PlayEndedPayload import com.fabledsword.minstrel.cache.mutations.PlayOfflinePayload import com.fabledsword.minstrel.di.ApplicationScope import com.fabledsword.minstrel.models.wire.PlayEndedRequest @@ -64,11 +65,24 @@ class PlayEventsReporter @Inject constructor( private var curLastPositionMs: Long = 0 private var curDurationMs: Long = 0 + // Whether the currently-tracked play is actively playing (vs paused). + // onStop reads this: a backgrounded play that's still playing is left + // to the live path (the foreground service keeps us alive), while a + // paused one is durably closed so a kill can't orphan its open row. + private var curPlaying: Boolean = false + // Server play_event_id when the live play_started succeeded; // null if start failed/offline — then the close is captured into // the offline mutation queue instead of a live ended/skipped. private var openPlayEventId: String? = null + // True from beginTrack until startLive's play_started resolves. When a + // close races an in-flight start (id still null), the server row is + // being created but we've lost its id — we must NOT also enqueue an + // offline play (that would duplicate); the server auto-closes the + // orphan on the next play_started instead. + private var startInFlight: Boolean = false + private var prevTrackId: String? = null init { @@ -95,7 +109,7 @@ class PlayEventsReporter @Inject constructor( ) { // Track-change → close the prior tracked play. if (trackId != prevTrackId && curTrackId != null) { - closeCurrent(viaOffline = false) + closeCurrent() } // Entered playing for a new track → begin tracking it. @@ -110,6 +124,7 @@ class PlayEventsReporter @Inject constructor( if (curTrackId != null && trackId == curTrackId) { curLastPositionMs = positionMs if (durationMs > 0) curDurationMs = durationMs + curPlaying = playing } prevTrackId = trackId @@ -121,12 +136,18 @@ class PlayEventsReporter @Inject constructor( curSource = source curLastPositionMs = 0 curDurationMs = durationMs + curPlaying = true openPlayEventId = null + startInFlight = true startLive(trackId = trackId, source = source) } private fun startLive(trackId: String, source: String?) { - val cid = resolveClientId() ?: return + val cid = resolveClientId() + if (cid == null) { + startInFlight = false + return + } scope.launch { try { val response = api.playStarted( @@ -144,56 +165,72 @@ class PlayEventsReporter @Inject constructor( ) { // Offline / flaky — openPlayEventId stays null; close // path will enqueue the completed play for replay. + } finally { + startInFlight = false } } } - private fun closeCurrent(viaOffline: Boolean) { + /** + * Closes the tracked play. Routes by whether the server already has an + * open row, NOT by caller context — this is what keeps background / + * track-change / failure paths from each minting a second row: + * - open server row (id) → close it by id; on failure enqueue a durable + * close-by-id (never an offline play). + * - start in flight (no id yet) → leave it; the server auto-closes the + * orphan on the next play_started. Enqueuing offline here would + * duplicate the in-flight row. + * - no server row at all → record the whole play offline. + * + * Always uses the actual position played; the server's skip rule + * (50%/30s defaults) classifies was_skipped from it. + */ + private fun closeCurrent() { val trackId = curTrackId val startedAt = curStartedAt if (trackId == null || startedAt == null) { resetCurrent() return } - // Always send the actual position played to; the server's - // skip rule (50%/30s defaults, configurable) classifies - // was_skipped from this. If the track ran to completion the - // last observed position is ~= duration, which gives ratio ~1 - // and the server records a non-skip. The COMPLETION_TOLERANCE - // rounding the prior shape did is no longer needed — the - // server's threshold is the source of truth. val durationMs = curLastPositionMs val source = curSource val id = openPlayEventId - - if (!viaOffline && id != null) { - scope.launch { - try { - api.playEnded( - PlayEndedRequest( - playEventId = id, - durationPlayedMs = durationMs, - ), - ) - } catch ( - @Suppress("TooGenericExceptionCaught", "SwallowedException") e: Throwable, - ) { - enqueueOffline(trackId, startedAt, source, durationMs) - } - } - } else { - enqueueOffline(trackId, startedAt, source, durationMs) + val inFlight = startInFlight + when { + id != null -> closeById(id, durationMs) + inFlight -> Unit + else -> enqueueOffline(trackId, startedAt, source, durationMs) } resetCurrent() } + private fun closeById(id: String, durationMs: Long) { + scope.launch { + try { + api.playEnded( + PlayEndedRequest(playEventId = id, durationPlayedMs = durationMs), + ) + } catch ( + @Suppress("TooGenericExceptionCaught", "SwallowedException") e: Throwable, + ) { + // Durable close-by-id (idempotent server-side) — survives a + // process kill without orphaning the open row or duplicating. + mutationQueue.enqueuePlayEnded( + PlayEndedPayload(playEventId = id, durationPlayedMs = durationMs), + ) + } + } + } + private fun resetCurrent() { curTrackId = null curStartedAt = null curSource = null curLastPositionMs = 0 curDurationMs = 0 + curPlaying = false openPlayEventId = null + startInFlight = false } private fun enqueueOffline( @@ -232,8 +269,17 @@ class PlayEventsReporter @Inject constructor( // ── Lifecycle: durable-close on app background / detach ──────── override fun onStop(owner: LifecycleOwner) { - if (curTrackId != null) { - closeCurrent(viaOffline = true) + // App backgrounded. If the tracked play is still PLAYING, leave it + // alone: the foreground media service keeps the process — and this + // app-scope uiState collector — alive, so the live state machine + // closes the play accurately on the next real track change. Closing + // here (the old always-offline behavior) raced the collector's + // re-begin and minted a duplicate, skip-misclassified row on every + // screen-lock. Only durably close a PAUSED play, where no further + // progress will happen and a process kill would otherwise orphan the + // open server row. See docs/superpowers/2026-06-11 contract audit. + if (curTrackId != null && !curPlaying) { + closeCurrent() } } } diff --git a/internal/playevents/writer.go b/internal/playevents/writer.go index 9d3a75f0..a65ba169 100644 --- a/internal/playevents/writer.go +++ b/internal/playevents/writer.go @@ -180,6 +180,12 @@ func (w *Writer) RecordPlayEnded( if err != nil { return err } + // Idempotency for the client's durable close-by-id replay: if the + // row is already closed, re-applying the close is harmless (the + // UPDATE is last-write-wins) but a second skip_events insert must + // NOT fire — a lost-response retry would otherwise double the + // skip signal. See the 2026-06-11 contract audit. + alreadyEnded := ev.EndedAt.Valid ratio := 0.0 if track.DurationMs > 0 { ratio = float64(durationPlayedMs) / float64(track.DurationMs) @@ -196,7 +202,7 @@ func (w *Writer) RecordPlayEnded( }); err != nil { return err } - if isSkip { + if isSkip && !alreadyEnded { if _, err := q.InsertSkipEvent(ctx, dbq.InsertSkipEventParams{ UserID: ev.UserID, TrackID: ev.TrackID, @@ -356,6 +362,24 @@ func (w *Writer) RecordOfflinePlay( if err != nil { return err } + // Idempotency: the offline queue is at-least-once, so a response lost + // after commit makes the client re-fire the same play. The replayed + // payload carries the original `at`, so an exact (user, track, + // started_at) match means we already recorded it — no-op instead of a + // duplicate history row. See the 2026-06-11 contract audit. + var dupID pgtype.UUID + err = tx.QueryRow(ctx, + `SELECT id FROM play_events + WHERE user_id = $1 AND track_id = $2 AND started_at = $3 + LIMIT 1`, + userID, trackID, pgtype.Timestamptz{Time: at, Valid: true}, + ).Scan(&dupID) + if err == nil { + return nil + } + if !errors.Is(err, pgx.ErrNoRows) { + return err + } if err := w.autoClosePriorOpen(ctx, q, userID, at); err != nil { return err } diff --git a/internal/playevents/writer_test.go b/internal/playevents/writer_test.go index e4133922..a0158c11 100644 --- a/internal/playevents/writer_test.go +++ b/internal/playevents/writer_test.go @@ -219,6 +219,55 @@ func TestRecordPlayEnded_DurationOver50Percent_NotSkip(t *testing.T) { } } +// A lost-response retry of an offline play (same user/track/started_at) +// must not create a second history row. Guards the at-least-once replay +// path added in the 2026-06-11 audit. +func TestRecordOfflinePlay_DedupsByUserTrackStartedAt(t *testing.T) { + f := newFixture(t, 200_000) + ctx := context.Background() + at := time.Now().UTC() + if err := f.w.RecordOfflinePlay(ctx, f.user, f.track, "c", "", at, 120_000); err != nil { + t.Fatalf("first offline: %v", err) + } + if err := f.w.RecordOfflinePlay(ctx, f.user, f.track, "c", "", at, 120_000); err != nil { + t.Fatalf("replay offline: %v", err) + } + var count int + if err := f.pool.QueryRow(ctx, + "SELECT COUNT(*) FROM play_events WHERE user_id=$1 AND track_id=$2 AND started_at=$3", + f.user, f.track, pgtype.Timestamptz{Time: at, Valid: true}).Scan(&count); err != nil { + t.Fatalf("count: %v", err) + } + if count != 1 { + t.Errorf("play_events count = %d, want 1 (deduped)", count) + } +} + +// Re-closing an already-ended row (the client's durable close-by-id replay) +// updates the row but must NOT insert a second skip_events row. +func TestRecordPlayEnded_IdempotentClose_NoDuplicateSkipEvent(t *testing.T) { + f := newFixture(t, 200_000) + ctx := context.Background() + t1 := time.Now().UTC() + res, _ := f.w.RecordPlayStarted(ctx, f.user, f.track, "c", t1) + t2 := t1.Add(10 * time.Second) // 5% / 10s -> skip + if err := f.w.RecordPlayEnded(ctx, res.PlayEventID, 10_000, t2); err != nil { + t.Fatalf("first end: %v", err) + } + if err := f.w.RecordPlayEnded(ctx, res.PlayEventID, 10_000, t2); err != nil { + t.Fatalf("replay end: %v", err) + } + var count int + if err := f.pool.QueryRow(ctx, + "SELECT COUNT(*) FROM skip_events WHERE user_id=$1 AND track_id=$2", + f.user, f.track).Scan(&count); err != nil { + t.Fatalf("count: %v", err) + } + if count != 1 { + t.Errorf("skip_events count = %d, want 1 (idempotent close)", count) + } +} + func TestRecordPlaySkipped_AlwaysSkipFlagAndSkipEventRow(t *testing.T) { f := newFixture(t, 200_000) t1 := time.Now().UTC()