From e6c2c1d21e80d5f7c8f59af8ba0b1bb952ece738 Mon Sep 17 00:00:00 2001 From: Taksh Date: Sat, 15 Aug 2026 14:24:45 +0530 Subject: [PATCH 1/2] Deliver mail cached after a peer was already served sendCachedMessages is latched per peer by cachedMessagesSentToPeer, and nothing ever removed a peer from that set. So a peer received whatever was held for it on its first connection of the session, and every later reconnection was refused at the top of the function. Mail queued while it was away sat in the cache 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, so release it there. A reconnect with nothing new still sends nothing, which is what the latch is for. Also stops the periodic cleanup wiping both tracking sets outright. Emptying them bounded memory by forgetting: deliveredMessages is what keeps an already-delivered message from being queued again, and the peer latch is what keeps a batch from being sent twice. Both are now insertion-ordered and trimmed to their cap oldest-first. --- .../android/mesh/StoreForwardManager.kt | 65 ++++++++++++++++--- 1 file changed, 56 insertions(+), 9 deletions(-) diff --git a/app/src/main/java/com/bitchat/android/mesh/StoreForwardManager.kt b/app/src/main/java/com/bitchat/android/mesh/StoreForwardManager.kt index d9dd18d0b..5fc2eab73 100644 --- a/app/src/main/java/com/bitchat/android/mesh/StoreForwardManager.kt +++ b/app/src/main/java/com/bitchat/android/mesh/StoreForwardManager.kt @@ -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()) private val favoriteMessageQueue = ConcurrentHashMap>() - private val deliveredMessages = Collections.synchronizedSet(mutableSetOf()) - private val cachedMessagesSentToPeer = Collections.synchronizedSet(mutableSetOf()) + // 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()) + private val cachedMessagesSentToPeer = Collections.synchronizedSet(LinkedHashSet()) // 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, 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 } } From eab61056dd32723def272977fb086e3e9b3a9bc6 Mon Sep 17 00:00:00 2001 From: Taksh Date: Sat, 15 Aug 2026 14:24:45 +0530 Subject: [PATCH 2/2] Cover reconnect delivery and the trim MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The reconnect case fails on main: a peer served once, then sent mail while away, receives nothing when it returns. The other three hold the properties the fix must not break — a reconnect with no new mail still sends nothing, the trim leaves the cap in place rather than emptying it, and a recently delivered message is not the one forgotten. --- .../android/mesh/StoreForwardManagerTest.kt | 124 ++++++++++++++++++ 1 file changed, 124 insertions(+) create mode 100644 app/src/test/kotlin/com/bitchat/android/mesh/StoreForwardManagerTest.kt diff --git a/app/src/test/kotlin/com/bitchat/android/mesh/StoreForwardManagerTest.kt b/app/src/test/kotlin/com/bitchat/android/mesh/StoreForwardManagerTest.kt new file mode 100644 index 000000000..9772923fe --- /dev/null +++ b/app/src/test/kotlin/com/bitchat/android/mesh/StoreForwardManagerTest.kt @@ -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() + 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" + } +}