Reject malformed stream batches before mutating the playback queue

This commit is contained in:
evgeny committed 2026-10-29 13:34:24 +01:00
1 parent 5fb6e72c43
commit a1dcf28bf9
2 files changed
+31 -6

No files matched your search

@@ -4,7 +4,10 @@ import java.nio.ByteBuffer
import java.nio.ByteOrder
/** Ближайшие записи viewer: порядок каждого трека независим, EOF не ждёт чужой трек.
* GAP/CLOSED отменяют прежние данные. Только player worker; ошибки логирует его владелец. */
* GAP/CLOSED отменяют прежние данные. add принимает до 16 BE записей (header 28 bytes + payload)
* и проверяет всю пачку до изменения очереди; при ошибке очередь сохраняется.
* take разрешает EOF обойти другой трек, сохраняя порядок собственного. Только player worker;
* ограничение общей очереди/backpressure задаёт caller, он же логирует исключения. */
internal class StreamReadQueue {
private val records = mutableListOf<ByteArray>()
val size: Int get() = records.size
@@ -14,19 +17,22 @@ internal class StreamReadQueue {
fun add(batch: ByteArray): Int {
val input = ByteBuffer.wrap(batch).order(ByteOrder.BIG_ENDIAN)
var count = 0
val parsed = mutableListOf<ByteArray>()
while (input.hasRemaining()) {
check(input.remaining() >= 28) { "Truncated stream record header" }
val start = input.position()
val kind = batch[start].toInt()
val length = input.getInt(start + 16)
check(kind in 1..8 && length >= 0 && length <= input.remaining() - 28) { "Invalid stream record kind=$kind length=$length" }
if (kind == 3 || kind == 6) records.clear()
records.add(batch.copyOfRange(start, start + 28 + length))
check(parsed.size < 16) { "Stream batch exceeds record budget" }
parsed.add(batch.copyOfRange(start, start + 28 + length))
input.position(start + 28 + length)
check(++count <= 16) { "Stream batch exceeds record budget" }
}
return count
for (record in parsed) {
if (record[0].toInt() in listOf(3, 6)) records.clear()
records.add(record)
}
return parsed.size
}
fun take(video: Boolean, audio: Boolean): ByteArray? {
@@ -52,4 +52,23 @@ class StreamReadQueueTest {
assertArrayEquals(audio, queue.take(video = false, audio = true))
assertArrayEquals(video, queue.take(video = true, audio = false))
}
@Test fun malformedTailCannotApplyItsEarlierGapOrReplacePendingData() {
val queue = StreamReadQueue(); val original = record(1, 42)
queue.add(original)
for (invalid in listOf(record(3) + byteArrayOf(1), record(6) + record(9),
record(2) + record(1).also { ByteBuffer.wrap(it).putInt(16, 100) })) {
assertThrows(IllegalStateException::class.java) { queue.add(invalid) }
assertEquals(1, queue.size)
}
assertArrayEquals(original, queue.take(video = true, audio = true))
}
@Test fun excessiveRecordCountIsRejectedBeforeAnyChange() {
val queue = StreamReadQueue(); val original = record(2, 123)
queue.add(original)
val invalid = List(17) { record(3) }.fold(byteArrayOf()) { bytes, next -> bytes + next }
assertThrows(IllegalStateException::class.java) { queue.add(invalid) }
assertEquals(1, queue.size); assertArrayEquals(original, queue.take(video = true, audio = true))
}
}