nik/bitchat-android
1
0
Fork 0
mirror of https://github.com/permissionlesstech/bitchat-android.git synced 2026-10-25 07:40:36 +00:00
Code Issues Packages Projects Releases Wiki Activity

Keep busy peer queues serialized during actor eviction

Browse Source
This commit is contained in:
Taksh 2026-10-05 16:16:07 +05:30
parent bf8c0c0e8a
commit e4c9881f36
3 changed files with 131 additions and 63 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -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<RoutedPacket>(
capacity = Channel.UNLIMITED
) {
for (packet in channel) {
handleReceivedPacket(packet)
}
private class PeerActor {
lateinit var channel: kotlinx.coroutines.channels.SendChannel<RoutedPacket>
// 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<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
private var acceptingPackets = true
private val actors = LinkedHashMap<String, PeerActor>(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()
}

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

@ -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<Unit>()
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

9
docs/client-rewrite-contracts.md
View File

@ -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