diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 3401a487c..449548888 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -180,6 +180,7 @@ jobs: buildScript/lib/naive.sh \ buildScript/lib/olcrtc.sh \ buildScript/lib/olcrtc-src/main.go \ + buildScript/lib/olcrtc-src/main_test.go \ buildScript/lib/olcrtc-src/go.mod \ buildScript/init/env.sh \ buildScript/init/env_ndk.sh \ diff --git a/app/src/main/java/io/nekohasekai/sagernet/bg/GuardedProcessPool.kt b/app/src/main/java/io/nekohasekai/sagernet/bg/GuardedProcessPool.kt index 9994353d0..55aac2211 100644 --- a/app/src/main/java/io/nekohasekai/sagernet/bg/GuardedProcessPool.kt +++ b/app/src/main/java/io/nekohasekai/sagernet/bg/GuardedProcessPool.kt @@ -12,12 +12,23 @@ import io.nekohasekai.sagernet.utils.Commandline import kotlinx.coroutines.* import kotlinx.coroutines.channels.Channel import kotlinx.coroutines.channels.trySendBlocking +import kotlinx.coroutines.selects.select import libcore.Libcore import java.io.File import java.io.IOException import java.io.InputStream import kotlin.concurrent.thread +private data class ProcessGenerationExit( + val exitCode: Int, + val readyAtMillis: Long? = null, +) + +private data class RestartReadinessResult( + val readyAtMillis: Long? = null, + val error: IOException? = null, +) + class GuardedProcessPool(private val onFatal: suspend (IOException) -> Unit) : CoroutineScope { companion object { private val pid by lazy { @@ -43,51 +54,148 @@ class GuardedProcessPool(private val onFatal: suspend (IOException) -> Unit) : C }.start() } + private fun watchProcess(cmdName: String, exitChannel: Channel) { + val proc = process + thread(name = "stderr-$cmdName") { + streamLogger(proc.errorStream) { + Libcore.nekoLogPrintln("[$cmdName] ${Commandline.redactProcessOutput(it)}") + } + } + thread(name = "stdout-$cmdName") { + streamLogger(proc.inputStream) { + Libcore.nekoLogPrintln("[$cmdName] ${Commandline.redactProcessOutput(it)}") + } + } + // The channel is generation-local and buffered, so this waiter never blocks a + // later generation and remains available to bounded NonCancellable teardown. + thread(name = "waitFor-$cmdName") { + val code = proc.waitFor() + if (exitChannel.trySendBlocking(code).isFailure) { + Logs.w("$cmdName: could not deliver exit code $code (channel closed)") + } + } + } + + private suspend fun observeRestart( + cmdName: String, + exitChannel: Channel, + onRestartCallback: suspend () -> Unit, + ): ProcessGenerationExit = coroutineScope { + val readiness = async { + try { + onRestartCallback() + RestartReadinessResult(readyAtMillis = SystemClock.elapsedRealtime()) + } catch (e: CancellationException) { + throw e + } catch (e: Exception) { + RestartReadinessResult( + error = if (e is IOException) e else IOException("restart readiness check failed", e), + ) + } + } + select { + exitChannel.onReceive { exitCode -> + readiness.cancelAndJoin() + ProcessGenerationExit(exitCode) + } + readiness.onAwait { result -> + val readinessError = result.error + if (readinessError == null) { + ProcessGenerationExit( + exitCode = exitChannel.receive(), + readyAtMillis = result.readyAtMillis, + ) + } else { + Logs.w("$cmdName restart readiness failed; restarting") + val exitCode = terminateProcess(exitChannel) + ?: throw IOException( + "$cmdName could not stop after restart readiness failure", + readinessError, + ) + ProcessGenerationExit(exitCode) + } + } + } + } + + private suspend fun terminateProcess(exitChannel: Channel): Int? = withContext(NonCancellable) { + exitChannel.tryReceive().getOrNull()?.let { return@withContext it } + if (Build.VERSION.SDK_INT < 24) { + try { + Os.kill(pid.get(process) as Int, OsConstants.SIGTERM) + } catch (e: ErrnoException) { + if (e.errno != OsConstants.ESRCH) Logs.w(e) + } catch (e: ReflectiveOperationException) { + Logs.w(e) + } + withTimeoutOrNull(500) { exitChannel.receive() }?.let { return@withContext it } + } + process.destroy() + if (Build.VERSION.SDK_INT >= 26) { + withTimeoutOrNull(1000) { exitChannel.receive() }?.let { return@withContext it } + process.destroyForcibly() + } + withTimeoutOrNull(1000) { exitChannel.receive() } + } + @DelicateCoroutinesApi - suspend fun looper(onRestartCallback: (suspend () -> Unit)?) { + suspend fun looper( + onRestartPrepare: (() -> Unit)?, + onRestartCallback: (suspend () -> Unit)?, + restartPolicy: GuardedProcessRestartPolicy?, + ) { var running = true + var restarted = false + var currentExitChannel: Channel? = null val cmdName = File(cmd.first()).nameWithoutExtension - val exitChannel = Channel() + val backoff = restartPolicy.createBackoff() try { while (true) { - thread(name = "stderr-$cmdName") { - streamLogger(process.errorStream) { - Libcore.nekoLogPrintln("[$cmdName] ${Commandline.redactProcessOutput(it)}") - } - } - thread(name = "stdout-$cmdName") { - streamLogger(process.inputStream) { - Libcore.nekoLogPrintln("[$cmdName] ${Commandline.redactProcessOutput(it)}") - } - } - // Dedicated waiter thread (lifecycle independent of the pool's Job) so the - // NonCancellable teardown below can still drain the exit code even after the - // pool is cancelled. Use trySendBlocking instead of runBlocking { send } to - // avoid spinning up a coroutine dispatcher on this raw thread. - val proc = process - thread(name = "waitFor-$cmdName") { - val code = proc.waitFor() - // If the channel is already closed/failed, log rather than silently drop - // (the NonCancellable teardown below also bounds its receive()). - if (exitChannel.trySendBlocking(code).isFailure) { - Logs.w("$cmdName: could not deliver exit code $code (channel closed)") - } - } + val exitChannel = Channel(capacity = 1) + currentExitChannel = exitChannel + watchProcess(cmdName, exitChannel) val startTime = SystemClock.elapsedRealtime() - val exitCode = exitChannel.receive() + val generation = if (restarted && onRestartCallback != null) { + observeRestart(cmdName, exitChannel, onRestartCallback) + } else { + ProcessGenerationExit(exitChannel.receive()) + } running = false - when { - SystemClock.elapsedRealtime() - startTime < 1000 -> throw IOException( - "$cmdName exits too fast (exit code: $exitCode)", + currentExitChannel = null + exitChannel.close() + + val exitTime = SystemClock.elapsedRealtime() + val processUptimeMillis = exitTime - startTime + if (restartPolicy == null && processUptimeMillis < 1000L) { + throw IOException("$cmdName exits too fast (exit code: ${generation.exitCode})") + } + when (generation.exitCode) { + 128 + OsConstants.SIGKILL -> Logs.w("$cmdName was killed") + else -> Logs.w( + IOException("$cmdName unexpectedly exits with code ${generation.exitCode}"), ) + } - exitCode == 128 + OsConstants.SIGKILL -> Logs.w("$cmdName was killed") - else -> Logs.w(IOException("$cmdName unexpectedly exits with code $exitCode")) + val readyDurationMillis = generation.readyAtMillis?.let { + (exitTime - it).coerceAtLeast(0L) + } + val restartDelayMillis = backoff?.delayAfterExit(readyDurationMillis) + onRestartPrepare?.invoke() + if (restartDelayMillis != null) { + Logs.i( + "restart process after ${restartDelayMillis}ms: " + + Commandline.toRedactedString(cmd), + ) + delay(restartDelayMillis) + } else { + Logs.i( + "restart process: ${Commandline.toRedactedString(cmd)} " + + "(last exit code: ${generation.exitCode})", + ) } - Logs.i("restart process: ${Commandline.toRedactedString(cmd)} (last exit code: $exitCode)") start() running = true - onRestartCallback?.invoke() + restarted = true } } catch (e: IOException) { Logs.w("error occurred. stop guard: ${Commandline.toRedactedString(cmd)}") @@ -95,26 +203,11 @@ class GuardedProcessPool(private val onFatal: suspend (IOException) -> Unit) : C // and stop a freshly-restarted instance. this@GuardedProcessPool.launch(Dispatchers.Main.immediate) { onFatal(e) } } finally { - if (running) { - withContext(NonCancellable) { // clean-up cannot be cancelled - if (Build.VERSION.SDK_INT < 24) { - try { - Os.kill(pid.get(process) as Int, OsConstants.SIGTERM) - } catch (e: ErrnoException) { - if (e.errno != OsConstants.ESRCH) Logs.w(e) - } catch (e: ReflectiveOperationException) { - Logs.w(e) - } - if (withTimeoutOrNull(500) { exitChannel.receive() } != null) return@withContext - } - process.destroy() // kill the process - if (Build.VERSION.SDK_INT >= 26) { - if (withTimeoutOrNull(1000) { exitChannel.receive() } != null) return@withContext - process.destroyForcibly() // Force to kill the process if it's still alive - } - // Bounded so a missed exit-code send (closed channel) can't hang teardown. - withTimeoutOrNull(1000) { exitChannel.receive() } - } // otherwise process already exited, nothing to be done + val exitChannel = currentExitChannel + if (running && exitChannel != null) { + terminateProcess(exitChannel) + } else if (running) { + process.destroy() } } } @@ -127,12 +220,14 @@ class GuardedProcessPool(private val onFatal: suspend (IOException) -> Unit) : C fun start( cmd: List, env: MutableMap = mutableMapOf(), + onRestartPrepare: (() -> Unit)? = null, onRestartCallback: (suspend () -> Unit)? = null, + restartPolicy: GuardedProcessRestartPolicy? = null, ) { Logs.i("start process: ${Commandline.toRedactedString(cmd)}") Guard(cmd, env).apply { start() // if start fails, IOException will be thrown directly - launch { looper(onRestartCallback) } + launch { looper(onRestartPrepare, onRestartCallback, restartPolicy) } } processCount += 1 } diff --git a/app/src/main/java/io/nekohasekai/sagernet/bg/GuardedProcessRestartPolicy.kt b/app/src/main/java/io/nekohasekai/sagernet/bg/GuardedProcessRestartPolicy.kt new file mode 100644 index 000000000..cba44092a --- /dev/null +++ b/app/src/main/java/io/nekohasekai/sagernet/bg/GuardedProcessRestartPolicy.kt @@ -0,0 +1,36 @@ +package io.nekohasekai.sagernet.bg + +data class GuardedProcessRestartPolicy( + val initialDelayMillis: Long = 1_000L, + val maximumDelayMillis: Long = 30_000L, + val stableAfterReadyMillis: Long = 60_000L, +) { + init { + require(initialDelayMillis > 0L) { "initial restart delay must be positive" } + require(maximumDelayMillis >= initialDelayMillis) { + "maximum restart delay must not be smaller than the initial delay" + } + require(stableAfterReadyMillis > 0L) { "stable readiness duration must be positive" } + } +} + +internal class GuardedProcessRestartBackoff( + private val policy: GuardedProcessRestartPolicy, +) { + private var nextDelayMillis = policy.initialDelayMillis + + fun delayAfterExit(readyDurationMillis: Long?): Long { + if (readyDurationMillis != null && readyDurationMillis >= policy.stableAfterReadyMillis) { + nextDelayMillis = policy.initialDelayMillis + } + val delayMillis = nextDelayMillis + nextDelayMillis = if (nextDelayMillis >= policy.maximumDelayMillis - nextDelayMillis) { + policy.maximumDelayMillis + } else { + nextDelayMillis * 2L + } + return delayMillis + } +} + +internal fun GuardedProcessRestartPolicy?.createBackoff() = this?.let(::GuardedProcessRestartBackoff) diff --git a/app/src/main/java/io/nekohasekai/sagernet/bg/proto/BoxInstance.kt b/app/src/main/java/io/nekohasekai/sagernet/bg/proto/BoxInstance.kt index ad7c32f3c..ff8ed376b 100644 --- a/app/src/main/java/io/nekohasekai/sagernet/bg/proto/BoxInstance.kt +++ b/app/src/main/java/io/nekohasekai/sagernet/bg/proto/BoxInstance.kt @@ -4,6 +4,7 @@ import android.os.SystemClock import io.nekohasekai.sagernet.SagerNet import io.nekohasekai.sagernet.bg.AbstractInstance import io.nekohasekai.sagernet.bg.GuardedProcessPool +import io.nekohasekai.sagernet.bg.GuardedProcessRestartPolicy import io.nekohasekai.sagernet.database.DataStore import io.nekohasekai.sagernet.database.ProxyEntity import io.nekohasekai.sagernet.fmt.ConfigBuildResult @@ -44,7 +45,15 @@ abstract class BoxInstance( val pluginConfigs = hashMapOf>() val externalInstances = hashMapOf() open lateinit var processes: GuardedProcessPool + protected open val enableOlcrtcRecovery = true + private val olcrtcReadyMarkers = hashMapOf() private var cacheFiles = ArrayList() + + private fun olcrtcReadyTimeoutMillis() = maxOf( + 60_000L, + DataStore.connectionTestTimeout.toLong(), + ) + fun isInitialized(): Boolean { return ::config.isInitialized && ::box.isInitialized } @@ -115,7 +124,9 @@ abstract class BoxInstance( // rather than start an unauthenticated listener if they are missing. val creds = config.localProxyCredentials[port] ?: error("olcRTC: missing loopback SOCKS credentials for port $port") - val readyTimeoutMs = maxOf(60_000L, DataStore.connectionTestTimeout.toLong()) + val readyTimeoutMs = olcrtcReadyTimeoutMillis() + val readyMarker = File(app.noBackupFilesDir, "olcrtc_ready_$port") + olcrtcReadyMarkers[port] = readyMarker val args = bean.buildOlcrtcArgs( port, File(app.noBackupFilesDir, "protect_path").absolutePath, @@ -124,7 +135,7 @@ abstract class BoxInstance( DataStore.logLevel >= 3, "9.9.9.9:53", readyTimeoutMs, - ) + ) + listOf("-ready-marker", readyMarker.absolutePath) pluginConfigs[port] = profile.type to args.joinToString("\u0000") } } @@ -293,7 +304,24 @@ abstract class BoxInstance( // signal-based preemption can fault during the first protected dial // in the VpnService process context. val env = mutableMapOf("GODEBUG" to "asyncpreemptoff=1") - processes.start(commands, env) + val readyMarker = olcrtcReadyMarkers.getValue(port) + clearOlcrtcReadyMarker(readyMarker) + if (enableOlcrtcRecovery) { + processes.start( + commands, + env, + onRestartPrepare = { clearOlcrtcReadyMarker(readyMarker) }, + onRestartCallback = { + awaitExternalPortReady( + port, + olcrtcReadyTimeoutMillis() + 5_000L, + ) + }, + restartPolicy = GuardedProcessRestartPolicy(), + ) + } else { + processes.start(commands, env) + } } } } @@ -302,15 +330,52 @@ abstract class BoxInstance( box.start() } + private fun clearOlcrtcReadyMarker(marker: File) { + if (marker.exists() && !marker.delete() && marker.exists()) { + throw IOException("olcRTC: could not reset readiness marker") + } + } + + private suspend fun pendingExternalPorts(ports: Collection, timeoutMillis: Long) = + withContext(Dispatchers.IO) { + val deadline = SystemClock.elapsedRealtime() + timeoutMillis + val pending = ports.toMutableSet() + while (pending.isNotEmpty() && SystemClock.elapsedRealtime() < deadline) { + ensureActive() + val iterator = pending.iterator() + while (iterator.hasNext()) { + val port = iterator.next() + val readyMarker = olcrtcReadyMarkers[port] + if (!readinessMarkerSatisfied(readyMarker != null, readyMarker?.isFile == true)) continue + try { + Socket().use { + it.connect(InetSocketAddress(LOCALHOST, port), 100) + } + iterator.remove() + } catch (_: IOException) { + // not ready yet + } + } + if (pending.isNotEmpty()) delay(50) + } + pending + } + + private suspend fun awaitExternalPortReady(port: Int, timeoutMillis: Long) { + if (pendingExternalPorts(listOf(port), timeoutMillis).isNotEmpty()) { + throw IOException("sidecar listener not ready on port: $port") + } + } + /** * Waits until every external sidecar's local SOCKS listener is accepting connections * before the service reports Connected, so the sing-box socks outbound (and any * connection test) doesn't race a sidecar that hasn't bound its port yet. * * Most sidecars open their listener immediately, so the short connection-test timeout - * is sufficient. MasterDnsVPN is the exception: it only starts listening after DNS - * MTU probing and session setup, which can take tens of seconds (with retries) on - * lossy or restricted links, so it gets a longer readiness window. + * is sufficient. MasterDnsVPN and olcRTC are exceptions: they only start listening + * after carrier setup, which can take tens of seconds (with retries) on lossy or + * restricted links, so they get a longer readiness window. * * @param strict when true (URL test), a sidecar that never binds is a hard failure with * a clear message, instead of the live-service behavior of logging and continuing @@ -348,52 +413,37 @@ abstract class BoxInstance( maxOf(1_000L, DataStore.connectionTestTimeout.toLong()) } - withContext(Dispatchers.IO) { - val deadline = SystemClock.elapsedRealtime() + readinessTimeoutMs - val pending = ports.toMutableSet() - while (pending.isNotEmpty() && SystemClock.elapsedRealtime() < deadline) { - // Honor cancellation promptly: if this start was superseded (reload/profile - // switch), the connect job is cancelled and the sidecars are torn down. Exiting - // here stops us from polling a now-dead port for the full (60s for MasterDnsVPN) - // window and then throwing a false "sidecar listener not ready". - ensureActive() - val iterator = pending.iterator() - while (iterator.hasNext()) { - val port = iterator.next() - try { - Socket().use { - it.connect(InetSocketAddress(LOCALHOST, port), 100) - } - iterator.remove() - } catch (_: IOException) { - // not ready yet - } - } - if (pending.isNotEmpty()) delay(50) + val pending = pendingExternalPorts(ports, readinessTimeoutMs) + if (pending.isNotEmpty()) { + // If the process pool is no longer active, its sidecars were torn down (e.g. a + // superseded start during reload). A port that never bound on a dead pool is an + // orphan, not a real failure - drop it instead of throwing. + if (!processes.isActive) { + Logs.w( + "sidecar listener not ready on port(s): ${pending.joinToString()}; " + + "process pool already stopped (superseded start), ignoring", + ) + return } - if (pending.isNotEmpty()) { - // If the process pool is no longer active, its sidecars were torn down (e.g. a - // superseded start during reload). A port that never bound on a dead pool is an - // orphan, not a real failure - drop it instead of throwing. - if (!processes.isActive) { - Logs.w( - "sidecar listener not ready on port(s): ${pending.joinToString()}; " + - "process pool already stopped (superseded start), ignoring", - ) - return@withContext - } - // MasterDnsVPN must have its listener up before the first dial (it crashed - // otherwise), so a timeout there is fatal. Other sidecars (Mieru/Naïve/ - // TrojanGo/Hysteria) were historically fire-and-forget: the first sing-box - // dial retries, so a slow bind shouldn't hard-fail VPN start - log and continue. - // For a URL test (strict), there is no retry window, so a listener that never - // binds is reported as a clear error instead of a flaky "connection refused". - val message = "sidecar listener not ready on port(s): ${pending.joinToString()}" - if (hasMasterDnsVpn || hasOlcrtc || strict) { - throw IOException(message) - } else { - Logs.w("$message; continuing (sing-box will retry the connection)") + // MasterDnsVPN and olcRTC must have their listeners up before the first dial, + // so a timeout on either is fatal. Other sidecars (Mieru/Naïve/ + // TrojanGo/Hysteria) were historically fire-and-forget: the first sing-box + // dial retries, so a slow bind shouldn't hard-fail VPN start - log and continue. + // For a URL test (strict), there is no retry window, so a listener that never + // binds is reported as a clear error instead of a flaky "connection refused". + val message = "sidecar listener not ready on port(s): ${pending.joinToString()}" + val requiredPorts = config.externalIndex.flatMap { idx -> + idx.chain.mapNotNull { (port, profile) -> + when (profile.requireBean()) { + is MasterDnsVpnBean, is OlcrtcBean -> port + else -> null + } } + }.toSet() + if (shouldFailSidecarReadiness(pending, requiredPorts, strict)) { + throw IOException(message) + } else { + Logs.w("$message; continuing (sing-box will retry the connection)") } } } @@ -412,6 +462,7 @@ abstract class BoxInstance( } if (::processes.isInitialized) processes.close(GlobalScope + Dispatchers.IO) + olcrtcReadyMarkers.values.forEach { it.delete() } if (::box.isInitialized) { box.close() diff --git a/app/src/main/java/io/nekohasekai/sagernet/bg/proto/SidecarReadinessPolicy.kt b/app/src/main/java/io/nekohasekai/sagernet/bg/proto/SidecarReadinessPolicy.kt new file mode 100644 index 000000000..d712e956f --- /dev/null +++ b/app/src/main/java/io/nekohasekai/sagernet/bg/proto/SidecarReadinessPolicy.kt @@ -0,0 +1,7 @@ +package io.nekohasekai.sagernet.bg.proto + +internal fun readinessMarkerSatisfied(markerRequired: Boolean, markerPresent: Boolean) = + !markerRequired || markerPresent + +internal fun shouldFailSidecarReadiness(pendingPorts: Set, requiredPorts: Set, strict: Boolean) = + pendingPorts.isNotEmpty() && (strict || pendingPorts.any { it in requiredPorts }) diff --git a/app/src/main/java/io/nekohasekai/sagernet/bg/proto/TestInstance.kt b/app/src/main/java/io/nekohasekai/sagernet/bg/proto/TestInstance.kt index e4dced4fe..2a7514e21 100644 --- a/app/src/main/java/io/nekohasekai/sagernet/bg/proto/TestInstance.kt +++ b/app/src/main/java/io/nekohasekai/sagernet/bg/proto/TestInstance.kt @@ -16,6 +16,8 @@ import kotlin.coroutines.resumeWithException class TestInstance(profile: ProxyEntity, val link: String, private val timeout: Int) : BoxInstance(profile) { + protected override val enableOlcrtcRecovery = false + // close() can be reached from two paths that may overlap on cancellation: the // suspendCancellableCoroutine's invokeOnCancellation and the `use { }` block's // exit. BoxInstance.close() is not safe to run twice (native box.close()), so diff --git a/app/src/test/java/io/nekohasekai/sagernet/bg/GuardedProcessRestartPolicyTest.kt b/app/src/test/java/io/nekohasekai/sagernet/bg/GuardedProcessRestartPolicyTest.kt new file mode 100644 index 000000000..809617a59 --- /dev/null +++ b/app/src/test/java/io/nekohasekai/sagernet/bg/GuardedProcessRestartPolicyTest.kt @@ -0,0 +1,62 @@ +package io.nekohasekai.sagernet.bg + +import org.junit.Assert.assertEquals +import org.junit.Assert.assertNull +import org.junit.Assert.assertThrows +import org.junit.Test + +class GuardedProcessRestartPolicyTest { + + @Test + fun delayAfterExit_progressesAndCaps() { + val backoff = GuardedProcessRestartPolicy().createBackoff()!! + + assertEquals( + listOf(1_000L, 2_000L, 4_000L, 8_000L, 16_000L, 30_000L, 30_000L), + List(7) { backoff.delayAfterExit(readyDurationMillis = null) }, + ) + } + + @Test + fun delayAfterExit_resetsOnlyAfterStableReadyDuration() { + val backoff = GuardedProcessRestartPolicy().createBackoff()!! + + assertEquals(1_000L, backoff.delayAfterExit(readyDurationMillis = null)) + assertEquals(2_000L, backoff.delayAfterExit(readyDurationMillis = 59_999L)) + assertEquals(1_000L, backoff.delayAfterExit(readyDurationMillis = 60_000L)) + assertEquals(2_000L, backoff.delayAfterExit(readyDurationMillis = null)) + } + + @Test + fun delayAfterExit_capsWithoutOverflow() { + val policy = GuardedProcessRestartPolicy( + initialDelayMillis = Long.MAX_VALUE - 1L, + maximumDelayMillis = Long.MAX_VALUE, + stableAfterReadyMillis = 1L, + ) + val backoff = policy.createBackoff()!! + + assertEquals(Long.MAX_VALUE - 1L, backoff.delayAfterExit(readyDurationMillis = null)) + assertEquals(Long.MAX_VALUE, backoff.delayAfterExit(readyDurationMillis = null)) + } + + @Test + fun policy_rejectsInvalidBounds() { + assertThrows(IllegalArgumentException::class.java) { + GuardedProcessRestartPolicy(initialDelayMillis = 0L) + } + assertThrows(IllegalArgumentException::class.java) { + GuardedProcessRestartPolicy(initialDelayMillis = 2L, maximumDelayMillis = 1L) + } + assertThrows(IllegalArgumentException::class.java) { + GuardedProcessRestartPolicy(stableAfterReadyMillis = 0L) + } + } + + @Test + fun absentPolicy_hasNoBackoff() { + val policy: GuardedProcessRestartPolicy? = null + + assertNull(policy.createBackoff()) + } +} diff --git a/app/src/test/java/io/nekohasekai/sagernet/bg/proto/SidecarReadinessPolicyTest.kt b/app/src/test/java/io/nekohasekai/sagernet/bg/proto/SidecarReadinessPolicyTest.kt new file mode 100644 index 000000000..46b7a69c1 --- /dev/null +++ b/app/src/test/java/io/nekohasekai/sagernet/bg/proto/SidecarReadinessPolicyTest.kt @@ -0,0 +1,55 @@ +package io.nekohasekai.sagernet.bg.proto + +import org.junit.Assert.assertFalse +import org.junit.Assert.assertTrue +import org.junit.Test + +class SidecarReadinessPolicyTest { + + @Test + fun markerIsRequiredOnlyForMarkedSidecars() { + assertTrue(readinessMarkerSatisfied(markerRequired = false, markerPresent = false)) + assertTrue(readinessMarkerSatisfied(markerRequired = true, markerPresent = true)) + assertFalse(readinessMarkerSatisfied(markerRequired = true, markerPresent = false)) + } + + @Test + fun unrelatedPendingPortDoesNotMakeMixedChainFatal() { + assertFalse( + shouldFailSidecarReadiness( + pendingPorts = setOf(20001), + requiredPorts = setOf(20002), + strict = false, + ), + ) + } + + @Test + fun strictModeWithNoPendingPortIsNotFatal() { + assertFalse( + shouldFailSidecarReadiness( + pendingPorts = emptySet(), + requiredPorts = emptySet(), + strict = true, + ), + ) + } + + @Test + fun requiredOrStrictPendingPortIsFatal() { + assertTrue( + shouldFailSidecarReadiness( + pendingPorts = setOf(20001, 20002), + requiredPorts = setOf(20002), + strict = false, + ), + ) + assertTrue( + shouldFailSidecarReadiness( + pendingPorts = setOf(20001), + requiredPorts = emptySet(), + strict = true, + ), + ) + } +} diff --git a/buildScript/lib/olcrtc-src/main.go b/buildScript/lib/olcrtc-src/main.go index 81f68cb4d..d68dcc732 100644 --- a/buildScript/lib/olcrtc-src/main.go +++ b/buildScript/lib/olcrtc-src/main.go @@ -34,6 +34,7 @@ import ( "net/netip" "os" "os/signal" + "path/filepath" "syscall" "time" @@ -56,11 +57,13 @@ func main() { vp8FPS = flag.Int("vp8-fps", 30, "vp8 fps") vp8Batch = flag.Int("vp8-batch", 8, "vp8 batch size") protectPath = flag.String("protect-path", "", "path to libcore protect unix socket") + readyMarker = flag.String("ready-marker", "", "app-private readiness marker path") readyMillis = flag.Int("ready-timeout-ms", 60000, "readiness wait in ms") debug = flag.Bool("debug", false, "verbose logging") ) flag.Parse() + removeReadyMarker(*readyMarker) mobile.SetDebug(*debug) if !*debug { // Quiet by default so room ids / carrier urls are not written to logs. @@ -91,13 +94,69 @@ func main() { mobile.Stop() log.Fatalf("olcrtc wait ready: %v", err) } + if err := publishReadyMarker(*readyMarker); err != nil { + mobile.Stop() + log.Fatalf("olcrtc ready marker: %v", err) + } + defer removeReadyMarker(*readyMarker) sig := make(chan os.Signal, 1) signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM) - <-sig + ticker := time.NewTicker(15 * time.Second) + defer ticker.Stop() + defer signal.Stop(sig) + + if !waitAfterReady(sig, ticker.C, mobile.IsRunning) { + mobile.Stop() + log.Fatal("olcrtc runtime stopped after readiness") + } mobile.Stop() } +// waitAfterReady returns true for a requested signal shutdown and false when the +// already-ready mobile runtime reaches its terminal stopped state. +func waitAfterReady(signals <-chan os.Signal, ticks <-chan time.Time, isRunning func() bool) bool { + for { + select { + case <-signals: + return true + case <-ticks: + if !isRunning() { + return false + } + } + } +} + +// publishReadyMarker atomically proves that this app-owned wrapper, rather than +// an unrelated loopback listener, reached mobile readiness. The parent keeps the +// marker in its private no-backup directory and clears it before each generation. +func publishReadyMarker(path string) error { + if path == "" { + return nil + } + temp, err := os.CreateTemp(filepath.Dir(path), ".olcrtc-ready-*") + if err != nil { + return err + } + tempPath := temp.Name() + defer os.Remove(tempPath) + if _, err := temp.WriteString("ready\n"); err != nil { + temp.Close() + return err + } + if err := temp.Close(); err != nil { + return err + } + return os.Rename(tempPath, path) +} + +func removeReadyMarker(path string) { + if path != "" { + _ = os.Remove(path) + } +} + // installProtectedDefaults replaces net.DefaultResolver and http.DefaultTransport // so that (1) hostname lookups use dnsServer over a protected UDP/TCP socket // instead of Android's VPN fake-IP resolver, and (2) every TCP socket dialed by diff --git a/buildScript/lib/olcrtc-src/main_test.go b/buildScript/lib/olcrtc-src/main_test.go new file mode 100644 index 000000000..3505ace0b --- /dev/null +++ b/buildScript/lib/olcrtc-src/main_test.go @@ -0,0 +1,80 @@ +package main + +import ( + "os" + "path/filepath" + "syscall" + "testing" + "time" +) + +func TestWaitAfterReadySignalIsGraceful(t *testing.T) { + signals := make(chan os.Signal, 1) + signals <- syscall.SIGTERM + + if !waitAfterReady(signals, make(chan time.Time), func() bool { return true }) { + t.Fatal("signal shutdown was reported as runtime failure") + } +} + +func TestWaitAfterReadyStopsOnFirstUnhealthyTick(t *testing.T) { + ticks := make(chan time.Time, 1) + ticks <- time.Now() + + if waitAfterReady(make(chan os.Signal), ticks, func() bool { return false }) { + t.Fatal("stopped runtime was reported as graceful shutdown") + } +} + +func TestPublishReadyMarkerReplacesStaleMarker(t *testing.T) { + path := filepath.Join(t.TempDir(), "ready") + if err := os.WriteFile(path, []byte("stale\n"), 0o600); err != nil { + t.Fatal(err) + } + + if err := publishReadyMarker(path); err != nil { + t.Fatal(err) + } + content, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + if string(content) != "ready\n" { + t.Fatalf("marker content = %q, want ready", content) + } + info, err := os.Stat(path) + if err != nil { + t.Fatal(err) + } + if mode := info.Mode().Perm(); mode != 0o600 { + t.Fatalf("marker mode = %o, want 600", mode) + } +} + +func TestPublishReadyMarkerAllowsDisabledMarker(t *testing.T) { + if err := publishReadyMarker(""); err != nil { + t.Fatal(err) + } +} + +func TestWaitAfterReadyHealthyTicksContinue(t *testing.T) { + signals := make(chan os.Signal) + ticks := make(chan time.Time) + result := make(chan bool, 1) + checks := 0 + go func() { + result <- waitAfterReady(signals, ticks, func() bool { + checks++ + return true + }) + }() + + ticks <- time.Now() + signals <- syscall.SIGTERM + if graceful := <-result; !graceful { + t.Fatal("healthy runtime tick stopped the wait loop") + } + if checks != 1 { + t.Fatalf("running checks = %d, want 1", checks) + } +} diff --git a/buildScript/lib/olcrtc.sh b/buildScript/lib/olcrtc.sh index c939a94c1..24ab5af0c 100755 --- a/buildScript/lib/olcrtc.sh +++ b/buildScript/lib/olcrtc.sh @@ -83,8 +83,9 @@ fi BUILD="$(pwd)/.olcrtc-wrapper" rm -rf "$BUILD" mkdir -p "$BUILD" -cp "$SRC/main.go" "$SRC/go.mod" "$BUILD/" +cp "$SRC/main.go" "$SRC/main_test.go" "$SRC/go.mod" "$BUILD/" ( cd "$BUILD" && go mod edit -replace "github.com/openlibrecommunity/olcrtc=$WORK" && go mod tidy ) +( cd "$BUILD" && go test . ) build_abi() { local abi="$1" goarch="$2" cc="$3" goarm="$4"