Bound the per-peer actor map and evict least-recently-used

Each distinct sender ID allocated a coroutine and an unbounded channel that
were released only at shutdown. Peer IDs are ephemeral and rotate, and the
ID is taken straight off the wire, so the map grows without limit in
ordinary use and can be grown deliberately by anyone in radio range.

Cap it at 128 live actors. The map is access-ordered, so the entry dropped
is the peer least recently *used*: evicting by insertion order would
discard a long-lived active peer in favour of a burst of new IDs, which is
the case worth protecting. Eviction closes the channel rather than
cancelling it, so the actor drains what it already holds before finishing.

Two of the four tests fail on main. The concurrency one does not — that
race is timing-dependent — so it is there to hold the fix rather than to
demonstrate the bug.

app/src/main/java/com/bitchat/android/mesh/PacketProcessor.kt

@@ -21,6 +21,17 @@ class PacketProcessor(private val myPeerID: String) {
     companion object {
         private const val TAG = "PacketProcessor"
 
+        /**
+         * Live per-peer actors to retain.
+         *
+         * Every distinct sender ID used to allocate a coroutine and an
+         * unbounded channel that were only released at shutdown. Peer IDs are
+         * ephemeral and rotate, and the ID is read straight off the wire, so
+         * the set of keys grows without limit during ordinary use and can be
+         * grown deliberately by anyone in radio range. Bounded here, evicting
+         * least-recently-used, which is the peer least likely to be mid-session.
+         */
+        internal const val MAX_PEER_ACTORS = 128
     }
     
     // Delegate for callbacks
@@ -62,8 +73,23 @@ class PacketProcessor(private val myPeerID: String) {
     // drop a long-lived active peer in favour of a burst of new IDs.
     private val actorsLock = Any()
     private val actors =
-        mutableMapOf<String, kotlinx.coroutines.channels.SendChannel<RoutedPacket>>()
-
+        object : LinkedHashMap<String, kotlinx.coroutines.channels.SendChannel<RoutedPacket>>(
+            16, 0.75f, true
+        ) {
+            override fun removeEldestEntry(
+                eldest: MutableMap.MutableEntry<String, kotlinx.coroutines.channels.SendChannel<RoutedPacket>>
+            ): Boolean {
+                if (size <= MAX_PEER_ACTORS) return false
+                // Closing lets the actor drain what it already holds and then
+                // finish, rather than cancelling mid-packet.
+                eldest.value.close()
+                // Deliberately not formatPeerForLog: that reaches into the
+                // delegate, and this runs under actorsLock.
+                Log.d(TAG, "Evicting least-recently-used peer actor for ${eldest.key}")
+                return true
+            }
+        }
+    
     init {
         // Set up the packet relay manager delegate immediately
         setupRelayManager()
@@ -250,6 +276,13 @@ class PacketProcessor(private val myPeerID: String) {
 //        delegate?.handleDeliveryAck(routed)
 //    }
     
+    /**
+     * Number of live per-peer actors. Test seam; `getDebugInfo` reports the
+     * same number alongside the cap.
+     */
+    internal val activePeerActorCount: Int
+        get() = synchronized(actorsLock) { actors.size }
+
     /**
      * Get debug information
      */
@@ -258,7 +291,7 @@ class PacketProcessor(private val myPeerID: String) {
             appendLine("=== Packet Processor Debug Info ===")
             appendLine("Processor Scope Active: ${processorScope.isActive}")
             val peerIDs = synchronized(actorsLock) { actors.keys.toList() }
-            appendLine("Active Peer Actors: ${peerIDs.size}")
+            appendLine("Active Peer Actors: ${peerIDs.size} (cap $MAX_PEER_ACTORS)")
             appendLine("My Peer ID: $myPeerID")
 
             if (peerIDs.isNotEmpty()) {

app/src/test/kotlin/com/bitchat/android/mesh/PacketProcessorActorBoundsTest.kt

@@ -0,0 +1,148 @@
+package com.bitchat.android.mesh
+
+import com.bitchat.android.model.RoutedPacket
+import com.bitchat.android.protocol.BitchatPacket
+import com.bitchat.android.protocol.MessageType
+import com.bitchat.android.protocol.SpecialRecipients
+import org.junit.After
+import org.junit.Assert.assertEquals
+import org.junit.Assert.assertTrue
+import org.junit.Test
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.Executors
+import java.util.concurrent.TimeUnit
+
+/**
+ * The per-peer actor map is keyed by an ID read straight off the wire and was
+ * only ever emptied at shutdown, so it grew for the life of the process. It
+ * was also a plain map mutated from the BLE callback threads, which meant the
+ * one-actor-per-peer property this class exists to provide was not actually
+ * guaranteed.
+ */
+class PacketProcessorActorBoundsTest {
+    private val processors = mutableListOf<PacketProcessor>()
+
+    @After
+    fun tearDown() {
+        processors.forEach(PacketProcessor::shutdown)
+    }
+
+    @Test
+    fun `actor count stays at the cap however many peers are seen`() {
+        val processor = processor()
+
+        // Peer IDs are ephemeral and rotate, and nothing stops a peer in radio
+        // range from minting new ones, so "distinct senders" is unbounded.
+        repeat(PacketProcessor.MAX_PEER_ACTORS + 200) { index ->
+            processor.processPacket(packetFrom(peerID(index)))
+        }
+
+        assertEquals(PacketProcessor.MAX_PEER_ACTORS, processor.activePeerActorCount)
+    }
+
+    @Test
+    fun `a peer still being used is not evicted by a burst of new ones`() {
+        val processor = processor()
+        val busy = peerID(0)
+
+        processor.processPacket(packetFrom(busy))
+        repeat(PacketProcessor.MAX_PEER_ACTORS * 2) { index ->
+            processor.processPacket(packetFrom(peerID(index + 1)))
+            // Keep touching the long-lived peer. Evicting by insertion order
+            // would drop it anyway; access order is what saves it.
+            processor.processPacket(packetFrom(busy))
+        }
+
+        assertEquals(PacketProcessor.MAX_PEER_ACTORS, processor.activePeerActorCount)
+        assertTrue(
+            "The peer in active use must survive eviction",
+            processor.getDebugInfo().contains(busy)
+        )
+    }
+
+    @Test
+    fun `concurrent packets from one peer create exactly one actor`() {
+        val processor = processor()
+        val peer = peerID(7)
+        val threads = 16
+        val pool = Executors.newFixedThreadPool(threads)
+        val start = CountDownLatch(1)
+        val done = CountDownLatch(threads)
+
+        repeat(threads) {
+            pool.execute {
+                start.await()
+                repeat(50) { processor.processPacket(packetFrom(peer)) }
+                done.countDown()
+            }
+        }
+        start.countDown()
+        assertTrue(done.await(10, TimeUnit.SECONDS))
+        pool.shutdown()
+
+        // Two actors for one peer would process that peer's packets in
+        // parallel — exactly the session race the actors were introduced to
+        // stop. Timing-dependent, so this locks the fix in rather than
+        // reproducing the failure on demand.
+        assertEquals(1, processor.activePeerActorCount)
+    }
+
+    @Test
+    fun `shutdown releases every actor`() {
+        val processor = PacketProcessor(MY_PEER_ID).also { it.delegate = NoopDelegate() }
+
+        repeat(10) { processor.processPacket(packetFrom(peerID(it))) }
+        assertEquals(10, processor.activePeerActorCount)
+
+        processor.shutdown()
+
+        assertEquals(0, processor.activePeerActorCount)
+    }
+
+    private fun processor(): PacketProcessor =
+        PacketProcessor(MY_PEER_ID).also {
+            it.delegate = NoopDelegate()
+            processors += it
+        }
+
+    private fun peerID(index: Int): String = String.format("%016x", index + 1)
+
+    private fun packetFrom(peerID: String): RoutedPacket {
+        val packet = BitchatPacket(
+            version = 1u,
+            type = MessageType.MESSAGE.value,
+            senderID = peerID.hexToBytes(),
+            recipientID = SpecialRecipients.BROADCAST,
+            timestamp = 1u,
+            payload = byteArrayOf(0x01),
+            ttl = 7u
+        )
+        return RoutedPacket(packet, peerID, "direct-link")
+    }
+
+    private class NoopDelegate : PacketProcessorDelegate {
+        override fun validatePacketSecurity(packet: BitchatPacket, peerID: String) = true
+        override fun updatePeerLastSeen(peerID: String) = Unit
+        override fun getPeerNickname(peerID: String): String? = null
+        override fun getNetworkSize() = 1
+        override fun getBroadcastRecipient(): ByteArray = SpecialRecipients.BROADCAST
+        override fun handleNoiseHandshake(routed: RoutedPacket) = true
+        override fun handleNoiseEncrypted(routed: RoutedPacket) = true
+        override suspend fun handleAnnounce(routed: RoutedPacket) = true
+        override fun handleMessage(routed: RoutedPacket) = Unit
+        override fun handleLeave(routed: RoutedPacket) = Unit
+        override fun handleFragment(packet: BitchatPacket): BitchatPacket? = null
+        override fun handleRequestSync(routed: RoutedPacket) = Unit
+        override fun sendAnnouncementToPeer(peerID: String) = Unit
+        override fun sendCachedMessages(peerID: String) = Unit
+        override fun relayPacket(routed: RoutedPacket) = Unit
+        override fun sendToPeer(peerID: String, routed: RoutedPacket) = false
+    }
+
+    private fun String.hexToBytes(): ByteArray =
+        chunked(2).map { it.toInt(16).toByte() }.toByteArray()
+
+    private companion object {
+        const val MY_PEER_ID = "1111222233334444"
+    }
+}