Reject regressed stream PTS before submitting it to the decoder

This commit is contained in:
evgeny committed 2026-10-29 13:34:24 +01:00
1 parent 5e6313377b
commit aa27892c12
2 files changed
+11 -2

No files matched your search

@@ -2,7 +2,8 @@ package com.utun.chat.data
import java.util.ArrayDeque
/** Один рабочий вход и один готовый выход. Codec владеет внутренним конвейером;
/** Один рабочий вход и один готовый выход; PTS/position/now/lead/period — микросекунды.
* Codec владеет внутренним конвейером;
* число поданных кадров и часы показа не ограничивают заполнение этого конвейера.
* Все вызовы принадлежат одному player worker. Ошибки логирует владелец вместе с session ID. */
internal class StreamDecodePump<F : StreamDecodePump.Frame>(
@@ -57,8 +58,9 @@ internal class StreamDecodePump<F : StreamDecodePump.Frame>(
drain(position, now)
if (output != null || ended) return
val frame = input
if (frame != null && submit(frame)) {
if (frame != null) {
check(lastInputPts < frame.pts) { "Decoder input PTS did not advance" }
if (!submit(frame)) return
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)
@@ -106,4 +106,11 @@ class StreamDecodePumpTest {
assertTrue(pump.ended); assertEquals(0, pump.inflight); assertEquals(-1L, pump.oldestSubmittedPts)
assertEquals(1, released); pump.checkProgress(30_000_000, true)
}
@Test fun regressedInputPtsIsRejectedBeforeTheCodecReceivesIt() {
val codec = Codec(1); codec.pump.offer(Frame(100), 0); codec.pump.step(100, 0)
codec.pump.offer(Frame(100), 1)
assertThrows(IllegalStateException::class.java) { codec.pump.step(100, 1) }
assertTrue(codec.queued.isEmpty()); assertEquals(1L, codec.pump.inputCount)
}
}