Skip to content

fix(olcrtc): supervise sidecar recovery #156

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
hawkff wants to merge 4 commits into from
Closed

fix(olcrtc): supervise sidecar recovery #156

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
83d9ce8
fix(olcrtc): supervise sidecar recovery
hawkff Jul 13, 2026
4b13ea3
fix(olcrtc): harden restart readiness
hawkff Jul 13, 2026
10e3b1b
style: apply sidecar readiness formatting
hawkff Jul 13, 2026
c3cfc66
fix: handle empty sidecar readiness sets
hawkff Jul 13, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions .github/workflows/ci.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 \
Expand Down
203 changes: 149 additions & 54 deletions app/src/main/java/io/nekohasekai/sagernet/bg/GuardedProcessPool.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand All @@ -43,78 +54,160 @@ class GuardedProcessPool(private val onFatal: suspend (IOException) -> Unit) : C
}.start()
}

private fun watchProcess(cmdName: String, exitChannel: Channel<Int>) {
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<Int>,
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>): 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<Int>? = null
val cmdName = File(cmd.first()).nameWithoutExtension
val exitChannel = Channel<Int>()
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<Int>(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)}")
// Structured (cancelled with the pool) so a torn-down pool can't fire onFatal
// 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()
}
}
}
Expand All @@ -127,12 +220,14 @@ class GuardedProcessPool(private val onFatal: suspend (IOException) -> Unit) : C
fun start(
cmd: List<String>,
env: MutableMap<String, String> = 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
}
Expand Down
36 changes: 36 additions & 0 deletions app/src/main/java/io/nekohasekai/sagernet/bg/GuardedProcessRestartPolicy.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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)
Loading
Loading