fix: close peer connections outside the engine lock (0.6.10)

app/build.gradle.kts

@@ -37,8 +37,8 @@ android {
         minSdk = 33
         targetSdk = 35
         // Remain newer than the 0.6.3 rendezvous candidate (code 26).
-        versionCode = 32
-        versionName = "0.6.9"
+        versionCode = 33
+        versionName = "0.6.10"
 
         testInstrumentationRunner = "androidx.test.runner.AndroidJUnitRunner"
 

app/src/main/kotlin/dev/forgesworn/kithmoot/media/LinkTable.kt

@@ -0,0 +1,58 @@
+package dev.forgesworn.kithmoot.media
+
+import java.util.concurrent.ConcurrentHashMap
+
+/**
+ * The engine's table of one link per remote device, readable from any thread.
+ *
+ * The engine mutates it under its own lock, so two reconciles cannot open one
+ * device twice. What this class exists for is the other direction: a libwebrtc
+ * callback, which arrives on the signalling thread, may ask "is this link still
+ * the device's current one" WITHOUT that lock.
+ *
+ * Closing a peer connection waits for the signalling thread. If the signalling
+ * thread is waiting for the engine lock, and the engine is holding that lock
+ * while it closes, neither can move, and five seconds later Android calls the
+ * app not responding. That was every freeze on 0.6.9: three traces, each with
+ * the main thread inside `PeerConnection.close` under the lock and the
+ * signalling thread "waiting to lock, held by thread 1". So identity reads here
+ * take no lock, and [reconcile] hands back what it removed so the engine can
+ * close it after the lock has gone.
+ */
+internal class LinkTable<L : Any> {
+    private val links = ConcurrentHashMap<String, L>()
+
+    operator fun get(device: String): L? = links[device]
+
+    /** Read from any thread, never under the engine lock. */
+    fun isCurrent(device: String, link: L): Boolean = links[device] === link
+
+    val devices: Set<String> get() = links.keys.toSet()
+
+    fun values(): List<L> = links.values.toList()
+
+    fun snapshot(): List<Pair<String, L>> = links.entries.map { it.key to it.value }
+
+    fun put(device: String, link: L) {
+        links[device] = link
+    }
+
+    fun remove(device: String): L? = links.remove(device)
+
+    /** Empties the table and hands back what was in it, for closing outside the lock. */
+    fun clear(): List<L> {
+        val all = values()
+        links.clear()
+        return all
+    }
+
+    /**
+     * Applies a roster: opens, under [lock], every device not yet linked, and
+     * removes every link whose device has gone. The removed links are handed
+     * back, not closed: the caller closes them once it has released [lock].
+     */
+    fun reconcile(lock: Any, devices: Set<String>, open: (String) -> L): List<Pair<String, L>> = synchronized(lock) {
+        for (device in devices - links.keys) links[device] = open(device)
+        (links.keys - devices).mapNotNull { device -> links.remove(device)?.let { device to it } }
+    }
+}

app/src/main/kotlin/dev/forgesworn/kithmoot/media/WebRtcEngine.kt

@@ -105,9 +105,14 @@ class WebRtcEngine(
      * caller cancels this scope moments after asking for it. Anything launched
      * here would be cancelled before it ran, leaving every peer connection
      * alive behind a factory that had already gone.
+     *
+     * Two rules keep it from deadlocking against libwebrtc, whose callbacks
+     * arrive on its signalling thread and whose `close` waits for that thread:
+     * nothing that runs on a WebRTC callback takes this lock (see
+     * [LinkTable]), and a link is closed only after this lock is released.
      */
     private val lock = Any()
-    private val links = mutableMapOf<String, ManagedLink>()
+    private val links = LinkTable<ManagedLink>()
 
     /**
      * Which remote devices this device's media may be sent to.
@@ -156,15 +161,21 @@ class WebRtcEngine(
 
     fun setCallActive(active: Boolean) {
         synchronized(lock) { callActive = active }
-        if (active) reconcile(session.remoteDevices.value) else stop()
+        if (active) scope.launch(MediaDispatcher) { reconcile(session.remoteDevices.value) } else stop()
     }
 
+    /**
+     * Everything here runs on [MediaDispatcher], never on the caller's thread.
+     * Opening and closing a peer connection blocks until libwebrtc's signalling
+     * thread has done it, and a roster change used to do that on the main
+     * thread, which is where the input timeout is measured.
+     */
     fun start() {
         // The set of devices to connect to is derived from the roster, so a
         // device that joins, leaves or lapses is reconciled here rather than
         // being handled as an event.
-        scope.launch { session.remoteDevices.collect { reconcile(it) } }
-        scope.launch {
+        scope.launch(MediaDispatcher) { session.remoteDevices.collect { reconcile(it) } }
+        scope.launch(MediaDispatcher) {
             session.signals.collect { signal ->
                 val target = managedLinkFor(signal.from) ?: return@collect
                 try {
@@ -173,33 +184,31 @@ class WebRtcEngine(
                 catch (failure: Exception) {
                     // A signal can finish after its link was deliberately
                     // replaced. That old link's exception must not relabel the
-                    // replacement as failed. Check identity and publish the
-                    // state while holding the same lock used for replacement.
-                    synchronized(lock) {
-                        if (callActive && links[signal.from] === target) {
-                            Log.w("KithMootMedia", "signal failed peer=${signal.from.take(8)}", failure)
-                            _connections.update { it + (signal.from to "failed") }
-                        }
+                    // replacement as failed: publish only while this is still
+                    // the device's current link.
+                    if (callActive && links.isCurrent(signal.from, target)) {
+                        Log.w("KithMootMedia", "signal failed peer=${signal.from.take(8)}", failure)
+                        _connections.update { it + (signal.from to "failed") }
                     }
                 }
             }
         }
-        scope.launch { localMedia.tracks.collect { onLocalTracksChanged(it) } }
-        scope.launch {
+        scope.launch(MediaDispatcher) { localMedia.tracks.collect { onLocalTracksChanged(it) } }
+        scope.launch(MediaDispatcher) {
             while (isActive) {
                 delay(10_000)
-                synchronized(lock) { links.values.forEach { it.reportMediaProgress() } }
+                synchronized(lock) { links.values().forEach { it.reportMediaProgress() } }
             }
         }
         // One statistics sample per profile-2 pair every two seconds, which is
         // the whole input to the health ladder. Whether packets are moving is
         // the only honest evidence that a direction is alive: a connection can
         // be `connected`, the signalling quiet, every object healthy, and one
         // direction carrying nothing at all.
-        scope.launch {
+        scope.launch(MediaDispatcher) {
             while (isActive) {
                 delay(HEALTH_SAMPLE_MS)
-                val sampling = synchronized(lock) { links.values.filter { it.profileTwo } }
+                val sampling = synchronized(lock) { links.values().filter { it.profileTwo } }
                 for (link in sampling) runCatching { link.sampleHealth() }
             }
         }
@@ -230,27 +239,30 @@ class WebRtcEngine(
     }
 
     private fun closeLinks() {
-        val closing = synchronized(lock) {
-            val all = links.values.toList()
-            links.clear()
-            all
-        }
+        val closing = synchronized(lock) { links.clear() }
         for (link in closing) runCatching { link.close() }
         _remoteTracks.value = emptyList()
         _connections.value = emptyMap()
     }
 
-    private fun reconcile(devices: Set<String>) = synchronized(lock) {
-        if (!callActive) return@synchronized
-        for (device in devices - links.keys) links[device] = openLink(device)
-        for (device in links.keys - devices) {
-            links.remove(device)?.close()
+    /**
+     * Opened under the lock, so two reconciles cannot open one device twice;
+     * closed after it, because a close waits for the signalling thread, and
+     * the signalling thread must never find this lock held while it waits.
+     */
+    private fun reconcile(devices: Set<String>) {
+        val closing = synchronized(lock) {
+            if (!callActive) return
+            links.reconcile(lock, devices) { openLink(it) }
+        }
+        for ((device, link) in closing) {
+            runCatching { link.close() }
             _connections.update { it - device }
             _remoteTracks.update { current -> current.filterNot { it.device == device } }
         }
     }
 
-    private fun managedLinkFor(device: String): ManagedLink? = synchronized(lock) { links[device] }
+    private fun managedLinkFor(device: String): ManagedLink? = links[device]
 
     /**
      * Throw a pair's connection away and open another.
@@ -262,11 +274,16 @@ class WebRtcEngine(
      * rather than opening a generation of its own and glaring with it.
      */
     private fun rebuildLink(device: String, gen: Long, open: Boolean) {
-        synchronized(lock) {
-            if (!callActive || device !in links) return
-            links.remove(device)?.let { runCatching { it.close() } }
-            links[device] = openLink(device, gen, open)
+        val old = synchronized(lock) {
+            val current = links[device]
+            if (!callActive || current == null) return
+            links.remove(device)
+            links.put(device, openLink(device, gen, open))
+            current
         }
+        // Outside the lock: see `reconcile`. The old link's callbacks are
+        // already ignored, because it is no longer the device's current link.
+        runCatching { old.close() }
         _remoteTracks.update { current -> current.filterNot { it.device == device } }
     }
 
@@ -279,11 +296,14 @@ class WebRtcEngine(
      * it was addressed to a session the far end no longer has.
      */
     private fun reopenAsProfileOne(device: String) {
-        synchronized(lock) {
-            if (!callActive || device !in links) return
-            links.remove(device)?.let { runCatching { it.close() } }
-            links[device] = openLink(device, profileTwo = false)
+        val old = synchronized(lock) {
+            val current = links[device]
+            if (!callActive || current == null) return
+            links.remove(device)
+            links.put(device, openLink(device, profileTwo = false))
+            current
         }
+        runCatching { old.close() }
         _remoteTracks.update { current -> current.filterNot { it.device == device } }
     }
 
@@ -359,7 +379,7 @@ class WebRtcEngine(
         audience = rule
         val tracks = localMedia.tracks.value
         synchronized(lock) {
-            for ((device, link) in links) link.syncLocalTracks(tracksFor(device, tracks))
+            for ((device, link) in links.snapshot()) link.syncLocalTracks(tracksFor(device, tracks))
         }
     }
 
@@ -408,7 +428,7 @@ class WebRtcEngine(
         // byte-identical for everyone who never mutes.
         session.setTracks(tracks.map { TrackRef(it.trackId, it.role, if (it.muted) true else null) })
         synchronized(lock) {
-            for ((device, link) in links) link.syncLocalTracks(tracksFor(device, tracks))
+            for ((device, link) in links.snapshot()) link.syncLocalTracks(tracksFor(device, tracks))
         }
     }
 
@@ -565,15 +585,19 @@ class WebRtcEngine(
             }
         }
 
-        /** Publish state only while this is still the device's current link.
+        /**
+         * Publish state only while this is still the device's current link.
          * Native WebRTC callbacks can arrive after close/rebuild; without the
          * identity check an old callback can overwrite the new link's state.
+         *
+         * Never under the engine lock. This runs on libwebrtc's signalling
+         * thread, and closing a connection waits for that thread: taking the
+         * lock here while the engine held it through a close was the deadlock
+         * behind every freeze on 0.6.9.
          */
         private fun updateConnectionState(state: String) {
-            synchronized(lock) {
-                if (!closed && callActive && links[device] === this) {
-                    _connections.update { it + (device to state) }
-                }
+            if (!closed && callActive && links.isCurrent(device, this)) {
+                _connections.update { it + (device to state) }
             }
         }
 

app/src/test/kotlin/dev/forgesworn/kithmoot/media/LinkTableTest.kt

@@ -0,0 +1,80 @@
+package dev.forgesworn.kithmoot.media
+
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.TimeUnit
+import kotlin.concurrent.thread
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertFalse
+import kotlin.test.assertNotNull
+import kotlin.test.assertNull
+import kotlin.test.assertTrue
+
+/**
+ * The lock discipline that keeps the engine and libwebrtc's signalling thread
+ * from waiting on each other. Three freezes on 0.6.9 had the same shape: the
+ * main thread inside a peer-connection close under the engine lock, and the
+ * signalling thread waiting for that lock to publish a state.
+ */
+class LinkTableTest {
+    private class Link(val device: String)
+
+    @Test fun `opens under the lock and hands back what is gone for closing outside it`() {
+        val lock = Any()
+        val table = LinkTable<Link>()
+        var openedUnderLock = true
+        table.reconcile(lock, setOf("a", "b")) { device ->
+            if (!Thread.holdsLock(lock)) openedUnderLock = false
+            Link(device)
+        }
+        assertTrue(openedUnderLock, "a link is opened under the lock, so a device is never opened twice")
+        assertEquals(setOf("a", "b"), table.devices)
+        val b = assertNotNull(table["b"])
+
+        val gone = table.reconcile(lock, setOf("b")) { Link(it) }
+        assertFalse(Thread.holdsLock(lock), "the lock is released before the caller closes anything")
+        assertEquals(listOf("a"), gone.map { it.first })
+        assertNull(table["a"])
+        assertTrue(table.isCurrent("b", b))
+        assertFalse(table.isCurrent("a", gone.single().second))
+    }
+
+    @Test fun `a state read never waits for the engine lock`() {
+        val lock = Any()
+        val table = LinkTable<Link>()
+        val a = Link("a")
+        table.put("a", a)
+        val read = CountDownLatch(1)
+        var current = false
+        var completed = false
+        val engine = thread {
+            synchronized(lock) {
+                // The worst case, and the shape of every 0.6.9 freeze: the
+                // engine holds its lock and waits for the signalling thread.
+                val signalling = thread {
+                    current = table.isCurrent("a", a)
+                    read.countDown()
+                }
+                completed = read.await(5, TimeUnit.SECONDS)
+                signalling.join()
+            }
+        }
+        engine.join()
+        assertTrue(completed, "the callback waited for the engine lock")
+        assertTrue(current)
+    }
+
+    @Test fun `clear hands everything back and leaves nothing current`() {
+        val lock = Any()
+        val table = LinkTable<Link>()
+        val a = Link("a")
+        val b = Link("b")
+        table.put("a", a)
+        table.put("b", b)
+        val all = synchronized(lock) { table.clear() }
+        assertEquals(setOf(a, b), all.toSet())
+        assertTrue(table.devices.isEmpty())
+        assertFalse(table.isCurrent("a", a))
+        assertEquals(emptyList(), table.snapshot())
+    }
+}

docs/android-release.md

@@ -2,6 +2,25 @@
 
 The public website currently offers production-signed 0.6.7 (30), Android 13 or later. Signing, publication and physical acceptance are recorded separately.
 
+## 0.6.10 the freeze on calls
+
+Version code 33 fixes the "KithMoot isn't responding" that 0.6.9 produced three
+times in two days on the owner's Pixel 10 Pro XL, always during or around a
+call. All three system traces show the same deadlock: a roster change ran the
+engine's reconcile on the main thread, which closed the departed device's peer
+connection while holding the engine lock; that close waits for libwebrtc's
+signalling thread, and the signalling thread was delivering an ICE state
+callback that took the same lock. Now nothing that runs on a WebRTC callback
+takes the engine lock, a link is closed only after the lock is released, and
+the engine's collectors run off the main thread. `LinkTableTest` pins the two
+rules. It also carries the epoch request admission proof (PR #70), which the
+web keeper deployed on 22 September now requires: without this build an
+Android member who misses a removal in a keeper room cannot catch up.
+
+Not changed here, and worth knowing while judging a hot phone: the camera is
+captured at 1280 by 720 at 30 frames a second and every remote peer gets its own
+encoder with no bitrate, resolution or framerate cap.
+
 ## 0.6.8 sharing and listening candidate
 
 Version code 31 adds app playback capture during screen sharing, preserves an