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 92cdbaa65a93275135edddfbb3da9c8b89124099 into c127eb83ab94c069c32d37530d2faecd381cd2a8

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

65
app/src/main/java/com/bitchat/android/mesh/StoreForwardManager.kt
View File

@ -19,6 +19,8 @@ class StoreForwardManager {
private const val MESSAGE_CACHE_TIMEOUT = com.bitchat.android.util.AppConstants.StoreForward.MESSAGE_CACHE_TIMEOUT_MS // 12 hours for regular peers
private const val MAX_CACHED_MESSAGES = com.bitchat.android.util.AppConstants.StoreForward.MAX_CACHED_MESSAGES // For regular peers
private const val MAX_CACHED_MESSAGES_FAVORITES = com.bitchat.android.util.AppConstants.StoreForward.MAX_CACHED_MESSAGES_FAVORITES // For favorites
internal const val MAX_DELIVERED_MESSAGE_IDS = 1000
internal const val MAX_TRACKED_SENT_PEERS = 200
private const val CLEANUP_INTERVAL = com.bitchat.android.util.AppConstants.StoreForward.CLEANUP_INTERVAL_MS // 10 minutes
}
@ -35,8 +37,11 @@ class StoreForwardManager {
// Message storage
private val messageCache = Collections.synchronizedList(mutableListOf<StoredMessage>())
private val favoriteMessageQueue = ConcurrentHashMap<String, MutableList<StoredMessage>>()
private val deliveredMessages = Collections.synchronizedSet(mutableSetOf<String>())
private val cachedMessagesSentToPeer = Collections.synchronizedSet(mutableSetOf<String>())
// Insertion-ordered so the periodic trim can drop the oldest entries.
// These used to be wiped wholesale once they grew past their bound, which
// bought a memory bound by forgetting delivery state.
private val deliveredMessages = Collections.synchronizedSet(LinkedHashSet<String>())
private val cachedMessagesSentToPeer = Collections.synchronizedSet(LinkedHashSet<String>())
// Delegate for callbacks
var delegate: StoreForwardManagerDelegate? = null
@ -98,6 +103,7 @@ class StoreForwardManager {
favoriteMessageQueue[recipientPeerID]?.removeAt(0)
}
releaseAlreadySentLatch(recipientPeerID)
Log.d(TAG, "Cached message for favorite peer $recipientPeerID (${favoriteMessageQueue[recipientPeerID]?.size} total)")
} else {
@ -111,10 +117,27 @@ class StoreForwardManager {
messageCache.removeAt(0)
}
releaseAlreadySentLatch(recipientPeerID)
Log.d(TAG, "Cached message for peer $recipientPeerID (${messageCache.size} total in cache)")
}
}
/**
* `sendCachedMessages` runs at most once per peer, latched by
* [cachedMessagesSentToPeer]. Nothing ever removed a peer from that latch,
* so the *first* time a peer came online it received whatever was held for
* it and every later reconnection was refused — mail cached while it was
* away sat there until it aged out.
*
* The latch means "this peer has been handed everything we hold". Caching
* new mail for that peer is exactly the event that stops being true.
*/
private fun releaseAlreadySentLatch(recipientPeerID: String) {
if (cachedMessagesSentToPeer.remove(recipientPeerID)) {
Log.d(TAG, "New cached mail for $recipientPeerID; it may be sent again on reconnect")
}
}
/**
* Send cached messages to peer when they come online
*/
@ -267,14 +290,38 @@ class StoreForwardManager {
* Clean up delivered messages set (prevent memory leak)
*/
private fun cleanupDeliveredMessages() {
if (deliveredMessages.size > 1000) {
Log.d(TAG, "Clearing delivered messages set (${deliveredMessages.size} entries)")
deliveredMessages.clear()
// Trim the oldest, do not wipe. Emptying these sets bounds memory by
// forgetting: `deliveredMessages` is what stops an already-delivered
// message being queued again, and `cachedMessagesSentToPeer` is what
// stops a peer being handed the same batch twice in one session.
val droppedDeliveries = trimOldest(deliveredMessages, MAX_DELIVERED_MESSAGE_IDS)
if (droppedDeliveries > 0) {
Log.d(TAG, "Trimmed $droppedDeliveries oldest delivered message IDs")
}
if (cachedMessagesSentToPeer.size > 200) {
Log.d(TAG, "Clearing cached messages sent tracking (${cachedMessagesSentToPeer.size} entries)")
cachedMessagesSentToPeer.clear()
val droppedPeers = trimOldest(cachedMessagesSentToPeer, MAX_TRACKED_SENT_PEERS)
if (droppedPeers > 0) {
Log.d(TAG, "Trimmed $droppedPeers oldest already-sent peer entries")
}
}
/**
* Drops the oldest entries until [maxEntries] remain, returning how many
* went. `Collections.synchronizedSet` requires manual synchronization for
* iteration, hence the explicit block.
*/
private fun trimOldest(entries: MutableSet<String>, maxEntries: Int): Int {
synchronized(entries) {
var toRemove = entries.size - maxEntries
if (toRemove <= 0) return 0
val removed = toRemove
val iterator = entries.iterator()
while (toRemove > 0 && iterator.hasNext()) {
iterator.next()
iterator.remove()
toRemove--
}
return removed
}
}

124
app/src/test/kotlin/com/bitchat/android/mesh/StoreForwardManagerTest.kt Normal file
View File

@ -0,0 +1,124 @@
package com.bitchat.android.mesh
import com.bitchat.android.protocol.BitchatPacket
import com.bitchat.android.protocol.MessageType
import kotlinx.coroutines.runBlocking
import org.junit.After
import org.junit.Assert.assertEquals
import org.junit.Assert.assertTrue
import org.junit.Test
/**
* Store-and-forward exists so mail for an offline peer is handed over when it
* comes back. Two things stopped that working.
*/
class StoreForwardManagerTest {
private val manager = StoreForwardManager()
private val delegate = RecordingDelegate()
init {
manager.delegate = delegate
}
@After
fun tearDown() = manager.shutdown()
@Test
fun `mail cached after a peer was served is delivered on the next reconnect`() = runBlocking {
// First contact: nothing held, peer is marked as served.
manager.sendCachedMessages(PEER)
waitForDelivery()
assertEquals(0, delegate.sent.size)
// Peer goes away; a message is queued for it.
manager.cacheMessage(privateMessage("m-1"), "m-1")
// It comes back. Before this fix the send was refused outright,
// because nothing ever removed the peer from the already-sent latch,
// and the message sat in the cache until it aged out.
manager.sendCachedMessages(PEER)
waitForDelivery()
assertEquals(1, delegate.sent.size)
}
@Test
fun `a peer with nothing new is still not re-sent the same batch`() {
// The latch has to keep working, or every reconnect replays whatever
// is still cached.
manager.cacheMessage(privateMessage("m-1"), "m-1")
manager.sendCachedMessages(PEER)
waitForDelivery()
val afterFirst = delegate.sent.size
manager.sendCachedMessages(PEER)
waitForDelivery()
assertEquals("A second connect with no new mail must send nothing more", afterFirst, delegate.sent.size)
}
@Test
fun `trimming delivery state drops the oldest rather than all of it`() {
repeat(StoreForwardManager.MAX_DELIVERED_MESSAGE_IDS + 50) { index ->
manager.markMessageAsDelivered("delivered-$index")
}
manager.forceCleanup()
// Wiping the set bounded memory by forgetting which messages had
// already been handed over.
val debug = manager.getDebugInfo()
assertTrue(
"Delivery state must be trimmed to the cap, not emptied: $debug",
debug.contains("Delivered Messages: ${StoreForwardManager.MAX_DELIVERED_MESSAGE_IDS}")
)
}
@Test
fun `the newest delivery records are the ones kept`() {
repeat(StoreForwardManager.MAX_DELIVERED_MESSAGE_IDS + 10) { index ->
manager.markMessageAsDelivered("delivered-$index")
}
manager.forceCleanup()
// A message delivered a moment ago must not be the one forgotten.
manager.cacheMessage(privateMessage("delivered-${StoreForwardManager.MAX_DELIVERED_MESSAGE_IDS + 9}"),
"delivered-${StoreForwardManager.MAX_DELIVERED_MESSAGE_IDS + 9}")
manager.sendCachedMessages(PEER)
waitForDelivery()
assertEquals("An already-delivered message must not be queued again", 0, delegate.sent.size)
}
private fun waitForDelivery() {
// sendCachedMessages dispatches on its own scope with a 10ms/message
// stagger; this is well clear of the single-message case under test.
Thread.sleep(300)
}
private fun privateMessage(id: String): BitchatPacket = BitchatPacket(
version = 1u,
type = MessageType.MESSAGE.value,
senderID = "1111222233334444".hexToBytes(),
recipientID = PEER.toByteArray(),
timestamp = 1u,
payload = id.toByteArray(),
ttl = 7u
)
private class RecordingDelegate : StoreForwardManagerDelegate {
val sent = mutableListOf<BitchatPacket>()
override fun isFavorite(peerID: String) = false
override fun isPeerOnline(peerID: String) = false
override fun sendPacket(packet: BitchatPacket) {
synchronized(sent) { sent.add(packet) }
}
}
private fun String.hexToBytes(): ByteArray =
chunked(2).map { it.toInt(16).toByte() }.toByteArray()
private companion object {
const val PEER = "aaaabbbbccccdddd"
}
}