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..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,13 +13,15 @@ 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" 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,14 +38,17 @@ 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 // Coroutines - private val managerScope = CoroutineScope(Dispatchers.IO + SupervisorJob()) + private val managerScope = CoroutineScope(dispatcher + SupervisorJob()) init { startPeriodicCleanup() @@ -69,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()) { @@ -98,6 +104,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 +118,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 */ @@ -143,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) @@ -203,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 @@ -267,14 +291,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 } } 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..513678eb --- /dev/null +++ b/app/src/test/kotlin/com/bitchat/android/mesh/StoreForwardManagerTest.kt @@ -0,0 +1,148 @@ +package com.bitchat.android.mesh + +import com.bitchat.android.protocol.BitchatPacket +import com.bitchat.android.protocol.MessageType +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 scheduler = TestCoroutineScheduler() + private val manager = StoreForwardManager(StandardTestDispatcher(scheduler)) + 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`() { + // 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") + 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, + // and the message sat in the cache until it aged out. + manager.sendCachedMessages(PEER) + waitForDelivery() + + 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 + + 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() { + // 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.hexToBytes(), + timestamp = 1u, + payload = id.toByteArray(), + ttl = 7u + ) + + private class RecordingDelegate : StoreForwardManagerDelegate { + val sent = mutableListOf() + 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) } + } + } + + private fun String.hexToBytes(): ByteArray = + chunked(2).map { it.toInt(16).toByte() }.toByteArray() + + private companion object { + const val PEER = "aaaabbbbccccdddd" + } +} diff --git a/docs/client-rewrite-contracts.md b/docs/client-rewrite-contracts.md index 05d5e1ad..d8f7cb1e 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, 22h outbound envelope randomization | `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