fix: deliver store-and-forward mail on a peer's second and later reconnects

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

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