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..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,45 +15,65 @@ 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 { 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 idle actors. Busy peers keep their queues until + * all accepted packets finish. + */ + internal const val MAX_PEER_ACTORS = 128 } // 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()) - - // 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>() + private val processorScope = CoroutineScope(dispatcher + SupervisorJob()) + private class PeerActor { + lateinit var channel: kotlinx.coroutines.channels.SendChannel + // Accepted packets, including the one currently executing; guarded by actorsLock. + var pending = 0 + } + + private val actorsLock = Any() + private var acceptingPackets = true + private val actors = LinkedHashMap(16, 0.75f, true) + @OptIn(ObsoleteCoroutinesApi::class) - private fun getOrCreateActorForPeer(peerID: String) = processorScope.actor( - capacity = Channel.UNLIMITED - ) { - for (packet in channel) { - handleReceivedPacket(packet) + 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-- } + } + } } } - - // Cache actors to reuse them - private val actors = mutableMapOf>() - + init { // Set up the packet relay manager delegate immediately setupRelayManager() @@ -71,17 +91,28 @@ class PacketProcessor(private val myPeerID: String) { return } - // Get or create actor for this peer - val actor = 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") } } } @@ -238,6 +269,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 */ @@ -245,12 +283,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} (cap $MAX_PEER_ACTORS)") 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 +300,16 @@ 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) { + acceptingPackets = false + Log.d(TAG, "Shutting down PacketProcessor and ${actors.size} peer actors") + + // Close all peer actors gracefully + actors.values.forEach { actor -> + actor.channel.close() + } + actors.clear() } - actors.clear() // Shutdown the relay manager packetRelayManager.shutdown() diff --git a/app/src/test/kotlin/com/bitchat/android/mesh/PacketProcessorActorBoundsTest.kt b/app/src/test/kotlin/com/bitchat/android/mesh/PacketProcessorActorBoundsTest.kt new file mode 100644 index 00000000..dea5225c --- /dev/null +++ b/app/src/test/kotlin/com/bitchat/android/mesh/PacketProcessorActorBoundsTest.kt @@ -0,0 +1,213 @@ +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 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 +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() + + @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) + } + + @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() } + + repeat(10) { processor.processPacket(packetFrom(peerID(it))) } + assertEquals(10, processor.activePeerActorCount) + + processor.shutdown() + + assertEquals(0, processor.activePeerActorCount) + processor.processPacket(packetFrom(peerID(100))) + 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, type: MessageType = MessageType.MESSAGE): RoutedPacket { + val packet = BitchatPacket( + version = 1u, + type = type.value, + senderID = peerID.hexToBytes(), + recipientID = SpecialRecipients.BROADCAST, + timestamp = 1u, + payload = byteArrayOf(0x01), + ttl = 7u + ) + return RoutedPacket(packet, peerID, "direct-link") + } + + 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 + 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" + } +} diff --git a/docs/client-rewrite-contracts.md b/docs/client-rewrite-contracts.md index 05d5e1ad..d2839f67 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, 22h outbound envelope randomization | `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