fix: prepare server 1.2.4

build.gradle.kts

@@ -36,7 +36,7 @@ dependencies {
     implementation("io.ktor:ktor-server-call-logging-jvm")
     implementation("io.ktor:ktor-server-rate-limit-jvm")
     implementation("ch.qos.logback:logback-classic:1.5.38")
-    implementation("com.github.Priveetee.PipePipeExtractor:extractor:d1a723d382014c2428b273c92bc9fc8c15d59c83")
+    implementation("com.github.Priveetee.PipePipeExtractor:extractor:21ab9b6ac415d23ef88df19c5ad126c7952bb543")
     compileOnly("com.github.TeamNewPipe:nanojson:1d9e1aea9049fc9f85e68b43ba39fe7be1c1f751")
     implementation("org.json:json:20260522")
     implementation("com.squareup.okhttp3:okhttp:5.4.0")

gradle.properties

@@ -1,5 +1,5 @@
 org.gradle.jvmargs=-Xmx2g -XX:+UseG1GC
 kotlin.code.style=official
-appVersion=1.2.3
+appVersion=1.2.4
 systemProp.sun.net.client.defaultReadTimeout=180000
 systemProp.sun.net.client.defaultConnectTimeout=60000

settings.gradle.kts

@@ -6,6 +6,8 @@ if (localPipePipeExtractor.isDirectory) {
         dependencySubstitution {
             substitute(module("com.github.InfinityLoop1308.PipePipeExtractor:extractor"))
                 .using(project(":extractor"))
+            substitute(module("com.github.Priveetee.PipePipeExtractor:extractor"))
+                .using(project(":extractor"))
         }
     }
 }

src/main/kotlin/dev/typetype/server/services/SabrPlaybackInitializationPreloader.kt

@@ -1,20 +1,32 @@
 package dev.typetype.server.services
 
-import kotlinx.coroutines.Dispatchers
-import kotlinx.coroutines.async
-import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.withTimeoutOrNull
 
+internal data class SabrPlaybackInitializationPreloadResult(
+    val video: ByteArray?,
+    val audio: ByteArray?,
+) {
+    fun isComplete(audioOnly: Boolean): Boolean =
+        audio != null && (audioOnly || video != null)
+
+    fun missingTracks(audioOnly: Boolean, videoItag: Int, audioItag: Int): String =
+        buildList {
+            if (!audioOnly && video == null) add("video:$videoItag")
+            if (audio == null) add("audio:$audioItag")
+        }.joinToString()
+}
+
 internal object SabrPlaybackInitializationPreloader {
     suspend fun preload(
         sessionStore: SabrSessionStore,
         holder: SabrSessionHolder,
+        audioOnly: Boolean,
         timeoutMs: Long,
-    ): Pair<ByteArray?, ByteArray?> = withTimeoutOrNull(timeoutMs) {
-        coroutineScope {
-            val video = async(Dispatchers.IO) { sessionStore.fetchInitializationData(holder, holder.videoFormat) }
-            val audio = async(Dispatchers.IO) { sessionStore.fetchInitializationData(holder, holder.audioFormat) }
-            video.await() to audio.await()
-        }
-    } ?: (null to null)
+    ): SabrPlaybackInitializationPreloadResult = withTimeoutOrNull(timeoutMs) {
+        val video = holder.videoFormat
+            .takeUnless { audioOnly }
+            ?.let { sessionStore.fetchInitializationData(holder, it) }
+        val audio = sessionStore.fetchInitializationData(holder, holder.audioFormat)
+        SabrPlaybackInitializationPreloadResult(video, audio)
+    } ?: SabrPlaybackInitializationPreloadResult(null, null)
 }

src/main/kotlin/dev/typetype/server/services/SabrPlaybackSessionService.kt

@@ -35,7 +35,7 @@ internal class SabrPlaybackSessionService(
             audioOnly = audioOnly,
             initialGeneration = initialGeneration,
         )
-        if (isLive || prepared.isLive || prepared.isLiveContent) holder.markExpectedLive()
+        if (isLive || prepared.isLive) holder.markExpectedLive()
         if (holder.expectsLive()) {
             holder.setActiveTracks(videoActive = !audioOnly, audioActive = true)
             holder.session.streamState.setSelectVideoFormatBeforeAudio(!audioOnly)
@@ -45,7 +45,19 @@ internal class SabrPlaybackSessionService(
             }
             sessionStore.ensureWarmed(holder, LIVE_INITIAL_PUMPS)
         } else if (preloadInitialization) {
-            SabrPlaybackInitializationPreloader.preload(sessionStore, holder, INITIALIZATION_PRELOAD_TIMEOUT_MS)
+            val initialization = SabrPlaybackInitializationPreloader.preload(
+                sessionStore,
+                holder,
+                audioOnly,
+                INITIALIZATION_PRELOAD_TIMEOUT_MS,
+            )
+            if (!initialization.isComplete(audioOnly)) {
+                val missing = initialization.missingTracks(audioOnly, video.itag, audio.itag)
+                holder.setActiveTracks(videoActive = !audioOnly, audioActive = true)
+                holder.setPlayerTimeMs(startTimeMs)
+                holder.failTerminal(sabrRecoverableFailureMessage("SABR initialization unavailable for $missing"))
+                return SabrPlaybackPreparation(holder, startTimeMs, ready = false)
+            }
         }
         return SabrPlaybackStarter.start(
             sessionStore,
@@ -73,7 +85,7 @@ internal class SabrPlaybackSessionService(
             video = video,
             startTimeMs = playerTimeMs,
             audioOnly = audioOnly,
-            isLive = source.expectsLive() || prepared.isLive || prepared.isLiveContent,
+            isLive = source.expectsLive() || prepared.isLive,
             initialGeneration = source.nextReplacementGeneration(),
         )
     }

src/test/kotlin/dev/typetype/server/services/SabrLivePlaybackSessionServiceTest.kt

@@ -95,6 +95,71 @@ class SabrLivePlaybackSessionServiceTest {
         verify(exactly = 1) { store.startPump(holder) }
     }
 
+    @Test
+    fun `historical live content prepares as vod`() = runTest {
+        val audio = format(140, isAudio = true)
+        val video = format(137, isAudio = false)
+        val replacementVideo = format(248, isAudio = false)
+        val info = mockk<YoutubeSabrInfo>()
+        val prepared = SabrPreparedInfo(info, token(), isLive = false, isLiveContent = true)
+        val holder = holder(audio, video)
+        val replacement = holder(audio, replacementVideo, initialGeneration = 1L)
+        val state = holder.session.streamState
+        val store = mockk<SabrSessionStore>()
+        every {
+            store.getOrCreate(
+                "video",
+                "user",
+                info,
+                audio,
+                video,
+                prepared.initialToken,
+                0L,
+                false,
+                SabrSessionPurpose.PLAYBACK,
+                false,
+                0L,
+            )
+        } returns holder
+        every {
+            store.getOrCreate(
+                "video",
+                "user",
+                info,
+                audio,
+                replacementVideo,
+                prepared.initialToken,
+                30_000L,
+                false,
+                SabrSessionPurpose.PLAYBACK,
+                false,
+                1L,
+            )
+        } returns replacement
+        coEvery { store.fetchInitializationData(holder, video) } returns byteArrayOf(1)
+        coEvery { store.fetchInitializationData(holder, audio) } returns byteArrayOf(2)
+        coEvery { store.fetchInitializationData(replacement, replacementVideo) } returns byteArrayOf(3)
+        coEvery { store.fetchInitializationData(replacement, audio) } returns byteArrayOf(4)
+        every { store.startPump(holder) } returns Unit
+        every { store.startPump(replacement) } returns Unit
+
+        val service = SabrPlaybackSessionService(store)
+        val result = service.prepare("video", "user", prepared, audio, video, 0L)
+        service.seek(holder, prepared, audio, replacementVideo, 30_000L)
+
+        assertEquals(0L, result.startTimeMs)
+        assertFalse(holder.expectsLive())
+        assertFalse(replacement.expectsLive())
+        verify(exactly = 0) { state.setPlayerTimeMs(9_007_199_254_740_991L) }
+        coVerify(exactly = 0) { store.ensureWarmed(any(), any()) }
+        coVerify(exactly = 1) { store.fetchInitializationData(holder, video) }
+        coVerify(exactly = 1) { store.fetchInitializationData(holder, audio) }
+        coVerify(exactly = 1) { store.fetchInitializationData(replacement, replacementVideo) }
+        coVerify(exactly = 1) { store.fetchInitializationData(replacement, audio) }
+        verify(exactly = 1) { store.startPump(holder) }
+        verify(exactly = 1) { store.startPump(replacement) }
+    }
+
     @Test
     fun `live format change starts the replacement session from warmed track boundaries`() = runTest {
         val audio = format(140, isAudio = true)

src/test/kotlin/dev/typetype/server/services/SabrPlaybackInitializationFailureTest.kt

@@ -0,0 +1,203 @@
+package dev.typetype.server.services
+
+import dev.typetype.server.routes.SabrPlaybackRecovery
+import io.mockk.coEvery
+import io.mockk.coVerify
+import io.mockk.coVerifyOrder
+import io.mockk.every
+import io.mockk.mockk
+import io.mockk.verify
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.test.runTest
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Assertions.assertFalse
+import org.junit.jupiter.api.Test
+import org.schabi.newpipe.extractor.services.youtube.sabr.YoutubeSabrFormat
+import org.schabi.newpipe.extractor.services.youtube.sabr.YoutubeSabrInfo
+import org.schabi.newpipe.extractor.services.youtube.sabr.YoutubeSabrSession
+import org.schabi.newpipe.extractor.services.youtube.sabr.YoutubeSabrStreamState
+import java.time.Instant
+
+@OptIn(ExperimentalCoroutinesApi::class)
+class SabrPlaybackInitializationFailureTest {
+    @Test
+    fun `initialization timeout becomes a recoverable terminal failure`() = runTest {
+        val audio = format(140, isAudio = true)
+        val video = format(271, isAudio = false)
+        val info = mockk<YoutubeSabrInfo>()
+        val prepared = SabrPreparedInfo(info, token())
+        val holder = holder(audio, video)
+        val store = mockk<SabrSessionStore>()
+        every {
+            store.getOrCreate(
+                "video",
+                "user",
+                info,
+                audio,
+                video,
+                prepared.initialToken,
+                475_058L,
+                false,
+                SabrSessionPurpose.PLAYBACK,
+                false,
+                0L,
+            )
+        } returns holder
+        coEvery { store.fetchInitializationData(holder, video) } coAnswers {
+            delay(Long.MAX_VALUE)
+            null
+        }
+        coEvery { store.invalidatePlaybackInfo("video") } returns Unit
+
+        val result = SabrPlaybackSessionService(store).prepare(
+            "video",
+            "user",
+            prepared,
+            audio,
+            video,
+            475_058L,
+        )
+
+        assertFalse(result.ready)
+        assertEquals(6_000L, testScheduler.currentTime)
+        assertEquals(SabrPlaybackState.TERMINAL, holder.playbackState())
+        assertEquals("retry_fresh_session", SabrPlaybackRecovery(store).action(holder))
+        verify(exactly = 0) { store.startPump(any()) }
+        coVerify(exactly = 0) { store.fetchInitializationData(holder, audio) }
+    }
+
+    @Test
+    fun `missing required initialization terminates before starting pump`() = runBlocking {
+        val audio = format(140, isAudio = true)
+        val video = format(271, isAudio = false)
+        val info = mockk<YoutubeSabrInfo>()
+        val prepared = SabrPreparedInfo(info, token())
+        val holder = holder(audio, video)
+        val store = mockk<SabrSessionStore>()
+        every {
+            store.getOrCreate(
+                "video",
+                "user",
+                info,
+                audio,
+                video,
+                prepared.initialToken,
+                475_058L,
+                false,
+                SabrSessionPurpose.PLAYBACK,
+                false,
+                0L,
+            )
+        } returns holder
+        coEvery { store.fetchInitializationData(holder, video) } returns null
+        coEvery { store.fetchInitializationData(holder, audio) } returns byteArrayOf(1)
+        coEvery { store.invalidatePlaybackInfo("video") } returns Unit
+
+        val result = SabrPlaybackSessionService(store).prepare(
+            "video",
+            "user",
+            prepared,
+            audio,
+            video,
+            475_058L,
+        )
+
+        assertFalse(result.ready)
+        assertEquals(SabrPlaybackState.TERMINAL, holder.playbackState())
+        assertEquals(
+            "$SABR_RECOVERABLE_FAILURE_PREFIX SABR initialization unavailable for video:271",
+            holder.terminalFailure(),
+        )
+        assertEquals("retry_fresh_session", SabrPlaybackRecovery(store).action(holder))
+        verify(exactly = 0) { store.startPump(any()) }
+        coVerify(exactly = 1) { store.invalidatePlaybackInfo("video") }
+        coVerifyOrder {
+            store.fetchInitializationData(holder, video)
+            store.fetchInitializationData(holder, audio)
+        }
+    }
+
+    @Test
+    fun `audio only preparation does not require video initialization`() = runBlocking {
+        val audio = format(140, isAudio = true)
+        val video = format(271, isAudio = false)
+        val info = mockk<YoutubeSabrInfo>()
+        val prepared = SabrPreparedInfo(info, token())
+        val holder = holder(audio, video)
+        val store = mockk<SabrSessionStore>()
+        every {
+            store.getOrCreate(
+                "video",
+                "user",
+                info,
+                audio,
+                video,
+                prepared.initialToken,
+                0L,
+                false,
+                SabrSessionPurpose.PLAYBACK,
+                true,
+                0L,
+            )
+        } returns holder
+        coEvery { store.fetchInitializationData(holder, video) } returns null
+        coEvery { store.fetchInitializationData(holder, audio) } returns byteArrayOf(1)
+        every { store.startPump(holder) } returns Unit
+
+        SabrPlaybackSessionService(store).prepare(
+            "video",
+            "user",
+            prepared,
+            audio,
+            video,
+            0L,
+            audioOnly = true,
+        )
+
+        assertEquals(null, holder.terminalFailure())
+        verify(exactly = 1) { store.startPump(holder) }
+        coVerify(exactly = 0) { store.fetchInitializationData(holder, video) }
+        coVerify(exactly = 1) { store.fetchInitializationData(holder, audio) }
+    }
+
+    private fun holder(audio: YoutubeSabrFormat, video: YoutubeSabrFormat): SabrSessionHolder {
+        val session = mockk<YoutubeSabrSession>()
+        val state = mockk<YoutubeSabrStreamState>(relaxed = true)
+        every { session.streamState } returns state
+        every { session.getCachedSegment(any()) } returns null
+        every { session.isBeyondEnd(any()) } returns false
+        every { state.setActiveTrackTypes(any(), any()) } returns Unit
+        every { state.setSelectVideoFormatBeforeAudio(any()) } returns Unit
+        every { state.getMinBufferedEndMs() } returns 0L
+        return SabrSessionHolder(
+            session = session,
+            info = mockk<YoutubeSabrInfo>(),
+            audioFormat = audio,
+            videoFormat = video,
+            sessionToken = "session-token",
+            key = SabrSessionKey("video", "user", audio.itag, null, video.itag, 0L),
+            lastRequestAt = Instant.EPOCH,
+        )
+    }
+
+    private fun format(itag: Int, isAudio: Boolean): YoutubeSabrFormat {
+        val format = mockk<YoutubeSabrFormat>()
+        every { format.itag } returns itag
+        every { format.isAudio } returns isAudio
+        every { format.audioTrackId } returns null
+        every { format.mimeType } returns if (isAudio) "audio/mp4" else "video/mp4"
+        every { format.bitrate } returns if (isAudio) 128_000 else 2_000_000
+        return format
+    }
+
+    private fun token(): SabrTokenBundle = SabrTokenBundle(
+        videoId = "video",
+        visitorBoundPoToken = "visitor-token",
+        visitorBoundPoTokenBytes = byteArrayOf(1),
+        visitorData = "visitor-data",
+        videoBoundPoToken = "video-token",
+        videoBoundPoTokenBytes = byteArrayOf(2),
+    )
+}