fix(service): unwedge the inbound pipeline behind stale-Connected zombies (#6587)

Co-authored-by: Claude Opus 5 <noreply@anthropic.com>

Author: jamesarich

core/data/src/commonMain/kotlin/org/meshtastic/core/data/manager/MeshConnectionManagerImpl.kt

@@ -33,6 +33,7 @@ import org.meshtastic.core.common.di.ServiceScope
 import org.meshtastic.core.common.util.handledLaunch
 import org.meshtastic.core.common.util.nowMillis
 import org.meshtastic.core.common.util.nowSeconds
+import org.meshtastic.core.common.util.safeCatching
 import org.meshtastic.core.common.util.safeCatchingAll
 import org.meshtastic.core.model.ConnectionState
 import org.meshtastic.core.model.DeviceType
@@ -136,7 +137,14 @@ class MeshConnectionManagerImpl(
         // Bridge transport-level state into the canonical app-level state.
         // This is the ONLY consumer of RadioInterfaceService.connectionState — it applies
         // light-sleep policy and handshake awareness before writing to ServiceRepository.
-        radioInterfaceService.connectionState.onEach(::onRadioConnectionState).launchIn(scope)
+        // Guarded per-emission: one uncaught throw here would kill the sole bridge collector and
+        // permanently freeze the app-level state (a stuck-"Connected" UI no transport event can fix).
+        radioInterfaceService.connectionState
+            .onEach { state ->
+                safeCatching { onRadioConnectionState(state) }
+                    .onFailure { Logger.e(it) { "Connection state bridge failed for $state; collector kept alive" } }
+            }
+            .launchIn(scope)
 
         // Ensure notification title and content stay in sync with state changes
         serviceRepository.connectionState.onEach { updateStatusNotification() }.launchIn(scope)

core/service/src/commonMain/kotlin/org/meshtastic/core/service/SharedRadioInterfaceService.kt

@@ -27,10 +27,13 @@ import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.NonCancellable
 import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.TimeoutCancellationException
 import kotlinx.coroutines.cancel
 import kotlinx.coroutines.channels.BufferOverflow
 import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.currentCoroutineContext
 import kotlinx.coroutines.delay
+import kotlinx.coroutines.ensureActive
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.MutableStateFlow
@@ -50,6 +53,8 @@ import kotlinx.coroutines.launch
 import kotlinx.coroutines.sync.Mutex
 import kotlinx.coroutines.sync.withLock
 import kotlinx.coroutines.withContext
+import kotlinx.coroutines.withTimeout
+import kotlinx.coroutines.withTimeoutOrNull
 import okio.ByteString.Companion.toByteString
 import org.koin.core.annotation.Named
 import org.koin.core.annotation.Single
@@ -271,7 +276,23 @@ class SharedRadioInterfaceService(
     }
 
     override suspend fun runWhileSessionActive(session: RadioSessionContext, block: suspend () -> Unit): Boolean =
-        sessionOperationMutex.withLock { runWithSessionLease(session) { block() } }
+        sessionOperationMutex.withLock {
+            runWithSessionLease(session) {
+                // Bound the handler: it holds sessionOperationMutex (the whole inbound pipeline) and an admitted
+                // lease (which teardown's drain awaits), so an indefinite suspension here is a total wedge, not a
+                // slow packet. Cancelling the block releases both. Only OUR timeout is swallowed — ensureActive()
+                // rethrows if the surrounding scope was cancelled concurrently.
+                try {
+                    withTimeout(SESSION_HANDLER_TIMEOUT_MILLIS) { block() }
+                } catch (timeout: TimeoutCancellationException) {
+                    currentCoroutineContext().ensureActive()
+                    Logger.e(timeout) {
+                        "Session handler exceeded ${SESSION_HANDLER_TIMEOUT_MILLIS}ms and was cancelled; " +
+                            "dropping its packet to keep the receive pipeline alive"
+                    }
+                }
+            }
+        }
 
     /** Runs a callback only while [session] still owns admission, atomically with session teardown. */
     private inline fun runIfTransportSessionActive(session: RadioTransportSession, block: () -> Unit): Boolean =
@@ -299,7 +320,21 @@ class SharedRadioInterfaceService(
                         sessionDrainWaiter ?: CompletableDeferred<Unit>().also { sessionDrainWaiter = it }
                     }
                 }
-            drainWaiter?.await()
+            if (drainWaiter != null) {
+                // The drain must complete before a replacement generation is admitted (DB-atomicity contract), so we
+                // keep waiting — but never silently. A lease stuck past the handler timeout means a handler ignored
+                // cancellation; these error-level reports are the observability surface for that wedge (this wait
+                // previously blocked disconnect()/restart forever with no telemetry at all).
+                var waitedMillis = 0L
+                while (withTimeoutOrNull(DRAIN_WAIT_LOG_INTERVAL_MILLIS) { drainWaiter.await() } == null) {
+                    waitedMillis += DRAIN_WAIT_LOG_INTERVAL_MILLIS
+                    val outstanding = synchronized(sessionCallbackLock) { admittedSessionOperations }
+                    Logger.e {
+                        "Transport teardown blocked ${waitedMillis}ms waiting for $outstanding admitted session " +
+                            "operation(s) to release (generation=${session.generation})"
+                    }
+                }
+            }
             synchronized(sessionCallbackLock) {
                 if (activeTransportSession === session) {
                     check(admittedSessionOperations == 0) { "Session revoked before admitted operations drained" }
@@ -432,6 +467,18 @@ class SharedRadioInterfaceService(
          * flaky GATT connection. Serial and TCP typically flush well under this window.
          */
         private const val POLITE_DISCONNECT_DRAIN_MS = 500L
+
+        /**
+         * Ceiling on a single [runWhileSessionActive] handler. The block holds [sessionOperationMutex] — the whole
+         * inbound pipeline — so a handler that suspends indefinitely wedges packet processing, fills [_receivedData],
+         * and deadlocks teardown's lease drain (a Connected-looking zombie only a force-stop clears; observed in the
+         * field as multi-hour "receive queue at capacity" sessions). 2 minutes is far above any legitimate handler
+         * (large-mesh config DB installs run seconds) while still bounding the wedge.
+         */
+        private const val SESSION_HANDLER_TIMEOUT_MILLIS = 2 * 60 * 1000L
+
+        /** How often [revokeTransportSession] reports a lease drain that has not completed. */
+        private const val DRAIN_WAIT_LOG_INTERVAL_MILLIS = 15 * 1000L
     }
 
     private val initLock = Mutex()
@@ -979,7 +1026,6 @@ class SharedRadioInterfaceService(
     @Suppress("TooGenericExceptionCaught")
     private fun enqueueReceivedData(bytes: ByteArray, session: RadioTransportSession) {
         try {
-            lastDataReceivedMillis = now()
             // trySend synchronously onto the Channel so packet order matches arrival order. The
             // previous `launch { emit() }` pattern dispatched each packet onto a fresh coroutine,
             // letting the scheduler reorder them — which broke the firmware config handshake
@@ -993,6 +1039,12 @@ class SharedRadioInterfaceService(
             }
             val frame = ReceivedRadioFrame(payload = bytes.toByteString(), session = session.context)
             val result = _receivedData.trySend(frame)
+            if (result.isSuccess) {
+                // Stamp liveness only for frames actually admitted to the queue. Stamping on arrival kept
+                // checkLiveness() satisfied while a wedged consumer dropped every frame — a Connected-looking zombie
+                // the watchdog existed to catch. A full queue now reads as silence and trips liveness recovery.
+                lastDataReceivedMillis = now()
+            }
             if (result.isFailure) {
                 // Rate-limited on purpose: drops only happen under sustained inbound traffic, and Kermit forwards to
                 // Datadog/Crashlytics, so logging every drop would turn a bounded memory problem into unbounded

core/service/src/commonTest/kotlin/org/meshtastic/core/service/SharedRadioInterfaceServiceLivenessTest.kt

@@ -820,6 +820,132 @@ class SharedRadioInterfaceServiceLivenessTest {
         }
     }
 
+    /**
+     * Regression for the field wedge behind "app shows Connected but the node stops updating": frames dropped by a full
+     * receive queue must NOT feed the liveness timer. Before the fix, [SharedRadioInterfaceService] stamped
+     * `lastDataReceivedMillis` on arrival (even for dropped frames), so a wedged consumer kept liveness satisfied
+     * forever while every frame was discarded.
+     */
+    @Test
+    fun `frames dropped by a full receive queue do not reset the liveness timer`() = runTest(testDispatcher) {
+        clock = 0L
+        val service = createConnectedService("xAA:BB:CC:DD:EE:FF")
+        try {
+            // No collector attached: fill the channel to capacity at t=0. All of these are admitted
+            // and stamp liveness at 0.
+            val payload = byteArrayOf(1)
+            repeat(SharedRadioInterfaceService.RECEIVE_QUEUE_CAPACITY) { service.handleFromRadio(payload) }
+
+            // A frame arriving at t=30s is DROPPED (queue full). It must not count as liveness data.
+            clock = 30_000L
+            service.handleFromRadio(payload)
+
+            // At t=65s the silence is 65s if the drop was correctly ignored (fires), but only 35s if
+            // the drop stamped the timer (must not happen).
+            clock = 65_000L
+            service.checkLiveness()
+            testDispatcher.scheduler.runCurrent()
+            advanceTimeBy(1_000L)
+
+            assertTrue(
+                createdTransports.first().closeCalled,
+                "Liveness must fire on queue-full silence — dropped frames must not feed the timer",
+            )
+        } finally {
+            service.disconnect()
+            advanceTimeBy(1_000L)
+        }
+    }
+
+    // ─── Session handler timeout: the pipeline must not wedge forever ───────────────────────────
+
+    /**
+     * Regression for the 2.8.0 stale-connection wedge: a handler that suspends indefinitely inside
+     * [SharedRadioInterfaceService.runWhileSessionActive] holds the session-operation lane (the whole inbound
+     * pipeline). It must be cancelled at the handler timeout so queued work behind it can run.
+     */
+    @Test
+    fun `wedged session handler is cancelled at the timeout and the pipeline continues`() = runTest(testDispatcher) {
+        clock = 0L
+        val service = createConnectedService("xAA:BB:CC:DD:EE:FF")
+        val session = requireNotNull(service.activeSession.value)
+        val wedgeStarted = CompletableDeferred<Unit>()
+        val neverReleased = CompletableDeferred<Unit>()
+        var wedgeRanToCompletion = false
+
+        val wedged = launch {
+            service.runWhileSessionActive(session) {
+                wedgeStarted.complete(Unit)
+                neverReleased.await() // simulates a handler stuck on an unbounded suspension
+                wedgeRanToCompletion = true
+            }
+        }
+        wedgeStarted.await()
+        val nextStarted = CompletableDeferred<Unit>()
+        val next = launch { service.runWhileSessionActive(session) { nextStarted.complete(Unit) } }
+        try {
+            testDispatcher.scheduler.runCurrent()
+            assertFalse(nextStarted.isCompleted, "ordered work is serialized behind the wedged handler")
+
+            // Cross the 2-minute handler timeout: the wedged block is cancelled, releasing the lane.
+            advanceTimeBy(121_000L)
+            wedged.join()
+            next.join()
+
+            assertFalse(wedgeRanToCompletion, "the wedged handler must have been cancelled, not completed")
+            assertTrue(nextStarted.isCompleted, "the handler timeout must release the pipeline for queued work")
+        } finally {
+            neverReleased.complete(Unit)
+            wedged.cancel()
+            next.cancel()
+            service.disconnect()
+            advanceTimeBy(1_000L)
+        }
+    }
+
+    /**
+     * The user-facing half of the same wedge: disconnect() drains admitted leases before teardown, so a handler stuck
+     * forever previously made disconnect unreachable (only a force-stop recovered). The handler timeout must bound that
+     * wait.
+     */
+    @Test
+    fun `disconnect completes after a wedged handler is timed out`() = runTest(testDispatcher) {
+        clock = 0L
+        val service = createConnectedService("xAA:BB:CC:DD:EE:FF")
+        val session = requireNotNull(service.activeSession.value)
+        val wedgeStarted = CompletableDeferred<Unit>()
+        val neverReleased = CompletableDeferred<Unit>()
+
+        val wedged = launch {
+            service.runWhileSessionActive(session) {
+                wedgeStarted.complete(Unit)
+                neverReleased.await()
+            }
+        }
+        wedgeStarted.await()
+
+        val disconnectJob = launch { service.disconnect() }
+        try {
+            testDispatcher.scheduler.runCurrent()
+            assertFalse(disconnectJob.isCompleted, "disconnect must wait while the lease is admitted")
+
+            // Cross the handler timeout (cancels the wedged block, draining the lease) plus the
+            // polite-disconnect drain window inside stopTransportLocked.
+            advanceTimeBy(121_000L)
+            testDispatcher.scheduler.runCurrent()
+            advanceTimeBy(1_000L)
+            disconnectJob.join()
+            wedged.join()
+
+            assertNull(service.activeSession.value, "teardown must complete once the wedged lease is released")
+        } finally {
+            neverReleased.complete(Unit)
+            wedged.cancel()
+            service.disconnect()
+            advanceTimeBy(1_000L)
+        }
+    }
+
     @Test
     fun `USB permission denial emits error and permanent disconnected state`() = runTest(testDispatcher) {
         clock = 0L