feat(protocol): improve delivery by persisting inbound pending-decryption queue

CHANGELOG.md

@@ -34,6 +34,40 @@ archived by series under [docs/changelog/](docs/changelog/); see the
 
 ### Added
 
+- **The inbound pending-decryption queue is persisted.** A frame that arrives
+  before its sender's MLS session is ready is written to protocol-state storage
+  (`pending_decrypt_entries`, sealed like the outbound pending queue) when it
+  is admitted, deleted when it is drained, evicted or discarded, and restored
+  on the next launch. A restart during a slow handshake used to lose it, and a
+  relay that pushes ciphertext without store-and-forward holds no second copy
+  to ask for. A record older than seven days on disk is dropped on restore and
+  reported as `PENDING_QUEUE_DROPPED` with reason `expired_persisted`.
+- **The deduplicator's seen set is persisted.** Up to 2000 inbound ids, newest
+  first, written in batches (32 changes or 5 seconds on the process tick,
+  flushed on stop) and restored beside the Lamport clock, so the relay-socket
+  copy of a message the app already consumed from a push injection is
+  recognised as a duplicate across a restart instead of reaching a spent
+  ratchet generation and surfacing as a decryption failure. A runtime
+  `updateDedupConfig` carries the set across instead of clearing it. A
+  sender's own outgoing ids are tracked for the process only and never
+  written.
+- **`PENDING_QUEUE_DROPPED` is emitted for text messages**, not only for media
+  chunks, whenever the pending-decryption queue evicts a frame (overflow, TTL,
+  or the persisted age bound). Text evictions were metrics-only, which left an
+  app unable to tell "the sender went quiet" from "the SDK evicted their
+  message". The event is advisory: the frame was never ACKed, so a sender
+  still retrying resends it.
+- **Inbound message events are held while nothing can take them.**
+  `message_received`, `file_received` and `message_decryption_failed` each
+  report one message the core has already ACKed and will never restate, so a
+  drop between the core and the app's handler was a lost message. Each React
+  Native bridge now holds up to 256 of them, keyed by message id, while
+  JavaScript has no listener or no live React instance (a backgrounded app, a
+  push injection before the first subscribe), and replays them on subscribe
+  and on foreground; the TypeScript layer does the same across the gap to the
+  app's first `on()`. Held events are scoped to the session that produced
+  them, so a message held for a torn-down account never surfaces in the next
+  one.
 - **`enableTelemetry` / `enable_telemetry` on every binding.** Takes the API key
   and app id the developer portal issued; the binding fills in the platform
   and owns the application lifecycle, so there is no per-event or per-lifecycle
@@ -107,6 +141,29 @@ the client's answers were wrong on a device rather than merely different:
 
 ### Changed
 
+- **A DM the relay only pushed is parked, not left awaiting an ACK.** The relay
+  answers `MessageSent { pushed: true }` when the recipient has no live socket
+  and the ciphertext went out in a device push; it keeps no copy. The React
+  Native bridges now report that id to the core as `relay_pushed`, which drops
+  the pending ACK, keeps the outbox entry, schedules the reachability probe
+  and presence-watches the recipient, as a `DeliveryError` does, with two
+  differences because the push may have delivered the frame: connection
+  requests and Welcomes stay on their ordinary path, and no
+  `MessageUndeliverable` is emitted, on the push itself or on any later probe
+  verdict for that frame. The outbox entry remembers it was pushed (persisted
+  with the entry), because the relay answers a probe of a pushed message with
+  `DeliveryError` rather than a second notification and the core would
+  otherwise read that as an ordinary unreachable verdict.
+- **`pendingQueue.pendingTtlMs` defaults to 24 hours** (was 30 minutes) on
+  every binding, so a frame parked for a session still being set up outlives
+  a handshake on a phone opened once a day. The in-memory TTL restarts with
+  the process; the persisted copy is bounded separately at seven days. Memory
+  stays bounded by the per-peer and global caps.
+- **`reliability.dedup` defaults to 2000 ids and 24 hours** (were 1000 and
+  1 hour), sized for the persisted seen set. The Nostr replay overlap
+  (1 h 5 min) now fits inside the retention window, so a reconnect the next
+  day deduplicates it instead of re-processing it; `docs/nostr.md`, the
+  transport's subscription doc and the drift guard now state the absorption.
 - **`TelemetryConfig` is reshaped** on every binding: `apiKey` and `appId` are
   required, `appVersion`, `debug`, `flushIntervalMs`, `maxBatchBytes`,
   `maxBufferedRecords` and `includeDeviceId` are new, `enablePollQueue` is
@@ -232,6 +289,18 @@ the client's answers were wrong on a device rather than merely different:
   stays per-event and unaggregated by design; the rollup and session summary
   exist only inside the pipe, and a test pins that neither is ever an event.
 
+### Fixed
+
+- **A peer-requested session reset deletes the persisted pending-decryption
+  records** of the frames it discards. Left on disk, the next launch restored
+  frames sealed to the deleted session and drained them into its replacement
+  as spurious decrypt failures.
+- **Releasing a dropped group entry's replay protection reaches the persisted
+  seen set.** The release unmarked the id in memory only, so a record written
+  while the entry was buffered kept the id, the next launch restored it, and
+  the sender's redelivery was swallowed and re-ACKed as delivered for the whole
+  retention window.
+
 ## [0.25.0] — 2026-09-08
 
 > **A device on Reticulum could never receive anything. It can now.** Both

README.md

@@ -82,7 +82,7 @@ config = ProtocolConfig(
     require_encryption=False,
     max_pending_per_peer=64,
     max_pending_global=4096,
-    pending_ttl_ms=1_800_000,  # 30 min — matches the SDK default
+    pending_ttl_ms=86_400_000,  # 24 h — matches the SDK default
     overflow_policy=OverflowPolicy.DROP_OLDEST,
 )
 protocol = ProtocolManager(

bindings/react-native/MeshSdk.podspec

@@ -28,6 +28,7 @@ Pod::Spec.new do |s|
     "ios/BleManager.swift",
     "ios/InternetManager.swift",
     "ios/ForcedPresenceCheckQueue.swift",
+    "ios/InboundEventBuffer.swift",
     "ios/InboundFragmentBuffer.swift",
     "ios/OutboundFragmentQueue.swift",
     "ios/AddressDeclarationPolicy.swift",

bindings/react-native/README.md

@@ -569,7 +569,7 @@ interface EncryptionConfig {
 interface PendingQueueConfig {
   maxPendingPerPeer?: number;  // default: 64
   maxPendingGlobal?: number;   // default: 4096
-  pendingTtlMs?: number;       // default: 1800000 (30 min)
+  pendingTtlMs?: number;       // default: 86400000 (24 h)
   overflowPolicy?: 'drop_oldest' | 'drop_newest';  // default: drop_oldest
 }
 ```
@@ -605,8 +605,8 @@ interface ReliabilityConfig {
     pendingMessageMaxLifetimeMs?: number; // default: 604800000
   };
   dedup?: {
-    maxTrackedMessages?: number; // default: 1000, must be > 0
-    retentionTimeSecs?: number;  // default: 3600, must be > 0
+    maxTrackedMessages?: number; // default: 2000, must be > 0
+    retentionTimeSecs?: number;  // default: 86400, must be > 0
   };
 }
 ```

bindings/react-native/android/src/main/java/com/offlineprotocol/InternetManager.kt

@@ -1348,14 +1348,32 @@ class InternetManager(
                     )
                 }
 
+                // `pushed: true` is the relay saying it had no live socket for
+                // the recipient and handed the ciphertext to a push
+                // notification instead. The relay does not store-and-forward,
+                // so this is the same fact a DeliveryError carries — the
+                // recipient is not on the relay right now — with the message
+                // *possibly* arriving through the push. Report this one id to
+                // the core as `relay_pushed`, which parks a plain DM as the
+                // unreachable path would (no ACK budget burnt against an
+                // offline peer, a reachability probe scheduled, the recipient
+                // watched) so that if the push is lost the presence edge
+                // re-drives it. Absent on older relays, which reads as
+                // `false`.
+                val pushed = json.optBoolean("pushed", false)
+                if (pushed && messageId != null && messageId.isNotEmpty()) {
+                    parkPushedMessage(recipient, messageId)
+                }
+
                 if (messageId != null && messageId.isNotEmpty()) {
                     // The server has confirmed the message was sent with this message_id
                     // We need to notify the protocol so it can update the message ID
                     // The protocol will emit a message_sent event with this server-generated ID
                     emitDiagnostic("debug", "MessageSent from relay server", mapOf(
                         "messageId" to messageId,
                         "recipient" to recipient,
-                        "timestamp" to timestamp
+                        "timestamp" to timestamp,
+                        "pushed" to pushed
                     ))
                     // Note: The protocol SDK will handle the message_sent event internally
                     // The frontend will receive it via the normal event stream
@@ -2219,6 +2237,46 @@ class InternetManager(
         ))
     }
 
+    /**
+     * Parks one message the relay reports as `MessageSent { pushed: true }`.
+     *
+     * The narrower sibling of [handleRecipientUnreachable]: the relay named
+     * exactly one frame, so only that id is reported to the core and the
+     * recipient's other in-flight frames are left alone — a later
+     * `MessageSent` for each of them says what became of it. The reason is
+     * the exact `relay_pushed` token, not a `recipient_unreachable` tail: that
+     * prefix fast-fails connection requests and fails Welcomes, while
+     * `relay_pushed` parks a plain DM and leaves both of those alone, since
+     * the push may have delivered them. The recipient is presence-watched
+     * and an offline presence is fed to the core exactly as the DeliveryError
+     * path does, so the presence-online edge is what re-drives the parked
+     * message if the push never reaches the device.
+     */
+    private fun parkPushedMessage(recipient: String, messageId: String) {
+        // Sentinel entries track app-authored raw SendMessage frames; their
+        // outcomes belong to the app, not the core (see handleRecipientUnreachable).
+        if (messageId.isEmpty() || messageId.startsWith(RAW_SEND_SENTINEL_PREFIX)) return
+        try {
+            protocol.internetSendFailedWithReason(messageId, "relay_pushed")
+        } catch (e: Exception) {
+            Log.e(TAG, "Failed to park pushed message $messageId", e)
+        }
+        // Same self guard as handleRecipientUnreachable: never watch self and
+        // never feed "self is offline" into the core.
+        if (recipient.isNotEmpty() && !isSelfPeer(recipient)) {
+            presenceWatch.watch(recipient, monotonicNowMs())
+            try {
+                protocol.internetPeerPresence(recipient, false, null)
+            } catch (e: Exception) {
+                Log.e(TAG, "Failed to ingest offline presence for $recipient", e)
+            }
+        }
+        emitDiagnostic("info", "Relay pushed message to offline recipient; parked", mapOf(
+            "recipient" to recipient,
+            "messageId" to messageId
+        ))
+    }
+
     /**
      * True when [peerId] names this device in either namespace — the profile
      * (relay username by convention) or the derived address the core stamps

bindings/react-native/android/src/main/java/com/offlineprotocol/OfflineProtocolModule.kt

@@ -113,6 +113,33 @@ class OfflineProtocolModule(reactContext: ReactApplicationContext) :
         },
     )
 
+    /**
+     * Redelivers *inbound message* events that could not be handed to JS —
+     * `message_received`, `file_received`, `message_decryption_failed` — on
+     * the next subscribe or foreground. See [BUFFERED_INBOUND_EVENT_TYPES] for
+     * why these three qualify and [holdInboundEventIfBuffered] for the gate.
+     *
+     * A second [StickyEventDispatcher] rather than more keys in [stickyEvents]
+     * because the two hold different things. A one-shot event collapses per
+     * type and the buffer is sized for two keys; an inbound event is one
+     * message the core has already ACKed and will never restate, so every one
+     * must survive on its own key (`type:message_id`) and the cap has to be a
+     * real capacity — 256, the oldest dropped past it — rather than a backstop.
+     */
+    private val inboundEvents = StickyEventDispatcher(
+        buffer = StickyEventBuffer(maxEntries = INBOUND_EVENT_BUFFER_CAPACITY),
+        canEmit = { canEmitToJs() },
+        emit = { eventJson -> sendEvent(EVENT_NAME, eventParams(eventJson)) },
+        schedule = { runnable ->
+            try {
+                reactApplicationContext.runOnJSQueueThread(runnable)
+            } catch (e: Exception) {
+                android.util.Log.w(NAME, "Could not schedule inbound event flush", e)
+                false
+            }
+        },
+    )
+
     /**
      * The started activities, which is what tells the process apart from an
      * activity, and the watcher that maintains them. Held by identity rather
@@ -200,6 +227,26 @@ class OfflineProtocolModule(reactContext: ReactApplicationContext) :
 
         /** How long `destroy` waits for an in-flight process tick to finish. */
         private const val PROCESS_SHUTDOWN_TIMEOUT_MS = 2_000L
+
+        /**
+         * The inbound event tags [inboundEvents] holds when JS cannot take
+         * them. Each reports one message the core has already ACKed,
+         * dedup-marked and dropped its queued copy of — the sender will not
+         * resend and nothing will restate it — so a drop here is a lost
+         * message, and the drop is ordinary: the React instance is down while
+         * the app is backgrounded, or a push injection lands before JS has
+         * subscribed. Must match `BUFFERED_INBOUND_EVENT_TYPES` in
+         * `src/constants.ts` and `InboundEventBuffer.bufferedEventTypes` on
+         * iOS; pinned by `react_native_buffered_inbound_event_set_matches_native`.
+         */
+        private val BUFFERED_INBOUND_EVENT_TYPES: Set<String> = setOf(
+            "message_received",
+            "file_received",
+            "message_decryption_failed",
+        )
+
+        /** Inbound events held at most; the oldest is dropped past it. */
+        private const val INBOUND_EVENT_BUFFER_CAPACITY = 256
     }
 
     private object Constants {
@@ -688,10 +735,14 @@ class OfflineProtocolModule(reactContext: ReactApplicationContext) :
             // Set up event callback
             proto.setEventCallback(object : EventCallback {
                 override fun onEvent(eventJson: String) {
-                    val params = Arguments.createMap().apply {
-                        putString("eventJson", eventJson)
+                    // The gate is checked first, before the payload is built
+                    // or the JSON is parsed: with JS reachable the common path
+                    // costs what it always did, and only a shut gate pays for
+                    // the type/id read that decides whether to hold.
+                    if (!canEmitToJs() && holdInboundEventIfBuffered(eventJson)) {
+                        return
                     }
-                    sendEvent(EVENT_NAME, params)
+                    sendEvent(EVENT_NAME, eventParams(eventJson))
                 }
             })
 
@@ -935,11 +986,45 @@ class OfflineProtocolModule(reactContext: ReactApplicationContext) :
         stickyEvents.send(key, eventJson)
     }
 
-    /** Redelivers held one-shot events, if JS looks able to take them now. */
+    /**
+     * Redelivers held one-shot and inbound events, if JS looks able to take
+     * them now.
+     */
     private fun flushStickyEvents() {
         stickyEvents.flush()
+        inboundEvents.flush()
     }
 
+    /**
+     * Holds an inbound message event for redelivery when JS could not take
+     * it, keyed `type:message_id` so every message survives on its own.
+     * Returns whether the event was one of [BUFFERED_INBOUND_EVENT_TYPES] and
+     * was handed to [inboundEvents]; anything else is the caller's to emit
+     * (and, for a shut gate, to drop) as before.
+     *
+     * Only called with the gate already known to be shut, so the parse here
+     * is off the hot path. An event of a buffered type that carries no id at
+     * all is keyed by arrival order rather than dropped: losing a message to
+     * a missing field would be the failure this buffer exists to prevent.
+     */
+    private fun holdInboundEventIfBuffered(eventJson: String): Boolean {
+        val json = try {
+            JSONObject(eventJson)
+        } catch (e: Exception) {
+            return false
+        }
+        val type = json.optString("type", "")
+        if (type !in BUFFERED_INBOUND_EVENT_TYPES) return false
+        val id = json.optString("message_id", "")
+            .ifEmpty { json.optString("file_id", "") }
+            .ifEmpty { "seq-${inboundEventSequence.incrementAndGet()}" }
+        inboundEvents.send("$type:$id", eventJson)
+        return true
+    }
+
+    /** Fallback key source for a buffered inbound event that carries no id. */
+    private val inboundEventSequence = AtomicInteger(0)
+
     private fun emitDiagnostic(level: String, message: String, context: Map<String, Any?> = emptyMap()) {
         try {
             val json = JSONObject()
@@ -1067,6 +1152,10 @@ class OfflineProtocolModule(reactContext: ReactApplicationContext) :
             // that closes it. Collapsing the two is how an event emitted by the
             // session destroy() just ended ends up held for the next one.
             stickyEvents.beginSession()
+            // The inbound buffer follows the same session edges: nothing
+            // buffered can be produced before this start(), and a message held
+            // for the previous session belongs to whatever identity ran it.
+            inboundEvents.beginSession()
             emitDiagnostic("info", "Starting protocol")
             protocol?.start()
             emitDiagnostic("info", "Protocol core started")
@@ -2261,6 +2350,7 @@ class OfflineProtocolModule(reactContext: ReactApplicationContext) :
             // exotic one — and would otherwise be held under the new generation
             // for whichever session subscribes next.
             stickyEvents.endSession()
+            inboundEvents.endSession()
             currentConfig = null
             promise.resolve(null)
         } catch (e: Exception) {
@@ -3381,7 +3471,7 @@ class OfflineProtocolModule(reactContext: ReactApplicationContext) :
             val json = JSONObject(configJson)
             val dedupConfig = DedupConfig(
                 maxTrackedMessages = json.optLong("maxTrackedMessages", 10000).toULong(),
-                retentionTimeSecs = json.optLong("retentionTimeSecs", 3600).toULong()
+                retentionTimeSecs = json.optLong("retentionTimeSecs", 86400).toULong()
             )
 
             protocol?.updateDedupConfig(dedupConfig)

bindings/react-native/android/src/main/java/com/offlineprotocol/ProtocolConfigParser.kt

@@ -109,7 +109,7 @@ internal object ProtocolConfigParser {
         val pendingTtlMs = pendingQueueJson?.optLongCompat(
             "pendingTtlMs",
             "pending_ttl_ms"
-        ) ?: json.optLongCompat("pendingTtlMs", "pending_ttl_ms") ?: 1_800_000L
+        ) ?: json.optLongCompat("pendingTtlMs", "pending_ttl_ms") ?: 86_400_000L
         val overflowPolicyRaw = pendingQueueJson?.optStringCompat(
             "overflowPolicy",
             "overflow_policy"

bindings/react-native/android/src/test/java/com/offlineprotocol/ProtocolConfigParserTest.kt

@@ -146,11 +146,11 @@ class ProtocolConfigParserTest {
 
     @Test
     fun pendingTtlFallsBackToTheRustDefault() {
-        // Mirrors DEFAULT_PENDING_TTL_MS (30 min); the iOS reader asserts the
+        // Mirrors DEFAULT_PENDING_TTL_MS (24 h); the iOS reader asserts the
         // same, and `rn_bridge_pending_ttl_fallbacks_match_rust_default` pins
         // all three bridge literals to the Rust constant.
         val config = parse("""{"appId":"app","userId":"alice"}""")
-        assertEquals(1_800_000L, config.pendingTtlMs.toLong())
+        assertEquals(86_400_000L, config.pendingTtlMs.toLong())
     }
 
     @Test