mirror of
https://github.com/permissionlesstech/bitchat-android.git
synced 2026-10-18 07:33:01 +00:00
Merge e4c9881f361345ca3ddf1f7a0c62ec10f0d7616e into 6ade68256cf32ca327047ba7f618ac6514f205ff
This commit is contained in:
Taksh Kothari
GitHub
commit
5ab35eb7ce
@ -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<String, CompletableDeferred<Unit>>()
|
||||
private val processorScope = CoroutineScope(dispatcher + SupervisorJob())
|
||||
|
||||
private class PeerActor {
|
||||
lateinit var channel: kotlinx.coroutines.channels.SendChannel<RoutedPacket>
|
||||
// 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<String, PeerActor>(16, 0.75f, true)
|
||||
|
||||
@OptIn(ObsoleteCoroutinesApi::class)
|
||||
private fun getOrCreateActorForPeer(peerID: String) = processorScope.actor<RoutedPacket>(
|
||||
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<String, kotlinx.coroutines.channels.SendChannel<RoutedPacket>>()
|
||||
|
||||
|
||||
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()
|
||||
|
||||
@ -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<PacketProcessor>()
|
||||
|
||||
@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<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() }
|
||||
|
||||
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"
|
||||
}
|
||||
}
|
||||
@ -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
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user