Merge 3ed439384240cd6ed5d8c5134afc335a8e285d9e into 6ade68256cf32ca327047ba7f618ac6514f205ff

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<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
     
     // 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<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,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<BitchatPacket>()
+        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"
+    }
+}

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