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>() - @OptIn(ObsoleteCoroutinesApi::class) private fun getOrCreateActorForPeer(peerID: String) = processorScope.actor( capacity = Channel.UNLIMITED @@ -51,9 +50,20 @@ class PacketProcessor(private val myPeerID: String) { } } - // Cache actors to reuse them - private val actors = mutableMapOf>() - + // 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>() + 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()