From 58e635ccf113f095f3232fb2dfe04a10e7a0c9dc Mon Sep 17 00:00:00 2001 From: Taksh Date: Mon, 5 Oct 2026 16:16:07 +0530 Subject: [PATCH 1/3] 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 d9dd18d0..5fc2eab7 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 328ff03ed0467231f289c5771f8f3d78909f002a Mon Sep 17 00:00:00 2001 From: Taksh Date: Mon, 5 Oct 2026 16:16:07 +0530 Subject: [PATCH 2/3] 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 00000000..9772923f --- /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" + } +} From 3ed439384240cd6ed5d8c5134afc335a8e285d9e Mon Sep 17 00:00:00 2001 From: Taksh Date: Mon, 5 Oct 2026 16:16:07 +0530 Subject: [PATCH 3/3] Use binary recipient IDs in cached delivery and deterministic tests --- .../android/mesh/StoreForwardManager.kt | 11 ++--- .../android/mesh/StoreForwardManagerTest.kt | 40 +++++++++++++++---- docs/client-rewrite-contracts.md | 10 +++++ 3 files changed, 48 insertions(+), 13 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 5fc2eab7..80896f8a 100644 --- a/app/src/main/java/com/bitchat/android/mesh/StoreForwardManager.kt +++ b/app/src/main/java/com/bitchat/android/mesh/StoreForwardManager.kt @@ -4,6 +4,7 @@ import android.util.Log import com.bitchat.android.protocol.BitchatPacket import com.bitchat.android.protocol.MessageType import com.bitchat.android.protocol.SpecialRecipients +import com.bitchat.android.util.hexEncodedString import kotlinx.coroutines.* import java.util.* import java.util.concurrent.ConcurrentHashMap @@ -12,7 +13,7 @@ import java.util.concurrent.ConcurrentHashMap * Manages store-and-forward messaging for offline peers * Extracted from BluetoothMeshService for better separation of concerns */ -class StoreForwardManager { +class StoreForwardManager(dispatcher: CoroutineDispatcher = Dispatchers.IO) { companion object { private const val TAG = "StoreForwardManager" @@ -47,7 +48,7 @@ class StoreForwardManager { var delegate: StoreForwardManagerDelegate? = null // Coroutines - private val managerScope = CoroutineScope(Dispatchers.IO + SupervisorJob()) + private val managerScope = CoroutineScope(dispatcher + SupervisorJob()) init { startPeriodicCleanup() @@ -74,7 +75,7 @@ class StoreForwardManager { // Determine if this is for a favorite peer val recipientPeerID = packet.recipientID?.let { recipientID -> - String(recipientID).replace("\u0000", "") + recipientID.hexEncodedString() } if (recipientPeerID.isNullOrEmpty()) { @@ -166,7 +167,7 @@ class StoreForwardManager { val recipientMessages = messageCache.filter { storedMessage -> !deliveredMessages.contains(storedMessage.messageID) && storedMessage.packet.recipientID?.let { recipientID -> - String(recipientID).replace("\u0000", "") == peerID + recipientID.hexEncodedString() == peerID } == true } messagesToSend.addAll(recipientMessages) @@ -226,7 +227,7 @@ class StoreForwardManager { val favoriteCount = favoriteMessageQueue[peerID]?.size ?: 0 val regularCount = messageCache.count { storedMessage -> storedMessage.packet.recipientID?.let { recipientID -> - String(recipientID).replace("\u0000", "") == peerID + recipientID.hexEncodedString() == peerID } == true } return favoriteCount + regularCount diff --git a/app/src/test/kotlin/com/bitchat/android/mesh/StoreForwardManagerTest.kt b/app/src/test/kotlin/com/bitchat/android/mesh/StoreForwardManagerTest.kt index 9772923f..513678eb 100644 --- a/app/src/test/kotlin/com/bitchat/android/mesh/StoreForwardManagerTest.kt +++ b/app/src/test/kotlin/com/bitchat/android/mesh/StoreForwardManagerTest.kt @@ -2,18 +2,25 @@ package com.bitchat.android.mesh import com.bitchat.android.protocol.BitchatPacket import com.bitchat.android.protocol.MessageType -import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.test.StandardTestDispatcher +import kotlinx.coroutines.test.TestCoroutineScheduler import org.junit.After import org.junit.Assert.assertEquals import org.junit.Assert.assertTrue import org.junit.Test +import org.junit.runner.RunWith +import org.robolectric.RobolectricTestRunner /** * Store-and-forward exists so mail for an offline peer is handed over when it * comes back. Two things stopped that working. */ +@OptIn(ExperimentalCoroutinesApi::class) +@RunWith(RobolectricTestRunner::class) class StoreForwardManagerTest { - private val manager = StoreForwardManager() + private val scheduler = TestCoroutineScheduler() + private val manager = StoreForwardManager(StandardTestDispatcher(scheduler)) private val delegate = RecordingDelegate() init { @@ -24,7 +31,7 @@ class StoreForwardManagerTest { fun tearDown() = manager.shutdown() @Test - fun `mail cached after a peer was served is delivered on the next reconnect`() = runBlocking { + fun `mail cached after a peer was served is delivered on the next reconnect`() { // First contact: nothing held, peer is marked as served. manager.sendCachedMessages(PEER) waitForDelivery() @@ -32,6 +39,7 @@ class StoreForwardManagerTest { // Peer goes away; a message is queued for it. manager.cacheMessage(privateMessage("m-1"), "m-1") + assertEquals(1, manager.getCachedMessageCount(PEER)) // It comes back. Before this fix the send was refused outright, // because nothing ever removed the peer from the already-sent latch, @@ -42,11 +50,26 @@ class StoreForwardManagerTest { assertEquals(1, delegate.sent.size) } + @Test + fun `favorite recipient keys use the same binary wire identity`() { + delegate.favorite = true + manager.sendCachedMessages(PEER) + waitForDelivery() + manager.cacheMessage(privateMessage("favorite-1"), "favorite-1") + assertEquals(1, manager.getCachedMessageCount(PEER)) + + manager.sendCachedMessages(PEER) + waitForDelivery() + assertEquals(1, delegate.sent.size) + assertEquals(0, manager.getCachedMessageCount(PEER)) + } + @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") + assertEquals(1, manager.getCachedMessageCount(PEER)) manager.sendCachedMessages(PEER) waitForDelivery() val afterFirst = delegate.sent.size @@ -91,16 +114,16 @@ class StoreForwardManagerTest { } 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) + // Each tested batch contains one message, whose stagger is zero. + // Run queued work deterministically without advancing periodic cleanup. + scheduler.runCurrent() } private fun privateMessage(id: String): BitchatPacket = BitchatPacket( version = 1u, type = MessageType.MESSAGE.value, senderID = "1111222233334444".hexToBytes(), - recipientID = PEER.toByteArray(), + recipientID = PEER.hexToBytes(), timestamp = 1u, payload = id.toByteArray(), ttl = 7u @@ -108,7 +131,8 @@ class StoreForwardManagerTest { private class RecordingDelegate : StoreForwardManagerDelegate { val sent = mutableListOf() - override fun isFavorite(peerID: String) = false + var favorite = false + override fun isFavorite(peerID: String) = favorite override fun isPeerOnline(peerID: String) = false override fun sendPacket(packet: BitchatPacket) { synchronized(sent) { sent.add(packet) } diff --git a/docs/client-rewrite-contracts.md b/docs/client-rewrite-contracts.md index 1abed243..25278157 100644 --- a/docs/client-rewrite-contracts.md +++ b/docs/client-rewrite-contracts.md @@ -20,6 +20,16 @@ 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` | +## Store-forward recipient identity + +The recipient of a cached mesh packet is its eight-byte wire ID, rendered as +16 lowercase hexadecimal characters for cache and peer lookups. It is not a +UTF-8 string. The synthetic bytes `aa aa bb bb cc cc dd dd` therefore identify +`aaaabbbbccccdddd` for regular queues, favorite queues, cache counts, and the +already-sent latch. Caching new mail for that identity releases its latch so a +later reconnect can deliver it. `StoreForwardManagerTest` uses this fixed wire +identity and a controlled coroutine scheduler for the reconnect contract. + ## Golden-vector policy Golden vectors compare literal externally visible bytes or hashes. Do not update