Drive Android decoding by codec readiness and bounded working frames

2026-10-14 13:36:38 +02:00
1 parent 223c9a153c
commit f72e2fca94
@@ -0,0 +1,94 @@
+package com.utun.chat.data
+
+import java.util.ArrayDeque
+
+/** Один рабочий вход и один готовый выход. Codec владеет внутренним конвейером;
+ * число поданных кадров и часы показа не ограничивают заполнение этого конвейера.
+ * Все вызовы принадлежат одному player worker. Ошибки логирует владелец вместе с session ID. */
+internal class StreamDecodePump<F : StreamDecodePump.Frame>(
+    private val submit: (F) -> Boolean,
+    private val receive: () -> Output?,
+    private val release: (Output, Boolean) -> Unit
+) {
+    interface Frame { val pts: Long }
+    data class Output(val index: Int, val pts: Long, val eos: Boolean = false, val empty: Boolean = false)
+    private val submitted = ArrayDeque<Long>()
+    var input: F? = null
+        private set
+    var output: Output? = null
+        private set
+    var firstOutputPts: Long? = null
+        private set
+    var lastInputPts = -1L
+        private set
+    var inputCount = 0L
+        private set
+    var outputCount = 0L
+        private set
+    var shown = 0L
+        private set
+    var discarded = 0L
+        private set
+    var decoderLeadUs = 0L
+        private set
+    var framePeriodUs = 33333L
+        private set
+    var ended = false
+        private set
+    private var waitingSince = -1L
+    private var lastInputAt = -1L
+    val wantsInput: Boolean get() = input == null && output == null && !ended
+    val oldestSubmittedPts: Long get() = submitted.firstOrNull() ?: -1L
+    val inflight: Int get() = submitted.size
+
+    fun offer(frame: F, now: Long) {
+        // Уже запрошенный ответ может прийти после появления будущего output.
+        check(input == null && !ended) { "Decoder already owns a working frame" }
+        // Новые данные после сетевой паузы получают собственное окно ожидания decoder progress.
+        if (lastInputAt < 0 || now - lastInputAt > 1_000_000) waitingSince = now
+        input = frame
+    }
+
+    /** Выход проверяется перед каждым входом. Свободный input buffer берут лишь для готового кадра. */
+    fun step(position: Long, now: Long) {
+        drain(position, now)
+        if (output != null || ended) return
+        val frame = input
+        if (frame != null && submit(frame)) {
+            check(lastInputPts < frame.pts) { "Decoder input PTS did not advance" }
+            if (lastInputPts >= 0) framePeriodUs = (frame.pts - lastInputPts).coerceIn(1, 1_000_000)
+            lastInputPts = frame.pts; lastInputAt = now; inputCount++; submitted.addLast(frame.pts); input = null
+            drain(position, now)
+        }
+    }
+
+    private fun drain(position: Long, now: Long) {
+        repeat(8) {
+            if (ended) return
+            if (output == null) {
+                val next = receive() ?: return
+                output = next
+                if (!next.empty) {
+                    if (firstOutputPts == null) firstOutputPts = next.pts
+                    outputCount++; decoderLeadUs = (lastInputPts - next.pts).coerceAtLeast(0)
+                    while (submitted.isNotEmpty() && submitted.first <= next.pts) submitted.removeFirst()
+                }
+                waitingSince = now
+            }
+            val next = output!!
+            if (!next.empty && next.pts > position) return
+            val render = !next.empty && next.pts >= position - 100000
+            release(next, render); output = null
+            if (!next.empty) { if (render) shown++ else discarded++ }
+            if (next.eos) ended = true
+        }
+    }
+
+    /** Нехватка сети и ожидание PTS допустимы; зависший decoder при наличии данных — ошибка. */
+    fun checkProgress(now: Long, finishing: Boolean) {
+        val hasWork = input != null || (lastInputAt >= 0 && now - lastInputAt < 1_000_000) || finishing
+        check(ended || output != null || !hasWork || waitingSince < 0 || now - waitingSince <= 10_000_000) {
+            "Decoder made no output progress: inputs=$inputCount outputs=$outputCount inflight=$inflight lastPTS=$lastInputPts"
+        }
+    }
+}
@@ -5,14 +5,16 @@ import android.media.AudioFormat
 import android.media.AudioTimestamp
 import android.media.AudioTrack
 import android.media.MediaCodec
+import android.media.MediaCodecInfo
+import android.media.MediaCodecList
 import android.media.MediaFormat
+import android.os.Build
 import android.view.Surface
 import com.utun.chat.R
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.StateFlow
 import java.nio.ByteBuffer
 import java.nio.ByteOrder
-import java.util.ArrayDeque
 import java.util.concurrent.locks.ReentrantLock
 import kotlin.concurrent.withLock

@@ -23,9 +25,9 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv
                         private val localPreview: StreamPreview? = null, private val displayDegrees: Int = 0) {
    companion object { private val viewerLock = ReentrantLock() }
    private class UnsupportedCodec(message: String) : IllegalStateException(message)
-    private data class Frame(val pts: Long, val key: Boolean, val width: Int, val height: Int,
+    private data class Frame(override val pts: Long, val key: Boolean, val width: Int, val height: Int,
                             val rotation: Int, val top: Int?, val revision: Long, val codec: StreamVideoCodec,
-                             val config: ByteArray, val data: ByteArray)
+                             val config: ByteArray, val data: ByteArray) : StreamDecodePump.Frame
    private val mutableAspect = MutableStateFlow(16f / 9f)
    val aspect: StateFlow<Float> = mutableAspect
    private val mutableTop = MutableStateFlow<Int?>(null)
@@ -47,8 +49,7 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv
        var decoder: MediaCodec? = null
        var track: AudioTrack? = null
        var playback = 0L
-        val videos = ArrayDeque<Frame>()
-        val submitted = ArrayDeque<Long>()
+        lateinit var video: StreamDecodePump<Frame>
        var waitingKey = true
        var audioEnabled = false
        var opened = false
@@ -62,14 +63,8 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv
        var previousHead = 0L
        var headEpoch = 0L
        var generation = 0L
-        var decoded = 0L
-        var discarded = 0L
-        var lastSubmittedPts = -1L
-        var pendingOutput = -1
-        var pendingPts = 0L
-        var pendingEos = false
+        var playbackStarted = false
        var eosQueued = false
-        var decoderEnded = false
        var endUntil = 0L
        var reported = 0L
        var deadline = System.nanoTime() + 10_000_000_000
@@ -77,6 +72,53 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv
        val info = MediaCodec.BufferInfo()
        var clockState: LongArray? = null

+        fun createVideoPump() = StreamDecodePump<Frame>(
+            submit = { frame ->
+                if (decoder == null) {
+                    check(frame.key)
+                    val format = frame.codec.decoderFormat(frame.width, frame.height, frame.rotation, frame.config)
+                    val name = MediaCodecList(MediaCodecList.REGULAR_CODECS).findDecoderForFormat(format)
+                        ?: throw UnsupportedCodec("Unsupported decoder ${frame.codec.label} ${frame.width}x${frame.height}")
+                    decoder = MediaCodec.createByCodecName(name)
+                    val lowLatency = Build.VERSION.SDK_INT >= 30 && decoder!!.codecInfo.getCapabilitiesForType(frame.codec.mime)
+                        .isFeatureSupported(MediaCodecInfo.CodecCapabilities.FEATURE_LowLatency)
+                    if (lowLatency) format.setInteger(MediaFormat.KEY_LOW_LATENCY, 1)
+                    LogManager.addLog("INFO", "video", "viewer codec id=$id type=${frame.codec.label} decoder=$name " +
+                        "config=${frame.config.size} lowLatency=$lowLatency revision=$videoRevision")
+                    decoder!!.configure(format, surface, null, 0); decoder!!.start()
+                }
+                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); 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)
+                        }
+                        break
+                    }
+                }
+                output
+            },
+            release = { output, render ->
+                decoder!!.releaseOutputBuffer(output.index, render)
+                if (render) mutableStatus.value = R.string.stream_live
+            }
+        )
+
        fun observeStatus(meta: LongArray?) {
            if (meta == null || meta[5] and 1 == 0L) return
            ready = if (localPreview != null) meta[1] else meta[0]
@@ -90,8 +132,8 @@ 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; videos.clear(); submitted.clear(); pendingOutput = -1; lastSubmittedPts = -1; pendingEos = false
-            waitingKey = true; eosQueued = false; decoderEnded = false
+            decoder = null; video = createVideoPump()
+            waitingKey = true; eosQueued = false
            mutableStatus.value = R.string.stream_waiting_key
        }
        fun resetTimeline(request: Boolean) {
@@ -99,7 +141,7 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv
            catch (e: Exception) { LogManager.addLog("WARN", "stream", "audio reset id=$id: ${e.message}") }
            track = null; pausedAudio = false; written = 0; previousHead = 0; headEpoch = 0
            if (playback != 0L) NativeLib.streamPlaybackClose(playback)
-            playback = 0; position = -1; generation = 0; needsAudio = true; clockState = null
+            playback = 0; position = -1; generation = 0; needsAudio = true; clockState = null; playbackStarted = false
            resetVideo(); deadline = System.nanoTime() + 10_000_000_000
            if (request) check(NativeLib.streamResync(id)) { "Media resync refused" }
        }
@@ -137,6 +179,7 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv
                else { check(System.nanoTime() < until) { "AudioTrack stalled" }; Thread.sleep(2) }
            }
        }
+        video = createVideoPump()
        try {
            check(NativeLib.streamOpen(id, group, source, audio = localPreview == null, epoch = epoch)) { "Viewer open refused" }
            while (running && surface.isValid) {
@@ -145,11 +188,11 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv
                val now = System.nanoTime()
                val played = playedFrames(now)
                val deviceLead = (written - played).coerceAtLeast(0) * 1_000_000 / 48000
-                val videoAhead = if (videos.size < 2 && submitted.size < 3) 80000L else -1L
-                val audioAhead = if (audioEnabled && needsAudio) deviceLead + 80000 else -1L
-                repeat(4) {
-                    val raw = if (ended) null else NativeLib.streamPull(id, position, videoAhead, audioAhead)
-                    if (raw == null) return@repeat
+                for (work in 0 until 4) {
+                    video.step(position, System.nanoTime() / 1000)
+                    val audioAhead = if (audioEnabled && needsAudio) deviceLead + 80000 else -1L
+                    val raw = if (ended) null else NativeLib.streamPull(id, position, video.wantsInput, audioAhead)
+                    if (raw == null) break
                    check(raw.size >= 28)
                    val b = ByteBuffer.wrap(raw).order(ByteOrder.BIG_ENDIAN)
                    val kind = b.get().toInt(); val flags = b.get().toInt() and 255
@@ -165,7 +208,7 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv
                            val camera = localPreview?.camera(revision)
                            if (localPreview != null && camera == null) {
                                LogManager.addLog("WARN", "video", "preview camera unavailable id=$id revision=$revision")
-                                resetTimeline(true); return@repeat
+                                resetTimeline(true); continue
                            }
                            val display = StreamVideoDisplay.forFrame(broadcastRotation, camera, displayDegrees)
                            if (revision != videoRevision) {
@@ -180,9 +223,11 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv
                                if (waitingKey) {
                                    waitingKey = false; mutableTop.value = display.top
                                    mutableAspect.value = if (display.rotation % 180 == 0) width.toFloat() / height else height.toFloat() / width
+                                    mutableStatus.value = R.string.stream_waiting_decoder
+                                    LogManager.addLog("INFO", "video", "viewer KEY accepted id=$id revision=$revision PTS=$pts bytes=$size")
                                }
-                                check(videos.size < 4) { "Video preparation queue exceeded" }
-                                videos.addLast(Frame(pts, key, width, height, display.rotation, display.top, revision, codec, access.first, access.second))
+                                video.offer(Frame(pts, key, width, height, display.rotation, display.top, revision, codec, access.first, access.second),
+                                    System.nanoTime() / 1000)
                            }
                        }
                        2 -> if (audioEnabled && playback != 0L) {
@@ -204,7 +249,12 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv
                    }
                }
                if (!ended && (!opened || waitingKey) && System.nanoTime() > deadline) error("Viewer open timeout")
-                if (playback != 0L && audioEnabled && !pausedAudio) {
+                if (!playbackStarted && video.firstOutputPts != null) {
+                    playbackStarted = true
+                    LogManager.addLog("INFO", "video", "viewer first decoded frame id=$id revision=$videoRevision " +
+                        "PTS=${video.firstOutputPts} inputs=${video.inputCount} decoderLeadUs=${video.decoderLeadUs}")
+                }
+                if (playbackStarted && playback != 0L && audioEnabled && !pausedAudio) {
                    val head = playedFrames(System.nanoTime())
                    val t = track
                    val prime = t == null || (head == 0L && written < t.bufferSizeInFrames)
@@ -212,73 +262,39 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv
                        NativeLib.streamPlaybackPull(playback, 480, if (ended) Long.MAX_VALUE else ready)?.let { writePcm(it) }
                    }
                }
-                if (playback != 0L && ready >= 0) {
-                    val lateVideo = submitted.isNotEmpty() && submitted.first < position - 10000
+                if (playbackStarted && playback != 0L && ready >= 0) {
+                    val lateVideo = video.firstOutputPts == null || (video.oldestSubmittedPts >= 0 && video.oldestSubmittedPts < position - 10000)
                    if (lateVideo && track != null && !pausedAudio) { track!!.pause(); pausedAudio = true }
                    if (pausedAudio && !lateVideo) { track!!.play(); pausedAudio = false }
                    val head = playedFrames(System.nanoTime())
                    val audioEmpty = audioEnabled && written > 0 && head >= written && !ended
                    clockState = NativeLib.streamPlaybackUpdate(playback, System.nanoTime() / 1000, head,
                        if (ended) ready.coerceAtLeast(position) else ready,
-                        (deviceLead + 10000).coerceAtLeast(33333), lateVideo || audioEmpty)
+                        maxOf(deviceLead + 10000, video.decoderLeadUs + video.framePeriodUs), lateVideo || audioEmpty)
                    check(clockState != null) { "Playback timeline failed" }
                    position = clockState!![0]; needsAudio = clockState!![5] != 0L
                }
-                repeat(4) {
-                    if (videos.isEmpty()) return@repeat
-                    val frame = videos.first
-                    if (decoder == null) {
-                        check(frame.key)
-                        val format = frame.codec.decoderFormat(frame.width, frame.height, frame.rotation, frame.config)
-                        val name = android.media.MediaCodecList(android.media.MediaCodecList.REGULAR_CODECS).findDecoderForFormat(format)
-                            ?: throw UnsupportedCodec("Unsupported decoder ${frame.codec.label} ${frame.width}x${frame.height}")
-                        decoder = MediaCodec.createByCodecName(name)
-                        LogManager.addLog("INFO", "video", "viewer codec id=$id type=${frame.codec.label} decoder=$name config=${frame.config.size}")
-                        decoder!!.configure(format, surface, null, 0); decoder!!.start()
-                    }
-                    val index = decoder!!.dequeueInputBuffer(0)
-                    if (index >= 0) {
-                        val input = decoder!!.getInputBuffer(index) ?: error("Missing decoder input")
-                        check(frame.data.size <= input.capacity()); input.clear(); input.put(frame.data)
-                        decoder!!.queueInputBuffer(index, 0, frame.data.size, frame.pts, 0)
-                        submitted.addLast(frame.pts); lastSubmittedPts = frame.pts; videos.removeFirst(); decoded++
-                    }
-                }
                val codec = decoder
                if (codec != null) {
-                    if (ended && videos.isEmpty() && !eosQueued) {
+                    if (ended && video.input == null && video.output == null && !eosQueued) {
                        val index = codec.dequeueInputBuffer(0)
-                        if (index >= 0) { codec.queueInputBuffer(index, 0, 0, lastSubmittedPts.coerceAtLeast(0),
+                        if (index >= 0) { codec.queueInputBuffer(index, 0, 0, video.lastInputPts.coerceAtLeast(0),
                            MediaCodec.BUFFER_FLAG_END_OF_STREAM); eosQueued = true }
                    }
-                    repeat(8) {
-                        if (pendingOutput < 0) {
-                            val index = codec.dequeueOutputBuffer(info, 0)
-                            if (index < 0) return@repeat
-                            if (info.flags and MediaCodec.BUFFER_FLAG_END_OF_STREAM != 0 && info.size == 0) {
-                                codec.releaseOutputBuffer(index, false); decoderEnded = true; return@repeat
-                            }
-                            pendingOutput = index; pendingPts = info.presentationTimeUs
-                            pendingEos = info.flags and MediaCodec.BUFFER_FLAG_END_OF_STREAM != 0
-                            while (submitted.isNotEmpty() && submitted.first <= pendingPts) submitted.removeFirst()
-                        }
-                        if (pendingPts > position) return@repeat
-                        val render = pendingPts >= position - 100000
-                        codec.releaseOutputBuffer(pendingOutput, render); pendingOutput = -1
-                        if (pendingEos) { decoderEnded = true; pendingEos = false }
-                        if (!render) discarded++ else mutableStatus.value = R.string.stream_live
-                    }
+                    video.step(position, System.nanoTime() / 1000); video.checkProgress(System.nanoTime() / 1000, ended)
                }
-                if (ended && videos.isEmpty() && (decoder == null || decoderEnded) && NativeLib.streamPlaybackEmpty(playback)
+                if (ended && video.input == null && (decoder == null || video.ended) && NativeLib.streamPlaybackEmpty(playback)
                    && playedFrames(System.nanoTime()) >= written) {
                    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 decoded=$decoded discarded=$discarded " +
-                        "videoQ=${videos.size} submitted=${submitted.size} clock=$position reserve=${clockState?.get(1)} " +
-                        "minimum=${clockState?.get(2)} tempo=${clockState?.get(3)} written=$written waitingKey=$waitingKey")
+                    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} " +
+                        "clock=$position reserve=${clockState?.get(1)} minimum=${clockState?.get(2)} " +
+                        "tempo=${clockState?.get(3)} written=$written waitingKey=$waitingKey")
                }
                Thread.sleep(5)
            }
@@ -296,12 +312,13 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv
            if (playback != 0L) NativeLib.streamPlaybackClose(playback)
            val until = System.nanoTime() + 500_000_000
            while (NativeLib.isReady() && System.nanoTime() < until) {
-                val event = NativeLib.streamPull(id)
+                val event = NativeLib.streamPull(id, wantVideo = false, audioAheadUs = -1)
                if (event != null && event[0].toInt() in 5..6) break
                Thread.sleep(5)
            }
            running = false
-            LogManager.addLog("INFO", "stream", "viewer released id=$id decoded=$decoded discarded=$discarded")
+            LogManager.addLog("INFO", "stream", "viewer released id=$id inputs=${video.inputCount} " +
+                "decoded=${video.outputCount} shown=${video.shown} discarded=${video.discarded}")
        }
    }
 }
@@ -0,0 +1,79 @@
+package com.utun.chat.data
+
+import org.junit.Assert.*
+import org.junit.Test
+import java.util.ArrayDeque
+
+class StreamDecodePumpTest {
+    private data class Frame(override val pts: Long) : StreamDecodePump.Frame
+    private class Codec(val depth: Int) {
+        val queued = ArrayDeque<Frame>()
+        var released = 0
+        var blocked = false
+        var finishing = false
+        var held = false
+        var nextIndex = 0
+        val pump = StreamDecodePump<Frame>(
+            submit = { if (blocked) false else { queued.addLast(it); true } },
+            receive = {
+                if (held || (queued.size < depth && !finishing) || queued.isEmpty()) null
+                else { held = true; StreamDecodePump.Output(nextIndex++, queued.removeFirst().pts) }
+            },
+            release = { _, _ -> check(held); held = false; released++ }
+        )
+    }
+
+    @Test fun fourAndEightFramePipelinesStartWithFrozenPresentationClock() {
+        for (depth in listOf(4, 8)) {
+            val codec = Codec(depth)
+            repeat(depth) { i ->
+                assertTrue(codec.pump.wantsInput)
+                codec.pump.offer(Frame(i * 66666L), i * 66666L)
+                codec.pump.step(0, i * 66666L)
+            }
+            assertEquals(depth.toLong(), codec.pump.inputCount)
+            assertEquals(1L, codec.pump.outputCount)
+            assertEquals(1, codec.released)
+            assertEquals(0L, codec.pump.firstOutputPts)
+            assertEquals((depth - 1) * 66666L, codec.pump.decoderLeadUs)
+        }
+    }
+
+    @Test fun futureOutputStopsReadsAndNeverReleasesEarly() {
+        val codec = Codec(1)
+        codec.pump.offer(Frame(100000), 0); codec.pump.step(0, 0)
+        assertFalse(codec.pump.wantsInput); assertNotNull(codec.pump.output); assertEquals(0, codec.released)
+        codec.pump.step(99999, 15_000_000); codec.pump.checkProgress(15_000_000, false)
+        assertEquals(0, codec.released)
+        codec.pump.step(100000, 15_000_001)
+        assertEquals(1, codec.released); assertTrue(codec.pump.wantsInput)
+    }
+
+    @Test fun blockedCodecKeepsOneInputAndReportsStall() {
+        val codec = Codec(4); codec.blocked = true
+        codec.pump.offer(Frame(0), 0); codec.pump.step(0, 0)
+        assertNotNull(codec.pump.input); assertFalse(codec.pump.wantsInput)
+        try { codec.pump.checkProgress(10_000_001, false); fail("Stalled decoder must fail") }
+        catch (expected: IllegalStateException) { assertTrue(expected.message!!.contains("no output progress")) }
+    }
+
+    @Test fun networkPauseKeepsReferencesAndAllowsIncompletePipelineToResume() {
+        val codec = Codec(4)
+        repeat(2) { i -> codec.pump.offer(Frame(i * 40000L), i * 40000L); codec.pump.step(0, i * 40000L) }
+        codec.pump.checkProgress(20_000_000, false)
+        repeat(2) { i ->
+            val now = 20_000_000 + i * 40000L
+            codec.pump.offer(Frame((i + 2) * 40000L), now); codec.pump.step(0, now); codec.pump.checkProgress(now, false)
+        }
+        assertEquals(4L, codec.pump.inputCount); assertEquals(1, codec.released)
+    }
+
+    @Test fun finishDrainsShortStreamAndPreservesTheLastPicture() {
+        val codec = Codec(8)
+        repeat(3) { i -> codec.pump.offer(Frame(i * 40000L), i * 40000L); codec.pump.step(0, i * 40000L) }
+        assertEquals(0, codec.released); codec.finishing = true
+        codec.pump.step(0, 120000); assertEquals(1, codec.released); assertFalse(codec.pump.wantsInput)
+        codec.pump.step(40000, 160000); assertEquals(2, codec.released)
+        codec.pump.step(80000, 200000); assertEquals(3, codec.released)
+    }
+}