refactor(player): demux through random-access IO instead of unwinding libavformat

Resuming or seeking a large MKV on the Android FFmpeg demux path buffered for minutes with no error and no fallback. The cause was the seam, not the container: libavformat's demuxers own their seeking — end-of-file index, back to the header, binary search — while media3's Extractor is forward only, so every backward jump had to abort the in-flight libavformat call with a synthetic IO error and replay the whole open. matroskadec attempts its deferred Cues parse exactly once per context, so that abort burned the index for the session and every later seek fell back to libavformat's linear generic scan, reading the file up to the target. That protocol had produced a bug of this shape repeatedly (a NULL deref resuming find_stream_info across an abort, a use-after-free on fallback teardown, matroskadec resync skipping the keyframe cluster) and it was held together by four empirical retry budgets whose exhaustion degraded silently — which is why #2096 looked like an unbounded spinner instead of an error. Give libavformat what it requires instead. FfmpegRandomAccessSource serves any absolute position from a second DataSource built by the same factory media3 uses, cloning the DataSpec media3 opened for the item so Cronet, download caches, SAF and per-item request headers still apply. Reads that the loader's ExtractorInput can serve still go through it, so sample delivery keeps feeding media3's byte accounting, back-pressure and load-error policy; the extractor nudges the loader to follow the demuxer with one RESULT_SEEK, which is an optimization that can never stall playback because the read succeeds either way. Deleted with the protocol: the synthetic AVERROR_NEED_SEEK unwind, the 24 MB replay cache, the Cues priming seek and its read guard, the sticky AVIO error clearing, and all four budgets (seekAttempts, stickyRecovers, MAX_RECONCILES, MAX_OPEN_LOADER_ROUND_TRIPS). Opens run straight through, seeks are one avformat_seek_file on the loader thread, and an IO failure is an IOException that reaches media3's retry policy instead of a silent degradation. The demuxer is now the only component that resolves seeks, so the extractor's sample-derived seek index is gone too. Verified on a SEI Robotics Android TV box against Jellyfin: the DV P8.1 28.4 GB MKV resume that started at a 4m28s spinner opens in 1.1 s and resolves its seek in one call, a mid-playback scrub resolves in 1 ms from the in-memory index, a 2 GB MKV resume opens in 0.5 s, fresh playback needs no index read at all, and MP4/AVI still open and deliver. New coverage: FfmpegRandomAccessSourceTest pins the read/reopen contract, and FfmpegExtractorSeekTest asserts on-device that a cued seek lands on target without walking the file and that a cueless file still reaches it. close #2096
2026-08-24 23:02:23 +02:00
parent defb0f4983
commit 1995fc6b75
11 changed files with 1024 additions and 563 deletions
@@ -7,6 +7,7 @@ import android.os.HandlerThread
 import androidx.media3.common.MediaItem
 import androidx.media3.common.PlaybackException
 import androidx.media3.common.Player
+import androidx.media3.datasource.DataSpec
 import androidx.media3.datasource.DefaultDataSource
 import androidx.media3.exoplayer.ExoPlayer
 import androidx.media3.exoplayer.source.ProgressiveMediaSource
@@ -66,12 +67,14 @@ class FfmpegExtractorAudioCsdPlaybackTest {

    handler.post {
      try {
+        val fixtureUri = Uri.fromFile(fixtureFile)
        val ffmpegOnlyExtractors = ExtractorsFactory {
          val extractor = FfmpegExtractor.create(
            { FfmpegDemuxerPolicy.Preference.FFMPEG },
            DvConversionMode.DISABLED,
            DefaultSubtitleParserFactory(),
-            AssHandler()
+            AssHandler(),
+            FfmpegRandomAccessSource(DefaultDataSource.Factory(context)) { DataSpec(fixtureUri) }
          )
          assertNotNull("FFmpeg demuxer JNI library is unavailable", extractor)
          arrayOf(extractor!!)
@@ -95,7 +98,7 @@ class FfmpegExtractorAudioCsdPlaybackTest {
        val source = ProgressiveMediaSource.Factory(
          DefaultDataSource.Factory(context),
          ffmpegOnlyExtractors
-        ).createMediaSource(MediaItem.fromUri(Uri.fromFile(fixtureFile)))
+        ).createMediaSource(MediaItem.fromUri(fixtureUri))
        player.setMediaSource(source)
        player.prepare()
        player.play()
@@ -0,0 +1,321 @@
+package com.edde746.plezy.exoplayer
+
+import android.content.Context
+import android.net.Uri
+import androidx.media3.common.C
+import androidx.media3.common.DataReader
+import androidx.media3.common.Format
+import androidx.media3.common.util.ParsableByteArray
+import androidx.media3.datasource.DataSource
+import androidx.media3.datasource.DataSpec
+import androidx.media3.datasource.DefaultDataSource
+import androidx.media3.datasource.TransferListener
+import androidx.media3.extractor.DefaultExtractorInput
+import androidx.media3.extractor.Extractor
+import androidx.media3.extractor.ExtractorInput
+import androidx.media3.extractor.ExtractorOutput
+import androidx.media3.extractor.PositionHolder
+import androidx.media3.extractor.SeekMap
+import androidx.media3.extractor.TrackOutput
+import androidx.media3.extractor.text.DefaultSubtitleParserFactory
+import androidx.test.ext.junit.runners.AndroidJUnit4
+import androidx.test.platform.app.InstrumentationRegistry
+import com.edde746.plezy.libass.media.AssHandler
+import java.io.File
+import org.junit.Assert.assertEquals
+import org.junit.Assert.assertTrue
+import org.junit.Test
+import org.junit.runner.RunWith
+
+/**
+ * Regression coverage for #2096 and for the seam that caused it.
+ *
+ * libavformat resolves seeks through its own index, reading it via
+ * [FfmpegRandomAccessSource]. When that read cannot be served, matroskadec
+ * burns its one-shot Cues parse and every later seek degrades to libavformat's
+ * linear generic scan — which is what buffered a resume for minutes. The cued
+ * fixture here has the same geometry as the file from the report (front
+ * SeekHead, Cues at 99.5% of the file), so a scan is trivially distinguishable
+ * from an index seek: it has to read most of the file.
+ */
+@RunWith(AndroidJUnit4::class)
+class FfmpegExtractorSeekTest {
+
+  private companion object {
+    // Regenerate with:
+    //   ffmpeg -f lavfi -i "testsrc=size=160x120:rate=10:duration=30" \
+    //     -c:v libx264 -g 10 -pix_fmt yuv420p -preset veryfast seek_cued.mkv
+    //   mkvmerge --no-cues -o seek_cueless.mkv seek_cued.mkv
+    // The cued fixture keeps ffmpeg's default layout: a SeekHead at byte 52
+    // pointing at Cues written after the clusters, at 99.5% of the file.
+    const val CUED = "ffmpeg/seek_cued.mkv"
+    const val CUELESS = "ffmpeg/seek_cueless.mkv"
+
+    /** The fixtures are 30 s at 10 fps with a keyframe every second. */
+    const val TARGET_US = 25_000_000L
+    const val KEYFRAME_TOLERANCE_US = 1_500_000L
+  }
+
+  @Test
+  fun cuedSeekLandsOnTheTargetWithoutWalkingTheFile() {
+    withHarness(CUED) { harness ->
+      harness.pumpUntilSampleDelivered()
+      assertTrue(
+        "playback should start at the beginning, got ${harness.lastSampleUs}us",
+        harness.lastSampleUs < 500_000L
+      )
+
+      harness.resetByteCounter()
+      harness.seekTo(TARGET_US)
+      val landed = harness.pumpUntilSampleAtLeast(TARGET_US - KEYFRAME_TOLERANCE_US)
+
+      assertTrue("no sample delivered after seeking to ${TARGET_US}us", landed)
+      assertTrue(
+        "seek landed at ${harness.lastSampleUs}us, expected within " +
+          "${KEYFRAME_TOLERANCE_US}us of ${TARGET_US}us",
+        harness.lastSampleUs <= TARGET_US + KEYFRAME_TOLERANCE_US
+      )
+      // An index seek reads the Cues plus the target cluster. A generic scan
+      // has to read everything up to 83% of the file to get here.
+      val budget = harness.fixtureSize / 3
+      assertTrue(
+        "seek read ${harness.bytesRead} bytes of a ${harness.fixtureSize}-byte file; " +
+          "expected under $budget (index seek, not a scan)",
+        harness.bytesRead < budget
+      )
+      // Reopening per read would mean one request per 64 KB in steady state.
+      assertTrue("random-access opens after seek: ${harness.randomAccessOpens}", harness.randomAccessOpens <= 4)
+    }
+  }
+
+  /**
+   * Without Cues libavformat falls back to its generic scan, exactly as media3's
+   * own MatroskaExtractor scans clusters. That is slow but it must still land on
+   * the target rather than stalling or silently delivering from the wrong place.
+   */
+  @Test
+  fun cuelessSeekStillReachesTheTarget() {
+    withHarness(CUELESS) { harness ->
+      harness.pumpUntilSampleDelivered()
+      harness.seekTo(TARGET_US)
+      val landed = harness.pumpUntilSampleAtLeast(TARGET_US - KEYFRAME_TOLERANCE_US)
+
+      assertTrue("cueless seek to ${TARGET_US}us delivered nothing", landed)
+      assertTrue(
+        "cueless seek landed at ${harness.lastSampleUs}us",
+        harness.lastSampleUs <= TARGET_US + KEYFRAME_TOLERANCE_US
+      )
+    }
+  }
+
+  @Test
+  fun playbackFromTheStartDeliversEverySampleFromTheBeginning() {
+    withHarness(CUED) { harness ->
+      val timestamps = harness.pumpSamples(count = 40)
+
+      assertTrue("expected samples from the start, got ${timestamps.size}", timestamps.size >= 40)
+      assertTrue("first sample at ${timestamps.first()}us", timestamps.first() < 500_000L)
+      // Delivery is in decode order, which is not presentation order once the
+      // encoder emits B-frames; what must hold is that the set of samples is
+      // gapless from zero (this fixture is a steady 10 fps).
+      val sorted = timestamps.sorted()
+      assertEquals(sorted.indices.map { it * 100_000L }, sorted)
+    }
+  }
+
+  private fun withHarness(fixture: String, block: (Harness) -> Unit) {
+    val instrumentation = InstrumentationRegistry.getInstrumentation()
+    val context = instrumentation.targetContext
+    val file = copyFixture(instrumentation.context, context, fixture)
+    val harness = Harness(context, file)
+    try {
+      harness.prepare()
+      block(harness)
+    } finally {
+      harness.close()
+      file.delete()
+    }
+  }
+
+  private fun copyFixture(instrumentationContext: Context, targetContext: Context, fixture: String): File {
+    val output = File.createTempFile("ffmpeg-seek-fixture-", ".mkv", targetContext.cacheDir)
+    instrumentationContext.assets.open(fixture).use { input ->
+      output.outputStream().use(input::copyTo)
+    }
+    return output
+  }
+
+  /**
+   * Minimal stand-in for ProgressiveMediaPeriod's loader: it opens the data
+   * source, feeds an ExtractorInput, and honours RESULT_SEEK by re-opening —
+   * the same contract media3 implements in ExtractingLoadable.load().
+   */
+  private class Harness(context: Context, private val fixture: File) {
+    val fixtureSize = fixture.length()
+    private val uri: Uri = Uri.fromFile(fixture)
+    private val counting = CountingDataSourceFactory(DefaultDataSource.Factory(context))
+    private val io = FfmpegRandomAccessSource(counting) { DataSpec(uri) }
+    private val output = CapturingOutput()
+    private val positionHolder = PositionHolder()
+    private val extractor = requireNotNull(
+      FfmpegExtractor.create(
+        { FfmpegDemuxerPolicy.Preference.FFMPEG },
+        DvConversionMode.DISABLED,
+        DefaultSubtitleParserFactory(),
+        AssHandler(),
+        io
+      )
+    ) { "native ffmpeg demuxer unavailable" }
+
+    private var dataSource: DataSource? = null
+    private var input: ExtractorInput? = null
+    private var opensBaseline = 0
+
+    val lastSampleUs: Long
+      get() = output.lastSampleUs
+
+    val bytesRead: Long
+      get() = counting.bytesRead
+
+    val randomAccessOpens: Int
+      get() = io.openCount - opensBaseline
+
+    fun prepare() {
+      openAt(0)
+      assertTrue("fixture should be sniffed by the ffmpeg demuxer", extractor.sniff(input!!))
+      input!!.resetPeekPosition()
+      extractor.init(output)
+    }
+
+    fun resetByteCounter() {
+      counting.reset()
+      opensBaseline = io.openCount
+    }
+
+    fun seekTo(timeUs: Long) {
+      // media3 restarts the load at the SeekMap's byte hint, then calls seek()
+      // on the loader thread before the next read.
+      val hint = output.seekMap?.getSeekPoints(timeUs)?.first?.position ?: 0L
+      openAt(hint)
+      extractor.seek(hint, timeUs)
+    }
+
+    fun pumpUntilSampleDelivered(): Boolean {
+      val before = output.sampleCount
+      return pump { output.sampleCount > before }
+    }
+
+    fun pumpUntilSampleAtLeast(timeUs: Long): Boolean = pump { output.lastSampleUs >= timeUs }
+
+    fun pumpSamples(count: Int): List<Long> {
+      pump { output.sampleTimestamps.size >= count }
+      return output.sampleTimestamps.toList()
+    }
+
+    private fun pump(until: () -> Boolean): Boolean {
+      repeat(200_000) {
+        if (until()) return true
+        when (extractor.read(input!!, positionHolder)) {
+          Extractor.RESULT_SEEK -> openAt(positionHolder.position)
+          Extractor.RESULT_END_OF_INPUT -> return until()
+          else -> Unit
+        }
+      }
+      return until()
+    }
+
+    private fun openAt(position: Long) {
+      dataSource?.close()
+      val source = counting.createDataSource()
+      val remaining = source.open(DataSpec.Builder().setUri(uri).setPosition(position).build())
+      dataSource = source
+      input = DefaultExtractorInput(
+        source,
+        position,
+        if (remaining == C.LENGTH_UNSET.toLong()) C.LENGTH_UNSET.toLong() else position + remaining
+      )
+    }
+
+    fun close() {
+      extractor.release()
+      dataSource?.close()
+    }
+  }
+
+  private class CapturingOutput : ExtractorOutput {
+    val sampleTimestamps = mutableListOf<Long>()
+    var seekMap: SeekMap? = null
+      private set
+
+    val sampleCount: Int
+      get() = sampleTimestamps.size
+
+    val lastSampleUs: Long
+      get() = sampleTimestamps.lastOrNull() ?: C.TIME_UNSET
+
+    override fun track(id: Int, type: Int): TrackOutput = CapturingTrack(type, sampleTimestamps)
+
+    override fun endTracks() = Unit
+
+    override fun seekMap(seekMap: SeekMap) {
+      this.seekMap = seekMap
+    }
+  }
+
+  private class CapturingTrack(private val type: Int, private val timestamps: MutableList<Long>) : TrackOutput {
+    private val scratch = ByteArray(64 * 1024)
+
+    override fun format(format: Format) = Unit
+
+    override fun sampleData(input: DataReader, length: Int, allowEndOfInput: Boolean): Int = sampleData(input, length, allowEndOfInput, TrackOutput.SAMPLE_DATA_PART_MAIN)
+
+    override fun sampleData(input: DataReader, length: Int, allowEndOfInput: Boolean, sampleDataPart: Int): Int {
+      val read = input.read(scratch, 0, minOf(length, scratch.size))
+      return if (read == C.RESULT_END_OF_INPUT) 0 else read
+    }
+
+    override fun sampleData(data: ParsableByteArray, length: Int) {
+      sampleData(data, length, TrackOutput.SAMPLE_DATA_PART_MAIN)
+    }
+
+    override fun sampleData(data: ParsableByteArray, length: Int, sampleDataPart: Int) {
+      data.skipBytes(length)
+    }
+
+    override fun sampleMetadata(timeUs: Long, flags: Int, size: Int, offset: Int, cryptoData: TrackOutput.CryptoData?) {
+      if (type == C.TRACK_TYPE_VIDEO) timestamps.add(timeUs)
+    }
+  }
+
+  /** Counts every byte the demuxer pulls, through either read path. */
+  private class CountingDataSourceFactory(private val delegate: DataSource.Factory) : DataSource.Factory {
+    @Volatile private var counter = 0L
+
+    val bytesRead: Long
+      get() = counter
+
+    fun reset() {
+      counter = 0L
+    }
+
+    override fun createDataSource(): DataSource = Counting(delegate.createDataSource())
+
+    private inner class Counting(private val inner: DataSource) : DataSource {
+      override fun open(dataSpec: DataSpec): Long = inner.open(dataSpec)
+
+      override fun read(buffer: ByteArray, offset: Int, length: Int): Int {
+        val read = inner.read(buffer, offset, length)
+        if (read > 0) counter += read
+        return read
+      }
+
+      override fun addTransferListener(transferListener: TransferListener) = inner.addTransferListener(transferListener)
+
+      override fun getUri(): Uri? = inner.uri
+
+      override fun getResponseHeaders(): Map<String, List<String>> = inner.responseHeaders
+
+      override fun close() = inner.close()
+    }
+  }
+}