diff --git a/android-shared/src/androidMain/kotlin/org/siloserver/silo/common/player/PlaybackSessionLifecycle.kt b/android-shared/src/androidMain/kotlin/org/siloserver/silo/common/player/PlaybackSessionLifecycle.kt index 39a3327c1..941d1034b 100644 --- a/android-shared/src/androidMain/kotlin/org/siloserver/silo/common/player/PlaybackSessionLifecycle.kt +++ b/android-shared/src/androidMain/kotlin/org/siloserver/silo/common/player/PlaybackSessionLifecycle.kt @@ -101,6 +101,15 @@ class PlaybackSessionLifecycle( private data class ActiveSessionSnapshot( val state: SessionState, + /** + * The ownership token as it stood, captured rather than derived. + * + * [SessionState] carries a session id only while Active, but + * Reconnecting and Failed deliberately keep owning theirs — so + * reconstructing the token from the restored state alone erases + * ownership exactly for the states that exist to survive an outage. + */ + val lastAdoptedSessionId: String?, val notice: PlayerNotice?, val lastStartParams: StartParams?, val lastReportedPosition: Double?, @@ -366,6 +375,7 @@ class PlaybackSessionLifecycle( private fun captureActiveSessionSnapshot(): ActiveSessionSnapshot = ActiveSessionSnapshot( state = _state.value, + lastAdoptedSessionId = lastAdoptedSessionId, notice = _notice.value, lastStartParams = lastStartParams, lastReportedPosition = lastReportedPosition, @@ -393,8 +403,15 @@ class PlaybackSessionLifecycle( renewMissingSessionWithLegacyStart = snapshot.renewMissingSessionWithLegacyStart diagnosticsRecording = snapshot.diagnosticsRecording _notice.value = snapshot.notice - // A rollback to the predecessor hands ownership back to that session. - (snapshot.state as? SessionState.Active)?.let { lastAdoptedSessionId = it.session.sessionId } + // Restore the token the snapshot captured, rather than deriving it from + // the restored state. Deriving gets both ends wrong: reading it only + // from Active leaves a rolled-back first deferred adoption naming the + // discarded replacement, while clearing everything that is not Active + // erases ownership for Reconnecting and Failed — which hold a session + // precisely so an outage does not lose it. A predecessor restored as + // Reconnecting would then have no id for stop() to name, and its + // transcode would run until the server expired it. + lastAdoptedSessionId = snapshot.lastAdoptedSessionId _state.value = snapshot.state if ( restartReporter && @@ -527,7 +544,30 @@ class PlaybackSessionLifecycle( * Push a position update from the player. Non-suspend — the actual server * report happens on the internal 10s debounce loop (see [PROGRESS_REPORT_INTERVAL_MS]). */ - fun reportPosition(positionSec: Double, durationSec: Double, isPaused: Boolean) { + fun reportPosition( + positionSec: Double, + durationSec: Double, + isPaused: Boolean, + /** + * The session the caller believes produced this sample; null when the + * caller owns none, as downloaded and local playback do not. + * + * These fields are process-global and the reporter loop pairs them with + * whichever session is current when it next fires — so a final callback + * from an outgoing player, arriving after the next screen has adopted, + * would otherwise flush the previous episode's position under the new + * episode's id. That is the "resume jumped to the last episode's time" + * shape. + * + * Note that null is NOT "skip the check": both exit paths clear the UI + * session id while player callbacks are still draining, so treating null + * as permission is exactly the hole this closes. A caller with no + * session may only write these fields while the lifecycle owns none + * either. + */ + expectedSessionId: String?, + ) { + if (expectedSessionId != lastAdoptedSessionId) return if (positionSec.isFinite() && positionSec >= 0) { lastReportedPosition = positionSec } diff --git a/android-shared/src/androidMain/kotlin/org/siloserver/silo/common/player/PlaybackSessionManager.kt b/android-shared/src/androidMain/kotlin/org/siloserver/silo/common/player/PlaybackSessionManager.kt index 8cf2e3c41..356fb4423 100644 --- a/android-shared/src/androidMain/kotlin/org/siloserver/silo/common/player/PlaybackSessionManager.kt +++ b/android-shared/src/androidMain/kotlin/org/siloserver/silo/common/player/PlaybackSessionManager.kt @@ -35,6 +35,7 @@ import org.siloserver.silo.network.TokenManager import org.siloserver.silo.repository.PlaybackRepository import java.util.IdentityHashMap import java.util.UUID +import java.util.concurrent.atomic.AtomicLong import java.util.concurrent.atomic.AtomicReference import kotlinx.coroutines.CancellationException import kotlinx.coroutines.CompletableDeferred @@ -180,7 +181,44 @@ open class PlaybackSessionManager( private val activeVideoAttempt = AtomicReference() private val stagedVideoReplans = IdentityHashMap() - private val orphanedSessionIds = mutableSetOf() + // Insertion-ordered so the bound in [rememberOrphanedSessionLocked] evicts + // the oldest unconfirmed session rather than an arbitrary one. + private val orphanedSessionIds = LinkedHashSet() + + /** + * Sessions registered as orphans under [videoAttemptMutex] whose stop still + * has to be issued, each tagged with the release claim of the lock holder + * that queued it. Guarded by that same mutex; drained immediately after it + * is released. See [stopRetainingFailureLocked]. + */ + private val pendingOwnershipReleases = mutableListOf>() + + /** + * Sessions whose stop is in flight via the queued-release or orphan-drain + * paths, counted rather than flagged. + * + * Two callers can legitimately be releasing the same id — a queued release + * and an orphan drain that selected it before the queue existed. The count + * keeps the marker honest for that overlap, so the first to finish cannot + * clear protection while the second is still running. It does NOT prevent + * the duplicate request itself: the second stop still goes out, which is + * tolerable only because stopping an already-stopped session is harmless. + * + * Not a register of every stop in the manager: committed-session cleanup and + * the direct retaining-stop helpers issue their own unmarked stops, so this + * excludes duplicates between the two paths that consult it, not globally. + * Guarded by [videoAttemptMutex]. + */ + private val releasesInFlight = mutableMapOf() + + private val releaseClaims = AtomicLong() + + /** + * The claim of whoever currently holds [videoAttemptMutex]. Only read and + * written under that lock, which is what makes a plain field safe here — + * exactly one coroutine can be inside the lock at a time. + */ + private var currentReleaseClaim = 0L private var pendingVideoPublication: PendingVideoPublication? = null private var contentResetInProgress = false @@ -191,10 +229,16 @@ open class PlaybackSessionManager( videoAttemptMutex.lock() val pending = pendingVideoPublication if (pending == null) { + val claim = releaseClaims.incrementAndGet() + currentReleaseClaim = claim try { return block() } finally { videoAttemptMutex.unlock() + // After the unlock, deliberately: the block may have queued + // stops for sessions it discarded, and issuing them under + // the lock would hold every other caller behind network I/O. + drainPendingOwnershipReleases(claim) } } videoAttemptMutex.unlock() @@ -225,6 +269,34 @@ open class PlaybackSessionManager( subtitleFidelityPreference: SubtitleFidelityPreference = SubtitleFidelityPreference.PRESERVE, deferPublication: Boolean = false, ): ApiResult = contentStartMutex.withLock { + /** + * The session this call is currently answerable for. + * + * Once the server responds it has allocated a session, but every branch + * below still suspends — acquiring [videoAttemptMutex], emitting a route + * event, issuing its own stop — before that id is either published into + * [activeVideoAttempt] or stopped. A cancellation in that window leaves + * the id owned by nobody: the manager never published it, and the + * callers never learn it, because they only see an id when this function + * returns. The transcode then runs on until the server's own expiry, + * holding a stream slot; a retry can produce a second session for the + * same screen, or fail outright as "too many streams". + * + * So: arm this the moment the response decodes, and clear it only where + * responsibility genuinely moves — to the manager on publication, or to + * a stop the *server acknowledged*. A branch that takes ownership back + * (the replan error path) re-arms it. The finally releases whatever is + * still held, uncancellably. + * + * Scope: this covers ids allocated by *this* call. The internal replan + * reached from the ReplanRequired branch allocates its own candidates + * and clears `activeVideoAttempt` before its own suspending cleanup; + * those windows are held by [stopRetainingFailureLocked] instead, which + * is the same register-before-stop discipline expressed against the + * manager's orphan set because those paths already hold + * [videoAttemptMutex]. + */ + var leasedSessionId: String? = null try { beginContentReset() val predecessorForPublication = videoAttemptMutex.withLock { @@ -263,6 +335,7 @@ open class PlaybackSessionManager( return@withLock when (val result = playbackRepository.startPlaybackV3(request)) { is ApiResult.Success -> when (val validated = result.data.validateForMedia3()) { is PlaybackV3Validation.Playable -> { + leasedSessionId = validated.sessionId val planAttemptId = UUID.randomUUID().toString() val active = newActiveAttempt( request = request, @@ -279,6 +352,9 @@ open class PlaybackSessionManager( deferPublication = deferPublication, ) } + // Published: the manager owns this id now, so teardown + // is its problem rather than this call's. + leasedSessionId = null PassthroughSuppressionRegistry.beginAttempt(active.planAttemptKey) reportActiveVideoEvent("plan_selected", network.asRouteDiagnostics()) ApiResult.Success( @@ -292,6 +368,8 @@ open class PlaybackSessionManager( ) } is PlaybackV3Validation.Terminal -> { + leasedSessionId = + result.data.playbackPlan?.sessionId ?: result.data.sessionId if (!deferPublication) { videoAttemptMutex.withLock { activeVideoAttempt.set(null) @@ -306,25 +384,34 @@ open class PlaybackSessionManager( outputRouteGeneration = request.outputRouteGeneration, ), ) - (result.data.playbackPlan?.sessionId ?: result.data.sessionId) + // Only a stop the server acknowledged discharges the + // lease. An Error/NetworkError does not throw, so + // clearing on the call alone would drop the session on + // exactly the failure the lease exists to survive. + val stopped = (result.data.playbackPlan?.sessionId ?: result.data.sessionId) ?.let { playbackRepository.stopPlayback(it) } + if (stopped.isStopDischarged()) leasedSessionId = null ApiResult.Success( VideoSessionStartV3.Terminal(validated.reason, validated.message, validated.retryable), ) } is PlaybackV3Validation.Incompatible -> { + leasedSessionId = validated.allocatedSessionId if (!deferPublication) { videoAttemptMutex.withLock { activeVideoAttempt.set(null) } } - validated.allocatedSessionId?.let { playbackRepository.stopPlayback(it) } + val stopped = validated.allocatedSessionId + ?.let { playbackRepository.stopPlayback(it) } + if (stopped.isStopDischarged()) leasedSessionId = null ApiResult.Success(VideoSessionStartV3.ServerUpgradeRequired) } is PlaybackV3Validation.ReplanRequired -> { // Decode stale engine enums, but never execute them. Preserve // the allocated session and give the v3 planner exactly one // opportunity to replace the route with a Media3 plan. + leasedSessionId = validated.sessionId val planAttemptId = UUID.randomUUID().toString() val active = newActiveAttempt( request = request, @@ -335,6 +422,16 @@ open class PlaybackSessionManager( planAttemptId = planAttemptId, ) videoAttemptMutex.withLock { activeVideoAttempt.set(active) } + // The lease STAYS ARMED across the nested replan. Being + // installed as the active attempt is not the same as + // being findable: nothing outside this call has the id + // yet, and the replan below suspends — on + // finishContentReset, then on its own mutex — before it + // reaches any cancellation-safe cleanup of its own. A + // cancellation in that window used to leave the session + // installed, unknown to every caller, and running until + // the server expired it. The branches after the replan + // clear or re-arm it once its fate is decided. PassthroughSuppressionRegistry.beginAttempt(active.planAttemptKey) finishContentReset() val replanResult = replanActiveVideoSession( @@ -365,7 +462,14 @@ open class PlaybackSessionManager( } } } - abandonedSessionId?.let { playbackRepository.stopPlayback(it) } + // The lock above decided this id's fate. Either it + // was abandoned — in which case the lease names it + // until the stop is acknowledged — or it is the + // published replacement and the manager owns it. + leasedSessionId = abandonedSessionId + val stopped = abandonedSessionId + ?.let { playbackRepository.stopPlayback(it) } + if (stopped.isStopDischarged()) leasedSessionId = null } else if ( replanResult is ApiResult.Error || replanResult is ApiResult.NetworkError @@ -374,8 +478,21 @@ open class PlaybackSessionManager( activeVideoAttempt.compareAndSet(active, null) } if (cleared) { - playbackRepository.stopPlayback(validated.sessionId) + // Same reasoning as the deferred branch: the CAS + // above removed the manager's only reference. + leasedSessionId = validated.sessionId + val stopped = + playbackRepository.stopPlayback(validated.sessionId) + if (stopped.isStopDischarged()) leasedSessionId = null } + } else { + // Replan succeeded and published through the manager. + // The base id is either the committed attempt or was + // stopped by the replan itself; either way this call + // is no longer answerable for it, and leaving the + // lease armed would have the finally stop a session + // that is playing. + leasedSessionId = null } replanResult } @@ -385,6 +502,14 @@ open class PlaybackSessionManager( } } finally { finishContentReset() + // NonCancellable because this runs precisely when the surrounding + // work was cancelled. Failures stay queued in orphanedSessionIds so + // the next content reset drains them. + leasedSessionId?.let { orphan -> + withContext(NonCancellable) { + stopSessionsRetainingFailures(listOf(orphan)) + } + } } } @@ -915,7 +1040,24 @@ open class PlaybackSessionManager( suspend fun commitStagedVideoReplan( staged: StagedVideoReplan, deferPublication: Boolean = false, + ): ApiResult { + val claim = releaseClaims.incrementAndGet() + return try { + commitStagedVideoReplanLocked(staged, deferPublication, claim) + } finally { + // Same contract as withSettledVideoAttempt: stops queued while the + // lock was held are issued once it is released, and this awaits only + // the ones this call queued. + drainPendingOwnershipReleases(claim) + } + } + + private suspend fun commitStagedVideoReplanLocked( + staged: StagedVideoReplan, + deferPublication: Boolean, + claim: Long, ): ApiResult = videoAttemptMutex.withLock { + currentReleaseClaim = claim val prepared = stagedVideoReplans.remove(staged) ?: return@withLock stagedVideoReplanUnavailable() val active = activeVideoAttempt.get() @@ -945,7 +1087,9 @@ open class PlaybackSessionManager( // This is the commit point. Everything after it is best-effort, // non-blocking bookkeeping: callers must always receive the committed - // candidate once manager ownership has moved to [next]. + // candidate once manager ownership has moved to [next]. A commit that + // reaches here has queued no release, so the caller's drain finds + // nothing of its own and returns without waiting. activeVideoAttempt.set(next) if (deferPublication) { pendingVideoPublication = PendingVideoPublication( @@ -972,7 +1116,7 @@ open class PlaybackSessionManager( pendingVideoPublication = null val predecessorSessionId = pending.predecessor?.sessionId ?.takeIf { it != sessionId } - predecessorSessionId?.let { orphanedSessionIds += it } + predecessorSessionId?.let { rememberOrphanedSessionLocked(it) } pending.settled.complete(Unit) true to predecessorSessionId } ?: return false @@ -1026,39 +1170,51 @@ open class PlaybackSessionManager( } /** - * Fire-and-forget abandonment on the manager's own scope. + * Fire-and-forget [abandonActiveVideoSession] on the manager's own scope. * * Callers reach this exactly when their own scope is being torn down, which * rules out doing the work inline. `viewModelScope.launch(NonCancellable)` * looks like the answer and does run, but it severs the parent link to - * produce an untracked coroutine nothing can await or observe failures - * from. This scope already outlives any screen and is what the - * committed-session cleanup uses, so a release belongs here rather than in - * a ViewModel on its way out. - * - * Only stops the session while the manager still owns it. The unconditional - * variant stops even when it failed to disown, and [stopSession]'s - * predecessor branch then clears a newer pending publication and stops its - * replacement — a real hazard once the release is dispatched rather than - * inline, because the window widens. When ownership has moved on, the id is - * recorded as an orphan and the next drain stops it with a plain repository - * call that cannot disturb whoever owns playback now. + * produce an untracked coroutine nothing can await or observe failures from + * — the pattern the coroutines documentation warns against. The manager's + * cleanup scope already outlives any screen and is what the committed-session + * cleanup path uses, so ownership of a release belongs there rather than in + * a ViewModel that is on its way out. */ fun abandonActiveVideoSessionAsync(sessionId: String) { sessionCleanupScope.launch { - runCatching { - val disowned = videoAttemptMutex.withLock { - val active = activeVideoAttempt.get() - if (active?.sessionId != sessionId) { - orphanedSessionIds += sessionId - false - } else { - activeVideoAttempt.compareAndSet(active, null) - } - } - if (disowned) stopSession(sessionId) + runCatching { abandonActiveVideoSessionIfCurrent(sessionId) } + } + } + + /** + * [abandonActiveVideoSession], but only while this session is still the one + * the manager holds. + * + * The unconditional variant stops the session even when it failed to disown + * it, and [stopSession]'s predecessor branch then clears a *newer* pending + * publication and stops its replacement. Running abandonment on a dispatched + * scope widens that window enough to matter: a stale result scheduled for + * release can land after a newer deferred publication has installed itself + * with this id as its predecessor, and tear the new one down. + * + * When ownership has already moved on, the id is recorded as an orphan + * instead. The drain stops it with a plain repository call that cannot + * disturb whoever owns playback now. + */ + suspend fun abandonActiveVideoSessionIfCurrent(sessionId: String): Boolean { + val disowned = videoAttemptMutex.withLock { + val active = activeVideoAttempt.get() + if (active?.sessionId != sessionId) { + rememberOrphanedSessionLocked(sessionId) + false + } else { + activeVideoAttempt.compareAndSet(active, null) } } + if (!disowned) return false + stopSession(sessionId) + return true } suspend fun rollbackUnpublishedVideoSession(sessionId: String): Boolean { @@ -1155,7 +1311,7 @@ open class PlaybackSessionManager( activeSessionId: String, ) { if (oldSessionId == activeSessionId) return - orphanedSessionIds += oldSessionId + rememberOrphanedSessionLocked(oldSessionId) scheduleRegisteredCommittedSessionCleanup( oldSessionId = oldSessionId, activeSessionId = activeSessionId, @@ -1178,7 +1334,7 @@ open class PlaybackSessionManager( } catch (_: Throwable) { null } - if (result is ApiResult.Success) { + if (result.isStopDischarged()) { stopped = true break } @@ -1193,21 +1349,48 @@ open class PlaybackSessionManager( } private suspend fun drainOrphanedSessions(protectedSessionIds: Set) { + // Claim the ids under the lock, marking them in flight in the same + // critical section. Filtering alone only holds for the instant of the + // snapshot — the stops below run unlocked, and without a marker another + // drain could select the same session and stop it concurrently. val orphanIds = videoAttemptMutex.withLock { val live = setOfNotNull(activeVideoAttempt.get()?.sessionId) - orphanedSessionIds.filterNot { it in protectedSessionIds || it in live } + orphanedSessionIds.filterNot { + it in protectedSessionIds || + it in live || + // A queued release already owns this one, and its own drain + // will remove it on discharge. + it in releasesInFlight || + pendingOwnershipReleases.any { pending -> pending.second == it } + }.onEach { markReleaseInFlightLocked(it) } } orphanIds.forEach { sessionId -> - val result = try { - playbackRepository.stopPlayback(sessionId) - } catch (_: CancellationException) { - return@forEach - } catch (_: Throwable) { - null - } - if (result is ApiResult.Success) { - videoAttemptMutex.withLock { - orphanedSessionIds -= sessionId + try { + val result = try { + playbackRepository.stopPlayback(sessionId) + } catch (_: CancellationException) { + return@forEach + } catch (_: Throwable) { + null + } + if (result.isStopDischarged()) { + videoAttemptMutex.withLock { + orphanedSessionIds -= sessionId + } + } + } finally { + // Including the cancellation return above: a marker left behind + // would hide this session from every future drain. + withContext(NonCancellable) { + videoAttemptMutex.withLock { + clearReleaseInFlightLocked(sessionId) + // Re-trim here too. Trimming skips in-flight ids, so + // whichever release path clears the last marker has to + // re-apply the cap — otherwise a burst drained from here + // leaves the ledger over its bound until some unrelated + // future orphan happens to trigger a trim. + trimOrphanedSessionsLocked() + } } } } @@ -1217,7 +1400,7 @@ open class PlaybackSessionManager( val uniqueSessionIds = sessionIds.distinct() if (uniqueSessionIds.isEmpty()) return videoAttemptMutex.withLock { - orphanedSessionIds += uniqueSessionIds + uniqueSessionIds.forEach { rememberOrphanedSessionLocked(it) } } uniqueSessionIds.forEach { sessionId -> val result = try { @@ -1225,7 +1408,7 @@ open class PlaybackSessionManager( } catch (_: Throwable) { null } - if (result is ApiResult.Success) { + if (result.isStopDischarged()) { videoAttemptMutex.withLock { orphanedSessionIds -= sessionId } @@ -1245,7 +1428,7 @@ open class PlaybackSessionManager( stagedVideoReplans.keys.none { it.candidateSessionId == candidateSessionId } - }?.also { orphanedSessionIds += it } + }?.also { rememberOrphanedSessionLocked(it) } } if (candidateSessionId == null) return @@ -1255,7 +1438,7 @@ open class PlaybackSessionManager( } catch (_: Throwable) { null } - if (result is ApiResult.Success) { + if (result.isStopDischarged()) { videoAttemptMutex.withLock { orphanedSessionIds -= candidateSessionId } @@ -1269,6 +1452,145 @@ open class PlaybackSessionManager( message = "The staged playback replan was already consumed or no longer matches the active content.", ) + /** + * Stops a session this caller is discarding, keeping a record of it until + * the server confirms it is gone. Caller must hold [videoAttemptMutex]. + * + * Some callers reach here having already cleared `activeVideoAttempt` or + * removed the staged handle, and for those the id exists nowhere else in the + * process from that moment until the server replies. A bare suspending stop + * there is cancellable — and these run from ViewModel recovery jobs that + * exit, content replacement and teardown all cancel — so the id would simply + * be lost and the transcode would hold its stream slot until the server's + * own expiry. Registering first makes the worst case a retry on the next + * content reset rather than an orphan nobody remembers. Callers that have + * not given up ownership (a rejected validation candidate, say) are + * registered on the same path because it costs nothing. + */ + private fun stopRetainingFailureLocked(sessionId: String) { + // The stop is queued rather than issued. Requests time out at 60s, and + // awaiting one while holding videoAttemptMutex would serialise every + // start, replan, content reset and staged commit behind a dying + // session's teardown. [drainPendingOwnershipReleases] runs it once the + // lock is released, still awaited by the caller that queued it. + // + // Tagged with the claim of the lock holder that queued it. Without that + // tag one shared queue lets any concurrent drain take another caller's + // work: the queuing caller then returns before its own stop ran, while + // an unrelated caller — which queued nothing — blocks for a full network + // timeout on someone else's teardown. + // + // Queued BEFORE registering, because rememberOrphanedSessionLocked + // trims and trimming protects only ids already queued or in flight — + // so with a full ledger of protected entries this session could be the + // one evictable entry and get dropped before its stop had even started. + pendingOwnershipReleases += currentReleaseClaim to sessionId + // Registration is what makes the id survivable: if the queued stop + // fails, this is the record the next content reset retries from. + rememberOrphanedSessionLocked(sessionId) + } + + /** + * Issues the stops [stopRetainingFailureLocked] queued under [claim]. Must + * be called with [videoAttemptMutex] NOT held. + * + * NonCancellable throughout: these sessions are already registered as + * orphans and unreferenced anywhere else, and the callers reaching here are + * frequently being cancelled. Anything that fails stays registered for the + * next content reset to drain — unless the ledger is at its cap and the + * entry has already been evicted, in which case that session falls back to + * the server's own expiry. + */ + private suspend fun drainPendingOwnershipReleases(claim: Long) { + withContext(NonCancellable) { + while (true) { + val sessionId = videoAttemptMutex.withLock { + val index = pendingOwnershipReleases.indexOfFirst { it.first == claim } + if (index < 0) { + null + } else { + pendingOwnershipReleases.removeAt(index).second.also { + // Visible to drainOrphanedSessions for as long as the + // stop is in flight, so a concurrent content reset + // does not issue a second stop for the same session. + markReleaseInFlightLocked(it) + } + } + } ?: return@withContext + val result = try { + playbackRepository.stopPlayback(sessionId) + } catch (_: Throwable) { + null + } + videoAttemptMutex.withLock { + clearReleaseInFlightLocked(sessionId) + if (result.isStopDischarged()) { + orphanedSessionIds -= sessionId + } + // The cap can only skip entries that were mid-release, so + // re-apply it once one finishes; otherwise a burst of + // concurrent releases leaves the ledger permanently over + // its bound with nothing to bring it back down. + trimOrphanedSessionsLocked() + } + } + } + } + + /** + * Records a session whose stop has not been confirmed, oldest evicted first. + * + * The ledger has to be bounded. Only a discharged stop removes an entry, so + * a server that keeps failing this call — while playback keeps producing new + * sessions — would otherwise grow it without limit and make every later + * content reset retry an ever-larger collection. Dropping the oldest entry + * costs that session its explicit stop and falls back to the server's own + * expiry, which is exactly what happens today when a stop never succeeds. + */ + private fun markReleaseInFlightLocked(sessionId: String) { + releasesInFlight[sessionId] = (releasesInFlight[sessionId] ?: 0) + 1 + } + + private fun clearReleaseInFlightLocked(sessionId: String) { + val remaining = (releasesInFlight[sessionId] ?: 0) - 1 + if (remaining > 0) releasesInFlight[sessionId] = remaining else releasesInFlight -= sessionId + } + + private fun rememberOrphanedSessionLocked(sessionId: String) { + orphanedSessionIds += sessionId + trimOrphanedSessionsLocked() + } + + private fun trimOrphanedSessionsLocked() { + while (orphanedSessionIds.size > MAX_RETAINED_ORPHANED_SESSIONS) { + // Never evict an id whose release this manager is tracking — the + // queued and in-flight sets. Committed-session cleanup and the + // direct retaining-stop helpers issue unmarked stops, so this is not + // protection against every release in flight, only the ones the + // queue knows about. + // When everything over the cap is mid-release there is nothing + // safe to drop, so the set stays over its bound until one of those + // releases completes and re-runs this. + val oldest = orphanedSessionIds.firstOrNull { + it !in releasesInFlight && + pendingOwnershipReleases.none { pending -> pending.second == it } + } ?: break + orphanedSessionIds -= oldest + } + } + + /** + * True once the server owes us nothing more for this session. + * + * A typed session-missing 404 counts: the session is already gone, and + * treating that as a failure would keep the id in [orphanedSessionIds] + * forever and retry it on every single drain. A bare 404 does not — routing, + * proxy and compatibility 404s prove nothing about the session, so this uses + * the same predicate the rest of the manager uses for absence. + */ + private fun ApiResult?.isStopDischarged(): Boolean = + this is ApiResult.Success || this?.isPlaybackSessionMissingError() == true + private suspend fun stopCandidateSessionIfUnowned( activeSessionId: String?, candidateSessionId: String?, @@ -1280,7 +1602,7 @@ open class PlaybackSessionManager( ) { return } - playbackRepository.stopPlayback(candidateSessionId) + stopRetainingFailureLocked(candidateSessionId) } private suspend fun stopCandidateSessionsIfUnowned( @@ -1297,7 +1619,9 @@ open class PlaybackSessionManager( stopActiveSession: Boolean, ) { if (stopActiveSession) { - playbackRepository.stopPlayback(activeSessionId) + // Ownership was cleared immediately above, so this is the same + // register-before-stop case as the candidates below. + stopRetainingFailureLocked(activeSessionId) } candidateSessionIds.filterNotNull().distinct() .filter { it != activeSessionId } @@ -2080,6 +2404,15 @@ open class PlaybackSessionManager( private const val TAG = "PlaybackSessionMgr" private const val COMMITTED_SESSION_CLEANUP_ATTEMPTS = 2 + /** + * Ceiling on unconfirmed orphaned sessions kept for retry. + * + * Generous relative to how many sessions one viewing session produces, + * so it only bites when stops are persistently failing — the case where + * retrying an unbounded backlog on every content reset is pure cost. + */ + private const val MAX_RETAINED_ORPHANED_SESSIONS = 64 + /** * How long a content reset waits for a deferred publication to settle * before rolling it back itself. Comfortably above the 30s local-mount @@ -2209,11 +2542,11 @@ open class PlaybackSessionManager( stopSessionsRetainingFailures(candidateSessionIds) if (sessionId != null) { videoAttemptMutex.withLock { - orphanedSessionIds += sessionId + rememberOrphanedSessionLocked(sessionId) } try { result = playbackRepository.stopPlayback(sessionId) - if (result is ApiResult.Success) { + if (result.isStopDischarged()) { videoAttemptMutex.withLock { orphanedSessionIds -= sessionId } diff --git a/android-shared/src/androidUnitTest/kotlin/org/siloserver/silo/common/player/PlaybackSessionLifecycleTest.kt b/android-shared/src/androidUnitTest/kotlin/org/siloserver/silo/common/player/PlaybackSessionLifecycleTest.kt index 075ad1f73..aa141a694 100644 --- a/android-shared/src/androidUnitTest/kotlin/org/siloserver/silo/common/player/PlaybackSessionLifecycleTest.kt +++ b/android-shared/src/androidUnitTest/kotlin/org/siloserver/silo/common/player/PlaybackSessionLifecycleTest.kt @@ -238,7 +238,7 @@ class PlaybackSessionLifecycleTest { assertEquals("sess-adopted", (active as SessionState.Active).session.sessionId) assertEquals(listOf("sess-adopted"), recordedSessions) - lifecycle.reportPosition(positionSec = 33.0, durationSec = 100.0, isPaused = false) + lifecycle.reportOwnedPosition(positionSec = 33.0, durationSec = 100.0, isPaused = false) advanceTimeBy(PlaybackSessionLifecycle.PROGRESS_REPORT_INTERVAL_MS + 100) assertEquals(0, sessionMgr.startCallCount) @@ -264,7 +264,7 @@ class PlaybackSessionLifecycleTest { stopSessionOnStop = false, ) - lifecycle.reportPosition(positionSec = 33.0, durationSec = 100.0, isPaused = false) + lifecycle.reportOwnedPosition(positionSec = 33.0, durationSec = 100.0, isPaused = false) advanceTimeBy(PlaybackSessionLifecycle.PROGRESS_REPORT_INTERVAL_MS + 100) lifecycle.stop() advanceUntilIdle() @@ -507,7 +507,7 @@ class PlaybackSessionLifecycleTest { assertEquals("sess-original", (first as SessionState.Active).session.sessionId) // Simulate the player advancing. - lifecycle.reportPosition(positionSec = 42.5, durationSec = 100.0, isPaused = false) + lifecycle.reportOwnedPosition(positionSec = 42.5, durationSec = 100.0, isPaused = false) // Trigger the 10s reporter; first call returns 404 -> recovery -> re-start. advanceTimeBy(PlaybackSessionLifecycle.PROGRESS_REPORT_INTERVAL_MS + 100) @@ -552,7 +552,7 @@ class PlaybackSessionLifecycleTest { val active = lifecycle.start(defaultStartParams()) assertTrue(active is SessionState.Active) - lifecycle.reportPosition(10.0, 100.0, isPaused = false) + lifecycle.reportOwnedPosition(10.0, 100.0, isPaused = false) // Trigger the 10s reporter -> NetworkError -> beginOutageRecovery. advanceTimeBy(PlaybackSessionLifecycle.PROGRESS_REPORT_INTERVAL_MS + 100) @@ -596,7 +596,7 @@ class PlaybackSessionLifecycleTest { val lifecycle = newLifecycle(sessionMgr, healthApi = healthApi, scope = backgroundScope) val active = lifecycle.start(defaultStartParams()) assertTrue(active is SessionState.Active) - lifecycle.reportPosition(5.0, 100.0, isPaused = false) + lifecycle.reportOwnedPosition(5.0, 100.0, isPaused = false) advanceTimeBy(PlaybackSessionLifecycle.PROGRESS_REPORT_INTERVAL_MS + 100) advanceUntilIdle() @@ -634,7 +634,7 @@ class PlaybackSessionLifecycleTest { } val lifecycle = newLifecycle(sessionMgr, healthApi = healthApi, scope = backgroundScope) lifecycle.start(defaultStartParams()) - lifecycle.reportPosition(0.0, 100.0, isPaused = false) + lifecycle.reportOwnedPosition(0.0, 100.0, isPaused = false) advanceTimeBy(PlaybackSessionLifecycle.PROGRESS_REPORT_INTERVAL_MS + 100) advanceUntilIdle() assertTrue(lifecycle.state.value is SessionState.Reconnecting) @@ -682,7 +682,7 @@ class PlaybackSessionLifecycleTest { } val lifecycle = newLifecycle(sessionMgr, healthApi = healthApi, scope = backgroundScope) lifecycle.start(defaultStartParams()) - lifecycle.reportPosition(0.0, 100.0, isPaused = false) + lifecycle.reportOwnedPosition(0.0, 100.0, isPaused = false) advanceTimeBy(PlaybackSessionLifecycle.PROGRESS_REPORT_INTERVAL_MS + 100) advanceUntilIdle() @@ -718,7 +718,7 @@ class PlaybackSessionLifecycleTest { } val lifecycle = newLifecycle(sessionMgr, healthApi = healthApi, scope = backgroundScope) lifecycle.start(defaultStartParams()) - lifecycle.reportPosition(0.0, 100.0, isPaused = false) + lifecycle.reportOwnedPosition(0.0, 100.0, isPaused = false) advanceTimeBy(PlaybackSessionLifecycle.PROGRESS_REPORT_INTERVAL_MS + 100) advanceUntilIdle() @@ -753,7 +753,7 @@ class PlaybackSessionLifecycleTest { } val lifecycle = newLifecycle(sessionMgr, healthApi = healthApi, scope = backgroundScope) lifecycle.start(defaultStartParams()) - lifecycle.reportPosition(12.0, 100.0, isPaused = false) + lifecycle.reportOwnedPosition(12.0, 100.0, isPaused = false) advanceTimeBy(PlaybackSessionLifecycle.PROGRESS_REPORT_INTERVAL_MS + 100) advanceUntilIdle() @@ -779,7 +779,7 @@ class PlaybackSessionLifecycleTest { } val lifecycle = newLifecycle(sessionMgr, scope = backgroundScope) lifecycle.start(defaultStartParams()) - lifecycle.reportPosition(15.0, 100.0, isPaused = false) + lifecycle.reportOwnedPosition(15.0, 100.0, isPaused = false) advanceTimeBy(PlaybackSessionLifecycle.PROGRESS_REPORT_INTERVAL_MS + 100) advanceUntilIdle() @@ -832,7 +832,7 @@ class PlaybackSessionLifecycleTest { } val lifecycle = newLifecycle(sessionMgr, scope = backgroundScope) lifecycle.start(defaultStartParams()) - lifecycle.reportPosition(7.0, 100.0, isPaused = false) + lifecycle.reportOwnedPosition(7.0, 100.0, isPaused = false) // Four reporter ticks — every tick reads state.value's session, which // is still sess-original because the renewal start() is gated. Each @@ -1022,6 +1022,24 @@ class PlaybackSessionLifecycleTest { playbackSessions = playbackSessions, ) + /** + * Reports a sample as the session the lifecycle currently owns. + * + * reportPosition requires the caller to name its session — null is "I own + * none", not "skip the check" — so these tests have to say which session + * they are reporting for, exactly as the players do. + */ + private fun PlaybackSessionLifecycle.reportOwnedPosition( + positionSec: Double, + durationSec: Double, + isPaused: Boolean, + ) = reportPosition( + positionSec = positionSec, + durationSec = durationSec, + isPaused = isPaused, + expectedSessionId = (state.value as? SessionState.Active)?.session?.sessionId, + ) + private fun defaultStartParams(startPosition: Double? = null) = StartParams( contentId = "content-1", fileId = 42, diff --git a/android-shared/src/androidUnitTest/kotlin/org/siloserver/silo/common/player/PlaybackSessionManagerStagedReplanTest.kt b/android-shared/src/androidUnitTest/kotlin/org/siloserver/silo/common/player/PlaybackSessionManagerStagedReplanTest.kt index cb7d4fab9..3f163a032 100644 --- a/android-shared/src/androidUnitTest/kotlin/org/siloserver/silo/common/player/PlaybackSessionManagerStagedReplanTest.kt +++ b/android-shared/src/androidUnitTest/kotlin/org/siloserver/silo/common/player/PlaybackSessionManagerStagedReplanTest.kt @@ -70,7 +70,9 @@ class PlaybackSessionManagerStagedReplanTest { .substringAfter("suspend fun confirmVideoSessionPublication(") .substringBefore("suspend fun rollbackUnpublishedVideoSession(") - val orphanRegistration = confirmation.indexOf("orphanedSessionIds +=") + // Every insertion goes through the bounded helper now, so the ledger + // cannot grow without limit when stops keep failing. + val orphanRegistration = confirmation.indexOf("rememberOrphanedSessionLocked(") val waiterRelease = confirmation.indexOf("pending.settled.complete(Unit)") val registeredCleanup = confirmation.indexOf( "scheduleRegisteredCommittedSessionCleanup(", diff --git a/androidApp/src/androidMain/kotlin/org/siloserver/silo/android/ui/screens/player/PlayerViewModel.kt b/androidApp/src/androidMain/kotlin/org/siloserver/silo/android/ui/screens/player/PlayerViewModel.kt index 8598c0bd2..1220be263 100644 --- a/androidApp/src/androidMain/kotlin/org/siloserver/silo/android/ui/screens/player/PlayerViewModel.kt +++ b/androidApp/src/androidMain/kotlin/org/siloserver/silo/android/ui/screens/player/PlayerViewModel.kt @@ -17,6 +17,7 @@ import org.siloserver.silo.common.player.FinalPlaybackPositionWriter import org.siloserver.silo.common.player.Playability import org.siloserver.silo.common.player.PlaybackSessionLifecycle import org.siloserver.silo.common.player.PlaybackSessionManager +import org.siloserver.silo.common.player.PlaybackTeardownGate import org.siloserver.silo.common.player.VideoSessionStartV3 import org.siloserver.silo.common.player.cast.CastMediaSpec import org.siloserver.silo.common.player.cast.CastPrepareRequest @@ -101,6 +102,7 @@ import kotlinx.coroutines.channels.BufferOverflow import kotlinx.coroutines.channels.Channel import kotlinx.coroutines.delay import kotlinx.coroutines.ensureActive +import kotlinx.coroutines.isActive import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.MutableSharedFlow import kotlinx.coroutines.flow.MutableStateFlow @@ -723,6 +725,19 @@ class PlayerViewModel( * player-to-player navigation is somebody else's session. */ private var retainedOwnedSessionId: String? = null + + /** + * Makes this screen's teardown of the process-scoped lifecycle one-shot. + * + * Naming the session is necessary but not sufficient. onExit() runs an + * ordered stop and onCleared() then schedules a detached one for the same + * id; the second passes the lifecycle's ownership guard because the first + * already cleared the owner, and bumps `stopEpoch` on its way through. A + * screen that acquired its start epoch between the two — but has not yet + * adopted its session — is then rejected as superseded. TV has been behind + * this gate since auto-advance broke on exactly that race; phone was not. + */ + private val lifecycleTeardown = PlaybackTeardownGate(sessionLifecycle) private var finalPositionScope: PlaybackWriteScope? = null private val initialPlayerLoadGate = InitialPlayerLoadGate() @@ -1623,22 +1638,81 @@ class PlayerViewModel( capabilities = capabilities, clientPlaybackContext = playbackContext, ) + // Returning on a stale generation is not enough on its own. By the + // time this call returns, the manager has already committed and + // taken ownership of the replacement session — so dropping the + // result quietly leaves a transcode running on the server that + // nothing will ever stop. The viewer sees playback exit; the server + // holds the stream slot until it times out. Release it when the + // generation moved on, the way TV already does; the adoption below + // owns the cancellation windows past this point. + val abandonedSessionId = (result as? ApiResult.Success) + ?.data + ?.let { it as? VideoSessionStartV3.Ready } + ?.session + ?.sessionId + if (!isActive || recoveryGeneration != playbackRecoveryGeneration) { + // Released on the manager's own scope, which outlives this + // screen: the whole point is to run after the reason for + // abandoning, and this ViewModel's scope may already be gone. + abandonedSessionId?.let(playbackSessionManager::abandonActiveVideoSessionAsync) + } + currentCoroutineContext().ensureActive() if (recoveryGeneration != playbackRecoveryGeneration) return@launch when (result) { is ApiResult.Success -> when (val decision = result.data) { is VideoSessionStartV3.Ready -> { - sessionLifecycle.adoptActiveSession( - params = StartParams( - contentId = state.contentId, - fileId = fileId, - capabilities = capabilities, - audioTrackIndex = decision.session.audioTrackIndex, - subtitleTrackIndex = selectedSubtitleTrackIndex, - startPosition = decision.session.position, - ), - session = decision.session, - renewMissingSessionWithLegacyStart = false, - ) + // Conditional adoption, evaluated inside the lifecycle + // lock: an unconditional adopt can hand the lifecycle a + // session this screen has already stopped owning, and + // then the manager owns the replacement while the + // lifecycle still owns its predecessor and the UI owns + // neither. + var adopted = false + try { + adopted = sessionLifecycle.adoptActiveSessionIfCurrent( + params = StartParams( + contentId = state.contentId, + fileId = fileId, + capabilities = capabilities, + audioTrackIndex = decision.session.audioTrackIndex, + subtitleTrackIndex = selectedSubtitleTrackIndex, + startPosition = decision.session.position, + ), + session = decision.session, + renewMissingSessionWithLegacyStart = false, + isCurrent = { + recoveryGeneration == playbackRecoveryGeneration && isActive + }, + ) + } finally { + // Covers refusal AND cancellation while waiting for + // the lifecycle mutex, which throws before isCurrent + // ever runs. NonCancellable because the usual reason + // for being here is that this coroutine was + // cancelled, and a cancelled coroutine cannot make + // the call that releases the server's stream slot — + // runCatching would only swallow the failure. + if (!adopted) { + withContext(NonCancellable) { + runCatching { + playbackSessionManager.stopSession( + decision.session.sessionId, + ) + } + } + } + } + if (!adopted) return@launch + // Ownership moved at adoption, so the exit token moves + // with it — and both exit routes read this token ahead + // of UI state precisely so this write wins. Publishing + // it only in the UI update below leaves a cancellation + // in between with the lifecycle owning the replacement + // while exit still names the predecessor, and the + // lifecycle then correctly refuses to stop it. + retainedOwnedSessionId = decision.session.sessionId + currentCoroutineContext().ensureActive() if (recoveryGeneration != playbackRecoveryGeneration) return@launch val mountGeneration = expectNextMediaMount() _uiState.update { current -> @@ -1948,6 +2022,7 @@ class PlayerViewModel( positionSec = positionSec, durationSec = _uiState.value.duration, isPaused = _uiState.value.isPaused, + expectedSessionId = _uiState.value.sessionId, ) // Track B: durably record the position (local resume + outbox sync) for @@ -2533,7 +2608,11 @@ class PlayerViewModel( .takeIf { it.isFinite() && it >= 0.0 } ?: request.targetSourceSec seekRecoveryRollbackInvalidated = false - sessionLifecycle.adoptActiveSession( + // Conditional, evaluated inside the lifecycle lock. An unconditional + // adopt only checks currency before and after, so a seek superseded + // while this awaited the lock still handed the lifecycle a session this + // screen had stopped owning. + val adopted = sessionLifecycle.adoptActiveSessionIfCurrent( params = StartParams( contentId = before.contentId, fileId = fileId, @@ -2546,7 +2625,20 @@ class PlayerViewModel( ), session = decision.session, renewMissingSessionWithLegacyStart = false, + isCurrent = { isCurrentServerSeek(request, recoveryGeneration) }, ) + // Deliberately no stop on refusal, unlike the replan paths. A seek + // re-anchor is validated to reuse the SAME session id — the manager + // rejects any response that changes it — so there is no disposable + // candidate here. The id names the session still playing, and the + // ordinary reason for refusal is that a newer seek was queued, which + // needs that very session as its base. + if (!adopted) return + // Same rule as the other two adoption paths: the exit token names what + // the lifecycle owns, from the moment it owns it. Supersession or + // cancellation before the UI publication below would otherwise leave + // exit naming the predecessor and the replacement running. + retainedOwnedSessionId = decision.session.sessionId if (!isCurrentServerSeek(request, recoveryGeneration)) return currentCoroutineContext().ensureActive() val mountGeneration = expectNextMediaMount() @@ -2758,6 +2850,15 @@ class PlayerViewModel( renewMissingSessionWithLegacyStart = false, isCurrent = adoption::isCurrent, ) + // The lifecycle owns this session from here, so the exit token has to + // name it from here — not from the UI publication below. Supersession + // between the two abandons the manager's session without rolling the + // lifecycle back, and exit would otherwise name the predecessor, be + // rightly refused by the ownership guard, and leave the replacement + // running with the teardown gate stopping onCleared from retrying. + if (lifecycleAdopted) { + playback.sessionId?.let { retainedOwnedSessionId = it } + } if (!lifecycleAdopted || !adoption.isCurrent()) { return MobileSubtitleAdoptionResult.Superseded } @@ -3756,7 +3857,14 @@ class PlayerViewModel( // one finishes tearing down, and an unqualified stop then kills the // playback the viewer is currently watching. TV already qualifies both // of its exits; phone did not. - val ownedSessionId = _uiState.value.sessionId ?: retainedOwnedSessionId + // Retained first, UI second. Every path that publishes a session id into + // UI state writes this token no later, and the three adoption paths + // (protocol-V3 replan, seek recovery, subtitle replan) write it earlier — + // at the moment the lifecycle takes ownership. That gap is the whole + // point: reading UI first inside it names the predecessor, the lifecycle + // rightly refuses to stop a session it no longer owns, and the one-shot + // gate stops onCleared from trying again. The replacement runs on. + val ownedSessionId = retainedOwnedSessionId ?: _uiState.value.sessionId retainedOwnedSessionId = ownedSessionId viewModelScope.launch { mobileSubtitleTransactions.persistCommittedSelectionAndFlush() @@ -3764,7 +3872,7 @@ class PlayerViewModel( // guard entirely, which is the opposite of what a missing token // should mean — if we cannot say which session was ours, we have no // business stopping anyone's. - ownedSessionId?.let { sessionLifecycle.stop(expectedSessionId = it) } + ownedSessionId?.let { lifecycleTeardown.stopOrdered(expectedSessionId = it) } } val state = _uiState.value val cid = state.contentId.takeIf { it.isNotBlank() } @@ -3968,13 +4076,14 @@ class PlayerViewModel( } override fun onCleared() { - // The RETAINED token, not the live state. An explicit back/remote exit + // The RETAINED token first, for the reason onExit gives. An explicit + // back/remote exit // calls onExit() before navigation, which clears sessionId — so by the // time onCleared runs, a "snapshot" of UI state is already null, and a // null token disables the ownership guard and stops whatever session is // current. That is precisely the session a replacement screen may have // just adopted. - val clearedSessionId = _uiState.value.sessionId ?: retainedOwnedSessionId + val clearedSessionId = retainedOwnedSessionId ?: _uiState.value.sessionId org.siloserver.silo.common.player.debug.PlaybackDebugState.screenError = null org.siloserver.silo.common.player.ActivePlaybackFile.clear(_uiState.value.mediaFileId) loadOwners.invalidate() @@ -3984,10 +4093,11 @@ class PlayerViewModel( mobileSubtitleTransactions.invalidate() onExit() // viewModelScope is cancelling here, so onExit's ordered stop may not run. - // stopAsync() is app-scoped and de-duplicates against an in-flight stop. + // The gate is what decides: if that stop already claimed teardown this + // is a no-op, and otherwise the app-scoped async stop takes ownership. // Qualified for the same reason as the ordered stop above: by the time // onCleared runs, a replacement screen may already own playback. - clearedSessionId?.let { sessionLifecycle.stopAsync(expectedSessionId = it) } + clearedSessionId?.let { lifecycleTeardown.stopDetached(expectedSessionId = it) } controlsHideJob?.cancel() introObserverJob?.cancel() lifecycleObserverJob?.cancel() diff --git a/androidApp/src/androidUnitTest/kotlin/org/siloserver/silo/android/ui/performance/MobilePlayerLifecyclePerformanceSourceTest.kt b/androidApp/src/androidUnitTest/kotlin/org/siloserver/silo/android/ui/performance/MobilePlayerLifecyclePerformanceSourceTest.kt index 0d8615592..9a01e6993 100644 --- a/androidApp/src/androidUnitTest/kotlin/org/siloserver/silo/android/ui/performance/MobilePlayerLifecyclePerformanceSourceTest.kt +++ b/androidApp/src/androidUnitTest/kotlin/org/siloserver/silo/android/ui/performance/MobilePlayerLifecyclePerformanceSourceTest.kt @@ -29,8 +29,13 @@ class MobilePlayerLifecyclePerformanceSourceTest { // Still the non-blocking teardown this test exists to protect, now // qualified by the session this view model owned — phone navigation // replaces the player entry, so an unqualified stop could kill the - // session a newer screen had already adopted. - assertTrue(viewModel.contains("sessionLifecycle.stopAsync(expectedSessionId =")) + // session a newer screen had already adopted. It goes through the + // one-shot gate as well: the ordered stop and this one target the same + // session, and the second to run would otherwise bump the lifecycle's + // stop epoch and supersede whichever screen started next. + assertTrue(viewModel.contains("lifecycleTeardown.stopDetached(expectedSessionId =")) + assertTrue(viewModel.contains("lifecycleTeardown.stopOrdered(expectedSessionId =")) + assertTrue(viewModel.contains("PlaybackTeardownGate(sessionLifecycle)")) assertTrue(!viewModel.contains("runBlocking(")) assertTrue(!screen.contains("onDispose { viewModel.onExit() }")) assertTrue(screen.contains("viewModel.claimInitialRouteLoad()")) diff --git a/androidTvApp/src/androidMain/kotlin/org/siloserver/silo/tv/ui/screens/player/TvPlayerViewModel.kt b/androidTvApp/src/androidMain/kotlin/org/siloserver/silo/tv/ui/screens/player/TvPlayerViewModel.kt index a5354ea1f..b3b7452fc 100644 --- a/androidTvApp/src/androidMain/kotlin/org/siloserver/silo/tv/ui/screens/player/TvPlayerViewModel.kt +++ b/androidTvApp/src/androidMain/kotlin/org/siloserver/silo/tv/ui/screens/player/TvPlayerViewModel.kt @@ -1530,6 +1530,11 @@ class TvPlayerViewModel( isCurrent = adoption::isCurrent, ) if (!adopted) return TvSubtitleAdoptionResult.Superseded + // The exit token names what the lifecycle owns, from the moment it owns + // it — not from the UI publication further down. Supersession in the gap + // otherwise leaves teardown naming the predecessor, the ownership guard + // rightly refusing it, and the one-shot gate blocking any retry. + lastAdoptedSessionId = ready.session.sessionId if (!adoption.isCurrent()) return TvSubtitleAdoptionResult.Superseded unpublishedSubtitleUi[ready.session.sessionId] = before @@ -1588,6 +1593,14 @@ class TvPlayerViewModel( restoreUi: Boolean, ): Boolean { val predecessor = unpublishedSubtitleUi.remove(playback.sessionId) + // The token follows lifecycle ownership in BOTH directions. Rollback + // hands ownership back to the predecessor, so leaving the token on the + // discarded replacement would make teardown name a session the + // lifecycle no longer holds — and be refused. This runs regardless of + // restoreUi: ownership reverts either way. + if (lastAdoptedSessionId == playback.sessionId) { + lastAdoptedSessionId = predecessor?.sessionId + } if (restoreUi && predecessor != null) { val identity = predecessor.committedSubtitleIdentity _uiState.value = predecessor.copy( @@ -1609,6 +1622,11 @@ class TvPlayerViewModel( if (!jointlyRolledBack) { playbackSessionManager.rollbackUnpublishedVideoSession(sessionId) } + // Same rule as the subtitle rollback: ownership reverted to the + // predecessor, so the exit token has to revert with it. + if (lastAdoptedSessionId == sessionId) { + lastAdoptedSessionId = predecessor?.state?.sessionId + } try { if (predecessor != null && _uiState.value.sessionId == sessionId) { val identity = predecessor.state.committedSubtitleIdentity @@ -1754,6 +1772,15 @@ class TvPlayerViewModel( return@launch } unpublishedReadySession.acquire(allocatedSessionId) + // Fresh load is a fourth lifecycle-first path: the + // starter already adopted this session before returning, + // and several suspending hydration steps stand between + // here and the UI publication below. Advance the exit + // token now, or an exit landing in that gap names the + // predecessor, is refused, and permanently claims the + // one-shot gate while this load goes on to publish. + // The rollback paths revert it if this never publishes. + lastAdoptedSessionId = allocatedSessionId if (!loadOwners.owns(loadOwner)) { loadOwners.publishReadyIfOwned( owner = loadOwner, @@ -1888,6 +1915,11 @@ class TvPlayerViewModel( predecessorSessionId = predecessorUi.sessionId, ) val transportMountNonce = nextTransportMountNonce(null) + // Paired with the UI publication so the exit + // token is never staler than UI state — the + // invariant that lets exitSessionId read it + // first. + result.sessionId?.let { lastAdoptedSessionId = it } _uiState.update { it.copy( isLoading = false, @@ -2242,28 +2274,42 @@ class TvPlayerViewModel( ?: effectiveVersion?.duration?.takeIf { it > 0.0 } ?: state.duration.takeIf { effectiveFileId == fileId } ?: 0.0 - val adopted = sessionLifecycle.adoptActiveSessionIfCurrent( - params = StartParams( - contentId = contentId, - fileId = effectiveFileId, - capabilities = capabilities, - audioTrackIndex = decision.session.audioTrackIndex, - subtitleTrackIndex = selectedSubtitle, - startPosition = decision.session.position, - ), - session = decision.session, - renewMissingSessionWithLegacyStart = false, - isCurrent = { - recoveryContentGeneration == contentLoadGeneration && - isActive - }, - ) - if (!adopted) { - runCatching { - playbackSessionManager.stopSession(decision.session.sessionId) + var adopted = false + try { + adopted = sessionLifecycle.adoptActiveSessionIfCurrent( + params = StartParams( + contentId = contentId, + fileId = effectiveFileId, + capabilities = capabilities, + audioTrackIndex = decision.session.audioTrackIndex, + subtitleTrackIndex = selectedSubtitle, + startPosition = decision.session.position, + ), + session = decision.session, + renewMissingSessionWithLegacyStart = false, + isCurrent = { + recoveryContentGeneration == contentLoadGeneration && + isActive + }, + ) + } finally { + // Covers refusal AND cancellation while awaiting the + // lifecycle mutex, which throws before isCurrent runs. + // NonCancellable because the usual reason for being + // here is that this coroutine was cancelled, and a + // cancelled one cannot make the releasing call. + if (!adopted) { + withContext(NonCancellable) { + runCatching { + playbackSessionManager.stopSession( + decision.session.sessionId, + ) + } + } } - return@launch } + if (!adopted) return@launch + lastAdoptedSessionId = decision.session.sessionId coroutineContext.ensureActive() if (recoveryContentGeneration != contentLoadGeneration) return@launch val transportMountNonce = nextTransportMountNonce(selectedSubtitle) @@ -2489,6 +2535,7 @@ class TvPlayerViewModel( positionSec = positionSec, durationSec = _uiState.value.duration, isPaused = _uiState.value.isPaused, + expectedSessionId = _uiState.value.sessionId, ) // Track B: durably record (local resume + outbox sync) for both streaming @@ -3117,10 +3164,14 @@ class TvPlayerViewModel( renewMissingSessionWithLegacyStart = false, isCurrent = { isCurrentSeekRecovery(request) }, ) - if (!adopted) { - runCatching { playbackSessionManager.stopSession(decision.session.sessionId) } - return - } + // Deliberately no stop on refusal. A seek re-anchor is validated to + // reuse the SAME session id — the manager rejects any response that + // changes it — so this id names the session still playing, not a + // disposable candidate. Refusal normally means a newer seek was queued, + // and that seek needs this very session as its base; stopping it here + // left the manager with no active attempt to re-anchor. + if (!adopted) return + lastAdoptedSessionId = decision.session.sessionId if (!isCurrentSeekRecovery(request)) return val transportMountNonce = nextTransportMountNonce(selectedSubtitle) _uiState.update { @@ -4387,8 +4438,19 @@ class TvPlayerViewModel( @Volatile private var lastAdoptedSessionId: String? = null + /** + * Retained token first, UI second. + * + * The token tracks *lifecycle ownership*, which is what teardown has to + * name, and it moves in both directions: forward at each adoption and at + * the load publication, back to the predecessor on either rollback path. + * That is strictly better than UI state here, because the three adoption + * paths take ownership before they publish and a cancellation in between + * would otherwise leave teardown naming a session the lifecycle has already + * let go of. + */ private val exitSessionId: String? - get() = _uiState.value.sessionId ?: lastAdoptedSessionId + get() = lastAdoptedSessionId ?: _uiState.value.sessionId /** * Keeps this screen's lifecycle teardown to exactly one stop. Without it, @@ -4422,7 +4484,12 @@ class TvPlayerViewModel( introObserveJob?.cancel() nextUpCountdownJob?.cancel() introAutoSkipController.reset() - _uiState.value.sessionId?.let { lastAdoptedSessionId = it } + // Only fills a gap; never overwrites. The adoption paths publish this + // token ahead of UI state on purpose, and taking the UI value here would + // put the older id back. + if (lastAdoptedSessionId == null) { + _uiState.value.sessionId?.let { lastAdoptedSessionId = it } + } _uiState.update { it.copy( isLoading = false, @@ -4721,7 +4788,6 @@ class TvPlayerViewModel( override fun onCleared() { episodeSelectionHandoffSlot.invalidate() - val teardownSessionId = exitSessionId val subtitlePersistenceReservation = subtitleTransactions.reserveDurableFinalPersistence() subtitleTransactions.invalidateAndSettleAsync(restoreUi = false) { @@ -4729,7 +4795,18 @@ class TvPlayerViewModel( subtitleTransactions::requestDurableFinalPersistence, ) playbackMutationFence.invalidateAll() - lifecycleTeardown.stopDetached(expectedSessionId = teardownSessionId) + // Read AFTER settlement, not snapshotted before it. Settlement can + // roll a subtitle publication back, and that rollback returns + // ownership to the predecessor — so a value captured before this + // callback names the discarded replacement, and the predecessor is + // left running with the one-shot gate already consumed. + // + // Never unqualified either. A null expectedSessionId disables the + // lifecycle's ownership guard entirely, and this callback is + // deliberately delayed behind subtitle settlement — long enough for + // a newer screen to have adopted its own session. A screen that + // never owned one has nothing to tear down. + exitSessionId?.let { lifecycleTeardown.stopDetached(expectedSessionId = it) } } subtitleSnapshotSettlement.reset() org.siloserver.silo.common.player.debug.PlaybackDebugState.screenError = null