nik/bitchat-android
1
0
Fork 0
mirror of https://github.com/permissionlesstech/bitchat-android.git synced 2026-09-19 04:59:59 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge ab5eccc01fc96fd8cb85d337b05198d88e9e4694 into c127eb83ab94c069c32d37530d2faecd381cd2a8

Browse Source
This commit is contained in:
Taksh Kothari 2026-09-14 09:43:25 +05:30 committed by GitHub
commit 1b400ceadf
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 211 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -20,6 +20,18 @@ class PacketProcessor(private val myPeerID: String) {
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, which is the peer least likely to be mid-session.
*/
internal const val MAX_PEER_ACTORS = 128
}
// Delegate for callbacks
@ -40,8 +52,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<String, CompletableDeferred<Unit>>()
@OptIn(ObsoleteCoroutinesApi::class)
private fun getOrCreateActorForPeer(peerID: String) = processorScope.actor<RoutedPacket>(
capacity = Channel.UNLIMITED
@ -51,8 +61,34 @@ class PacketProcessor(private val myPeerID: String) {
}
}
// Cache actors to reuse them
private val actors = mutableMapOf<String, kotlinx.coroutines.channels.SendChannel<RoutedPacket>>()
// 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
}
}
init {
// Set up the packet relay manager delegate immediately
@ -72,7 +108,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 {
@ -238,6 +276,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 +290,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 +307,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()

148
app/src/test/kotlin/com/bitchat/android/mesh/PacketProcessorActorBoundsTest.kt Normal file
View File

@ -0,0 +1,148 @@
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 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)
}
@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)
}
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): RoutedPacket {
val packet = BitchatPacket(
version = 1u,
type = MessageType.MESSAGE.value,
senderID = peerID.hexToBytes(),
recipientID = SpecialRecipients.BROADCAST,
timestamp = 1u,
payload = byteArrayOf(0x01),
ttl = 7u
)
return RoutedPacket(packet, peerID, "direct-link")
}
private 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"
}
}