feat(server): implement port fallback mechanism for playback proxy server

This commit is contained in:
Kingkor Roy Tirtho 2026-10-16 20:06:12 +06:00
parent 2474c26a83
commit 9db35e092b
5 changed files with 192 additions and 14 deletions

View File

@ -154,17 +154,26 @@ class LocalServer(
logger.d { "Restarting playback proxy server on $host:$port (remoteControl=$allowRemoteControl)" } logger.d { "Restarting playback proxy server on $host:$port (remoteControl=$allowRemoteControl)" }
stopServerLocked() stopServerLocked()
serverState.value = embeddedServer( val bound = runCatching {
factory = CIO, startServerWithPortFallback(
host = host, host = host,
port = port, preferredPort = port,
module = { configureRoutes() } onBindFailure = { failedPort, cause ->
).also { engine -> logger.w(cause) { "Playback proxy could not bind to $host:$failedPort" }
engine.start(wait = false) }
) { configureRoutes() }
}.getOrElse { throwable ->
logger.e(throwable) { "Playback proxy failed to start on $host (requested port $port)" }
return
} }
activePort.value = port
if (bound.usedFallback) {
logger.w { "Port $port is unavailable; playback proxy is using $host:${bound.port} instead" }
}
serverState.value = bound.engine
activePort.value = bound.port
activeHost.value = host activeHost.value = host
logger.i { "Playback proxy server started at ${baseUrl.value ?: "http://$host:$port"}" } logger.i { "Playback proxy server started at ${baseUrl.value ?: "http://$host:${bound.port}"}" }
} }
} }

View File

@ -0,0 +1,66 @@
/*
* Copyright (C) 2026 Kingkor Roy Tirtho and Spotube Contributors
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU Affero General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU Affero General Public License for more details.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
package org.spotube.spotube.core.server
import io.ktor.server.application.Application
import io.ktor.server.cio.CIO
import io.ktor.server.engine.EmbeddedServer
import io.ktor.server.engine.embeddedServer
import kotlinx.coroutines.currentCoroutineContext
import kotlinx.coroutines.ensureActive
import kotlinx.coroutines.withTimeout
import kotlin.time.Duration.Companion.seconds
internal class BoundServer(val engine: EmbeddedServer<*, *>, val port: Int, val usedFallback: Boolean)
/**
* Starts the playback proxy on [preferredPort] and confirms the socket is actually bound.
*
* CIO's `start(wait = false)` returns before binding, so a port that is already taken fails
* later on a worker thread and the server looks started while nothing is listening. If the
* preferred port can't be bound, retry on an OS-assigned free port (port 0).
*
* The caller must read [BoundServer.port] instead of assuming [preferredPort].
*/
internal suspend fun startServerWithPortFallback(
host: String,
preferredPort: Int,
onBindFailure: (port: Int, cause: Throwable) -> Unit = { _, _ -> },
module: Application.() -> Unit
): BoundServer {
val candidates = listOf(preferredPort, 0).distinct()
var lastFailure: Throwable? = null
for (candidate in candidates) {
val engine = embeddedServer(factory = CIO, host = host, port = candidate, module = module)
try {
engine.start(wait = false)
val boundPort = withTimeout(5.seconds) { engine.engine.resolvedConnectors().first().port }
return BoundServer(engine, boundPort, usedFallback = candidate != preferredPort)
} catch (e: Throwable) {
runCatching { engine.stop(gracePeriodMillis = 0, timeoutMillis = 500) }
// A failed bind surfaces as the engine job's CancellationException; only rethrow
// when it is the caller's own coroutine that was cancelled.
currentCoroutineContext().ensureActive()
onBindFailure(candidate, e)
lastFailure = e
}
}
throw lastFailure ?: IllegalStateException("No port candidates to bind")
}

View File

@ -58,6 +58,7 @@ import okio.buffer
import okio.use import okio.use
import org.koin.core.component.KoinComponent import org.koin.core.component.KoinComponent
import kotlin.time.Duration.Companion.milliseconds import kotlin.time.Duration.Companion.milliseconds
import kotlin.time.TimeSource
internal class StreamProxy( internal class StreamProxy(
private val httpClient: HttpClient, private val httpClient: HttpClient,
@ -92,6 +93,13 @@ internal class StreamProxy(
return return
} }
logger.i {
"[Req] $requestMethod /stream/$trackId from=${call.request.local.remoteHost} " +
"version=${call.request.local.version} range=${call.request.headers[HttpHeaders.Range]} " +
"ua=${call.request.headers[HttpHeaders.UserAgent]} " +
"connection=${call.request.headers[HttpHeaders.Connection]} caching=${isCachingEnabled()}"
}
if (isCachingEnabled()) { if (isCachingEnabled()) {
val cachedEntry = cacheManager.findCachedEntry(trackId) val cachedEntry = cacheManager.findCachedEntry(trackId)
if (cachedEntry != null) { if (cachedEntry != null) {
@ -152,7 +160,13 @@ internal class StreamProxy(
} }
private suspend fun doStreamFromUpstream(call: ApplicationCall, trackId: String, requestMethod: HttpMethod) { private suspend fun doStreamFromUpstream(call: ApplicationCall, trackId: String, requestMethod: HttpMethod) {
val resolveStart = TimeSource.Monotonic.markNow()
val streamInfo = streamingUrlRepository.resolveStreamInfo(trackId) val streamInfo = streamingUrlRepository.resolveStreamInfo(trackId)
logger.i {
"[Req] resolved stream for $trackId in ${resolveStart.elapsedNow()}: " +
"protocol=${streamInfo?.protocol} codec=${streamInfo?.codec} container=${streamInfo?.container} " +
"url=${streamInfo?.url?.take(120)}"
}
if (streamInfo == null) { if (streamInfo == null) {
logger.w { "Unable to resolve stream for track $trackId" } logger.w { "Unable to resolve stream for track $trackId" }
call.respond(HttpStatusCode.NotFound, "Unable to resolve stream for track $trackId") call.respond(HttpStatusCode.NotFound, "Unable to resolve stream for track $trackId")
@ -319,6 +333,7 @@ internal class StreamProxy(
var currentUrl = streamUrl var currentUrl = streamUrl
var attemptedRefresh = false var attemptedRefresh = false
val upstreamStart = TimeSource.Monotonic.markNow()
// `request()` buffers the whole body before returning, which stalls the player until // `request()` buffers the whole body before returning, which stalls the player until
// the full track is downloaded. `execute {}` hands over the response as soon as headers // the full track is downloaded. `execute {}` hands over the response as soon as headers
@ -333,6 +348,11 @@ internal class StreamProxy(
call.forwardRequestHeaderIfPresent(headers, HttpHeaders.Accept) call.forwardRequestHeaderIfPresent(headers, HttpHeaders.Accept)
call.forwardRequestHeaderIfPresent(headers, HttpHeaders.UserAgent) call.forwardRequestHeaderIfPresent(headers, HttpHeaders.UserAgent)
}.execute { response -> }.execute { response ->
logger.i {
"[Req] upstream headers for $trackId after ${upstreamStart.elapsedNow()}: " +
"status=${response.status} version=${response.version} " +
"headers=${response.headers.entries().joinToString { "${it.key}=${it.value}" }}"
}
if (response.status.value < 400) { if (response.status.value < 400) {
respondWithUpstream(call, response, trackId, requestMethod) respondWithUpstream(call, response, trackId, requestMethod)
return@execute return@execute
@ -356,7 +376,7 @@ internal class StreamProxy(
if (failure != null) { if (failure != null) {
if (failure is CancellationException) throw failure if (failure is CancellationException) throw failure
if (failure.isClientDisconnect()) { if (failure.isClientDisconnect()) {
logger.d { "Player closed the connection while streaming track $trackId" } logger.i(failure) { "[Req] player closed the connection while streaming track $trackId" }
return return
} }
logger.w(failure) { "Failed to fetch upstream stream for track $trackId" } logger.w(failure) { "Failed to fetch upstream stream for track $trackId" }
@ -432,6 +452,18 @@ internal class StreamProxy(
downstreamHeaders[HttpHeaders.ContentType]?.let(ContentType.Companion::parse) downstreamHeaders[HttpHeaders.ContentType]?.let(ContentType.Companion::parse)
override suspend fun writeTo(channel: io.ktor.utils.io.ByteWriteChannel) { override suspend fun writeTo(channel: io.ktor.utils.io.ByteWriteChannel) {
val writeStart = TimeSource.Monotonic.markNow()
logger.i { "[Req] start writing body for $trackId status=${upstream.status} cache=${cacheFilename != null} expected=$expectedLength" }
try {
writeBody(channel)
logger.i { "[Req] finished writing body for $trackId in ${writeStart.elapsedNow()}" }
} catch (e: Throwable) {
logger.i(e) { "[Req] body write for $trackId aborted after ${writeStart.elapsedNow()}: ${e::class.simpleName}: ${e.message}" }
throw e
}
}
private suspend fun writeBody(channel: io.ktor.utils.io.ByteWriteChannel) {
if (cacheFilename != null) { if (cacheFilename != null) {
val cacheDir = cacheManager.resolveCacheDir() val cacheDir = cacheManager.resolveCacheDir()
val cacheFile = cacheDir / cacheFilename.toPath() val cacheFile = cacheDir / cacheFilename.toPath()

View File

@ -0,0 +1,72 @@
/*
* Copyright (C) 2026 Kingkor Roy Tirtho and Spotube Contributors
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU Affero General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU Affero General Public License for more details.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
package org.spotube.spotube.core.server
import io.ktor.server.response.respondText
import io.ktor.server.routing.get
import io.ktor.server.routing.routing
import kotlinx.coroutines.runBlocking
import java.net.InetAddress
import java.net.ServerSocket
import java.net.URI
import kotlin.test.Test
import kotlin.test.assertEquals
import kotlin.test.assertNotEquals
import kotlin.test.assertTrue
class ServerBinderTest {
private val host = "127.0.0.1"
private fun health(port: Int) =
URI("http://$host:$port/health").toURL().readText()
@Test
fun usesPreferredPortWhenFree() = runBlocking {
val free = ServerSocket(0, 1, InetAddress.getByName(host)).use { it.localPort }
val bound = startServerWithPortFallback(host, free) {
routing { get("/health") { call.respondText("ok") } }
}
try {
assertEquals(free, bound.port)
assertTrue(!bound.usedFallback)
assertEquals("ok", health(bound.port))
} finally {
bound.engine.stop(0, 500)
}
}
@Test
fun fallsBackToAnotherPortWhenPreferredIsTaken() = runBlocking {
ServerSocket(0, 1, InetAddress.getByName(host)).use { blocker ->
val taken = blocker.localPort
val failures = mutableListOf<Int>()
val bound = startServerWithPortFallback(host, taken, { port, _ -> failures += port }) {
routing { get("/health") { call.respondText("ok") } }
}
try {
assertNotEquals(taken, bound.port)
assertTrue(bound.usedFallback)
assertEquals(listOf(taken), failures)
assertEquals("ok", health(bound.port))
} finally {
bound.engine.stop(0, 500)
}
}
}
}

View File

@ -55,7 +55,6 @@ vlcjNative = "4.12.0"
ziplineVersion = "1.27.0" ziplineVersion = "1.27.0"
ktor = "3.5.1" ktor = "3.5.1"
dnssdkt = "1.1.0" dnssdkt = "1.1.0"
kotlinStdlib = "2.4.10"
runner = "1.7.0" runner = "1.7.0"
core = "1.7.0" core = "1.7.0"
filekit = "0.14.2" filekit = "0.14.2"
@ -148,7 +147,7 @@ ktor-client-serialization-kotlinx-json = { module = "io.ktor:ktor-serialization-
kotlinx-coroutines-core = { module = "org.jetbrains.kotlinx:kotlinx-coroutines-core", version.ref = "kotlinx-coroutines" } kotlinx-coroutines-core = { module = "org.jetbrains.kotlinx:kotlinx-coroutines-core", version.ref = "kotlinx-coroutines" }
kotlinx-coroutines-android = { module = "org.jetbrains.kotlinx:kotlinx-coroutines-android", version.ref = "kotlinx-coroutines" } kotlinx-coroutines-android = { module = "org.jetbrains.kotlinx:kotlinx-coroutines-android", version.ref = "kotlinx-coroutines" }
kotlinx-coroutines-test = { module = "org.jetbrains.kotlinx:kotlinx-coroutines-test", version.ref = "kotlinx-coroutines" } kotlinx-coroutines-test = { module = "org.jetbrains.kotlinx:kotlinx-coroutines-test", version.ref = "kotlinx-coroutines" }
kotlin-stdlib = { group = "org.jetbrains.kotlin", name = "kotlin-stdlib", version.ref = "kotlinStdlib" } kotlin-stdlib = { group = "org.jetbrains.kotlin", name = "kotlin-stdlib", version.ref = "kotlin" }
androidx-runner = { group = "androidx.test", name = "runner", version.ref = "runner" } androidx-runner = { group = "androidx.test", name = "runner", version.ref = "runner" }
androidx-core = { group = "androidx.test", name = "core", version.ref = "core" } androidx-core = { group = "androidx.test", name = "core", version.ref = "core" }
filekit-core = { group = "io.github.vinceglb", name = "filekit-core", version.ref = "filekit" } filekit-core = { group = "io.github.vinceglb", name = "filekit-core", version.ref = "filekit" }
@ -186,8 +185,8 @@ jetbrainsKotlinJvm = { id = "org.jetbrains.kotlin.jvm", version.ref = "kotlin" }
androidKotlinMultiplatformLibrary = { id = "com.android.kotlin.multiplatform.library", version.ref = "agp" } androidKotlinMultiplatformLibrary = { id = "com.android.kotlin.multiplatform.library", version.ref = "agp" }
zipline-gradle-plugin = { id = "app.cash.zipline", version.ref = "ziplineVersion" } zipline-gradle-plugin = { id = "app.cash.zipline", version.ref = "ziplineVersion" }
mavenPublish = { id = "maven-publish" } mavenPublish = { id = "maven-publish" }
spotubeGradle = { id = "dev.krtirtho.spotube.gradle-plugin", version.ref = "spotube-gradle" } spotubeGradle = { id = "org.spotube.gradle-plugin", version.ref = "spotube-gradle" }
vlcjBundler = { id = "dev.krtirtho.vlcj-bundler.gradle-plugin", version.ref = "vlcjBundler" } vlcjBundler = { id = "org.spotube.vlcj-bundler", version.ref = "vlcjBundler" }
kotlinxAtomicFu = { id = "org.jetbrains.kotlinx.atomicfu", version.ref = "kotlinx-atomicfu" } kotlinxAtomicFu = { id = "org.jetbrains.kotlinx.atomicfu", version.ref = "kotlinx-atomicfu" }
uniffi = { id = "dev.gobley.uniffi", version.ref = "uniffi" } uniffi = { id = "dev.gobley.uniffi", version.ref = "uniffi" }
cargo = { id = "dev.gobley.cargo", version.ref = "uniffi" } cargo = { id = "dev.gobley.cargo", version.ref = "uniffi" }