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 } } 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" + } +}