Retry core shutdown until pending frontend resources are released
This commit is contained in:
1 parent
1529dfd04b
commit
d7fb8568b2
2 files changed
+45
-16
No files matched your search
@@ -4,14 +4,16 @@ import kotlinx.coroutines.CancellationException
|
||||
import kotlinx.coroutines.CoroutineScope
|
||||
import kotlinx.coroutines.Deferred
|
||||
import kotlinx.coroutines.async
|
||||
import kotlinx.coroutines.delay
|
||||
|
||||
/** Один незавершённый stop на процесс. Scope принадлежит приложению, переживает onDestroy Service.
|
||||
* Методы вызываются на main thread. Новый Service ждёт await перед init; отмена этого ожидания не отменяет stop.
|
||||
* Ошибка сохраняется для await, повторный request после завершения позволяет явно повторить очистку. */
|
||||
* Ошибка не подтверждает stop: та же задача повторяет очистку с backoff 250 мс..30 с до успеха. */
|
||||
internal class CoreShutdown(
|
||||
private val scope: CoroutineScope,
|
||||
private val stop: suspend () -> Unit,
|
||||
private val log: (String, String) -> Unit,
|
||||
private val retryWait: suspend (Long) -> Unit = { delay(it) },
|
||||
) {
|
||||
private var pending: Deferred<Unit>? = null
|
||||
|
||||
@@ -20,19 +22,24 @@ internal class CoreShutdown(
|
||||
pending?.takeUnless { it.isCompleted }?.let { return it }
|
||||
return scope.async {
|
||||
log("INFO", "core shutdown started")
|
||||
try {
|
||||
stop()
|
||||
log("INFO", "core shutdown completed")
|
||||
} catch (e: CancellationException) {
|
||||
log("ERROR", "core shutdown cancelled before completion")
|
||||
throw e
|
||||
} catch (e: Exception) {
|
||||
log("ERROR", "core shutdown failed: ${e.javaClass.simpleName}: ${e.message}")
|
||||
throw e
|
||||
var retryMs = 250L
|
||||
while (true) {
|
||||
try {
|
||||
stop()
|
||||
log("INFO", "core shutdown completed")
|
||||
break
|
||||
} catch (e: CancellationException) {
|
||||
log("ERROR", "core shutdown cancelled before completion")
|
||||
throw e
|
||||
} catch (e: Exception) {
|
||||
log("ERROR", "core shutdown failed; retry in ${retryMs}ms: ${e.javaClass.simpleName}: ${e.message}")
|
||||
}
|
||||
retryWait(retryMs)
|
||||
retryMs = (retryMs * 2).coerceAtMost(30_000)
|
||||
}
|
||||
}.also { pending = it }
|
||||
}
|
||||
|
||||
/** Нет незавершённой остановки — сразу вернуть управление; ошибка предыдущей очистки запрещает новый init. */
|
||||
/** Нет остановки — вернуть управление; иначе новый init ждёт успешную очистку, включая все повторы. */
|
||||
suspend fun await() { pending?.await() }
|
||||
}
|
||||
@@ -57,13 +57,35 @@ class CoreShutdownTest {
|
||||
|
||||
@Test fun failedStopBlocksInitializationUntilSuccessfulRetry() = runBlocking {
|
||||
var attempts = 0
|
||||
val retryGate = CompletableDeferred<Unit>()
|
||||
val delays = mutableListOf<Long>()
|
||||
val retry = CoreShutdown(scope, { check(++attempts > 1) { "join failed" } },
|
||||
{ level, message -> logs.add(level to message) })
|
||||
retry.request()
|
||||
assertTrue(runCatching { retry.await() }.exceptionOrNull() is IllegalStateException)
|
||||
{ level, message -> logs.add(level to message) }, { delays.add(it); retryGate.await() })
|
||||
val operation = retry.request()
|
||||
val replacement = async(start = CoroutineStart.UNDISPATCHED) { retry.await(); "init" }
|
||||
assertFalse(operation.isCompleted); assertFalse(replacement.isCompleted)
|
||||
assertSame(operation, retry.request()); assertEquals(1, attempts)
|
||||
assertTrue(logs.any { it.first == "ERROR" && it.second.contains("join failed") })
|
||||
retry.request().await()
|
||||
retryGate.complete(Unit)
|
||||
operation.await(); assertEquals("init", replacement.await())
|
||||
retry.await()
|
||||
assertEquals(2, attempts)
|
||||
assertEquals(2, attempts); assertEquals(listOf(250L), delays)
|
||||
}
|
||||
|
||||
@Test fun cancelledSettingsWaiterCannotInterruptAutomaticShutdownRetry() = runBlocking {
|
||||
val gate = CompletableDeferred<Unit>(); var attempts = 0
|
||||
val retry = CoreShutdown(scope, { check(++attempts > 1) { "release pending" } }, { _, _ -> }, { gate.await() })
|
||||
val operation = retry.request()
|
||||
val settings = async(start = CoroutineStart.UNDISPATCHED) { retry.await() }
|
||||
settings.cancel(); assertFalse(operation.isCancelled); assertEquals(1, attempts)
|
||||
gate.complete(Unit); operation.await(); assertEquals(2, attempts)
|
||||
}
|
||||
|
||||
@Test fun repeatedFailuresUseBoundedBackoffWithoutRequiringAnotherService() = runBlocking {
|
||||
val delays = mutableListOf<Long>(); var attempts = 0
|
||||
val retry = CoreShutdown(scope, { check(++attempts > 12) { "release pending" } }, { _, _ -> }, { delays.add(it) })
|
||||
retry.request().await()
|
||||
assertEquals(13, attempts); assertEquals(250L, delays.first()); assertEquals(30_000L, delays.last())
|
||||
assertTrue(delays.all { it in 250L..30_000L })
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user