From 9db35e092b043456b5617a74de1488ef0c46a98d Mon Sep 17 00:00:00 2001 From: Kingkor Roy Tirtho Date: Fri, 16 Oct 2026 20:06:12 +0600 Subject: [PATCH] feat(server): implement port fallback mechanism for playback proxy server --- .../spotube/core/server/LocalServer.kt | 27 ++++--- .../spotube/core/server/ServerBinder.kt | 66 +++++++++++++++++ .../spotube/core/server/StreamProxy.kt | 34 ++++++++- .../spotube/core/server/ServerBinderTest.kt | 72 +++++++++++++++++++ gradle/libs.versions.toml | 7 +- 5 files changed, 192 insertions(+), 14 deletions(-) create mode 100644 composeApp/src/commonMain/kotlin/org/spotube/spotube/core/server/ServerBinder.kt create mode 100644 composeApp/src/jvmTest/kotlin/org/spotube/spotube/core/server/ServerBinderTest.kt diff --git a/composeApp/src/commonMain/kotlin/org/spotube/spotube/core/server/LocalServer.kt b/composeApp/src/commonMain/kotlin/org/spotube/spotube/core/server/LocalServer.kt index 69c9ae48..d0a700e9 100644 --- a/composeApp/src/commonMain/kotlin/org/spotube/spotube/core/server/LocalServer.kt +++ b/composeApp/src/commonMain/kotlin/org/spotube/spotube/core/server/LocalServer.kt @@ -154,17 +154,26 @@ class LocalServer( logger.d { "Restarting playback proxy server on $host:$port (remoteControl=$allowRemoteControl)" } stopServerLocked() - serverState.value = embeddedServer( - factory = CIO, - host = host, - port = port, - module = { configureRoutes() } - ).also { engine -> - engine.start(wait = false) + val bound = runCatching { + startServerWithPortFallback( + host = host, + preferredPort = port, + onBindFailure = { failedPort, cause -> + logger.w(cause) { "Playback proxy could not bind to $host:$failedPort" } + } + ) { 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 - 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}"}" } } } diff --git a/composeApp/src/commonMain/kotlin/org/spotube/spotube/core/server/ServerBinder.kt b/composeApp/src/commonMain/kotlin/org/spotube/spotube/core/server/ServerBinder.kt new file mode 100644 index 00000000..756cc73e --- /dev/null +++ b/composeApp/src/commonMain/kotlin/org/spotube/spotube/core/server/ServerBinder.kt @@ -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 . + */ + + +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") +} diff --git a/composeApp/src/commonMain/kotlin/org/spotube/spotube/core/server/StreamProxy.kt b/composeApp/src/commonMain/kotlin/org/spotube/spotube/core/server/StreamProxy.kt index ec6ef668..900d830a 100644 --- a/composeApp/src/commonMain/kotlin/org/spotube/spotube/core/server/StreamProxy.kt +++ b/composeApp/src/commonMain/kotlin/org/spotube/spotube/core/server/StreamProxy.kt @@ -58,6 +58,7 @@ import okio.buffer import okio.use import org.koin.core.component.KoinComponent import kotlin.time.Duration.Companion.milliseconds +import kotlin.time.TimeSource internal class StreamProxy( private val httpClient: HttpClient, @@ -92,6 +93,13 @@ internal class StreamProxy( 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()) { val cachedEntry = cacheManager.findCachedEntry(trackId) if (cachedEntry != null) { @@ -152,7 +160,13 @@ internal class StreamProxy( } private suspend fun doStreamFromUpstream(call: ApplicationCall, trackId: String, requestMethod: HttpMethod) { + val resolveStart = TimeSource.Monotonic.markNow() 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) { logger.w { "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 attemptedRefresh = false + val upstreamStart = TimeSource.Monotonic.markNow() // `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 @@ -333,6 +348,11 @@ internal class StreamProxy( call.forwardRequestHeaderIfPresent(headers, HttpHeaders.Accept) call.forwardRequestHeaderIfPresent(headers, HttpHeaders.UserAgent) }.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) { respondWithUpstream(call, response, trackId, requestMethod) return@execute @@ -356,7 +376,7 @@ internal class StreamProxy( if (failure != null) { if (failure is CancellationException) throw failure 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 } 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) 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) { val cacheDir = cacheManager.resolveCacheDir() val cacheFile = cacheDir / cacheFilename.toPath() diff --git a/composeApp/src/jvmTest/kotlin/org/spotube/spotube/core/server/ServerBinderTest.kt b/composeApp/src/jvmTest/kotlin/org/spotube/spotube/core/server/ServerBinderTest.kt new file mode 100644 index 00000000..79e3a4c2 --- /dev/null +++ b/composeApp/src/jvmTest/kotlin/org/spotube/spotube/core/server/ServerBinderTest.kt @@ -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 . + */ + + +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() + 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) + } + } + } +} diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml index 24df3449..4186c354 100644 --- a/gradle/libs.versions.toml +++ b/gradle/libs.versions.toml @@ -55,7 +55,6 @@ vlcjNative = "4.12.0" ziplineVersion = "1.27.0" ktor = "3.5.1" dnssdkt = "1.1.0" -kotlinStdlib = "2.4.10" runner = "1.7.0" core = "1.7.0" 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-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" } -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-core = { group = "androidx.test", name = "core", version.ref = "core" } 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" } zipline-gradle-plugin = { id = "app.cash.zipline", version.ref = "ziplineVersion" } mavenPublish = { id = "maven-publish" } -spotubeGradle = { id = "dev.krtirtho.spotube.gradle-plugin", version.ref = "spotube-gradle" } -vlcjBundler = { id = "dev.krtirtho.vlcj-bundler.gradle-plugin", version.ref = "vlcjBundler" } +spotubeGradle = { id = "org.spotube.gradle-plugin", version.ref = "spotube-gradle" } +vlcjBundler = { id = "org.spotube.vlcj-bundler", version.ref = "vlcjBundler" } kotlinxAtomicFu = { id = "org.jetbrains.kotlinx.atomicfu", version.ref = "kotlinx-atomicfu" } uniffi = { id = "dev.gobley.uniffi", version.ref = "uniffi" } cargo = { id = "dev.gobley.cargo", version.ref = "uniffi" }