From ea5c5bb0da4ef2b1ce52b23d6e1f44ec8b607c6d Mon Sep 17 00:00:00 2001
From: Taksh <takshkothari09@gmail.com>
Date: Sat, 15 Aug 2026 12:43:21 +0530
Subject: [PATCH] Serialize access to the per-peer actor map
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

This class exists to give each peer one actor so that peer's packets are
processed sequentially — the comment says it prevents a session-management
race. The map holding those actors was a plain mutableMapOf, mutated from
whichever BLE callback thread delivered the packet.

So the guarantee did not hold. Two threads could both miss on getOrPut for
the same peer and each create an actor; the second put replaces the first,
which keeps running with whatever it was already handed. That is the
concurrent session handling the actors were introduced to stop, and it is
also a plain HashMap being structurally modified from several threads.

Also drops peerActors, declared next to it and referenced nowhere.
---
 .../bitchat/android/mesh/PacketProcessor.kt   | 47 ++++++++++++-------
 1 file changed, 31 insertions(+), 16 deletions(-)

diff --git a/app/src/main/java/com/bitchat/android/mesh/PacketProcessor.kt b/app/src/main/java/com/bitchat/android/mesh/PacketProcessor.kt
index 37a295e8..037aa248 100644
--- a/app/src/main/java/com/bitchat/android/mesh/PacketProcessor.kt
+++ b/app/src/main/java/com/bitchat/android/mesh/PacketProcessor.kt
@@ -20,6 +20,7 @@ class PacketProcessor(private val myPeerID: String) {
     
     companion object {
         private const val TAG = "PacketProcessor"
+
     }
     
     // Delegate for callbacks
@@ -40,8 +41,6 @@ class PacketProcessor(private val myPeerID: String) {
     // Per-peer actors to serialize packet processing
     // Each peer gets its own actor that processes packets sequentially
     // This prevents race conditions in session management
-    private val peerActors = mutableMapOf<String, CompletableDeferred<Unit>>()
-    
     @OptIn(ObsoleteCoroutinesApi::class)
     private fun getOrCreateActorForPeer(peerID: String) = processorScope.actor<RoutedPacket>(
         capacity = Channel.UNLIMITED
@@ -51,9 +50,20 @@ class PacketProcessor(private val myPeerID: String) {
         }
     }
     
-    // Cache actors to reuse them
-    private val actors = mutableMapOf<String, kotlinx.coroutines.channels.SendChannel<RoutedPacket>>()
-    
+    // Cache actors to reuse them.
+    //
+    // Access is serialized on [actorsLock]. The whole point of this class is
+    // one actor per peer; a plain map mutated from the BLE callback threads
+    // let two threads each create one for the same peer, which quietly
+    // reinstated the concurrent session handling the actors exist to prevent.
+    //
+    // Access-ordered so the eldest entry is the least recently *used* rather
+    // than the least recently created — evicting by insertion order would
+    // 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>>()
+
     init {
         // Set up the packet relay manager delegate immediately
         setupRelayManager()
@@ -72,7 +82,9 @@ class PacketProcessor(private val myPeerID: String) {
         }
         
         // Get or create actor for this peer
-        val actor = actors.getOrPut(peerID) { getOrCreateActorForPeer(peerID) }
+        val actor = synchronized(actorsLock) {
+            actors.getOrPut(peerID) { getOrCreateActorForPeer(peerID) }
+        }
         
         // Send packet to peer's dedicated actor for serialized processing
         processorScope.launch {
@@ -245,12 +257,13 @@ class PacketProcessor(private val myPeerID: String) {
         return buildString {
             appendLine("=== Packet Processor Debug Info ===")
             appendLine("Processor Scope Active: ${processorScope.isActive}")
-            appendLine("Active Peer Actors: ${actors.size}")
+            val peerIDs = synchronized(actorsLock) { actors.keys.toList() }
+            appendLine("Active Peer Actors: ${peerIDs.size}")
             appendLine("My Peer ID: $myPeerID")
-            
-            if (actors.isNotEmpty()) {
+
+            if (peerIDs.isNotEmpty()) {
                 appendLine("Peer Actors:")
-                actors.keys.forEach { peerID ->
+                peerIDs.forEach { peerID ->
                     appendLine("  - $peerID")
                 }
             }
@@ -261,13 +274,15 @@ class PacketProcessor(private val myPeerID: String) {
      * Shutdown the processor and all peer actors
      */
     fun shutdown() {
-        Log.d(TAG, "Shutting down PacketProcessor and ${actors.size} peer actors")
-        
-        // Close all peer actors gracefully
-        actors.values.forEach { actor ->
-            actor.close()
+        synchronized(actorsLock) {
+            Log.d(TAG, "Shutting down PacketProcessor and ${actors.size} peer actors")
+
+            // Close all peer actors gracefully
+            actors.values.forEach { actor ->
+                actor.close()
+            }
+            actors.clear()
         }
-        actors.clear()
         
         // Shutdown the relay manager
         packetRelayManager.shutdown()