From e4c9881f361345ca3ddf1f7a0c62ec10f0d7616e Mon Sep 17 00:00:00 2001 From: Taksh Date: Mon, 5 Oct 2026 16:16:07 +0530 Subject: [PATCH] Keep busy peer queues serialized during actor eviction --- .../bitchat/android/mesh/PacketProcessor.kt | 114 +++++++++--------- .../mesh/PacketProcessorActorBoundsTest.kt | 71 ++++++++++- docs/client-rewrite-contracts.md | 9 ++ 3 files changed, 131 insertions(+), 63 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 f7740d9e..88416136 100644 --- a/app/src/main/java/com/bitchat/android/mesh/PacketProcessor.kt +++ b/app/src/main/java/com/bitchat/android/mesh/PacketProcessor.kt @@ -15,7 +15,10 @@ import kotlinx.coroutines.channels.actor * Prevents race condition where multiple threads process packets * from the same peer simultaneously, causing session management conflicts. */ -class PacketProcessor(private val myPeerID: String) { +class PacketProcessor( + private val myPeerID: String, + dispatcher: CoroutineDispatcher = Dispatchers.IO +) { private val debugManager by lazy { try { com.bitchat.android.ui.debug.DebugSettingsManager.getInstance() } catch (e: Exception) { null } } companion object { @@ -29,7 +32,8 @@ class PacketProcessor(private val myPeerID: String) { * 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. + * least-recently-used idle actors. Busy peers keep their queues until + * all accepted packets finish. */ internal const val MAX_PEER_ACTORS = 128 } @@ -37,59 +41,39 @@ class PacketProcessor(private val myPeerID: String) { // Delegate for callbacks var delegate: PacketProcessorDelegate? = null - // Helper function to format peer ID with nickname for logging - private fun formatPeerForLog(peerID: String): String { - val nickname = delegate?.getPeerNickname(peerID) - return if (nickname != null) "$peerID ($nickname)" else peerID - } - // Packet relay manager for centralized relay decisions private val packetRelayManager = PacketRelayManager(myPeerID) // Coroutines - private val processorScope = CoroutineScope(Dispatchers.IO + SupervisorJob()) + private val processorScope = CoroutineScope(dispatcher + SupervisorJob()) - // Per-peer actors to serialize packet processing - // Each peer gets its own actor that processes packets sequentially - // This prevents race conditions in session management - @OptIn(ObsoleteCoroutinesApi::class) - private fun getOrCreateActorForPeer(peerID: String) = processorScope.actor( - capacity = Channel.UNLIMITED - ) { - for (packet in channel) { - handleReceivedPacket(packet) - } + private class PeerActor { + lateinit var channel: kotlinx.coroutines.channels.SendChannel + // Accepted packets, including the one currently executing; guarded by actorsLock. + var pending = 0 } - - // 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 = - object : LinkedHashMap>( - 16, 0.75f, true - ) { - override fun removeEldestEntry( - eldest: MutableMap.MutableEntry> - ): 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 + private var acceptingPackets = true + private val actors = LinkedHashMap(16, 0.75f, true) + + @OptIn(ObsoleteCoroutinesApi::class) + private fun createPeerActor(): PeerActor = PeerActor().also { state -> + state.channel = processorScope.actor(capacity = Channel.UNLIMITED) { + for (packet in channel) { + try { + handleReceivedPacket(packet) + } catch (cancelled: CancellationException) { + throw cancelled + } catch (error: Exception) { + Log.w(TAG, "Packet handler failed", error) + } finally { + synchronized(actorsLock) { state.pending-- } + } } } - + } + init { // Set up the packet relay manager delegate immediately setupRelayManager() @@ -107,19 +91,28 @@ class PacketProcessor(private val myPeerID: String) { return } - // Get or create actor for this peer - val actor = synchronized(actorsLock) { - actors.getOrPut(peerID) { getOrCreateActorForPeer(peerID) } - } - - // Send packet to peer's dedicated actor for serialized processing - processorScope.launch { - try { - actor.send(routed) - } catch (e: Exception) { - Log.w(TAG, "Failed to send packet to actor for ${formatPeerForLog(peerID)}: ${e.message}") - // Fallback to direct processing if actor fails - handleReceivedPacket(routed) + synchronized(actorsLock) { + if (!acceptingPackets || !processorScope.isActive) return + val actor = actors[peerID] ?: run { + if (actors.size >= MAX_PEER_ACTORS) { + val idle = actors.entries.firstOrNull { it.value.pending == 0 } + if (idle == null) { + // Keep the memory bound without processing a busy peer on + // a second actor. Transport retries may admit this peer later. + Log.w(TAG, "Dropping new peer packet while all actor slots are busy") + return + } + actors.remove(idle.key) + idle.value.channel.close() + } + createPeerActor().also { actors[peerID] = it } + } + actor.pending++ + if (actor.channel.trySend(routed).isFailure) { + actor.pending-- + // Never fall back to direct processing: it can overlap this + // peer's already accepted packets (including during shutdown). + Log.w(TAG, "Peer actor no longer accepts packets") } } } @@ -308,11 +301,12 @@ class PacketProcessor(private val myPeerID: String) { */ fun shutdown() { synchronized(actorsLock) { + acceptingPackets = false Log.d(TAG, "Shutting down PacketProcessor and ${actors.size} peer actors") // Close all peer actors gracefully actors.values.forEach { actor -> - actor.close() + actor.channel.close() } actors.clear() } diff --git a/app/src/test/kotlin/com/bitchat/android/mesh/PacketProcessorActorBoundsTest.kt b/app/src/test/kotlin/com/bitchat/android/mesh/PacketProcessorActorBoundsTest.kt index bf0d8a82..dea5225c 100644 --- a/app/src/test/kotlin/com/bitchat/android/mesh/PacketProcessorActorBoundsTest.kt +++ b/app/src/test/kotlin/com/bitchat/android/mesh/PacketProcessorActorBoundsTest.kt @@ -4,6 +4,11 @@ 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 kotlinx.coroutines.CompletableDeferred +import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.test.StandardTestDispatcher +import kotlinx.coroutines.test.runCurrent +import kotlinx.coroutines.test.runTest import org.junit.After import org.junit.Assert.assertEquals import org.junit.Assert.assertTrue @@ -87,6 +92,64 @@ class PacketProcessorActorBoundsTest { assertEquals(1, processor.activePeerActorCount) } + @OptIn(ExperimentalCoroutinesApi::class) + @Test + fun `peer churn cannot overlap packets already being processed`() = runTest { + val peer = peerID(0) + val releaseFirst = CompletableDeferred() + var handled = 0 + val processor = PacketProcessor(MY_PEER_ID, StandardTestDispatcher(testScheduler)) + processors += processor + processor.delegate = object : NoopDelegate() { + override suspend fun handleAnnounce(routed: RoutedPacket): Boolean { + if (routed.peerID == peer) { + handled++ + if (handled == 1) releaseFirst.await() + } + return true + } + } + processor.processPacket(packetFrom(peer, MessageType.ANNOUNCE)) + runCurrent() + assertEquals(1, handled) + repeat(PacketProcessor.MAX_PEER_ACTORS) { index -> + processor.processPacket(packetFrom(peerID(index + 1), MessageType.ANNOUNCE)) + } + runCurrent() + processor.processPacket(packetFrom(peer, MessageType.ANNOUNCE)) + runCurrent() + assertEquals("The second packet must wait for the first", 1, handled) + releaseFirst.complete(Unit) + runCurrent() + assertEquals(2, handled) + assertTrue(processor.activePeerActorCount <= PacketProcessor.MAX_PEER_ACTORS) + } + + @OptIn(ExperimentalCoroutinesApi::class) + @Test + fun `a full busy pool admits a new peer after a slot becomes idle`() = runTest { + val newcomer = peerID(PacketProcessor.MAX_PEER_ACTORS + 1) + var newcomerPackets = 0 + val processor = PacketProcessor(MY_PEER_ID, StandardTestDispatcher(testScheduler)) + processors += processor + processor.delegate = object : NoopDelegate() { + override suspend fun handleAnnounce(routed: RoutedPacket): Boolean { + if (routed.peerID == newcomer) newcomerPackets++ + return true + } + } + repeat(PacketProcessor.MAX_PEER_ACTORS) { index -> + processor.processPacket(packetFrom(peerID(index), MessageType.ANNOUNCE)) + } + processor.processPacket(packetFrom(newcomer, MessageType.ANNOUNCE)) + runCurrent() + assertEquals(0, newcomerPackets) + processor.processPacket(packetFrom(newcomer, MessageType.ANNOUNCE)) + runCurrent() + assertEquals(1, newcomerPackets) + assertEquals(PacketProcessor.MAX_PEER_ACTORS, processor.activePeerActorCount) + } + @Test fun `shutdown releases every actor`() { val processor = PacketProcessor(MY_PEER_ID).also { it.delegate = NoopDelegate() } @@ -97,6 +160,8 @@ class PacketProcessorActorBoundsTest { processor.shutdown() assertEquals(0, processor.activePeerActorCount) + processor.processPacket(packetFrom(peerID(100))) + assertEquals(0, processor.activePeerActorCount) } private fun processor(): PacketProcessor = @@ -107,10 +172,10 @@ class PacketProcessorActorBoundsTest { private fun peerID(index: Int): String = String.format("%016x", index + 1) - private fun packetFrom(peerID: String): RoutedPacket { + private fun packetFrom(peerID: String, type: MessageType = MessageType.MESSAGE): RoutedPacket { val packet = BitchatPacket( version = 1u, - type = MessageType.MESSAGE.value, + type = type.value, senderID = peerID.hexToBytes(), recipientID = SpecialRecipients.BROADCAST, timestamp = 1u, @@ -120,7 +185,7 @@ class PacketProcessorActorBoundsTest { return RoutedPacket(packet, peerID, "direct-link") } - private class NoopDelegate : PacketProcessorDelegate { + private open class NoopDelegate : PacketProcessorDelegate { override fun validatePacketSecurity(packet: BitchatPacket, peerID: String) = true override fun updatePeerLastSeen(peerID: String) = Unit override fun getPeerNickname(peerID: String): String? = null diff --git a/docs/client-rewrite-contracts.md b/docs/client-rewrite-contracts.md index 1abed243..5cdfb6af 100644 --- a/docs/client-rewrite-contracts.md +++ b/docs/client-rewrite-contracts.md @@ -20,6 +20,15 @@ The remaining implementation work and milestone progress are tracked in | Nostr | Bech32, secp256k1 key derivation, NIP-01 event IDs/signatures, NIP-44 authenticated encryption, NIP-13 PoW, authenticated NIP-17 seals | `ClientRewriteNostrContractTest`, `NostrProtocolTest` | | Application state | Peer unions, canonical private conversations, chronological history, delivery/read behavior, media migration policy | `AppStateStoreTest`, `PrivateChatManagerTest`, `MediaSendingManagerMigrationTest` | +## Packet actor admission + +Each peer's accepted packets are processed sequentially. At the actor limit, +only an actor with no queued or executing packets may be evicted. When every +slot is busy, packets from new peers are dropped until a slot becomes idle; +existing peers retain their queues. Closing or rejecting an actor never falls +back to concurrent direct processing. `PacketProcessorActorBoundsTest` covers +peer churn while an earlier packet handler is suspended and shutdown rejection. + ## Golden-vector policy Golden vectors compare literal externally visible bytes or hashes. Do not update