Isolate Android live video decoding from audio preparation

2026-10-21 01:23:35 +02:00
1 parent 4fa3b18cc2
commit 4e5530aca0
@@ -1,10 +1,5 @@
 package com.utun.chat.data

-import android.media.MediaCodec
-import android.media.MediaCodecInfo
-import android.media.MediaCodecList
-import android.media.MediaFormat
-import android.os.Build
 import android.os.Handler
 import android.os.Looper
 import android.view.Surface
@@ -16,8 +11,8 @@ import kotlinx.coroutines.withContext
 import java.nio.ByteBuffer
 import java.nio.ByteOrder

-/** Запас кодированного потока остаётся в общем RS-кольце. Worker запрашивает ближайшие кадры,
- * общий native controller ведёт минутную историю и темп, AudioTrack задаёт фактические часы.
+/** Audio worker читает независимые треки из RS-кольца и готовит PCM; MediaCodec работает отдельно.
+ * Общий native controller убирает избыточный запас ускорением, AudioTrack задаёт часы каждого источника.
 * Временная нехватка сохраняет decoder references; GAP/вытеснение начинают новую точку KEY. */
 class StreamPlayerEngine(private val group: Long, private val source: Long, private val epoch: Long) {
    companion object {
@@ -42,10 +37,6 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv
            LogManager.addLog("INFO", "stream", "viewers released for call count=${stopped.size}")
        }
    }
-    private class UnsupportedCodec(message: String) : IllegalStateException(message)
-    private data class Frame(override val pts: Long, val key: Boolean, val width: Int, val height: Int,
-                             val rotation: Int, val revision: Long, val codec: StreamVideoCodec,
-                             val config: ByteArray, val data: ByteArray) : StreamDecodePump.Frame
    private val mutableAspect = MutableStateFlow(0f) // Ориентация неизвестна до первого KEY с форматом.
    val aspect: StateFlow<Float> = mutableAspect
    private val mutableStatus = MutableStateFlow(R.string.stream_connecting)
@@ -80,7 +71,7 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv
                catch (e: Exception) { LogManager.addLog("WARN", "stream", "viewer surface release id=$id: ${e.message}") }
                main.post { players.remove(this); mutableViewing.value = players.isNotEmpty() }
            }
-        }, "stream-player").apply { start() }
+        }, "stream-audio-$id").apply { start() }
    }
    /** Surface/SurfaceTexture освобождаются callback только после остановки decoder и native DSP. */
    fun stop(onReleased: (() -> Unit)? = null) {
@@ -94,9 +85,8 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv
    }

    private fun play(surface: Surface) {
-        var decoder: MediaCodec? = null
        var playback: StreamMixedPlayback? = null
-        lateinit var video: StreamDecodePump<Frame>
+        var video: StreamVideoPlayback? = null
        var waitingKey = true
        var audioEnabled = false
        var opened = false
@@ -115,130 +105,15 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv
        var lastAudioPts = -1L
        var generation = 0L
        var playbackStarted = false
-        var eosQueued = false
        var endUntil = 0L
        var reported = 0L
+        var workMaxUs = 0L
        var deadline = System.nanoTime() + 10_000_000_000
-        val info = MediaCodec.BufferInfo()
        var clockState: LongArray? = null
-        var firstInputAtUs = 0L
-        var firstShown = false
-        var lastShownPts = -1L
-
-        fun createVideoPump() = StreamDecodePump<Frame>(
-            submit = { frame ->
-                if (decoder == null) {
-                    val openingAtUs = System.nanoTime() / 1000
-                    check(frame.key)
-                    val format = frame.codec.decoderFormat(frame.width, frame.height, frame.rotation, frame.config)
-                    val codecs = MediaCodecList(MediaCodecList.REGULAR_CODECS)
-                    val preferred = codecs.findDecoderForFormat(format)
-                    val candidates = codecs.codecInfos.filter { !it.isEncoder && frame.codec.mime in it.supportedTypes }
-                        .sortedBy { if (it.name == preferred) 0 else 1 }
-                    if (candidates.isEmpty()) throw UnsupportedCodec("Unsupported decoder ${frame.codec.label}")
-                    var failure: Exception? = null
-                    for (candidate in candidates) {
-                        var attempt: MediaCodec? = null
-                        try {
-                            val capabilities = candidate.getCapabilitiesForType(frame.codec.mime)
-                            if (!capabilities.isFormatSupported(format)) continue
-                            val lowLatency = Build.VERSION.SDK_INT >= 30 &&
-                                capabilities.isFeatureSupported(MediaCodecInfo.CodecCapabilities.FEATURE_LowLatency)
-                            if (Build.VERSION.SDK_INT >= 30) format.setInteger(MediaFormat.KEY_LOW_LATENCY, if (lowLatency) 1 else 0)
-                            attempt = MediaCodec.createByCodecName(candidate.name)
-                            attempt.configure(format, surface, null, 0); attempt.start(); decoder = attempt
-                            var firstRendered = true
-                            val codecRevision = frame.revision
-                            try {
-                                attempt.setOnFrameRenderedListener({ codec, pts, renderedNs ->
-                                    if (firstRendered) {
-                                        firstRendered = false
-                                        val callbackUs = System.nanoTime() / 1000
-                                        val renderedUs = renderedNs / 1000
-                                        val renderClockValid = renderedUs in requestedAtUs..callbackUs
-                                        if (!renderClockValid) LogManager.addLog("WARN", "stream",
-                                            "viewer surface timestamp outside request interval id=$id PTS=$pts renderedNs=$renderedNs " +
-                                                "requestedUs=$requestedAtUs callbackUs=$callbackUs; using callback time")
-                                        LogManager.addLog("INFO", "stream", "android first surface frame id=$id revision=$codecRevision " +
-                                            "stage=surface_callback PTS=$pts elapsedUs=${callbackUs - requestedAtUs} " +
-                                            "renderElapsedUs=${if (renderClockValid) renderedUs - requestedAtUs else -1} " +
-                                            "callbackDelayUs=${if (renderClockValid) callbackUs - renderedUs else -1}")
-                                        try { codec.setOnFrameRenderedListener(null, null) }
-                                        catch (e: Exception) {
-                                            LogManager.addLog("WARN", "stream", "viewer surface diagnostics detach id=$id: ${e.message}")
-                                        }
-                                    }
-                                }, main)
-                            } catch (e: Exception) {
-                                LogManager.addLog("WARN", "stream", "viewer presentation diagnostics unavailable id=$id: ${e.message}")
-                            }
-                            LogManager.addLog("INFO", "video", "viewer codec id=$id type=${frame.codec.label} decoder=${candidate.name} " +
-                                "config=${frame.config.size} lowLatency=$lowLatency revision=$videoRevision " +
-                                "elapsedUs=${System.nanoTime() / 1000 - requestedAtUs} openUs=${System.nanoTime() / 1000 - openingAtUs}")
-                            break
-                        } catch (e: Exception) {
-                            failure = e
-                            LogManager.addLog("WARN", "video", "viewer decoder unavailable id=$id codec=${candidate.name}: ${e.message}")
-                            try { attempt?.release() }
-                            catch (release: Exception) { LogManager.addLog("WARN", "video", "decoder fallback release: ${release.message}") }
-                            if (!surface.isValid) throw e
-                        }
-                    }
-                    if (decoder == null) {
-                        if (failure == null) throw UnsupportedCodec("Unsupported decoder format ${frame.codec.label} ${frame.width}x${frame.height}")
-                        throw IllegalStateException("No available ${frame.codec.label} decoder", failure)
-                    }
-                }
-                val index = decoder!!.dequeueInputBuffer(0)
-                if (index < 0) false else {
-                    val input = decoder!!.getInputBuffer(index) ?: error("Missing decoder input")
-                    check(frame.data.size <= input.capacity()) { "Decoder input is too small: ${input.capacity()}/${frame.data.size}" }
-                    input.clear(); input.put(frame.data)
-                    if (firstInputAtUs == 0L) firstInputAtUs = System.nanoTime() / 1000
-                    decoder!!.queueInputBuffer(index, 0, frame.data.size, frame.pts, 0); true
-                }
-            },
-            receive = {
-                var output: StreamDecodePump.Output? = null
-                val codec = decoder
-                if (codec != null) {
-                    for (attempt in 0 until 3) {
-                        val index = codec.dequeueOutputBuffer(info, 0)
-                        if (index == MediaCodec.INFO_OUTPUT_FORMAT_CHANGED) {
-                            LogManager.addLog("INFO", "video", "viewer output format id=$id revision=$videoRevision ${codec.outputFormat}")
-                            continue
-                        }
-                        if (index >= 0) {
-                            val eos = info.flags and MediaCodec.BUFFER_FLAG_END_OF_STREAM != 0
-                            output = StreamDecodePump.Output(index, info.presentationTimeUs, eos, eos && info.size == 0)
-                            if (!output.empty && video.firstOutputPts == null) {
-                                val decodedAtUs = System.nanoTime() / 1000
-                                LogManager.addLog("INFO", "video", "viewer first decoded frame id=$id revision=$videoRevision " +
-                                    "PTS=${output.pts} inputs=${video.inputCount} decoderLeadUs=${maxOf(0, video.lastInputPts - output.pts)} " +
-                                    "elapsedUs=${decodedAtUs - requestedAtUs} decodeUs=${decodedAtUs - firstInputAtUs}")
-                            }
-                        }
-                        break
-                    }
-                }
-                output
-            },
-            release = { output, render ->
-                if (render && !firstShown) {
-                    firstShown = true
-                    LogManager.addLog("INFO", "stream", "android first frame id=$id stage=display_submit " +
-                        "elapsedUs=${System.nanoTime() / 1000 - requestedAtUs} PTS=${output.pts} clockUs=$position " +
-                        "readyUs=$ready readyAheadUs=${ready - output.pts} avUs=${output.pts - position} " +
-                        "deviceLeadUs=${playback?.leadUs}")
-                }
-                decoder!!.releaseOutputBuffer(output.index, render)
-                if (render) { lastShownPts = output.pts; mutableStatus.value = R.string.stream_live }
-            }
-        )

        fun observeStatus(meta: LongArray?) {
            if (meta == null || meta[5] and 1 == 0L) return
-            ready = meta[0]
+            ready = if (meta[5] and 2 != 0L && meta[5] and 16 == 0L) meta[2] else maxOf(meta[1], meta[2])
            if (playback != null) return
            audioEnabled = meta[5] and 2 != 0L
            playback = StreamMixedPlayback(id, meta[3], audioEnabled) { NativeLib.streamSignalNotify(signal) }
@@ -249,33 +124,34 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv
        }

        fun resetVideo() {
-            try { decoder?.stop(); decoder?.release() }
-            catch (e: Exception) { LogManager.addLog("WARN", "video", "decoder reset id=$id: ${e.message}") }
-            decoder = null; video = createVideoPump(); firstInputAtUs = 0
-            waitingKey = true; eosQueued = false
+            video?.reset(); waitingKey = true
            mutableStatus.value = R.string.stream_waiting_key
        }
        fun resetTimeline(request: Boolean) {
            try { playback?.close() }
            catch (e: Exception) { LogManager.addLog("WARN", "stream", "audio reset id=$id: ${e.message}") }
            playback = null; position = -1; generation = 0; needsAudio = true; clockState = null; playbackStarted = false
-            records.clear(); ended = false; videoInputEnded = false; audioInputEnded = false
+            records.clear(); ended = false; videoInputEnded = false; audioInputEnded = false; endUntil = 0
            resetVideo(); deadline = System.nanoTime() + 10_000_000_000
            if (request) check(NativeLib.streamResync(id)) { "Media resync refused" }
        }
-        video = createVideoPump()
        try {
            check(NativeLib.streamOpen(id, group, source, audio = true, epoch = epoch)) { "Viewer open refused" }
            signal = NativeLib.streamSignalCreate(id).also { check(it != 0L) { "Viewer notification initialization failed" } }
+            video = StreamVideoPlayback(id, surface, requestedAtUs,
+                wake = { NativeLib.streamSignalNotify(signal) }, onRendered = { if (running) mutableStatus.value = R.string.stream_live })
+            LogManager.addLog("INFO", "stream", "viewer audio worker started id=$id preparedPcmMs=250 audioAheadUs=500000")
+
            while (running && surface.isValid) {
                check(NativeLib.isReady()) { "Core stopped" }
+                video.failure?.let { throw it }
+                val workStarted = System.nanoTime() / 1000
                observeStatus(NativeLib.streamStatus(id))
                drainTimeoutMs = ((ready - position) / 1000 + 2000).coerceIn(2000, 30000)
                val deviceLead = playback?.leadUs ?: 0
                var consumed = 0
                for (work in 0 until 32) {
                    if (!running) break
-                    video.step(position, System.nanoTime() / 1000)
                    needsAudio = !audioInputEnded && (playback?.needsInput ?: true)
                    val audioAhead = if (!audioInputEnded && (!audioEnabled || needsAudio) && records.audios == 0) deviceLead + 500000 else -1L
                    val videoBudget = if (!videoInputEnded && video.wantsInput && records.videos == 0) 1 else 0
@@ -311,8 +187,7 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv
                                    LogManager.addLog("INFO", "video", "viewer KEY accepted id=$id revision=$revision PTS=$pts bytes=$size " +
                                        "elapsedUs=${System.nanoTime() / 1000 - requestedAtUs} readyAheadUs=${ready - pts}")
                                }
-                                video.offer(Frame(pts, key, width, height, broadcastRotation, revision, codec, access.first, access.second),
-                                    System.nanoTime() / 1000)
+                                video.offer(StreamVideoPlayback.Frame(pts, key, width, height, broadcastRotation, revision, codec, access.first, access.second))
                            }
                        }
                        2 -> if (audioEnabled && playback != null) {
@@ -323,7 +198,7 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv
                        3 -> { LogManager.addLog("INFO", "stream", "playback GAP id=$id"); resetTimeline(false) }
                        4 -> opened = true
                        5, 7, 8 -> {
-                            if (kind != 8) videoInputEnded = true
+                            if (kind != 8 && !videoInputEnded) { videoInputEnded = true; video.finish() }
                            if (kind != 7 && !audioInputEnded) { audioInputEnded = true; playback?.finish() }
                            if (kind == 5) ended = true
                            LogManager.addLog("INFO", "stream", "viewer track end id=$id kind=$kind videoEnd=$videoInputEnded " +
@@ -343,7 +218,8 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv
                }
                if (!running) break
                if (!ended && (!opened || waitingKey) && System.nanoTime() > deadline) error("Viewer open timeout")
-                if (!playbackStarted && video.firstOutputPts != null) playbackStarted = true
+                val decoded = video.state
+                if (!playbackStarted && decoded.firstOutputPts != null) playbackStarted = true
                if (playbackStarted && playback != null && ready >= 0) {
                    clockState = playback!!.update(System.nanoTime() / 1000,
                        if (ended) ready.coerceAtLeast(position) else ready,
@@ -352,54 +228,43 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv
                    position = clockState!![0]; needsAudio = clockState!![5] != 0L
                }
                check((playback?.prepare() ?: 0) >= 0) { "PCM preparation failed" }
-                val codec = decoder
-                if (codec != null) {
-                    if (videoInputEnded && video.input == null && video.output == null && !eosQueued) {
-                        val index = codec.dequeueInputBuffer(0)
-                        if (index >= 0) { codec.queueInputBuffer(index, 0, 0, video.lastInputPts.coerceAtLeast(0),
-                            MediaCodec.BUFFER_FLAG_END_OF_STREAM); eosQueued = true }
-                    }
-                    video.step(position, System.nanoTime() / 1000); video.checkProgress(System.nanoTime() / 1000, videoInputEnded)
-                    check(!video.ended || video.firstOutputPts != null || video.inputCount == 0L) { "Decoder ended without producing an image" }
-                }
-                if (ended && video.input == null && (decoder == null || video.ended) && playback?.empty != false
+                video.update(position, ready, playback?.leadUs ?: 0)
+                workMaxUs = maxOf(workMaxUs, System.nanoTime() / 1000 - workStarted)
+                if (ended && decoded.ended && playback?.empty != false
                    && playback?.drained != false) {
                    if (endUntil == 0L) endUntil = System.nanoTime() + 100_000_000
                    if (System.nanoTime() >= endUntil) { mutableStatus.value = R.string.stream_ended; break }
                }
                if (System.nanoTime() - reported >= 1_000_000_000) {
                    reported = System.nanoTime()
-                    LogManager.addLog("DEBUG", "stream", "viewer id=$id revision=$videoRevision inputs=${video.inputCount} " +
-                        "decoded=${video.outputCount} shown=${video.shown} discarded=${video.discarded} input=${video.input != null} " +
-                        "output=${video.output?.pts} inflight=${video.inflight} decoderLeadUs=${video.decoderLeadUs} " +
+                    LogManager.addLog("DEBUG", "stream", "android audio worker id=$id generation=$generation revision=$videoRevision " +
                        "clock=$position ready=$ready audio=$audioEnabled audioInputs=$audioInputs audioPTS=$lastAudioPts needsAudio=$needsAudio " +
-                        "reserve=${clockState?.get(1)} minimum=${clockState?.get(2)} " +
-                        "tempo=${clockState?.get(3)} ${playback?.description} waitingKey=$waitingKey " +
-                        "shownPTS=$lastShownPts displayLagUs=${if (lastShownPts >= 0) position - lastShownPts else -1} " +
+                        "reserve=${clockState?.get(1)} minimum=${clockState?.get(2)} tempo=${clockState?.get(3)} " +
+                        "${playback?.description} workMaxUs=$workMaxUs waitingKey=$waitingKey " +
                        "batches=$batches batchMax=$batchMax pending=${records.size} videoEnd=$videoInputEnded audioEnd=$audioInputEnded")
-                    batches = 0; batchMax = 0
+                    batches = 0; batchMax = 0; workMaxUs = 0
                }
-                // Синхронный MediaCodec требует проверки готовности; media/PCM/stop будят без ожидания deadline.
+                // PCM consumption, media и video state будят worker; без аудио часы идут по wall clock.
                if (running && consumed < 32) {
-                    val codecPending = video.input != null || video.output != null || video.inflight > 0 || (videoInputEnded && !video.ended)
-                    check(NativeLib.streamSignalWait(signal, if (codecPending) 5 else 50) >= 0) { "Viewer notification wait failed" }
+                    check(NativeLib.streamSignalWait(signal, if (playbackStarted && !audioEnabled) 5 else 50) >= 0) {
+                        "Viewer notification wait failed"
+                    }
                }
            }
        } catch (e: Exception) {
            if (running) {
-                mutableStatus.value = if (e is UnsupportedCodec) R.string.stream_decoder_unsupported else R.string.stream_failed
+                mutableStatus.value = if (e is StreamVideoPlayback.UnsupportedCodec) R.string.stream_decoder_unsupported else R.string.stream_failed
                LogManager.addLog("ERROR", "stream", "viewer failed id=$id: ${e.message}")
            }
        } finally {
            NativeLib.streamStop(id)
-            try { decoder?.stop(); decoder?.release() }
+            try { video?.close() }
            catch (e: Exception) { LogManager.addLog("WARN", "video", "decoder release id=$id: ${e.message}") }
            try { playback?.close() }
            catch (e: Exception) { LogManager.addLog("WARN", "stream", "audio release id=$id: ${e.message}") }
            NativeLib.streamSignalClose(signal)
            running = false
-            LogManager.addLog("INFO", "stream", "viewer released id=$id inputs=${video.inputCount} " +
-                "decoded=${video.outputCount} shown=${video.shown} discarded=${video.discarded}")
+            LogManager.addLog("INFO", "stream", "viewer released id=$id video=${video?.state}")
        }
    }
 }
@@ -0,0 +1,273 @@
+package com.utun.chat.data
+
+import android.media.MediaCodec
+import android.media.MediaCodecInfo
+import android.media.MediaCodecList
+import android.media.MediaFormat
+import android.os.Build
+import android.os.Handler
+import android.os.Looper
+import android.view.Surface
+import java.util.ArrayDeque
+
+/** MediaCodec и pump принадлежат только video worker. Audio worker передаёт кадры через ограниченный
+ * mailbox; reset меняет поколение, close дожидается release до освобождения Surface/notification.
+ * Ни один вызов MediaCodec не выполняется под mailbox lock или на потоке подготовки звука. */
+internal class StreamVideoPlayback(
+    private val id: Long, private val surface: Surface, private val requestedAtUs: Long,
+    private val wake: () -> Unit, private val onRendered: () -> Unit
+) {
+    class UnsupportedCodec(message: String) : IllegalStateException(message)
+    data class Frame(override val pts: Long, val key: Boolean, val width: Int, val height: Int,
+                     val rotation: Int, val revision: Long, val codec: StreamVideoCodec,
+                     val config: ByteArray, val data: ByteArray) : StreamDecodePump.Frame
+    data class State(
+        val firstOutputPts: Long? = null, val ended: Boolean = false, val inputs: Long = 0,
+        val decoded: Long = 0, val shown: Long = 0, val discarded: Long = 0, val inflight: Int = 0,
+        val decoderLeadUs: Long = 0, val shownPts: Long = -1, val outputPts: Long? = null
+    )
+    private data class Clock(val position: Long = -1, val ready: Long = -1, val leadUs: Long = 0)
+    private val lock = Object()
+    private val input = ArrayDeque<Frame>()
+    @Volatile private var running = true
+    @Volatile private var generation = 0L
+    @Volatile private var clock = Clock()
+    @Volatile var state = State()
+        private set
+    @Volatile var failure: Exception? = null
+        private set
+    private var inputEnded = false // mailbox lock
+    private var activeGeneration = -1L // Далее поля только video worker.
+    private var decoder: MediaCodec? = null
+    private var video: StreamDecodePump<Frame> = createVideoPump()
+    private val info = MediaCodec.BufferInfo()
+    private var videoRevision = 0L
+    private var firstInputAtUs = 0L
+    private var firstShown = false
+    private var lastShownPts = -1L
+    private var eosQueued = false
+    private val main = Handler(Looper.getMainLooper())
+    private val position: Long get() = clock.position
+    private val ready: Long get() = clock.ready
+    private val worker = Thread(::run, "stream-video-$id").apply { start() }
+
+    val wantsInput: Boolean get() = synchronized(lock) { running && !inputEnded && input.size < 2 }
+
+    fun offer(frame: Frame) = synchronized(lock) {
+        check(running && !inputEnded && input.size < 2) { "Video mailbox full or closed id=$id size=${input.size}" }
+        input.addLast(frame); lock.notifyAll()
+    }
+
+    fun update(position: Long, ready: Long, leadUs: Long) {
+        clock = Clock(position, ready, leadUs)
+        synchronized(lock) { lock.notifyAll() }
+    }
+
+    fun finish() = synchronized(lock) { inputEnded = true; lock.notifyAll() }
+
+    fun reset() = synchronized(lock) {
+        generation++; input.clear(); inputEnded = false; state = State(); clock = Clock()
+        lock.notifyAll()
+        LogManager.addLog("INFO", "video", "viewer video reset id=$id generation=$generation")
+    }
+
+    fun close() {
+        synchronized(lock) { running = false; lock.notifyAll() }
+        worker.join()
+    }
+
+    private fun releaseDecoder() {
+        try { decoder?.stop() }
+        catch (e: Exception) { LogManager.addLog("WARN", "video", "decoder stop id=$id: ${e.message}") }
+        try { decoder?.release() }
+        catch (e: Exception) { LogManager.addLog("WARN", "video", "decoder release id=$id: ${e.message}") }
+        decoder = null
+    }
+
+    private fun run() {
+        var reported = 0L
+        var workMaxUs = 0L
+        try {
+            LogManager.addLog("INFO", "video", "viewer video worker started id=$id mailbox=2")
+            while (running) {
+                val requestedGeneration = generation
+                if (activeGeneration != requestedGeneration) {
+                    releaseDecoder(); video = createVideoPump(); firstInputAtUs = 0
+                    eosQueued = false; videoRevision = 0; lastShownPts = -1
+                    activeGeneration = requestedGeneration
+                }
+                val started = System.nanoTime() / 1000
+                video.step(position, started)
+                val frame = synchronized(lock) {
+                    if (activeGeneration == generation && video.wantsInput) input.pollFirst() else null
+                }
+                if (frame != null) {
+                    if (videoRevision != frame.revision) {
+                        check(frame.key) { "Video revision does not start at KEY" }
+                        releaseDecoder(); video = createVideoPump(); firstInputAtUs = 0
+                        eosQueued = false; videoRevision = frame.revision
+                    }
+                    video.offer(frame, started); video.step(position, System.nanoTime() / 1000)
+                }
+                val finishing = synchronized(lock) { activeGeneration == generation && inputEnded && input.isEmpty() }
+                val codec = decoder
+                if (finishing && codec != null && video.input == null && video.output == null && !eosQueued) {
+                    val index = codec.dequeueInputBuffer(0)
+                    if (index >= 0) {
+                        codec.queueInputBuffer(index, 0, 0, video.lastInputPts.coerceAtLeast(0), MediaCodec.BUFFER_FLAG_END_OF_STREAM)
+                        eosQueued = true
+                    }
+                }
+                video.step(position, System.nanoTime() / 1000)
+                video.checkProgress(System.nanoTime() / 1000, finishing)
+                check(!video.ended || video.firstOutputPts != null || video.inputCount == 0L) { "Decoder ended without producing an image" }
+                val next = State(video.firstOutputPts, video.ended || (finishing && codec == null),
+                    video.inputCount, video.outputCount, video.shown, video.discarded, video.inflight,
+                    video.decoderLeadUs, lastShownPts, video.output?.pts)
+                val changed = synchronized(lock) {
+                    if (activeGeneration == generation && state != next) { state = next; true } else false
+                }
+                if (changed || frame != null) wake()
+                val now = System.nanoTime() / 1000
+                workMaxUs = maxOf(workMaxUs, now - started)
+                if (now - reported >= 1_000_000) {
+                    reported = now
+                    LogManager.addLog("DEBUG", "stream", "android video worker id=$id generation=$activeGeneration revision=$videoRevision " +
+                        "inputs=${video.inputCount} decoded=${video.outputCount} shown=${video.shown} discarded=${video.discarded} " +
+                        "inflight=${video.inflight} decoderLeadUs=${video.decoderLeadUs} workMaxUs=$workMaxUs " +
+                        "clock=$position shownPTS=$lastShownPts displayLagUs=${if (lastShownPts >= 0) position - lastShownPts else -1}")
+                    workMaxUs = 0
+                }
+                synchronized(lock) {
+                    if (running && activeGeneration == generation && (input.isEmpty() || !video.wantsInput)) {
+                        val pending = video.input != null || video.output != null || video.inflight > 0 || (finishing && !next.ended)
+                        lock.wait(if (pending) 5 else 50)
+                    }
+                }
+            }
+        } catch (e: Exception) {
+            if (running) {
+                failure = e
+                LogManager.addLog("ERROR", "video", "viewer video worker failed id=$id generation=$activeGeneration: ${e.message}")
+                wake()
+            }
+        } finally {
+            releaseDecoder()
+            LogManager.addLog("INFO", "video", "viewer video worker released id=$id")
+        }
+    }
+
+    private fun createVideoPump() = StreamDecodePump<Frame>(
+        submit = { frame ->
+            if (decoder == null) {
+                val openingAtUs = System.nanoTime() / 1000
+                check(frame.key)
+                val format = frame.codec.decoderFormat(frame.width, frame.height, frame.rotation, frame.config)
+                val codecs = MediaCodecList(MediaCodecList.REGULAR_CODECS)
+                val preferred = codecs.findDecoderForFormat(format)
+                val candidates = codecs.codecInfos.filter { !it.isEncoder && frame.codec.mime in it.supportedTypes }
+                    .sortedBy { if (it.name == preferred) 0 else 1 }
+                if (candidates.isEmpty()) throw UnsupportedCodec("Unsupported decoder ${frame.codec.label}")
+                var failure: Exception? = null
+                for (candidate in candidates) {
+                    var attempt: MediaCodec? = null
+                    try {
+                        val capabilities = candidate.getCapabilitiesForType(frame.codec.mime)
+                        if (!capabilities.isFormatSupported(format)) continue
+                        val lowLatency = Build.VERSION.SDK_INT >= 30 &&
+                            capabilities.isFeatureSupported(MediaCodecInfo.CodecCapabilities.FEATURE_LowLatency)
+                        if (Build.VERSION.SDK_INT >= 30) format.setInteger(MediaFormat.KEY_LOW_LATENCY, if (lowLatency) 1 else 0)
+                        attempt = MediaCodec.createByCodecName(candidate.name)
+                        attempt.configure(format, surface, null, 0); attempt.start(); decoder = attempt
+                        var firstRendered = true
+                        val codecRevision = frame.revision
+                        try {
+                            attempt.setOnFrameRenderedListener({ codec, pts, renderedNs ->
+                                if (firstRendered) {
+                                    firstRendered = false
+                                    val callbackUs = System.nanoTime() / 1000
+                                    val renderedUs = renderedNs / 1000
+                                    val renderClockValid = renderedUs in requestedAtUs..callbackUs
+                                    if (!renderClockValid) LogManager.addLog("WARN", "stream",
+                                        "viewer surface timestamp outside request interval id=$id PTS=$pts renderedNs=$renderedNs " +
+                                            "requestedUs=$requestedAtUs callbackUs=$callbackUs; using callback time")
+                                    LogManager.addLog("INFO", "stream", "android first surface frame id=$id revision=$codecRevision " +
+                                        "stage=surface_callback PTS=$pts elapsedUs=${callbackUs - requestedAtUs} " +
+                                        "renderElapsedUs=${if (renderClockValid) renderedUs - requestedAtUs else -1} " +
+                                        "callbackDelayUs=${if (renderClockValid) callbackUs - renderedUs else -1}")
+                                    try { codec.setOnFrameRenderedListener(null, null) }
+                                    catch (e: Exception) {
+                                        LogManager.addLog("WARN", "stream", "viewer surface diagnostics detach id=$id: ${e.message}")
+                                    }
+                                }
+                            }, main)
+                        } catch (e: Exception) {
+                            LogManager.addLog("WARN", "stream", "viewer presentation diagnostics unavailable id=$id: ${e.message}")
+                        }
+                        LogManager.addLog("INFO", "video", "viewer codec id=$id type=${frame.codec.label} decoder=${candidate.name} " +
+                            "config=${frame.config.size} lowLatency=$lowLatency revision=$videoRevision " +
+                            "elapsedUs=${System.nanoTime() / 1000 - requestedAtUs} openUs=${System.nanoTime() / 1000 - openingAtUs}")
+                        break
+                    } catch (e: Exception) {
+                        failure = e
+                        LogManager.addLog("WARN", "video", "viewer decoder unavailable id=$id codec=${candidate.name}: ${e.message}")
+                        try { attempt?.release() }
+                        catch (release: Exception) { LogManager.addLog("WARN", "video", "decoder fallback release: ${release.message}") }
+                        if (!surface.isValid) throw e
+                    }
+                }
+                if (decoder == null) {
+                    if (failure == null) throw UnsupportedCodec("Unsupported decoder format ${frame.codec.label} ${frame.width}x${frame.height}")
+                    throw IllegalStateException("No available ${frame.codec.label} decoder", failure)
+                }
+            }
+            val index = decoder!!.dequeueInputBuffer(0)
+            if (index < 0) false else {
+                val input = decoder!!.getInputBuffer(index) ?: error("Missing decoder input")
+                check(frame.data.size <= input.capacity()) { "Decoder input is too small: ${input.capacity()}/${frame.data.size}" }
+                input.clear(); input.put(frame.data)
+                if (firstInputAtUs == 0L) firstInputAtUs = System.nanoTime() / 1000
+                decoder!!.queueInputBuffer(index, 0, frame.data.size, frame.pts, 0); true
+            }
+        },
+        receive = {
+            var output: StreamDecodePump.Output? = null
+            val codec = decoder
+            if (codec != null) {
+                for (attempt in 0 until 3) {
+                    val index = codec.dequeueOutputBuffer(info, 0)
+                    if (index == MediaCodec.INFO_OUTPUT_FORMAT_CHANGED) {
+                        LogManager.addLog("INFO", "video", "viewer output format id=$id revision=$videoRevision ${codec.outputFormat}")
+                        continue
+                    }
+                    if (index >= 0) {
+                        val eos = info.flags and MediaCodec.BUFFER_FLAG_END_OF_STREAM != 0
+                        output = StreamDecodePump.Output(index, info.presentationTimeUs, eos, eos && info.size == 0)
+                        if (!output.empty && video.firstOutputPts == null) {
+                            val decodedAtUs = System.nanoTime() / 1000
+                            LogManager.addLog("INFO", "video", "viewer first decoded frame id=$id revision=$videoRevision " +
+                                "PTS=${output.pts} inputs=${video.inputCount} decoderLeadUs=${maxOf(0, video.lastInputPts - output.pts)} " +
+                                "elapsedUs=${decodedAtUs - requestedAtUs} decodeUs=${decodedAtUs - firstInputAtUs}")
+                        }
+                    }
+                    break
+                }
+            }
+            output
+        },
+        release = { output, render ->
+            val current = running && activeGeneration == generation
+            if (render && current && !firstShown) {
+                firstShown = true
+                LogManager.addLog("INFO", "stream", "android first frame id=$id stage=display_submit " +
+                    "elapsedUs=${System.nanoTime() / 1000 - requestedAtUs} PTS=${output.pts} clockUs=$position " +
+                    "readyUs=$ready readyAheadUs=${ready - output.pts} avUs=${output.pts - position} " +
+                    "deviceLeadUs=${clock.leadUs}")
+            }
+            decoder!!.releaseOutputBuffer(output.index, render && current)
+            if (render && current) { lastShownPts = output.pts; onRendered() }
+        }
+    )
+
+}
+