diff --git a/app/src/main/kotlin/app/getarcane/android/core/ActivityCenterStore.kt b/app/src/main/kotlin/app/getarcane/android/core/ActivityCenterStore.kt index 0a99446..e04f1fc 100644 --- a/app/src/main/kotlin/app/getarcane/android/core/ActivityCenterStore.kt +++ b/app/src/main/kotlin/app/getarcane/android/core/ActivityCenterStore.kt @@ -22,6 +22,7 @@ import kotlinx.coroutines.Job import kotlinx.coroutines.async import kotlinx.coroutines.awaitAll import kotlinx.coroutines.coroutineScope +import kotlinx.coroutines.currentCoroutineContext import kotlinx.coroutines.launch /** @@ -55,6 +56,11 @@ class ActivityCenterStore(private val scope: CoroutineScope) { private val activityBuckets = LinkedHashMap>() private val environmentNames = HashMap() private val streamJobs = HashMap() + private var loadJob: Job? = null + private var clientGeneration = 0L + private var loadGeneration = 0L + private var streamGeneration = 0L + private var streamingRequested = false val filteredActivities: List get() { @@ -74,41 +80,42 @@ class ActivityCenterStore(private val scope: CoroutineScope) { get() = sortedUnique(activities.mapNotNull { it.resourceType }) fun configure(client: ArcaneClient?) { - val changed = this.client == null + if (this.client === client) return + clientGeneration++ + loadGeneration++ + loadJob?.cancel() + loadJob = null + stopStream() this.client = client - if (changed) { - stopStream() - activities = emptyList() - activityBuckets.clear() - environmentNames.clear() - environmentIds = emptyList() - limit = PAGE_SIZE - hasMore = false - errorMessage = null - streamErrorMessage = null - } + activities = emptyList() + activityBuckets.clear() + environmentNames.clear() + environmentIds = emptyList() + limit = PAGE_SIZE + hasMore = false + isLoading = false + isLoadingMore = false + errorMessage = null + streamErrorMessage = null } /** Fan out `listPaginated` across all environments, bucket per env, merge + sort. */ suspend fun load(reset: Boolean = true, refresh: Boolean = false) { val client = client ?: return - if (reset) { - limit = PAGE_SIZE - hasMore = false - } + val owningJob = currentCoroutineContext()[Job] + loadJob?.takeIf { it !== owningJob }?.cancel() + loadJob = owningJob + val expectedClientGeneration = clientGeneration + val operationGeneration = ++loadGeneration + val pageLimit = if (reset) PAGE_SIZE else limit if (activities.isEmpty() || refresh) isLoading = true errorMessage = null try { val environments = resolveEnvironments(client) - environmentIds = environments.map { it.id.rawValue } - environmentNames.clear() - environments.forEach { environmentNames[it.id.rawValue] = it.name } - - val pageLimit = limit val results: List?>> = coroutineScope { environments.map { environment -> async { - val data = runCatching { + val data = runSuspendCatching { client.activities.listPaginated( envId = environment.id, order = SortOrder.DESCENDING, @@ -131,9 +138,14 @@ class ActivityCenterStore(private val scope: CoroutineScope) { } val normalized = data.map { normalize(it, environment) } buckets[environment.id.rawValue] = sortActivities(normalized) - if (data.size >= limit) anyHasMore = true + if (data.size >= pageLimit) anyHasMore = true } + if (!isCurrentLoad(client, expectedClientGeneration, operationGeneration)) return + limit = pageLimit + environmentIds = environments.map { it.id.rawValue } + environmentNames.clear() + environments.forEach { environmentNames[it.id.rawValue] = it.name } activityBuckets.clear() activityBuckets.putAll(buckets) hasMore = anyHasMore @@ -141,21 +153,30 @@ class ActivityCenterStore(private val scope: CoroutineScope) { if (failures > 0) { streamErrorMessage = "Some environments could not load. Pull to refresh." } + if (streamingRequested) restartStreams() } catch (e: CancellationException) { throw e } catch (e: Throwable) { - if (activities.isEmpty()) errorMessage = friendlyErrorMessage(e) + if (isCurrentLoad(client, expectedClientGeneration, operationGeneration) && activities.isEmpty()) { + errorMessage = friendlyErrorMessage(e) + } } finally { - isLoading = false + if (isCurrentLoad(client, expectedClientGeneration, operationGeneration)) isLoading = false + if (loadJob === owningJob) loadJob = null } } suspend fun loadMore() { if (isLoadingMore || !hasMore) return isLoadingMore = true + val previousLimit = limit + val requestedLimit = previousLimit + PAGE_SIZE try { - limit += PAGE_SIZE + limit = requestedLimit load(reset = false) + } catch (e: CancellationException) { + if (limit == requestedLimit) limit = previousLimit + throw e } finally { isLoadingMore = false } @@ -163,8 +184,13 @@ class ActivityCenterStore(private val scope: CoroutineScope) { /** One coroutine per environment collecting `client.activities.stream(...)`. */ fun startStream() { + streamingRequested = true + restartStreams() + } + + private fun restartStreams() { val client = client ?: return - stopStream() + cancelStreamJobs() streamErrorMessage = null val environments = environmentIds.map { id -> @@ -172,15 +198,23 @@ class ActivityCenterStore(private val scope: CoroutineScope) { } if (environments.isEmpty()) return + val expectedClientGeneration = clientGeneration + val expectedStreamGeneration = streamGeneration isStreaming = true for (environment in environments) { streamJobs[environment.id.rawValue] = scope.launch { - consumeStream(client, environment) + consumeStream(client, environment, expectedClientGeneration, expectedStreamGeneration) } } } fun stopStream() { + streamingRequested = false + cancelStreamJobs() + } + + private fun cancelStreamJobs() { + streamGeneration++ streamJobs.values.forEach { it.cancel() } streamJobs.clear() isStreaming = false @@ -215,7 +249,7 @@ class ActivityCenterStore(private val scope: CoroutineScope) { val results: List> = coroutineScope { targets.map { id -> async { - runCatching { + runSuspendCatching { client.activities.clearHistory(envId = EnvironmentId(id)).deleted }.fold( onSuccess = { it to true }, @@ -235,21 +269,46 @@ class ActivityCenterStore(private val scope: CoroutineScope) { return ClearHistoryResult(deleted = deleted, failed = failed) } - private suspend fun consumeStream(client: ArcaneClient, environment: ActivityEnvironment) { + private suspend fun consumeStream( + client: ArcaneClient, + environment: ActivityEnvironment, + expectedClientGeneration: Long, + expectedStreamGeneration: Long, + ) { + val owningJob = currentCoroutineContext()[Job] try { client.activities.stream(envId = environment.id, limit = PAGE_SIZE).collect { event -> - apply(event, environment) + if (isCurrentStream(client, expectedClientGeneration, expectedStreamGeneration)) { + apply(event, environment) + } } } catch (e: CancellationException) { throw e } catch (e: Throwable) { - streamErrorMessage = "Live updates paused. Pull to refresh." + if (isCurrentStream(client, expectedClientGeneration, expectedStreamGeneration)) { + streamErrorMessage = "Live updates paused. Pull to refresh." + } } finally { - streamJobs.remove(environment.id.rawValue) - isStreaming = streamJobs.isNotEmpty() + val id = environment.id.rawValue + if (isCurrentStream(client, expectedClientGeneration, expectedStreamGeneration) && + streamJobs[id] === owningJob + ) { + streamJobs.remove(id) + isStreaming = streamJobs.isNotEmpty() + } } } + private fun isCurrentLoad(client: ArcaneClient, clientGeneration: Long, loadGeneration: Long): Boolean = + this.client === client && + this.clientGeneration == clientGeneration && + this.loadGeneration == loadGeneration + + private fun isCurrentStream(client: ArcaneClient, clientGeneration: Long, streamGeneration: Long): Boolean = + this.client === client && + this.clientGeneration == clientGeneration && + this.streamGeneration == streamGeneration + private fun apply(event: ActivityStreamEvent, environment: ActivityEnvironment) { when (event.type) { ActivityStreamEventType.SNAPSHOT -> replaceSnapshot(event.activities, environment) diff --git a/app/src/main/kotlin/app/getarcane/android/core/CoroutineFailures.kt b/app/src/main/kotlin/app/getarcane/android/core/CoroutineFailures.kt new file mode 100644 index 0000000..565ac76 --- /dev/null +++ b/app/src/main/kotlin/app/getarcane/android/core/CoroutineFailures.kt @@ -0,0 +1,19 @@ +package app.getarcane.android.core + +import kotlinx.coroutines.CancellationException +import kotlinx.coroutines.currentCoroutineContext +import kotlinx.coroutines.ensureActive + +/** Result wrapper for suspending best-effort work that must never consume coroutine cancellation. */ +internal suspend inline fun runSuspendCatching( + crossinline block: suspend () -> T, +): Result = + try { + val value = block() + currentCoroutineContext().ensureActive() + Result.success(value) + } catch (error: CancellationException) { + throw error + } catch (error: Throwable) { + Result.failure(error) + } diff --git a/app/src/main/kotlin/app/getarcane/android/core/DashboardStreamStore.kt b/app/src/main/kotlin/app/getarcane/android/core/DashboardStreamStore.kt index 3a0b90f..bf585f6 100644 --- a/app/src/main/kotlin/app/getarcane/android/core/DashboardStreamStore.kt +++ b/app/src/main/kotlin/app/getarcane/android/core/DashboardStreamStore.kt @@ -9,11 +9,13 @@ import app.getarcane.sdk.errors.ArcaneError import app.getarcane.sdk.models.environment.Environment import app.getarcane.sdk.streaming.ndjsonFlow import kotlinx.coroutines.CancellationException +import kotlinx.coroutines.CoroutineStart import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Job import kotlinx.coroutines.async import kotlinx.coroutines.awaitAll import kotlinx.coroutines.coroutineScope +import kotlinx.coroutines.currentCoroutineContext import kotlinx.coroutines.delay import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.launch @@ -77,6 +79,9 @@ class DashboardStreamStore( private var streamJob: Job? = null private var shouldRun = false private var generation = 0 + private var nextSnapshotRequest = 0L + private val snapshotRequestsByEnvironmentId = HashMap() + private val snapshotJobsByEnvironmentId = HashMap() val aggregate: DashboardStreamAggregateCounts? get() { @@ -120,6 +125,7 @@ class DashboardStreamStore( ) } } + snapshotRequestsByEnvironmentId.clear() streamUnsupported = false streamFailed = false } @@ -139,6 +145,9 @@ class DashboardStreamStore( fun stop() { shouldRun = false generation += 1 + snapshotRequestsByEnvironmentId.clear() + snapshotJobsByEnvironmentId.values.forEach { it.cancel() } + snapshotJobsByEnvironmentId.clear() streamJob?.cancel() streamJob = null connected = false @@ -172,13 +181,14 @@ class DashboardStreamStore( } } val newIds = next.keys - statesByEnvironmentId.keys + val removedIds = statesByEnvironmentId.keys - targetIds statesByEnvironmentId = next + snapshotRequestsByEnvironmentId.keys.retainAll(targetIds) + removedIds.forEach { id -> snapshotJobsByEnvironmentId.remove(id)?.cancel() } if (streamJob != null) { val currentGeneration = generation - newIds.forEach { id -> - scope.launch { refreshEnvironment(id, currentGeneration) } - } + newIds.forEach { id -> launchSnapshotRefresh(id, currentGeneration) } } } @@ -261,6 +271,12 @@ class DashboardStreamStore( } private fun applySnapshot(snapshot: DashboardSnapshot, environmentId: String) { + snapshotRequestsByEnvironmentId.remove(environmentId) + snapshotJobsByEnvironmentId.remove(environmentId)?.cancel() + applySnapshotState(snapshot, environmentId) + } + + private fun applySnapshotState(snapshot: DashboardSnapshot, environmentId: String) { val state = statesByEnvironmentId[environmentId] ?: return statesByEnvironmentId = statesByEnvironmentId + (environmentId to state.copy( snapshot = snapshot, @@ -273,6 +289,12 @@ class DashboardStreamStore( } private fun applyError(message: String?, code: DashboardStreamErrorCode?, environmentId: String) { + snapshotRequestsByEnvironmentId.remove(environmentId) + snapshotJobsByEnvironmentId.remove(environmentId)?.cancel() + applyErrorState(message, code, environmentId) + } + + private fun applyErrorState(message: String?, code: DashboardStreamErrorCode?, environmentId: String) { val state = statesByEnvironmentId[environmentId] ?: return statesByEnvironmentId = statesByEnvironmentId + (environmentId to state.copy( loading = false, @@ -290,20 +312,51 @@ class DashboardStreamStore( private suspend fun refreshEnvironment(environmentId: String, currentGeneration: Int) { val activeClient = client ?: return + val request = ++nextSnapshotRequest + snapshotRequestsByEnvironmentId[environmentId] = request try { val snapshot = activeClient.snapshot(environmentId) - if (currentGeneration == generation && statesByEnvironmentId.containsKey(environmentId)) { - applySnapshot(snapshot, environmentId) + if (isCurrentSnapshotRequest(activeClient, environmentId, currentGeneration, request)) { + snapshotRequestsByEnvironmentId.remove(environmentId) + applySnapshotState(snapshot, environmentId) } } catch (e: CancellationException) { throw e } catch (e: Throwable) { - if (currentGeneration == generation && statesByEnvironmentId.containsKey(environmentId)) { - applyError(friendlyErrorMessage(e), null, environmentId) + if (isCurrentSnapshotRequest(activeClient, environmentId, currentGeneration, request)) { + snapshotRequestsByEnvironmentId.remove(environmentId) + applyErrorState(friendlyErrorMessage(e), null, environmentId) } } } + private fun launchSnapshotRefresh(environmentId: String, currentGeneration: Int) { + snapshotJobsByEnvironmentId.remove(environmentId)?.cancel() + val job = scope.launch(start = CoroutineStart.LAZY) { + val owningJob = currentCoroutineContext()[Job] + try { + refreshEnvironment(environmentId, currentGeneration) + } finally { + if (snapshotJobsByEnvironmentId[environmentId] === owningJob) { + snapshotJobsByEnvironmentId.remove(environmentId) + } + } + } + snapshotJobsByEnvironmentId[environmentId] = job + job.start() + } + + private fun isCurrentSnapshotRequest( + expectedClient: DashboardStreamClient, + environmentId: String, + expectedGeneration: Int, + request: Long, + ): Boolean = + client === expectedClient && + generation == expectedGeneration && + snapshotRequestsByEnvironmentId[environmentId] == request && + statesByEnvironmentId.containsKey(environmentId) + private companion object { const val MAX_RECONNECT_ATTEMPTS = 20 const val MAX_RECONNECT_DELAY_MILLIS = 15_000L diff --git a/app/src/main/kotlin/app/getarcane/android/core/DemoService.kt b/app/src/main/kotlin/app/getarcane/android/core/DemoService.kt index 410477f..02a7d71 100644 --- a/app/src/main/kotlin/app/getarcane/android/core/DemoService.kt +++ b/app/src/main/kotlin/app/getarcane/android/core/DemoService.kt @@ -72,7 +72,7 @@ object DemoService { while (isActive) { delay(15_000) if (!isActive) break - runCatching { post("demo-kuma/heartbeat", timeoutMs = 10_000, withCookie = true) } + runSuspendCatching { post("demo-kuma/heartbeat", timeoutMs = 10_000, withCookie = true) } } } } @@ -86,7 +86,7 @@ object DemoService { suspend fun endSession() { stopHeartbeat() withContext(Dispatchers.IO) { - runCatching { post("demo-kuma/end-session", timeoutMs = 10_000, withCookie = true) } + runSuspendCatching { post("demo-kuma/end-session", timeoutMs = 10_000, withCookie = true) } } sessionId = null } diff --git a/app/src/main/kotlin/app/getarcane/android/ui/screens/DashboardPinnedSection.kt b/app/src/main/kotlin/app/getarcane/android/ui/screens/DashboardPinnedSection.kt index aa018e1..1b594c3 100644 --- a/app/src/main/kotlin/app/getarcane/android/ui/screens/DashboardPinnedSection.kt +++ b/app/src/main/kotlin/app/getarcane/android/ui/screens/DashboardPinnedSection.kt @@ -91,7 +91,7 @@ fun DashboardPinnedSection( var runningId by remember { mutableStateOf(null) } var reloadKey by remember { mutableStateOf(0) } - LaunchedEffect(envId.rawValue, pinned.version, refreshToken, reloadKey) { + LaunchedEffect(client, envId.rawValue, pinned.version, refreshToken, reloadKey) { if (client == null) return@LaunchedEffect val pinnedContainers = pinned.pinnedIds(PinnedItemsStore.Kind.CONTAINER, envId) val pinnedProjects = pinned.pinnedIds(PinnedItemsStore.Kind.PROJECT, envId) diff --git a/app/src/main/kotlin/app/getarcane/android/ui/screens/DashboardScreen.kt b/app/src/main/kotlin/app/getarcane/android/ui/screens/DashboardScreen.kt index 04c57ed..f9de89f 100644 --- a/app/src/main/kotlin/app/getarcane/android/ui/screens/DashboardScreen.kt +++ b/app/src/main/kotlin/app/getarcane/android/ui/screens/DashboardScreen.kt @@ -79,6 +79,7 @@ import app.getarcane.android.core.DashboardEnvironmentStreamState import app.getarcane.android.core.DashboardStreamStore import app.getarcane.android.core.LocalArcaneManager import app.getarcane.android.core.friendlyErrorMessage +import app.getarcane.android.core.runSuspendCatching import app.getarcane.android.nav.AppTab import app.getarcane.android.ui.screens.activities.ActivitiesTab import app.getarcane.android.ui.screens.settings.FormErrorRow @@ -168,6 +169,8 @@ fun DashboardScreen( var loading by remember { mutableStateOf(false) } var refreshKey by remember { mutableStateOf(0) } var statsRestartKey by remember { mutableStateOf(0) } + var statsGeneration by remember { mutableStateOf(0) } + var statsClient by remember { mutableStateOf(client) } var showActivities by remember { mutableStateOf(false) } var showUpdateAll by remember { mutableStateOf(false) } var pruneEnvironmentId by remember { mutableStateOf(null) } @@ -181,6 +184,11 @@ fun DashboardScreen( val snackbar = remember { SnackbarHostState() } LaunchedEffect(client, enabledEnvironmentIds, refreshKey, statsRestartKey) { + val generation = ++statsGeneration + if (statsClient !== client) { + statsHistory.clear() + statsClient = client + } if (client == null) return@LaunchedEffect statsHistory.keys @@ -193,17 +201,21 @@ fun DashboardScreen( .forEachIndexed { index, id -> launch { delay(150L * (index + 1)) + if (generation != statsGeneration) return@launch val env = EnvironmentId(id) statsHistory[id] = (statsHistory[id] ?: DashboardStatsSeries()).reconnecting() - runCatching { + runSuspendCatching { client.system.statsStream(env).collect { stats -> - statsHistory[id] = (statsHistory[id] ?: DashboardStatsSeries()).append(stats) + if (generation == statsGeneration) { + statsHistory[id] = (statsHistory[id] ?: DashboardStatsSeries()).append(stats) + } } }.onFailure { error -> - if (error is CancellationException) throw error - statsHistory[id] = (statsHistory[id] ?: DashboardStatsSeries()).copy( - error = "Live stats unavailable: ${friendlyErrorMessage(error)}", - ) + if (generation == statsGeneration) { + statsHistory[id] = (statsHistory[id] ?: DashboardStatsSeries()).copy( + error = "Live stats unavailable: ${friendlyErrorMessage(error)}", + ) + } } } } @@ -239,7 +251,7 @@ fun DashboardScreen( streamStore.reconcile(environments) } - LaunchedEffect(refreshKey) { + LaunchedEffect(client, refreshKey) { if (client == null) return@LaunchedEffect loading = true val overview = try { diff --git a/app/src/main/kotlin/app/getarcane/android/ui/screens/EnvironmentDashboardCard.kt b/app/src/main/kotlin/app/getarcane/android/ui/screens/EnvironmentDashboardCard.kt index 4989d67..5b8cfd2 100644 --- a/app/src/main/kotlin/app/getarcane/android/ui/screens/EnvironmentDashboardCard.kt +++ b/app/src/main/kotlin/app/getarcane/android/ui/screens/EnvironmentDashboardCard.kt @@ -51,6 +51,7 @@ import app.getarcane.android.core.DashboardActionItem import app.getarcane.android.core.DashboardActionItemKind import app.getarcane.android.core.DashboardActionItemSeverity import app.getarcane.android.core.LocalArcaneManager +import app.getarcane.android.core.runSuspendCatching import app.getarcane.android.ui.theme.ArcaneBlue import app.getarcane.android.ui.theme.ArcaneGreen import app.getarcane.android.ui.theme.ArcaneOrange @@ -100,8 +101,9 @@ fun EnvironmentDashboardCard( var dockerInfo by remember(env.id) { mutableStateOf(null) } var showMenu by remember { mutableStateOf(false) } - LaunchedEffect(env.id, refreshToken) { - dockerInfo = runCatching { client?.system?.dockerInfo(envId) }.getOrNull() + LaunchedEffect(client, env.id, refreshToken) { + dockerInfo = null + dockerInfo = runSuspendCatching { client?.system?.dockerInfo(envId) }.getOrNull() } val stats = statsSeries?.latest diff --git a/app/src/main/kotlin/app/getarcane/android/ui/screens/activities/ActivitiesScreen.kt b/app/src/main/kotlin/app/getarcane/android/ui/screens/activities/ActivitiesScreen.kt index f66f117..303b565 100644 --- a/app/src/main/kotlin/app/getarcane/android/ui/screens/activities/ActivitiesScreen.kt +++ b/app/src/main/kotlin/app/getarcane/android/ui/screens/activities/ActivitiesScreen.kt @@ -98,9 +98,12 @@ fun ActivitiesScreen( val canClearHistory = clearableEnvironmentIds.isNotEmpty() // Load + start the live stream on appear; stop streams on dispose. - LaunchedEffect(supportsActivities) { - if (!supportsActivities) return@LaunchedEffect + LaunchedEffect(manager.client, supportsActivities) { store.configure(manager.client) + if (!supportsActivities) { + store.stopStream() + return@LaunchedEffect + } store.load(refresh = true) store.startStream() } @@ -157,7 +160,7 @@ fun ActivitiesScreen( Icons.Filled.Warning, store.errorMessage, "Retry", - ) { scope.launch { store.load(refresh = true); store.startStream() } } + ) { scope.launch { store.load(refresh = true) } } store.activities.isEmpty() -> ContentUnavailable( "No Activities", @@ -172,7 +175,6 @@ fun ActivitiesScreen( refreshing = true scope.launch { store.load(refresh = true) - store.startStream() refreshing = false } }, diff --git a/app/src/main/kotlin/app/getarcane/android/ui/screens/containers/ContainerStatsScreen.kt b/app/src/main/kotlin/app/getarcane/android/ui/screens/containers/ContainerStatsScreen.kt index 0f26e7c..5e919a6 100644 --- a/app/src/main/kotlin/app/getarcane/android/ui/screens/containers/ContainerStatsScreen.kt +++ b/app/src/main/kotlin/app/getarcane/android/ui/screens/containers/ContainerStatsScreen.kt @@ -57,6 +57,7 @@ import app.getarcane.sdk.models.base.int64Value import app.getarcane.sdk.models.base.objectValue import app.getarcane.sdk.models.base.stringValue import app.getarcane.sdk.models.container.ContainerStatsPayload +import kotlinx.coroutines.CancellationException /** A parsed point in the rolling stats window. Port of iOS `ContainerStatsFrame`. */ private data class StatsFrame( @@ -90,16 +91,24 @@ fun ContainerStatsScreen(id: String) { var error by remember { mutableStateOf(null) } var streaming by remember { mutableStateOf(false) } var retryKey by remember { mutableStateOf(0) } + var streamGeneration by remember { mutableStateOf(0) } val windowSize = 60 - LaunchedEffect(id, retryKey) { - if (client == null) return@LaunchedEffect + LaunchedEffect(client, envId.rawValue, id, retryKey) { + val generation = ++streamGeneration + frames.clear() + latest = null error = null + if (client == null) { + streaming = false + return@LaunchedEffect + } streaming = true - var previous: StatsFrame? = latest ?: frames.lastOrNull() + var previous: StatsFrame? = null try { client.containers.stats(envId = envId, id = id).collect { payload -> + if (generation != streamGeneration) return@collect parseFrame(payload, previous)?.let { frame -> previous = frame frames.add(frame) @@ -107,10 +116,14 @@ fun ContainerStatsScreen(id: String) { latest = frame } } + } catch (e: CancellationException) { + throw e } catch (e: Throwable) { - error = "Stats stream ended: ${friendlyErrorMessage(e)}" + if (generation == streamGeneration) { + error = "Stats stream ended: ${friendlyErrorMessage(e)}" + } } finally { - streaming = false + if (generation == streamGeneration) streaming = false } } diff --git a/app/src/main/kotlin/app/getarcane/android/ui/screens/containers/ContainerTerminalScreen.kt b/app/src/main/kotlin/app/getarcane/android/ui/screens/containers/ContainerTerminalScreen.kt index c2b01ff..efa40fb 100644 --- a/app/src/main/kotlin/app/getarcane/android/ui/screens/containers/ContainerTerminalScreen.kt +++ b/app/src/main/kotlin/app/getarcane/android/ui/screens/containers/ContainerTerminalScreen.kt @@ -34,7 +34,6 @@ import androidx.compose.material3.Text import androidx.compose.material3.TextButton import androidx.compose.material3.TopAppBar import androidx.compose.runtime.Composable -import androidx.compose.runtime.DisposableEffect import androidx.compose.runtime.LaunchedEffect import androidx.compose.runtime.getValue import androidx.compose.runtime.mutableStateOf @@ -51,12 +50,15 @@ import androidx.compose.ui.unit.dp import app.getarcane.android.core.AnsiSanitizer import app.getarcane.android.core.LocalArcaneManager import app.getarcane.android.core.friendlyErrorMessage +import app.getarcane.android.core.runSuspendCatching import app.getarcane.android.ui.components.BannerSeverity import app.getarcane.android.ui.components.ErrorBanner import app.getarcane.android.ui.theme.ArcaneBlue import app.getarcane.sdk.streaming.TerminalSession -import kotlinx.coroutines.Job +import kotlinx.coroutines.CancellationException +import kotlinx.coroutines.NonCancellable import kotlinx.coroutines.launch +import kotlinx.coroutines.withContext private val shells = listOf("/bin/sh", "/bin/bash", "/bin/zsh", "/bin/ash") private const val CHAR_BUDGET = 200_000 @@ -82,93 +84,118 @@ fun ContainerTerminalScreen(id: String, title: String, onClose: () -> Unit) { var output by remember { mutableStateOf("") } var input by remember { mutableStateOf("") } var session by remember { mutableStateOf(null) } - var outputJob by remember { mutableStateOf(null) } var connectError by remember { mutableStateOf(null) } var isConnecting by remember { mutableStateOf(false) } var isConnected by remember { mutableStateOf(false) } var shell by remember { mutableStateOf("/bin/sh") } var menuOpen by remember { mutableStateOf(false) } + var retryKey by remember { mutableStateOf(0) } + var outputClient by remember { mutableStateOf(client) } + var outputEnvironmentId by remember { mutableStateOf(envId.rawValue) } + var outputContainerId by remember { mutableStateOf(id) } val scrollState = rememberScrollState() - suspend fun teardown() { - outputJob?.cancel() - outputJob = null - runCatching { session?.close() } - session = null - isConnected = false - isConnecting = false + suspend fun closeSession(closingSession: TerminalSession? = session) { + if (closingSession != null) { + withContext(NonCancellable) { + runSuspendCatching { closingSession.close() } + } + } + if (session === closingSession) { + session = null + isConnected = false + isConnecting = false + } } - fun connect() { - if (client == null) { + LaunchedEffect(client, envId.rawValue, id, shell, retryKey) { + val activeClient = client + if (outputClient !== activeClient || + outputEnvironmentId != envId.rawValue || + outputContainerId != id + ) { + output = "" + input = "" + outputClient = activeClient + outputEnvironmentId = envId.rawValue + outputContainerId = id + } + if (activeClient == null) { connectError = "Not connected to a server." - return + return@LaunchedEffect } - if (isConnected || isConnecting) return isConnecting = true connectError = null + var ownedSession: TerminalSession? = null + try { + val activeSession = activeClient.containers.exec(envId = envId, id = id, shell = shell) + ownedSession = activeSession + session = activeSession + isConnected = true + isConnecting = false + activeSession.output.collect { chunk -> + val raw = chunk.decodeToString() + // Auto-reply to cursor-position requests (DSR ESC[6n). + if (raw.contains(ESC + "[6n")) { + runSuspendCatching { activeSession.send(ESC + "[1;1R") } + } + val stripped = AnsiSanitizer.strip(raw) + val combined = output + stripped + output = if (combined.length > CHAR_BUDGET) { + combined.substring(combined.length - CHAR_BUDGET + CHAR_BUDGET / 10) + } else { + combined + } + } + } catch (e: CancellationException) { + throw e + } catch (e: Throwable) { + connectError = if (isConnected) { + "Disconnected: ${friendlyErrorMessage(e)}" + } else { + friendlyErrorMessage(e) + } + } finally { + closeSession(ownedSession) + } + } + + fun send(text: String) { + val activeSession = session ?: return + if (!isConnected) return scope.launch { try { - val s = client.containers.exec(envId = envId, id = id, shell = shell) - session = s - isConnected = true - isConnecting = false - outputJob = scope.launch { - try { - s.output.collect { chunk -> - val raw = chunk.decodeToString() - // Auto-reply to cursor-position requests (DSR ESC[6n). - if (raw.contains(ESC + "[6n")) runCatching { s.send(ESC + "[1;1R") } - val stripped = AnsiSanitizer.strip(raw) - val combined = output + stripped - output = if (combined.length > CHAR_BUDGET) { - combined.substring(combined.length - CHAR_BUDGET + CHAR_BUDGET / 10) - } else { - combined - } - } - } catch (e: Throwable) { - connectError = "Disconnected: ${friendlyErrorMessage(e)}" - } finally { - isConnected = false - } - } + activeSession.send(text) + } catch (e: CancellationException) { + throw e } catch (e: Throwable) { - connectError = friendlyErrorMessage(e) - isConnecting = false + if (session === activeSession) { + connectError = "Send failed: ${friendlyErrorMessage(e)}" + } } } } fun sendShortcut(text: String) { - val s = session ?: return - if (!isConnected) return - scope.launch { runCatching { s.send(text) } } + send(text) } fun sendInput() { - val s = session ?: return - if (!isConnected || input.isEmpty()) return + if (input.isEmpty()) return val payload = input + "\n" - scope.launch { runCatching { s.send(payload) } } + send(payload) input = "" } - LaunchedEffect(Unit) { connect() } - LaunchedEffect(output) { scrollState.animateScrollTo(scrollState.maxValue) } - DisposableEffect(Unit) { - onDispose { scope.launch { teardown() } } - } - Scaffold( topBar = { TopAppBar( title = { Text(title, maxLines = 1) }, navigationIcon = { - TextButton(onClick = { scope.launch { teardown(); onClose() } }) { Text("Close") } + TextButton(onClick = { scope.launch { closeSession(); onClose() } }) { Text("Close") } }, actions = { Box { @@ -205,7 +232,7 @@ fun ContainerTerminalScreen(id: String, title: String, onClose: () -> Unit) { ErrorBanner( it, severity = BannerSeverity.Warning, - onRetry = { connect() }, + onRetry = { retryKey++ }, modifier = Modifier.padding(horizontal = 12.dp, vertical = 8.dp), ) } diff --git a/app/src/main/kotlin/app/getarcane/android/ui/screens/containers/LogsScreen.kt b/app/src/main/kotlin/app/getarcane/android/ui/screens/containers/LogsScreen.kt index 1da4343..e8da077 100644 --- a/app/src/main/kotlin/app/getarcane/android/ui/screens/containers/LogsScreen.kt +++ b/app/src/main/kotlin/app/getarcane/android/ui/screens/containers/LogsScreen.kt @@ -59,6 +59,7 @@ import app.getarcane.android.ui.theme.ArcaneGreen import app.getarcane.android.ui.theme.ArcaneOrange import app.getarcane.android.ui.theme.ArcaneRed import app.getarcane.sdk.streaming.LogLine +import kotlinx.coroutines.CancellationException /** Standalone logs screen (own back stack route). Wraps [LogsContent] in a Scaffold. */ @OptIn(ExperimentalMaterial3Api::class) @@ -127,6 +128,7 @@ private fun LogsContent( var autoScroll by remember { mutableStateOf(true) } var newWhilePaused by remember { mutableIntStateOf(0) } var error by remember { mutableStateOf(null) } + var streamGeneration by remember { mutableIntStateOf(0) } LaunchedEffect(clearSignal) { if (clearSignal > 0) { @@ -135,19 +137,29 @@ private fun LogsContent( } } - LaunchedEffect(id) { - if (client == null) return@LaunchedEffect + LaunchedEffect(client, envId.rawValue, id) { + val generation = ++streamGeneration + lines.clear() + newWhilePaused = 0 + error = null + if (client == null) { + onStreamingChange(false) + return@LaunchedEffect + } onStreamingChange(true) try { client.containers.logs(envId = envId, id = id, follow = true, tail = "200").collect { line -> + if (generation != streamGeneration) return@collect lines.add(line) if (lines.size > 5000) repeat(100) { if (lines.isNotEmpty()) lines.removeAt(0) } if (!autoScroll) newWhilePaused++ } + } catch (e: CancellationException) { + throw e } catch (e: Throwable) { - error = friendlyErrorMessage(e) + if (generation == streamGeneration) error = friendlyErrorMessage(e) } finally { - onStreamingChange(false) + if (generation == streamGeneration) onStreamingChange(false) } } diff --git a/app/src/main/kotlin/app/getarcane/android/ui/screens/images/PullImageSheet.kt b/app/src/main/kotlin/app/getarcane/android/ui/screens/images/PullImageSheet.kt index 84586a7..c2d51ad 100644 --- a/app/src/main/kotlin/app/getarcane/android/ui/screens/images/PullImageSheet.kt +++ b/app/src/main/kotlin/app/getarcane/android/ui/screens/images/PullImageSheet.kt @@ -25,6 +25,7 @@ import androidx.compose.material3.OutlinedTextField import androidx.compose.material3.Text import androidx.compose.material3.rememberModalBottomSheetState import androidx.compose.runtime.Composable +import androidx.compose.runtime.DisposableEffect import androidx.compose.runtime.getValue import androidx.compose.runtime.mutableStateListOf import androidx.compose.runtime.mutableStateMapOf @@ -70,6 +71,10 @@ internal fun PullImageSheet(onDismiss: () -> Unit, onComplete: () -> Unit) { val layers = remember { mutableStateMapOf() } var pullJob by remember { mutableStateOf(null) } + DisposableEffect(client, envId.rawValue) { + onDispose { pullJob?.cancel() } + } + fun parseNameAndTag(raw: String): Pair { val trimmed = raw.trim() val beforeDigest = trimmed.substringBefore("@") @@ -120,8 +125,9 @@ internal fun PullImageSheet(onDismiss: () -> Unit, onComplete: () -> Unit) { statusLine = "Pull complete" onComplete() } - } catch (_: CancellationException) { + } catch (e: CancellationException) { statusLine = "Cancelled" + throw e } catch (e: Throwable) { errorMessage = friendlyErrorMessage(e) statusLine = "" diff --git a/app/src/main/kotlin/app/getarcane/android/ui/screens/images/UploadImageSheet.kt b/app/src/main/kotlin/app/getarcane/android/ui/screens/images/UploadImageSheet.kt index d342f8f..4bb0085 100644 --- a/app/src/main/kotlin/app/getarcane/android/ui/screens/images/UploadImageSheet.kt +++ b/app/src/main/kotlin/app/getarcane/android/ui/screens/images/UploadImageSheet.kt @@ -27,6 +27,7 @@ import androidx.compose.material3.OutlinedButton import androidx.compose.material3.Text import androidx.compose.material3.rememberModalBottomSheetState import androidx.compose.runtime.Composable +import androidx.compose.runtime.DisposableEffect import androidx.compose.runtime.getValue import androidx.compose.runtime.mutableStateOf import androidx.compose.runtime.remember @@ -68,6 +69,10 @@ internal fun UploadImageSheet(onDismiss: () -> Unit, onComplete: () -> Unit) { var errorMessage by remember { mutableStateOf(null) } var uploadJob by remember { mutableStateOf(null) } + DisposableEffect(client, envId.rawValue) { + onDispose { uploadJob?.cancel() } + } + val picker = rememberLauncherForActivityResult(ActivityResultContracts.GetContent()) { uri -> if (uri == null) return@rememberLauncherForActivityResult var name = uri.lastPathSegment ?: "image.tar" @@ -121,8 +126,9 @@ internal fun UploadImageSheet(onDismiss: () -> Unit, onComplete: () -> Unit) { progress = 1f output = aggregated.toString().trim().ifEmpty { "Upload complete." } onComplete() - } catch (_: CancellationException) { + } catch (e: CancellationException) { errorMessage = "Cancelled" + throw e } catch (e: Throwable) { errorMessage = friendlyErrorMessage(e) } finally { diff --git a/app/src/main/kotlin/app/getarcane/android/ui/screens/projects/ProjectLogsScreen.kt b/app/src/main/kotlin/app/getarcane/android/ui/screens/projects/ProjectLogsScreen.kt index 0f324d4..079817e 100644 --- a/app/src/main/kotlin/app/getarcane/android/ui/screens/projects/ProjectLogsScreen.kt +++ b/app/src/main/kotlin/app/getarcane/android/ui/screens/projects/ProjectLogsScreen.kt @@ -38,6 +38,7 @@ import app.getarcane.android.ui.components.buildAnsiAnnotatedString import app.getarcane.android.ui.theme.ArcaneOrange import app.getarcane.android.ui.theme.ArcaneRed import app.getarcane.sdk.streaming.LogLine +import kotlinx.coroutines.CancellationException /** Live project log stream. Mirrors the iOS `LogsView` opened from the project detail. */ @OptIn(ExperimentalMaterial3Api::class) @@ -51,16 +52,23 @@ fun ProjectLogsScreen(projectId: String, title: String, onBack: () -> Unit) { val listState = rememberLazyListState() var autoScroll by remember { mutableStateOf(true) } var error by remember { mutableStateOf(null) } + var streamGeneration by remember { mutableStateOf(0) } - LaunchedEffect(projectId) { + LaunchedEffect(client, envId.rawValue, projectId) { + val generation = ++streamGeneration + lines.clear() + error = null if (client == null) return@LaunchedEffect try { client.projects.logs(envId = envId, projectId = projectId, follow = true, tail = "200").collect { line -> + if (generation != streamGeneration) return@collect lines.add(line) if (lines.size > 5000) repeat(100) { if (lines.isNotEmpty()) lines.removeAt(0) } } + } catch (e: CancellationException) { + throw e } catch (e: Throwable) { - error = friendlyErrorMessage(e) + if (generation == streamGeneration) error = friendlyErrorMessage(e) } } diff --git a/app/src/main/kotlin/app/getarcane/android/ui/screens/projects/StreamingActionScreen.kt b/app/src/main/kotlin/app/getarcane/android/ui/screens/projects/StreamingActionScreen.kt index 693acd7..faedf36 100644 --- a/app/src/main/kotlin/app/getarcane/android/ui/screens/projects/StreamingActionScreen.kt +++ b/app/src/main/kotlin/app/getarcane/android/ui/screens/projects/StreamingActionScreen.kt @@ -40,6 +40,7 @@ import app.getarcane.android.core.friendlyErrorMessage import app.getarcane.android.ui.theme.ArcaneGreen import app.getarcane.android.ui.theme.ArcaneRed import app.getarcane.sdk.models.project.PullProgressEvent +import kotlinx.coroutines.CancellationException import kotlinx.coroutines.flow.Flow /** A single rendered line of stream output. */ @@ -71,6 +72,7 @@ fun StreamingActionScreen( val lines = remember { mutableStateListOf() } var status by remember { mutableStateOf(StreamStatus.Running) } var currentPhase by remember { mutableStateOf(null) } + var streamGeneration by remember { mutableStateOf(0) } val listState = rememberLazyListState() fun append(text: String, isError: Boolean) { @@ -78,7 +80,11 @@ fun StreamingActionScreen( if (lines.size > 2000) repeat(200) { if (lines.isNotEmpty()) lines.removeAt(0) } } - LaunchedEffect(projectId, action) { + LaunchedEffect(client, envId.rawValue, projectId, action) { + val generation = ++streamGeneration + lines.clear() + status = StreamStatus.Running + currentPhase = null if (client == null) { status = StreamStatus.Failure("No client available") return@LaunchedEffect @@ -96,6 +102,7 @@ fun StreamingActionScreen( } try { stream.collect { event -> + if (generation != streamGeneration) return@collect val isError = event.error != null val display = displayText(event) if (display.isNotEmpty()) append(display, isError) @@ -104,13 +111,19 @@ fun StreamingActionScreen( if (!phase.isNullOrEmpty()) currentPhase = phase } } - status = StreamStatus.Success - currentPhase = "Complete" + if (generation == streamGeneration) { + status = StreamStatus.Success + currentPhase = "Complete" + } + } catch (e: CancellationException) { + throw e } catch (e: Throwable) { - val message = friendlyErrorMessage(e) - append(message, isError = true) - status = StreamStatus.Failure(message) - currentPhase = "Failed" + if (generation == streamGeneration) { + val message = friendlyErrorMessage(e) + append(message, isError = true) + status = StreamStatus.Failure(message) + currentPhase = "Failed" + } } } diff --git a/app/src/main/kotlin/app/getarcane/android/ui/screens/updates/UpdaterRunScreen.kt b/app/src/main/kotlin/app/getarcane/android/ui/screens/updates/UpdaterRunScreen.kt index 21a13a7..7500b23 100644 --- a/app/src/main/kotlin/app/getarcane/android/ui/screens/updates/UpdaterRunScreen.kt +++ b/app/src/main/kotlin/app/getarcane/android/ui/screens/updates/UpdaterRunScreen.kt @@ -60,6 +60,7 @@ import androidx.compose.ui.text.style.TextOverflow import androidx.compose.ui.unit.dp import app.getarcane.android.core.LocalArcaneManager import app.getarcane.android.core.friendlyErrorMessage +import app.getarcane.android.core.runSuspendCatching import app.getarcane.android.ui.components.ContentUnavailable import app.getarcane.android.ui.theme.ArcaneBlue import app.getarcane.android.ui.theme.ArcaneGray @@ -74,7 +75,6 @@ import app.getarcane.sdk.errors.ArcaneError import app.getarcane.sdk.models.updater.UpdaterResourceResult import app.getarcane.sdk.models.updater.UpdaterResult import app.getarcane.sdk.models.updater.UpdaterStatus -import kotlinx.coroutines.CancellationException import kotlinx.coroutines.async import kotlinx.coroutines.delay import kotlinx.coroutines.isActive @@ -139,12 +139,7 @@ internal fun shouldContinuePollingAfterRunFailure( ): Boolean = observedServerStart && latestStatus?.hasActiveWork == true internal suspend fun runUpdaterRequestCatching(block: suspend () -> UpdaterResult): Result = - try { - Result.success(block()) - } catch (error: Throwable) { - if (error is CancellationException && !coroutineContext.isActive) throw error - Result.failure(error) - } + runSuspendCatching(block) private fun logUpdaterDebug(message: String) { Log.d(UpdaterRunLogTag, message) @@ -202,12 +197,12 @@ fun UpdaterRunScreen(onBack: () -> Unit, environmentId: EnvironmentId? = null, e var observedServerStart = false var requestFailureHandled = false - val baselineStatus = runCatching { client.updater.status(envId = envId) } + val baselineStatus = runSuspendCatching { client.updater.status(envId = envId) } .onFailure { error -> logUpdaterError("Baseline status probe failed envId=${envId.rawValue}", error) } .getOrNull() val baselineSnapshot = baselineStatus?.let(UpdaterRunStatusSnapshot::from) logUpdaterDebug("Baseline status envId=${envId.rawValue} snapshot=$baselineSnapshot") - val baselineHistoryIds = runCatching { + val baselineHistoryIds = runSuspendCatching { loadUpdaterHistory(client = client, envId = envId, limit = 5).mapTo(mutableSetOf()) { it.id } } .onFailure { error -> logUpdaterError("Baseline history probe failed envId=${envId.rawValue}", error) } @@ -229,7 +224,7 @@ fun UpdaterRunScreen(onBack: () -> Unit, environmentId: EnvironmentId? = null, e while (coroutineContext.isActive) { if (runJob.isCompleted && !requestFailureHandled) { var finalStatusProbeSucceeded = false - val finalStatusSnapshot = runCatching { client.updater.status(envId = envId) } + val finalStatusSnapshot = runSuspendCatching { client.updater.status(envId = envId) } .onSuccess { finalStatusProbeSucceeded = true } .onFailure { error -> logUpdaterError("Final status probe failed envId=${envId.rawValue}", error) } .getOrNull() @@ -237,7 +232,7 @@ fun UpdaterRunScreen(onBack: () -> Unit, environmentId: EnvironmentId? = null, e ?.let(UpdaterRunStatusSnapshot::from) val finalStatusStartEvidence = finalStatusSnapshot?.isNewActiveWorkComparedTo(baselineSnapshot) ?: false var finalHistoryProbeSucceeded = false - val finalHistoryStartEvidence = runCatching { + val finalHistoryStartEvidence = runSuspendCatching { loadUpdaterHistory(client = client, envId = envId, limit = 5).mapTo(mutableSetOf()) { it.id } } .onSuccess { finalHistoryProbeSucceeded = true } @@ -278,7 +273,7 @@ fun UpdaterRunScreen(onBack: () -> Unit, environmentId: EnvironmentId? = null, e if (!requestFailureHandled) break } - runCatching { client.updater.status(envId = envId) } + runSuspendCatching { client.updater.status(envId = envId) } .onFailure { error -> logUpdaterError("Polling status failed envId=${envId.rawValue}", error) } .getOrNull() ?.let { status -> diff --git a/app/src/test/java/app/getarcane/android/core/CoroutineFailuresTest.kt b/app/src/test/java/app/getarcane/android/core/CoroutineFailuresTest.kt new file mode 100644 index 0000000..9185b2b --- /dev/null +++ b/app/src/test/java/app/getarcane/android/core/CoroutineFailuresTest.kt @@ -0,0 +1,52 @@ +package app.getarcane.android.core + +import java.io.IOException +import kotlinx.coroutines.CancellationException +import kotlinx.coroutines.cancel +import kotlinx.coroutines.currentCoroutineContext +import kotlinx.coroutines.runBlocking +import org.junit.Assert.assertEquals +import org.junit.Assert.assertFalse +import org.junit.Assert.assertSame +import org.junit.Assert.assertThrows +import org.junit.Test + +class CoroutineFailuresTest { + @Test + fun `suspending result wrapper preserves success and ordinary failure`() = runBlocking { + assertEquals("ok", runSuspendCatching { "ok" }.getOrThrow()) + + val failure = IOException("offline") + assertSame(failure, runSuspendCatching { throw failure }.exceptionOrNull()) + } + + @Test + fun `suspending result wrapper rethrows cancellation`() { + val cancellation = CancellationException("superseded") + + val actual = assertThrows(CancellationException::class.java) { + runBlocking { + runSuspendCatching { throw cancellation } + } + } + + assertSame(cancellation, actual) + } + + @Test + fun `a canceled context cannot publish a successful result`() { + val cancellation = CancellationException("target changed") + var published = false + + assertThrows(CancellationException::class.java) { + runBlocking { + runSuspendCatching { + currentCoroutineContext().cancel(cancellation) + "stale" + }.onSuccess { published = true } + } + } + + assertFalse(published) + } +} diff --git a/app/src/test/java/app/getarcane/android/core/DashboardStreamStoreTest.kt b/app/src/test/java/app/getarcane/android/core/DashboardStreamStoreTest.kt index 50bd3dd..f49cdba 100644 --- a/app/src/test/java/app/getarcane/android/core/DashboardStreamStoreTest.kt +++ b/app/src/test/java/app/getarcane/android/core/DashboardStreamStoreTest.kt @@ -4,11 +4,17 @@ import app.getarcane.sdk.models.container.ContainerStatusCounts import app.getarcane.sdk.models.environment.Environment import app.getarcane.sdk.models.image.ImageUsageCounts import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.CoroutineStart +import kotlinx.coroutines.CompletableDeferred import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.async import kotlinx.coroutines.cancel +import kotlinx.coroutines.cancelAndJoin +import kotlinx.coroutines.awaitCancellation import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.flow import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.yield import org.junit.Assert.assertEquals import org.junit.Assert.assertFalse import org.junit.Assert.assertNull @@ -115,6 +121,188 @@ class DashboardStreamStoreTest { } } + @Test + fun newerRefreshResultCannotBeOverwrittenByOlderRequest() = runBlocking { + val scope = CoroutineScope(Dispatchers.Unconfined) + val client = ControlledSnapshotClient() + val store = DashboardStreamStore(scope) + try { + store.configure(client) + store.reconcile(listOf(Environment(id = "edge", name = "Edge", apiUrl = "", status = "active"))) + + val first = async(start = CoroutineStart.UNDISPATCHED) { store.refresh() } + val second = async(start = CoroutineStart.UNDISPATCHED) { store.refresh() } + yield() + assertEquals(2, client.requests.size) + + client.requests[1].complete(snapshot(running = 9)) + second.await() + client.requests[0].complete(snapshot(running = 1)) + first.await() + + assertEquals(9, store.statesByEnvironmentId.getValue("edge").snapshot?.containers?.counts?.runningContainers) + } finally { + store.stop() + scope.cancel() + } + } + + @Test + fun cancelledRefreshPropagatesWithoutPublishingAnError() = runBlocking { + val scope = CoroutineScope(Dispatchers.Unconfined) + val client = ControlledSnapshotClient() + val store = DashboardStreamStore(scope) + try { + store.configure(client) + store.reconcile(listOf(Environment(id = "edge", name = "Edge", apiUrl = "", status = "active"))) + + val refresh = async(start = CoroutineStart.UNDISPATCHED) { store.refresh() } + yield() + refresh.cancelAndJoin() + + val state = store.statesByEnvironmentId.getValue("edge") + assertFalse(state.streamError) + assertNull(state.errorMessage) + assertFalse(state.hasLoaded) + } finally { + store.stop() + scope.cancel() + } + } + + @Test + fun removedEnvironmentRejectsAnInFlightSnapshot() = runBlocking { + val scope = CoroutineScope(Dispatchers.Unconfined) + val client = ControlledSnapshotClient() + val store = DashboardStreamStore(scope) + try { + store.configure(client) + store.reconcile(listOf(Environment(id = "edge", name = "Edge", apiUrl = "", status = "active"))) + val refresh = async(start = CoroutineStart.UNDISPATCHED) { store.refresh() } + yield() + + store.reconcile(emptyList()) + client.requests.single().complete(snapshot(running = 7)) + refresh.await() + + assertTrue(store.statesByEnvironmentId.isEmpty()) + } finally { + store.stop() + scope.cancel() + } + } + + @Test + fun replacementClientRejectsAnInFlightSnapshotFromThePriorServer() = runBlocking { + val scope = CoroutineScope(Dispatchers.Unconfined) + val originalClient = ControlledSnapshotClient() + val store = DashboardStreamStore(scope) + try { + store.configure(originalClient) + store.reconcile(listOf(Environment(id = "edge", name = "Edge", apiUrl = "", status = "active"))) + val refresh = async(start = CoroutineStart.UNDISPATCHED) { store.refresh() } + yield() + + store.configure(AlternateDashboardStreamClient) + originalClient.requests.single().complete(snapshot(running = 7)) + refresh.await() + + val state = store.statesByEnvironmentId.getValue("edge") + assertTrue(state.loading) + assertFalse(state.hasLoaded) + assertNull(state.snapshot) + } finally { + store.stop() + scope.cancel() + } + } + + @Test + fun environmentRemovalCancelsItsStoreOwnedSnapshotJob() = runBlocking { + val scope = CoroutineScope(Dispatchers.Unconfined) + val client = CountingSnapshotClient() + val store = DashboardStreamStore(scope) + try { + store.configure(client) + store.start() + store.reconcile(listOf(Environment(id = "edge", name = "Edge", apiUrl = "", status = "active"))) + assertEquals(1, client.active) + + store.reconcile(emptyList()) + yield() + + assertEquals(0, client.active) + } finally { + store.stop() + scope.cancel() + } + } + + @Test + fun clientReplacementCancelsStoreOwnedSnapshotJobs() = runBlocking { + val scope = CoroutineScope(Dispatchers.Unconfined) + val client = CountingSnapshotClient() + val store = DashboardStreamStore(scope) + try { + store.configure(client) + store.start() + store.reconcile(listOf(Environment(id = "edge", name = "Edge", apiUrl = "", status = "active"))) + assertEquals(1, client.active) + + store.configure(AlternateDashboardStreamClient) + yield() + + assertEquals(0, client.active) + } finally { + store.stop() + scope.cancel() + } + } + + @Test + fun liveSnapshotCancelsTheRedundantStoreOwnedFallback() = runBlocking { + val scope = CoroutineScope(Dispatchers.Unconfined) + val client = CountingSnapshotClient() + val store = DashboardStreamStore(scope) + try { + store.configure(client) + store.start() + store.reconcile(listOf(Environment(id = "edge", name = "Edge", apiUrl = "", status = "active"))) + assertEquals(1, client.active) + + store.applyForTest(snapshotEvent("edge", running = 3, stopped = 0, images = 2)) + yield() + + assertEquals(0, client.active) + assertEquals(3, store.statesByEnvironmentId.getValue("edge").snapshot?.containers?.counts?.runningContainers) + } finally { + store.stop() + scope.cancel() + } + } + + @Test + fun reconnectReplacesTheSingleOwnedStreamJob() = runBlocking { + val scope = CoroutineScope(Dispatchers.Unconfined) + val client = CountingStreamClient() + val store = DashboardStreamStore(scope) + try { + store.configure(client) + store.start() + assertEquals(1, client.started) + assertEquals(1, client.active) + + store.retry() + + assertEquals(2, client.started) + assertEquals(1, client.active) + } finally { + store.stop() + scope.cancel() + } + assertEquals(0, client.active) + } + private fun snapshotEvent(environmentId: String, running: Int, stopped: Int, images: Int): DashboardStreamEvent = DashboardStreamEvent( type = "snapshot", @@ -144,6 +332,9 @@ class DashboardStreamStoreTest { error = "unreachable", errorCode = "unreachable", ) + + private fun snapshot(running: Int): DashboardSnapshot = + snapshotEvent("edge", running = running, stopped = 0, images = 0).snapshot!! } private object HangingDashboardStreamClient : DashboardStreamClient { @@ -159,3 +350,45 @@ private object AlternateDashboardStreamClient : DashboardStreamClient { override suspend fun snapshot(environmentId: String): DashboardSnapshot = error("snapshot should not be called") } + +private class ControlledSnapshotClient : DashboardStreamClient { + val requests = mutableListOf>() + + override fun stream(): Flow = flow { awaitCancellation() } + + override suspend fun snapshot(environmentId: String): DashboardSnapshot = + CompletableDeferred().also(requests::add).await() +} + +private class CountingStreamClient : DashboardStreamClient { + var started = 0 + var active = 0 + + override fun stream(): Flow = flow { + started++ + active++ + try { + awaitCancellation() + } finally { + active-- + } + } + + override suspend fun snapshot(environmentId: String): DashboardSnapshot = + error("snapshot should not be called") +} + +private class CountingSnapshotClient : DashboardStreamClient { + var active = 0 + + override fun stream(): Flow = flow { awaitCancellation() } + + override suspend fun snapshot(environmentId: String): DashboardSnapshot { + active++ + try { + awaitCancellation() + } finally { + active-- + } + } +} diff --git a/app/src/test/java/app/getarcane/android/ui/screens/updates/UpdaterRunScreenTest.kt b/app/src/test/java/app/getarcane/android/ui/screens/updates/UpdaterRunScreenTest.kt index 02bfb65..5d546ee 100644 --- a/app/src/test/java/app/getarcane/android/ui/screens/updates/UpdaterRunScreenTest.kt +++ b/app/src/test/java/app/getarcane/android/ui/screens/updates/UpdaterRunScreenTest.kt @@ -1,11 +1,28 @@ package app.getarcane.android.ui.screens.updates import app.getarcane.sdk.errors.ArcaneError +import kotlinx.coroutines.CancellationException +import kotlinx.coroutines.runBlocking import org.junit.Assert.assertEquals +import org.junit.Assert.assertSame +import org.junit.Assert.assertThrows import org.junit.Assert.assertTrue import org.junit.Test class UpdaterRunScreenTest { + @Test + fun updaterRequestCancellationIsNotConvertedToAnOutcome() { + val cancellation = CancellationException("screen left") + + val actual = assertThrows(CancellationException::class.java) { + runBlocking { + runUpdaterRequestCatching { throw cancellation } + } + } + + assertSame(cancellation, actual) + } + @Test fun transportFailureBeforeServerStartsRemainsFailure() { val phase = updaterRunFailurePhase( diff --git a/docs/coroutine-stream-ownership.md b/docs/coroutine-stream-ownership.md new file mode 100644 index 0000000..25e9294 --- /dev/null +++ b/docs/coroutine-stream-ownership.md @@ -0,0 +1,48 @@ +# Coroutine and stream ownership + +This is the durable PAR-008 audit record for long-lived Android work. Its scope is stores, live +streams, and screen operations that continue long enough to cross a server, environment, or screen +change. Ordinary request/response actions remain owned by their calling Compose scope and are not +additional stream owners. + +## Ownership inventory + +| Operation | Owner | Replacement and cancellation rule | +| --- | --- | --- | +| Authentication, capability, and environment session work | `ArcaneClientManager` session job and scope | A new server/session advances the client generation and cancels the prior session work. | +| Dashboard aggregate stream | One `DashboardStreamStore` stream job | Stop, reconnect, or client replacement invalidates the generation and cancels the old job before starting another. | +| Dashboard per-environment snapshot fallback | One store-owned job per newly discovered environment, plus caller-owned explicit refresh children | Environment removal and store/client shutdown cancel store-owned work. Request tokens reject older concurrent responses and any response invalidated by a stream event, environment removal, or client replacement. | +| Dashboard live system stats | `LaunchedEffect(client, enabled environments, refresh, lifecycle restart)` | A key or lifecycle change cancels the entire structured child set before replacement. | +| Activity Center loading | Latest calling job, registered by `ActivityCenterStore` | A newer load or client replacement cancels the prior load; generation checks prevent late bucket publication. Canceled paging restores the previous requested limit. | +| Activity Center live events | One store-owned job per enabled environment | Reload reconciles and replaces the set. Client change, capability loss, or screen disposal cancels every prior job; job identity prevents an old `finally` block from removing its replacement. | +| Container logs and stats, project logs, and project streaming actions | A target-keyed `LaunchedEffect` | Client, environment, resource, or action changes cancel the old collector and clear target-specific output before starting the replacement. | +| Container terminal | One target/shell/retry-keyed `LaunchedEffect` | The effect owns both the terminal session and output collection. Its exact session is closed in `NonCancellable`; an old cleanup cannot close or reset a replacement session. | +| Image pull and upload streams | One remembered job in the sheet's composition scope | The action disables duplicate starts. Sheet disposal, client replacement, or environment change cancels the current job. | +| Updater run and status polling | One target-keyed `LaunchedEffect` with one structured request child | Leaving or changing the target cancels request and probes; cancellation is never mapped to a run outcome or polling failure. | +| Demo heartbeat | One `DemoService` job in the manager-owned session scope | Starting a heartbeat first cancels the previous job; ending or replacing the session stops it. | + +## Cancellation and stale-result rules + +- A broad catch around suspending work must rethrow `CancellationException`. Best-effort suspending + requests use `runSuspendCatching`; ordinary `runCatching` remains appropriate only for + non-suspending parsing, platform calls, and cleanup that cannot consume coroutine cancellation. +- A Compose stream key includes the client identity, environment identity, and resource/action + identity that determines the server request. +- Store-owned work uses generation, request-token, or exact-job identity checks before publishing + state. Cancellation is the primary stop mechanism; identity checks are the final defense against + transports that complete after cancellation. +- Cleanup acts on the resource captured by the owner. It must not read a newer shared session/job + and accidentally close or remove that replacement. +- Stream output is bounded by the screen's existing retention limit and cleared when its target + changes so content from one environment cannot appear under another. + +## Regression coverage + +- `CoroutineFailuresTest` proves the suspending result wrapper preserves success and ordinary + failures while rethrowing the original cancellation. +- `DashboardStreamStoreTest` covers canceled refresh, concurrent refresh ordering, reconnect with + one live owner, environment removal, client replacement, cancellation of store-owned snapshot + jobs, and rejection of stale snapshots. +- `CompleteListLoaderTest` retains paging cancellation coverage used by Activity Center's complete + environment discovery and the shared pagination boundary. +- `UpdaterRunScreenTest` proves cancellation cannot become an updater request outcome. diff --git a/docs/ios-parity-task-list.md b/docs/ios-parity-task-list.md index 8c8d315..7a26fe4 100644 --- a/docs/ios-parity-task-list.md +++ b/docs/ios-parity-task-list.md @@ -196,7 +196,7 @@ The standard checks are: - [x] **PAR-004 — Audit all complete-list call sites for silent pagination truncation** -- **Status:** Done +- **Status:** Complete - **Priority:** P0 - **Dependencies:** PAR-003 - **Scope:** Inventory every list call whose UI or calculation claims fleet-wide or complete @@ -274,18 +274,31 @@ The standard checks are: - [ ] Backup/restore behavior is checked on a supported emulator or documented platform test. - [ ] No machine-specific paths, secrets, or backup artifacts are committed. -- [ ] **PAR-008 — Audit coroutine cancellation and stream ownership** +- [x] **PAR-008 — Audit coroutine cancellation and stream ownership** -- **Status:** Ready +- **Status:** Complete - **Priority:** P0 - **Dependencies:** None - **Scope:** Find broad exception handling in stores and streams, rethrow `CancellationException`, and ensure environment/server/screen changes cancel the correct work. +- **Audit record:** [Coroutine and stream ownership](coroutine-stream-ownership.md) - **Acceptance criteria:** - - [ ] Broad catches no longer convert cancellation into user-visible failures or reconnect loops. - - [ ] Tests cover cancellation during refresh, paging, reconnect, and environment/server changes. - - [ ] At most one intended stream/job owner remains for each screen-level operation. - - [ ] No stale result from a canceled prior environment can overwrite current state. + - [x] Broad catches no longer convert cancellation into user-visible failures or reconnect loops. + - [x] Tests cover cancellation during refresh, paging, reconnect, and environment/server changes. + - [x] At most one intended stream/job owner remains for each screen-level operation. + - [x] No stale result from a canceled prior environment can overwrite current state. +- **Validation evidence (2026-08-21):** + - Source pins: Android base `b49f4d3c36b224b865d423a0febe93a26ca42689`, + libarcane-kotlin `991dfdc1ee747c171ebf1b5953fe5fb61ceadfb8`, and Arcane + `0fd8820822f49e2da25739306bc9bc401253fa9e`. + - Focused runs passed 33 tests across `CoroutineFailuresTest`, `DashboardStreamStoreTest`, + `CompleteListLoaderTest`, and `UpdaterRunScreenTest` (0 failures, 0 errors, 0 skipped), including + refresh, paging, reconnect, environment-removal, and client-replacement cancellation. + - `./gradlew :app:testDebugUnitTest :app:assembleDebug` passed all 143 unit tests and assembled the + debug APK; `git diff --check` passed. + - Michael's physical-device/live-server smoke test passed dashboard reconnect, environment + switching without stale stream content, Activity Center refresh and screen departure, container + Logs/Stats/Terminal departure and reopen, and recovery after force-stop. - [ ] **PAR-009 — Persist and apply Light/Dark/Auto appearance**