From f8ab0a7dc0223a7fc277a25d804d17d0d24b6c58 Mon Sep 17 00:00:00 2001 From: jack <212554440+jackjackbits@users.noreply.github.com> Date: Thu, 9 Jul 2026 18:23:46 +0200 Subject: [PATCH] =?UTF-8?q?Fix=20restore-path=20main=E2=86=94bleQueue=20de?= =?UTF-8?q?adlock=20and=20courier=20drop=20amplification=20(#1425)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * Fix restore-path main↔bleQueue deadlock and courier drop amplification A device froze permanently in a two-phone test. Debugger stacks showed an ABBA deadlock: the main actor was in bleQueue.sync (delivery-ack send → broadcastPacket → readLinkState) while bleQueue was in main.sync (captureBluetoothStatus reading backgroundTimeRemaining). The load that lined the two edges up came from a courier-drop amplification storm: drop dedup was in-memory only while the outbox driving 120s re-deposits is persisted, so every relaunch republished the same undelivered DM as a fresh 24h relay drop and every gateway relaunch re-fetched the whole backlog — ~20 copies of one DM delivered in 40ms, each triggering decrypt + delivery + ack + handshake work. Fixes, in rank order: - Edge B (P0): captureBluetoothStatus no longer main.syncs from bleQueue; backgroundTimeRemaining is sampled on main and cached behind a lock. Invariant documented: bleQueue must NEVER sync-dispatch to main. - Edge A (P0, defense in depth): sendDeliveryAck / sendReadReceipt / sendPrivateMessage / sendNoisePayload / triggerHandshake hop to messageQueue like sendMessage, so no main-actor call path reaches readLinkState's bleQueue.sync. - Drop dedup (P1): publishedDropKeys and seenDropEventIDs persist across relaunches (new BridgeDropDedupStore, entries expire with the 24h NIP-40 drop window; wiped on panic) — one drop per message ID per 24h regardless of relaunch count. - Receiver dedup (P1): openCourierEnvelope dedups on the inner private message ID before delivery, so a duplicate copy costs one decrypt and never re-delivers, re-acks, or re-triggers a handshake. - Handshake gating (P2): queued acks initiate a Noise handshake only for reachable peers; mail from absent/rotated identities no longer turns each copy into a mesh-wide handshake flood (the ack stays queued and flushes when a session eventually establishes). - Outbox (P3): re-enqueueing a queued message ID carries over its depositedCourierKeys so resends stop re-burning the same courier slots. Co-Authored-By: Claude Fable 5 * Review fixes: offline-drop durability, gateway handoff retry, restore-log freshness, coalesced persist Adversarial review of the storm/deadlock PR surfaced four issues: - Offline blackhole (must-fix): a deposit made while relays were down persisted its dedup key even though the drop only sat in the in-memory pending queue — app killed before reconnect meant the relaunch lost the drop but the persisted key blocked every re-deposit for 24h. The persisted snapshot now excludes keys still pending; they become durable only when flushPendingDrops actually publishes them. - Gateway handoff: seen-event IDs were consumed before the deliverToPeer handoff; a failed handoff (peer walked away) permanently dropped the event for a single-gateway island. deliverToPeer now reports whether the handoff was attempted, and a failure releases the seen slot so a relaunch or backlog redelivery retries. - Restore-path logs: central/peripheral-restore captures logged the init sentinel bgRemaining=∞. The cache is now seeded in init's main-thread branch and restore captures route through the sampler, which refreshes the cached budget before logging. - Persist cost: the dedup record was a full JSON encode + atomic write on the main actor per mutation (once per event during a backlog re-fetch). Writes now coalesce behind a 1s window, flushed immediately on background/terminate; panic wipe stays immediate. Co-Authored-By: Claude Fable 5 * Remove BoundedIDSet.remove, orphaned by the ExpiringIDSet migration The drop-dedup sets that needed slot release moved to ExpiringIDSet; remaining BoundedIDSet users only insert and check. Periphery caught it. Co-Authored-By: Claude Fable 5 --------- Co-authored-by: jack Co-authored-by: Claude Fable 5 --- bitchat/Services/BLE/BLEService.swift | 159 +++++++++++++++--- bitchat/Services/Gateway/BoundedIDSet.swift | 10 -- .../Gateway/BridgeCourierService.swift | 128 ++++++++++++-- .../Gateway/BridgeDropDedupStore.swift | 137 +++++++++++++++ bitchat/Services/MessageRouter.swift | 10 +- bitchat/Services/TransportConfig.swift | 3 + bitchat/ViewModels/ChatViewModel.swift | 1 + .../ChatViewModelBootstrapper.swift | 2 +- .../EndToEnd/CourierEndToEndTests.swift | 64 +++++++ .../EndToEnd/PrekeyEndToEndTests.swift | 11 +- .../Services/BridgeCourierServiceTests.swift | 138 ++++++++++++++- .../Services/BridgeDropDedupStoreTests.swift | 118 +++++++++++++ .../Services/MessageRouterTests.swift | 28 +++ 13 files changed, 754 insertions(+), 55 deletions(-) create mode 100644 bitchat/Services/Gateway/BridgeDropDedupStore.swift create mode 100644 bitchatTests/Services/BridgeDropDedupStoreTests.swift diff --git a/bitchat/Services/BLE/BLEService.swift b/bitchat/Services/BLE/BLEService.swift index 2152f316..56fee0b6 100644 --- a/bitchat/Services/BLE/BLEService.swift +++ b/bitchat/Services/BLE/BLEService.swift @@ -129,6 +129,15 @@ final class BLEService: NSObject { // Application state tracking (thread-safe) #if os(iOS) private var isAppActive: Bool = true // Assume active initially + /// Last `UIApplication.shared.backgroundTimeRemaining` sampled on the + /// main thread, cached so bleQueue status logs can read it without ever + /// dispatching to main (see `captureBluetoothStatus` for the invariant). + private let backgroundTimeLock = NSLock() + private var _cachedBackgroundTimeRemaining: TimeInterval = .greatestFiniteMagnitude + private var cachedBackgroundTimeRemaining: TimeInterval { + backgroundTimeLock.lock(); defer { backgroundTimeLock.unlock() } + return _cachedBackgroundTimeRemaining + } #endif // MARK: - Core BLE Objects @@ -176,6 +185,13 @@ final class BLEService: NSObject { // Ingress link tracking for duplicate and last-hop suppression private var ingressLinks = BLEIngressLinkRegistry() + // Inner message IDs of recently opened courier envelopes. Redundant + // copies of one message ride different envelopes (each seal uses a fresh + // ephemeral key, and bridge drops multiply across relays/couriers), so + // envelope-level dedup can't catch them; dedup on the inner ID before + // delivery so a duplicate costs one decrypt instead of a delivery + ack + // + handshake each. Owned by collectionsQueue barriers. + private var openedCourierMessageIDs = BoundedIDSet(capacity: TransportConfig.courierOpenedMessageIDCap) private let logRateLimiter = BLELogRateLimiter(defaultMinimumInterval: 5) private var pendingPeripheralWrites = BLEOutboundWriteBuffer() @@ -265,12 +281,18 @@ final class BLEService: NSObject { // Set up application state tracking (iOS only) #if os(iOS) - // Check initial state on main thread + // Check initial state on main thread. The background-budget cache is + // seeded here too: a background-restore launch captures Bluetooth + // status before any lifecycle notification fires, and the init-time + // sentinel would log a meaningless bgRemaining=∞ for exactly the + // wake window that matters. if Thread.isMainThread { isAppActive = UIApplication.shared.applicationState == .active + refreshCachedBackgroundTimeRemaining() } else { DispatchQueue.main.sync { isAppActive = UIApplication.shared.applicationState == .active + refreshCachedBackgroundTimeRemaining() } } @@ -777,7 +799,11 @@ final class BLEService: NSObject { } func triggerHandshake(with peerID: PeerID) { - initiateNoiseHandshake(with: peerID) + // Callers are on the main actor; the handshake broadcast sync-waits + // on bleQueue for link state, so hop off main first. + messageQueue.async { [weak self] in + self?.initiateNoiseHandshake(with: peerID) + } } // MARK: Noise identity/session access (narrow Transport wrappers) @@ -932,6 +958,15 @@ final class BLEService: NSObject { func sendReadReceipt(_ receipt: ReadReceipt, to peerID: PeerID) { + // Hop like sendMessage: callers are often on the main actor, and the + // send path sync-waits on bleQueue for link state — the main thread + // must never block on bleQueue (see captureBluetoothStatus). + if DispatchQueue.getSpecific(key: messageQueueKey) == nil { + messageQueue.async { [weak self] in + self?.sendReadReceipt(receipt, to: peerID) + } + return + } let payload = BLENoisePayloadFactory.readReceipt(originalMessageID: receipt.originalMessageID) if noiseService.hasEstablishedSession(with: peerID) { @@ -942,11 +977,15 @@ final class BLEService: NSObject { SecureLogger.error("Failed to send read receipt: \(error)") } } else { - // Queue for after handshake and initiate if needed + // Queue for after handshake; initiate only while the peer is + // around to answer (see sendDeliveryAck — absent senders must + // not turn queued acks into handshake floods). collectionsQueue.sync(flags: .barrier) { pendingNoiseSessionQueues.appendTypedPayload(payload, for: peerID) } - if !noiseService.hasSession(with: peerID) { initiateNoiseHandshake(with: peerID) } + if !noiseService.hasSession(with: peerID), isPeerReachable(peerID) { + initiateNoiseHandshake(with: peerID) + } SecureLogger.debug("🕒 Queued READ receipt for \(peerID.id.prefix(8))… until handshake completes", category: .session) } } @@ -1395,6 +1434,15 @@ final class BLEService: NSObject { } func sendDeliveryAck(for messageID: String, to peerID: PeerID) { + // Hop like sendMessage: callers are often on the main actor, and the + // send path sync-waits on bleQueue for link state — the main thread + // must never block on bleQueue (see captureBluetoothStatus). + if DispatchQueue.getSpecific(key: messageQueueKey) == nil { + messageQueue.async { [weak self] in + self?.sendDeliveryAck(for: messageID, to: peerID) + } + return + } let payload = BLENoisePayloadFactory.delivered(messageID: messageID) if noiseService.hasEstablishedSession(with: peerID) { @@ -1404,11 +1452,18 @@ final class BLEService: NSObject { SecureLogger.error("Failed to send delivery ACK: \(error)") } } else { - // Queue for after handshake and initiate if needed + // Queue for after handshake; initiate only while the peer is + // around to answer — couriered/bridged mail routinely arrives + // from absent (or rotated) identities, and every duplicate copy + // initiating a handshake broadcast turns one undeliverable ack + // into a mesh-wide flood. The queued ack flushes whenever a + // session eventually establishes. collectionsQueue.sync(flags: .barrier) { pendingNoiseSessionQueues.appendTypedPayload(payload, for: peerID) } - if !noiseService.hasSession(with: peerID) { initiateNoiseHandshake(with: peerID) } + if !noiseService.hasSession(with: peerID), isPeerReachable(peerID) { + initiateNoiseHandshake(with: peerID) + } SecureLogger.debug("🕒 Queued DELIVERED ack for \(peerID.id.prefix(8))… until handshake completes", category: .session) } } @@ -1680,7 +1735,10 @@ extension BLEService: CBCentralManagerDelegate { recentPeripheralCache.record(peripheral, peripheralID: identifier, at: Date()) } - captureBluetoothStatus(context: "central-restore") + // Via the sampler (not a direct capture): it refreshes the cached + // background budget on main first, so the restore log shows the real + // wake window instead of the init sentinel. + logBluetoothStatus("central-restore") if central.state == .poweredOn { startScanning() @@ -2439,7 +2497,8 @@ extension BLEService: CBPeripheralManagerDelegate { } } - captureBluetoothStatus(context: "peripheral-restore") + // Via the sampler for a fresh background budget (see central-restore). + logBluetoothStatus("peripheral-restore") if peripheral.state == .poweredOn && !peripheral.isAdvertising { peripheral.startAdvertising(buildAdvertisementData()) @@ -2720,19 +2779,37 @@ extension BLEService { } private func logBluetoothStatus(_ context: String) { - bleQueue.async { [weak self] in - guard let self = self else { return } - self.captureBluetoothStatus(context: context) - } + scheduleBluetoothStatusSample(after: 0, context: context) } private func scheduleBluetoothStatusSample(after delay: TimeInterval, context: String) { - bleQueue.asyncAfter(deadline: .now() + delay) { [weak self] in - guard let self = self else { return } - self.captureBluetoothStatus(context: context) + #if os(iOS) + // Sample the main-actor background budget first (async hop, never a + // sync wait), then log from bleQueue off the cache — bleQueue must + // never block on main (see captureBluetoothStatus). + DispatchQueue.main.asyncAfter(deadline: .now() + delay) { [weak self] in + guard let self else { return } + self.refreshCachedBackgroundTimeRemaining() + self.bleQueue.async { self.captureBluetoothStatus(context: context) } } + #else + bleQueue.asyncAfter(deadline: .now() + delay) { [weak self] in + self?.captureBluetoothStatus(context: context) + } + #endif } + #if os(iOS) + /// Main thread only (reads main-actor UIApplication state). + private func refreshCachedBackgroundTimeRemaining() { + dispatchPrecondition(condition: .onQueue(.main)) + let seconds = UIApplication.shared.backgroundTimeRemaining + backgroundTimeLock.lock() + _cachedBackgroundTimeRemaining = seconds + backgroundTimeLock.unlock() + } + #endif + private func captureBluetoothStatus(context: String) { assert(DispatchQueue.getSpecific(key: bleQueueKey) != nil, "captureBluetoothStatus must run on bleQueue") @@ -2750,11 +2827,15 @@ extension BLEService { } #if os(iOS) - var backgroundDescriptor = "" - var backgroundSeconds: TimeInterval = 0 - DispatchQueue.main.sync { - backgroundSeconds = UIApplication.shared.backgroundTimeRemaining - } + // INVARIANT: bleQueue must NEVER sync-dispatch to the main thread. + // The main actor sync-waits on bleQueue along the send paths + // (readLinkState), so a main.sync here completes an ABBA deadlock — + // field-verified as a permanent freeze when a courier-drop storm put + // an ack send (main → bleQueue.sync) up against a status capture + // (bleQueue → main.sync). backgroundTimeRemaining is main-actor + // state, so it is sampled on main and cached. + let backgroundSeconds = cachedBackgroundTimeRemaining + let backgroundDescriptor: String if backgroundSeconds == .greatestFiniteMagnitude { backgroundDescriptor = " bgRemaining=∞" } else { @@ -3044,6 +3125,15 @@ extension BLEService { private func sendNoisePayload(_ typedPayload: Data, to peerID: PeerID) { + // Hop like sendMessage: the Transport-facing wrappers (verify/vouch/ + // group payloads) call this from the main actor, and the send path + // sync-waits on bleQueue for link state. + if DispatchQueue.getSpecific(key: messageQueueKey) == nil { + messageQueue.async { [weak self] in + self?.sendNoisePayload(typedPayload, to: peerID) + } + return + } guard noiseService.hasSession(with: peerID) else { // No session yet - queue the payload SYNCHRONOUSLY before initiating handshake // to prevent race where fast handshake completion drains empty queue @@ -3291,6 +3381,23 @@ extension BLEService { SecureLogger.warning("⚠️ Courier envelope carried unsupported payload type", category: .session) return } + let payload = Data(typedPayload.dropFirst()) + guard let innerMessageID = PrivateMessagePacket.decode(from: payload)?.messageID else { + SecureLogger.warning("⚠️ Courier envelope carried undecodable private message", category: .session) + return + } + // Redundant copies of one message arrive as distinct envelopes + // (fresh seal each: mesh couriers, bridge drops across relays), + // so dedup here on the inner message ID — before delivery, ack, + // and handshake work. A duplicate costs only the decrypt above + // and at most one ack ever goes out per message ID. + let firstOpen = collectionsQueue.sync(flags: .barrier) { + openedCourierMessageIDs.insert(innerMessageID) + } + guard firstOpen else { + SecureLogger.debug("📦 Dropping duplicate courier envelope for message \(innerMessageID.prefix(8))…", category: .session) + return + } // Couriered mail arrives while the sender is absent, so the UI's // block check can't resolve their fingerprint from a live session. // Gate here, where the full static key is in hand. @@ -3306,7 +3413,6 @@ extension BLEService { let shortID = PeerID(publicKey: senderStaticKey) let isKnownOnMesh = collectionsQueue.sync { peerRegistry.info(for: shortID) != nil } let senderPeerID = isKnownOnMesh ? shortID : PeerID(hexData: senderStaticKey) - let payload = Data(typedPayload.dropFirst()) SecureLogger.debug("📦 Opened courier envelope from \(senderPeerID.id.prefix(8))…", category: .session) sfMetrics?.record(.courierOpened) notifyUI { [weak self] in @@ -3745,6 +3851,7 @@ extension BLEService { #if os(iOS) @objc private func appDidBecomeActive() { isAppActive = true + refreshCachedBackgroundTimeRemaining() // Restart scanning with allow duplicates when app becomes active if centralManager?.state == .poweredOn { centralManager?.stopScan() @@ -3758,6 +3865,7 @@ extension BLEService { @objc private func appDidEnterBackground() { isAppActive = false + refreshCachedBackgroundTimeRemaining() // Restart scanning without allow duplicates in background if centralManager?.state == .poweredOn { centralManager?.stopScan() @@ -3851,6 +3959,15 @@ extension BLEService { // MARK: Private Message Handling private func sendPrivateMessage(_ content: String, to recipientID: PeerID, messageID: String) { + // Hop like sendMessage: the Transport-facing wrappers call this from + // the main actor (router sends, favorite notifications), and the send + // path sync-waits on bleQueue for link state. + if DispatchQueue.getSpecific(key: messageQueueKey) == nil { + messageQueue.async { [weak self] in + self?.sendPrivateMessage(content, to: recipientID, messageID: messageID) + } + return + } // Sessions and wire recipient IDs are keyed by the short 16-hex form; // callers may pass the full 64-hex noise key (mirrors sendFilePrivate). let recipientID = recipientID.toShort() diff --git a/bitchat/Services/Gateway/BoundedIDSet.swift b/bitchat/Services/Gateway/BoundedIDSet.swift index 5cb61744..8509c295 100644 --- a/bitchat/Services/Gateway/BoundedIDSet.swift +++ b/bitchat/Services/Gateway/BoundedIDSet.swift @@ -32,14 +32,4 @@ struct BoundedIDSet { } return true } - - /// Releases a previously inserted ID so it can be re-added later. Used - /// when a tentatively-recorded item is later abandoned (e.g. a queued - /// drop evicted before it was ever published) and must become retryable. - mutating func remove(_ id: String) { - guard members.remove(id) != nil else { return } - if let index = order.firstIndex(of: id) { - order.remove(at: index) - } - } } diff --git a/bitchat/Services/Gateway/BridgeCourierService.swift b/bitchat/Services/Gateway/BridgeCourierService.swift index 8e777897..1f8bb1f9 100644 --- a/bitchat/Services/Gateway/BridgeCourierService.swift +++ b/bitchat/Services/Gateway/BridgeCourierService.swift @@ -9,6 +9,11 @@ import BitFoundation import BitLogger import Foundation +#if os(iOS) +import UIKit +#elseif os(macOS) +import AppKit +#endif /// Courier delivery over the internet bridge: sealed courier envelopes are /// parked on relays as kind-1401 "drops" tagged with their rotating @@ -48,6 +53,10 @@ final class BridgeCourierService: ObservableObject { /// Encoded envelope cap for a drop (16 KiB ciphertext + TLV slack). static let maxDropEnvelopeBytes = 20 * 1024 static let maxTrackedIDs = 512 + /// Coalescing window for dedup-record writes: a backlog re-fetch + /// mutates the seen set once per event, and each snapshot save is a + /// full JSON encode + atomic write on the main actor. + static let dedupPersistCoalesceSeconds: TimeInterval = 1.0 } static let shared = BridgeCourierService() @@ -70,7 +79,9 @@ final class BridgeCourierService: ObservableObject { /// Opens a drop addressed to us (tag verified inside). var openEnvelope: (@MainActor (CourierEnvelope) -> Void)? /// Hands a drop to a matching local peer as a directed courier packet. - var deliverToPeer: (@MainActor (CourierEnvelope, PeerID) -> Void)? + /// Returns false when the handoff could not even be attempted (peer no + /// longer reachable), so the drop event stays retryable. + var deliverToPeer: (@MainActor (CourierEnvelope, PeerID) -> Bool)? /// Held envelopes eligible for (re)publish, honoring the cooldown. var heldEnvelopes: (@MainActor (TimeInterval) -> [CourierEnvelope])? /// Timer injection for tests; nil arms a real `Task`. @@ -81,21 +92,90 @@ final class BridgeCourierService: ObservableObject { private(set) var myTagsHex: Set = [] private(set) var watchedPeerTags: [(peerID: PeerID, tagsHex: Set)] = [] private(set) var pendingDrops: [(envelope: CourierEnvelope, dedupKey: String?)] = [] - /// Message IDs already published as drops (sender-side dedup). - private var publishedDropKeys: BoundedIDSet - /// Drop event IDs already handled (multi-relay dedup). - private var seenDropEventIDs: BoundedIDSet + /// Message IDs already published as drops (sender-side dedup) and drop + /// event IDs already handled (multi-relay dedup). Both persist across + /// relaunches: relays hold drops for the full 24h NIP-40 window and the + /// persisted outbox keeps re-depositing, so in-memory-only dedup meant + /// every relaunch republished the same message as a fresh drop and every + /// gateway relaunch re-delivered the whole backlog (field-verified + /// amplification storm). Entries age out with the 24h drop window. + private var publishedDropKeys: ExpiringIDSet + private var seenDropEventIDs: ExpiringIDSet private var subscriptionOpen = false private var lastSubscribedTags: Set = [] private var refreshTimerArmed = false private var lastAnnounceRefresh = Date.distantPast private let now: () -> Date + private let dedupStore: BridgeDropDedupStore - init(now: @escaping () -> Date = Date.init) { + private var dedupPersistScheduled = false + + init(now: @escaping () -> Date = Date.init, dedupStore: BridgeDropDedupStore? = nil) { self.now = now - self.publishedDropKeys = BoundedIDSet(capacity: Limits.maxTrackedIDs) - self.seenDropEventIDs = BoundedIDSet(capacity: Limits.maxTrackedIDs) + self.dedupStore = dedupStore ?? BridgeDropDedupStore(persistsToDisk: !TestEnvironment.isRunningTests) + let snapshot = self.dedupStore.load() + let date = now() + self.publishedDropKeys = ExpiringIDSet( + capacity: Limits.maxTrackedIDs, + lifetime: CourierEnvelope.maxLifetimeSeconds, + entries: snapshot.publishedDropKeys, + now: date + ) + self.seenDropEventIDs = ExpiringIDSet( + capacity: Limits.maxTrackedIDs, + lifetime: CourierEnvelope.maxLifetimeSeconds, + entries: snapshot.seenDropEventIDs, + now: date + ) + // A coalesced dedup write scheduled just before a background kill + // would be lost; flush when the app backgrounds or terminates. + #if os(iOS) + let flushNotifications = [UIApplication.didEnterBackgroundNotification, UIApplication.willTerminateNotification] + #else + let flushNotifications = [NSApplication.willTerminateNotification] + #endif + for name in flushNotifications { + NotificationCenter.default.addObserver(forName: name, object: nil, queue: .main) { [weak self] _ in + MainActor.assumeIsolated { self?.flushDedupSnapshot() } + } + } + } + + /// Schedules a coalesced write of the dedup record (see + /// `Limits.dedupPersistCoalesceSeconds`); lifecycle notifications flush + /// any scheduled write before a background kill could drop it. + private func persistDedup() { + guard !dedupPersistScheduled else { return } + dedupPersistScheduled = true + Task { @MainActor [weak self] in + try? await Task.sleep(nanoseconds: UInt64(Limits.dedupPersistCoalesceSeconds * 1_000_000_000)) + guard let self else { return } + self.dedupPersistScheduled = false + self.flushDedupSnapshot() + } + } + + /// Writes the dedup record now. Sender keys still sitting in the + /// in-memory `pendingDrops` queue are excluded: their drop is not durable + /// until it actually reaches a relay, and persisting the key early would + /// turn "app killed before relays connected" into a silent 24h blackhole + /// (the relaunch loses the queued drop but the persisted key blocks every + /// re-deposit). `flushPendingDrops` re-persists once they publish. + func flushDedupSnapshot() { + let pendingKeys = Set(pendingDrops.compactMap(\.dedupKey)) + dedupStore.save(BridgeDropDedupStore.Snapshot( + publishedDropKeys: publishedDropKeys.entries.filter { !pendingKeys.contains($0.key) }, + seenDropEventIDs: seenDropEventIDs.entries + )) + } + + /// Panic wipe: forget queued drops and the persisted dedup record. + func wipe() { + pendingDrops.removeAll() + publishedDropKeys = ExpiringIDSet(capacity: Limits.maxTrackedIDs, lifetime: CourierEnvelope.maxLifetimeSeconds) + seenDropEventIDs = ExpiringIDSet(capacity: Limits.maxTrackedIDs, lifetime: CourierEnvelope.maxLifetimeSeconds) + dedupStore.wipe() } // MARK: - Sender role @@ -108,14 +188,15 @@ final class BridgeCourierService: ObservableObject { @discardableResult func depositDrop(content: String, messageID: String, recipientNoiseKey: Data) -> Bool { guard bridgeEnabled?() ?? false else { return false } - guard !publishedDropKeys.contains(messageID) else { return false } + guard !publishedDropKeys.contains(messageID, now: now()) else { return false } guard let envelope = sealEnvelope?(content, messageID, recipientNoiseKey) else { return false } // An envelope that can't encode within the drop size caps fails the // same way on every attempt (size is a function of the content, not // of the sealing); consume the dedup slot so the retry sweep stops // re-running Noise sealing on a drop that can never ship. guard let encoded = envelope.encode(), encoded.count <= Limits.maxDropEnvelopeBytes else { - publishedDropKeys.insert(messageID) + publishedDropKeys.insert(messageID, now: now()) + persistDedup() return false } // Only consume the sender-side dedup slot once the drop is durably @@ -124,7 +205,8 @@ final class BridgeCourierService: ObservableObject { // router's retry sweep can attempt a fresh deposit rather than // marking the message "carried" and blocking retries forever. guard publishDrop(envelope, messageID: messageID) else { return false } - publishedDropKeys.insert(messageID) + publishedDropKeys.insert(messageID, now: now()) + persistDedup() return true } @@ -153,7 +235,10 @@ final class BridgeCourierService: ObservableObject { let evicted = pendingDrops.removeFirst() // The oldest queued drop is being dropped before it ever // published; release its dedup slot so it stays retryable. - if let key = evicted.dedupKey { publishedDropKeys.remove(key) } + if let key = evicted.dedupKey { + publishedDropKeys.remove(key) + persistDedup() + } } return true } @@ -180,8 +265,13 @@ final class BridgeCourierService: ObservableObject { for item in queued where !publishDrop(item.envelope, messageID: item.dedupKey) { // Compose failed with relays up: release the slot so the router's // retry sweep can attempt a fresh deposit. - if let key = item.dedupKey { publishedDropKeys.remove(key) } + if let key = item.dedupKey { + publishedDropKeys.remove(key) + } } + // Flushed keys just became durable (published, so no longer excluded + // as pending) or were released above; either way the record changed. + persistDedup() } // MARK: - Subscription (recipient + gateway watch) @@ -265,7 +355,8 @@ final class BridgeCourierService: ObservableObject { func handleDropEvent(_ event: NostrEvent) { guard bridgeEnabled?() ?? false else { return } guard event.kind == NostrProtocol.EventKind.courierDrop.rawValue else { return } - guard seenDropEventIDs.insert(event.id) else { return } + guard seenDropEventIDs.insert(event.id, now: now()) else { return } + persistDedup() guard let data = Data(base64Encoded: event.content), data.count <= Limits.maxDropEnvelopeBytes, let envelope = CourierEnvelope.decode(data), @@ -285,7 +376,14 @@ final class BridgeCourierService: ObservableObject { } if let match = watchedPeerTags.first(where: { $0.tagsHex.contains(tagHex) }) { SecureLogger.info("📦🌉 Courier drop fetched for local peer \(match.peerID.id.prefix(8))…", category: .session) - deliverToPeer?(envelope, match.peerID) + if deliverToPeer?(envelope, match.peerID) != true { + // The best-effort handoff never left this device (the peer + // walked away between the relay fetch and the mesh send). + // Release the seen slot so a relaunch or backlog redelivery + // retries — a single-gateway island has no other carrier. + seenDropEventIDs.remove(event.id) + persistDedup() + } } } diff --git a/bitchat/Services/Gateway/BridgeDropDedupStore.swift b/bitchat/Services/Gateway/BridgeDropDedupStore.swift new file mode 100644 index 00000000..76a0fe28 --- /dev/null +++ b/bitchat/Services/Gateway/BridgeDropDedupStore.swift @@ -0,0 +1,137 @@ +// +// BridgeDropDedupStore.swift +// bitchat +// +// This is free and unencumbered software released into the public domain. +// For more information, see +// + +import BitLogger +import Foundation + +/// ID set with per-entry timestamps: entries expire after `lifetime` and the +/// oldest are evicted past `capacity`. The bridge's dedup caches use this +/// instead of `BoundedIDSet` because their contents persist across relaunches +/// and must age out with the 24h drop window they guard. +struct ExpiringIDSet { + private(set) var entries: [String: Date] + let capacity: Int + let lifetime: TimeInterval + + init(capacity: Int, lifetime: TimeInterval, entries: [String: Date] = [:], now: Date = Date()) { + self.capacity = capacity + self.lifetime = lifetime + self.entries = entries + prune(now: now) + } + + func contains(_ id: String, now: Date) -> Bool { + guard let recorded = entries[id] else { return false } + return now.timeIntervalSince(recorded) <= lifetime + } + + /// Returns false when the ID was already present (and unexpired). + @discardableResult + mutating func insert(_ id: String, now: Date) -> Bool { + guard !contains(id, now: now) else { return false } + entries[id] = now + prune(now: now) + return true + } + + /// Releases a previously inserted ID so it can be re-added later (e.g. a + /// queued drop evicted before it ever published must become retryable). + mutating func remove(_ id: String) { + entries.removeValue(forKey: id) + } + + private mutating func prune(now: Date) { + if entries.contains(where: { now.timeIntervalSince($0.value) > lifetime }) { + entries = entries.filter { now.timeIntervalSince($0.value) <= lifetime } + } + let overflow = entries.count - capacity + guard overflow > 0 else { return } + for (id, _) in entries.sorted(by: { $0.value < $1.value }).prefix(overflow) { + entries.removeValue(forKey: id) + } + } +} + +/// Disk persistence for the bridge courier's drop-dedup record. Relays hold +/// drops for the full 24h NIP-40 window and redeliver them on every launch, +/// and the 120s outbox sweep re-deposits anything undelivered — so with +/// in-memory-only dedup every relaunch republished the same message as a +/// fresh drop (fresh throwaway seal, undeduplicatable downstream) and every +/// gateway relaunch re-delivered the whole backlog. Field-verified: ~20 +/// copies of one DM delivered in 40ms fed the storm behind a permanent +/// device freeze. Persisting both sides caps this at one drop per message +/// ID per 24h regardless of relaunch count. +/// +/// Contents are opaque IDs (message UUIDs, relay event IDs) — no plaintext, +/// no peer identities — so until-first-unlock protection matches +/// `NostrProcessedEventStore`, and the file must load during a +/// locked-background restoration relaunch. Wiped on panic with the rest of +/// the courier state. +final class BridgeDropDedupStore { + struct Snapshot: Codable { + var publishedDropKeys: [String: Date] + var seenDropEventIDs: [String: Date] + } + + private let fileURL: URL? + + /// - Parameter fileURL: Overrides the on-disk location (tests). Ignored + /// when `persistsToDisk` is false. + init(persistsToDisk: Bool = true, fileURL: URL? = nil) { + self.fileURL = persistsToDisk ? (fileURL ?? Self.defaultFileURL()) : nil + } + + func load() -> Snapshot { + guard let fileURL, + let data = try? Data(contentsOf: fileURL), + let snapshot = try? JSONDecoder().decode(Snapshot.self, from: data) else { + return Snapshot(publishedDropKeys: [:], seenDropEventIDs: [:]) + } + return snapshot + } + + func save(_ snapshot: Snapshot) { + guard let fileURL else { return } + guard !(snapshot.publishedDropKeys.isEmpty && snapshot.seenDropEventIDs.isEmpty) else { + try? FileManager.default.removeItem(at: fileURL) + return + } + do { + try FileManager.default.createDirectory( + at: fileURL.deletingLastPathComponent(), + withIntermediateDirectories: true + ) + let data = try JSONEncoder().encode(snapshot) + var options: Data.WritingOptions = [.atomic] + #if os(iOS) + options.insert(.completeFileProtectionUntilFirstUserAuthentication) + #endif + try data.write(to: fileURL, options: options) + } catch { + SecureLogger.error("Failed to persist bridge drop dedup record: \(error)", category: .session) + } + } + + /// Panic wipe: forget which drops we published or handled. + func wipe() { + guard let fileURL else { return } + try? FileManager.default.removeItem(at: fileURL) + } + + private static func defaultFileURL() -> URL? { + guard let base = try? FileManager.default.url( + for: .applicationSupportDirectory, + in: .userDomainMask, + appropriateFor: nil, + create: true + ) else { return nil } + return base + .appendingPathComponent("courier", isDirectory: true) + .appendingPathComponent("bridge-drop-dedup.json") + } +} diff --git a/bitchat/Services/MessageRouter.swift b/bitchat/Services/MessageRouter.swift index 819c30c7..ff2656df 100644 --- a/bitchat/Services/MessageRouter.swift +++ b/bitchat/Services/MessageRouter.swift @@ -351,9 +351,15 @@ final class MessageRouter { } private func enqueue(_ message: QueuedMessage, for peerID: PeerID) { + var message = message var queue = outbox[peerID] ?? [] - // Re-sending an already-queued ID replaces the entry (keeps attempt count fresh) - queue.removeAll { $0.messageID == message.messageID } + // Re-sending an already-queued ID replaces the entry (keeps attempt + // count fresh) but must not forget which couriers already carry it, + // or the replacement re-burns the same courier slots. + if let existing = queue.firstIndex(where: { $0.messageID == message.messageID }) { + message.depositedCourierKeys.formUnion(queue[existing].depositedCourierKeys) + queue.remove(at: existing) + } queue.append(message) // Enforce per-peer size limit with FIFO eviction diff --git a/bitchat/Services/TransportConfig.swift b/bitchat/Services/TransportConfig.swift index ac42aa9f..8a80afee 100644 --- a/bitchat/Services/TransportConfig.swift +++ b/bitchat/Services/TransportConfig.swift @@ -345,6 +345,9 @@ enum TransportConfig { // Cooldown between speculative multi-hop handovers of the same envelope // toward a recipient heard only via relayed announces. static let courierRemoteHandoverCooldownSeconds: TimeInterval = 10 * 60 + // Recently opened courier inner message IDs kept for receiver-side dedup + // (redundant copies ride distinct seals, so only the inner ID matches). + static let courierOpenedMessageIDCap: Int = 512 // One-time prekey bundles (forward-secret courier sealing) // Own gossip-sync round for bundles: modest cadence, bounded peer count, diff --git a/bitchat/ViewModels/ChatViewModel.swift b/bitchat/ViewModels/ChatViewModel.swift index 082cfb63..49bed683 100644 --- a/bitchat/ViewModels/ChatViewModel.swift +++ b/bitchat/ViewModels/ChatViewModel.swift @@ -1173,6 +1173,7 @@ final class ChatViewModel: ObservableObject, BitchatDelegate, TransportEventDele // our own queued outbox, the carried public history, and the // counters describing all of it CourierStore.shared.wipe() + BridgeCourierService.shared.wipe() messageRouter.wipeOutbox() GossipMessageArchive.wipeDefault() StoreAndForwardMetrics.shared.reset() diff --git a/bitchat/ViewModels/ChatViewModelBootstrapper.swift b/bitchat/ViewModels/ChatViewModelBootstrapper.swift index c331fe00..40794ddc 100644 --- a/bitchat/ViewModels/ChatViewModelBootstrapper.swift +++ b/bitchat/ViewModels/ChatViewModelBootstrapper.swift @@ -567,7 +567,7 @@ private extension ChatViewModelBootstrapper { bleService?.openBridgedCourierEnvelope(envelope) } courier.deliverToPeer = { [weak bleService] envelope, peerID in - bleService?.deliverBridgedEnvelope(envelope, to: peerID) + bleService?.deliverBridgedEnvelope(envelope, to: peerID) ?? false } courier.heldEnvelopes = { cooldown in CourierStore.shared.envelopesForBridgePublish(cooldown: cooldown) diff --git a/bitchatTests/EndToEnd/CourierEndToEndTests.swift b/bitchatTests/EndToEnd/CourierEndToEndTests.swift index 00f210f3..af0af7a9 100644 --- a/bitchatTests/EndToEnd/CourierEndToEndTests.swift +++ b/bitchatTests/EndToEnd/CourierEndToEndTests.swift @@ -556,6 +556,70 @@ struct CourierEndToEndTests { #expect(!stored) } + /// Regression for the relaunch amplification storm: redundant copies of + /// one message arrive as *distinct* envelopes (every seal uses a fresh + /// ephemeral key), so only the inner message ID can dedup them. The first + /// copy delivers; duplicates must stop right after the decrypt — no + /// second delivery, no second ack/handshake trigger downstream. + @Test func duplicateCourierCopiesDeliverInnerMessageOnce() async throws { + let alice = makeService() + let bob = makeService() + let bobDelegate = NoiseCaptureDelegate() + bob.delegate = bobDelegate + + let bobKey = bob.noiseStaticPublicKeyData() + let first = try #require(alice.sealBridgeCourierEnvelope("once", messageID: "dup-1", recipientNoiseKey: bobKey)) + let second = try #require(alice.sealBridgeCourierEnvelope("once", messageID: "dup-1", recipientNoiseKey: bobKey)) + // Distinct seals: envelope-level dedup can never catch this pair. + #expect(first.ciphertext != second.ciphertext) + + #expect(bob.openBridgedCourierEnvelope(first)) + #expect(bob.openBridgedCourierEnvelope(second)) + + let delivered = await TestHelpers.waitUntil( + { !bobDelegate.snapshot().isEmpty }, + timeout: TestConstants.defaultTimeout + ) + #expect(delivered) + // Give a duplicate delivery a chance to surface, then confirm the + // second copy never reached the delegate. + let duplicated = await TestHelpers.waitUntil( + { bobDelegate.snapshot().count > 1 }, + timeout: TestConstants.shortTimeout + ) + #expect(!duplicated) + #expect(bobDelegate.snapshot().count == 1) + } + + /// Acks for mail from absent senders (the usual couriered/bridged case) + /// queue for a future session instead of initiating a handshake + /// broadcast — otherwise every duplicate copy of every drop turns into a + /// mesh-wide handshake flood at an identity that cannot answer. + @Test func deliveryAckForAbsentPeerQueuesWithoutHandshake() async throws { + let ble = makeService() + let outbound = PacketTap() + ble._test_onOutboundPacket = outbound.record + + // Nobody by this ID on the mesh (not connected, not reachable). + ble.sendDeliveryAck(for: "msg-1", to: PeerID(str: "00000000000000ee")) + + let initiated = await TestHelpers.waitUntil( + { outbound.count(ofType: .noiseHandshake) > 0 }, + timeout: TestConstants.shortTimeout + ) + #expect(!initiated) + + // Control: a peer that is actually around still gets the handshake. + let present = PeerID(str: "00000000000000ef") + ble._test_seedConnectedPeer(present, nickname: "present") + ble.sendDeliveryAck(for: "msg-2", to: present) + let initiatedForPresent = await TestHelpers.waitUntil( + { outbound.count(ofType: .noiseHandshake) > 0 }, + timeout: TestConstants.defaultTimeout + ) + #expect(initiatedForPresent) + } + private func makeUnsignedAnnounce(from service: BLEService) throws -> BitchatPacket { let announcement = AnnouncementPacket( nickname: "Unsigned", diff --git a/bitchatTests/EndToEnd/PrekeyEndToEndTests.swift b/bitchatTests/EndToEnd/PrekeyEndToEndTests.swift index 456932a7..79bd28b5 100644 --- a/bitchatTests/EndToEnd/PrekeyEndToEndTests.swift +++ b/bitchatTests/EndToEnd/PrekeyEndToEndTests.swift @@ -190,7 +190,11 @@ struct PrekeyEndToEndTests { #expect(message.content == "burn after reading") // 6. Redelivery tolerance: the same envelope arriving via another - // packet (spray-and-wait) still opens inside the grace window. + // packet (spray-and-wait) still decrypts inside the prekey grace + // window (asserted at the crypto layer in NoisePrekeyTests), but + // the duplicate is absorbed before delivery — the receiver dedups + // on the inner message ID, so redundant courier copies never + // re-deliver (or re-ack) the same message. let redelivery = BitchatPacket( type: MessageType.courierEnvelope.rawValue, senderID: Data(hexString: carol.myPeerID.id) ?? Data(), @@ -203,9 +207,10 @@ struct PrekeyEndToEndTests { bob._test_handlePacket(redelivery, fromPeerID: carol.myPeerID) let redelivered = await TestHelpers.waitUntil( { bobDelegate.snapshot().count == 2 }, - timeout: TestConstants.defaultTimeout + timeout: TestConstants.shortTimeout ) - #expect(redelivered) + #expect(!redelivered) + #expect(bobDelegate.snapshot().count == 1) } @Test func withoutBundleSealingFallsBackToStatic() async throws { diff --git a/bitchatTests/Services/BridgeCourierServiceTests.swift b/bitchatTests/Services/BridgeCourierServiceTests.swift index ebd25bf1..7a71170e 100644 --- a/bitchatTests/Services/BridgeCourierServiceTests.swift +++ b/bitchatTests/Services/BridgeCourierServiceTests.swift @@ -24,6 +24,7 @@ struct BridgeCourierServiceTests { var localPeers: [(peerID: PeerID, noiseKey: Data)] = [] var held: [CourierEnvelope] = [] var sealResult: CourierEnvelope? + var deliverResult = true private(set) var publishedEvents: [NostrEvent] = [] private(set) var openedSubscriptions: [[String]] = [] @@ -35,8 +36,8 @@ struct BridgeCourierServiceTests { let service: BridgeCourierService - init() { - service = BridgeCourierService() + init(dedupStore: BridgeDropDedupStore? = nil) { + service = BridgeCourierService(dedupStore: dedupStore) service.bridgeEnabled = { [weak self] in self?.bridgeOn ?? false } service.relaysConnected = { [weak self] in self?.relaysConnected ?? false } service.publishEvent = { [weak self] event in self?.publishedEvents.append(event) } @@ -49,7 +50,10 @@ struct BridgeCourierServiceTests { return self?.sealResult } service.openEnvelope = { [weak self] envelope in self?.openedEnvelopes.append(envelope) } - service.deliverToPeer = { [weak self] envelope, peer in self?.delivered.append((envelope, peer)) } + service.deliverToPeer = { [weak self] envelope, peer in + self?.delivered.append((envelope, peer)) + return self?.deliverResult ?? false + } service.heldEnvelopes = { [weak self] cooldown in self?.heldCooldowns.append(cooldown) return self?.held ?? [] @@ -175,6 +179,134 @@ struct BridgeCourierServiceTests { #expect(fixture.sealRequests.count == 1) } + @Test func publishedDropDedupSurvivesRelaunch() throws { + // Regression (field-verified amplification storm): the outbox that + // drives re-deposits is persisted, but the sender-side drop dedup was + // in-memory only — every relaunch republished the same undelivered + // message as a fresh drop, and relays hold each for 24h. + let fileURL = FileManager.default.temporaryDirectory + .appendingPathComponent("bridge-dedup-\(UUID().uuidString).json") + defer { try? FileManager.default.removeItem(at: fileURL) } + let recipientKey = Fixture.randomKey() + let messageID = UUID().uuidString + + let fixture = Fixture(dedupStore: BridgeDropDedupStore(fileURL: fileURL)) + fixture.sealResult = makeEnvelope(recipientKey: recipientKey) + #expect(fixture.service.depositDrop(content: "hello", messageID: messageID, recipientNoiseKey: recipientKey)) + #expect(fixture.publishedEvents.count == 1) + // Persistence is coalesced; a real launch flushes within a second or + // on backgrounding — tests flush explicitly. + fixture.service.flushDedupSnapshot() + + // "Relaunch": a fresh service over the same store must refuse to + // publish the same message ID again (before even re-sealing it). + let relaunched = Fixture(dedupStore: BridgeDropDedupStore(fileURL: fileURL)) + relaunched.sealResult = makeEnvelope(recipientKey: recipientKey) + #expect(!relaunched.service.depositDrop(content: "hello", messageID: messageID, recipientNoiseKey: recipientKey)) + #expect(relaunched.publishedEvents.isEmpty) + #expect(relaunched.sealRequests.isEmpty) + } + + @Test func seenDropEventDedupSurvivesRelaunch() throws { + // Same storm, gateway side: relays redeliver the whole 24h drop + // backlog on every launch; a relaunch must not re-open (and re-ack) + // events it already handled. + let fileURL = FileManager.default.temporaryDirectory + .appendingPathComponent("bridge-dedup-\(UUID().uuidString).json") + defer { try? FileManager.default.removeItem(at: fileURL) } + + let fixture = Fixture(dedupStore: BridgeDropDedupStore(fileURL: fileURL)) + let myKey = try #require(fixture.myKey) + fixture.service.refresh() + let event = try makeDropEvent(for: makeEnvelope(recipientKey: myKey)) + fixture.service.handleDropEvent(event) + #expect(fixture.openedEnvelopes.count == 1) + fixture.service.flushDedupSnapshot() + + let relaunched = Fixture(dedupStore: BridgeDropDedupStore(fileURL: fileURL)) + relaunched.myKey = myKey + relaunched.service.refresh() + relaunched.service.handleDropEvent(event) + #expect(relaunched.openedEnvelopes.isEmpty) + } + + @Test func offlineQueuedDropStaysRedepositableAfterRelaunch() throws { + // A deposit made while relays are down only joins the in-memory + // pending queue. Its dedup key must NOT be durable yet: if the app is + // killed before relays connect, the relaunch loses the queued drop — + // a persisted key would then block every 120s re-deposit for 24h and + // the message would silently never reach a relay. + let fileURL = FileManager.default.temporaryDirectory + .appendingPathComponent("bridge-dedup-\(UUID().uuidString).json") + defer { try? FileManager.default.removeItem(at: fileURL) } + let recipientKey = Fixture.randomKey() + let messageID = UUID().uuidString + + let fixture = Fixture(dedupStore: BridgeDropDedupStore(fileURL: fileURL)) + fixture.relaysConnected = false + fixture.sealResult = makeEnvelope(recipientKey: recipientKey) + #expect(fixture.service.depositDrop(content: "later", messageID: messageID, recipientNoiseKey: recipientKey)) + #expect(fixture.publishedEvents.isEmpty) + // Even a flush while the drop is still pending must exclude its key. + fixture.service.flushDedupSnapshot() + + // "App killed before relays connected": pendingDrops were memory-only. + let relaunched = Fixture(dedupStore: BridgeDropDedupStore(fileURL: fileURL)) + relaunched.sealResult = makeEnvelope(recipientKey: recipientKey) + #expect(relaunched.service.depositDrop(content: "later", messageID: messageID, recipientNoiseKey: recipientKey)) + #expect(relaunched.publishedEvents.count == 1) + } + + @Test func publishedPendingDropBecomesDurableAfterFlush() throws { + // Counterpart: once the queued drop actually publishes on reconnect, + // its key becomes durable and a relaunch must not republish. + let fileURL = FileManager.default.temporaryDirectory + .appendingPathComponent("bridge-dedup-\(UUID().uuidString).json") + defer { try? FileManager.default.removeItem(at: fileURL) } + let recipientKey = Fixture.randomKey() + let messageID = UUID().uuidString + + let fixture = Fixture(dedupStore: BridgeDropDedupStore(fileURL: fileURL)) + fixture.relaysConnected = false + fixture.sealResult = makeEnvelope(recipientKey: recipientKey) + #expect(fixture.service.depositDrop(content: "later", messageID: messageID, recipientNoiseKey: recipientKey)) + fixture.relaysConnected = true + fixture.service.flushPendingDrops() + #expect(fixture.publishedEvents.count == 1) + fixture.service.flushDedupSnapshot() + + let relaunched = Fixture(dedupStore: BridgeDropDedupStore(fileURL: fileURL)) + relaunched.sealResult = makeEnvelope(recipientKey: recipientKey) + #expect(!relaunched.service.depositDrop(content: "later", messageID: messageID, recipientNoiseKey: recipientKey)) + #expect(relaunched.publishedEvents.isEmpty) + } + + @Test func failedGatewayHandoffReleasesSeenSlot() throws { + // A gateway's deliverToPeer handoff is best-effort: when it fails + // (the peer walked away between relay fetch and mesh send), the drop + // event must stay retryable — for a single-gateway mesh island this + // gateway is the recipient's only carrier. + let fixture = Fixture() + let peerKey = Fixture.randomKey() + let peer = PeerID(str: "aabbccdd00112233") + fixture.localPeers = [(peer, peerKey)] + fixture.service.refresh() + let event = try makeDropEvent(for: makeEnvelope(recipientKey: peerKey)) + + fixture.deliverResult = false + fixture.service.handleDropEvent(event) + #expect(fixture.delivered.count == 1) + + // Redelivery (relaunch/backlog re-fetch) retries the handoff … + fixture.deliverResult = true + fixture.service.handleDropEvent(event) + #expect(fixture.delivered.count == 2) + + // … and a successful handoff consumes the event for good. + fixture.service.handleDropEvent(event) + #expect(fixture.delivered.count == 2) + } + @Test func distinctDropsUseDistinctThrowawayKeys() { let fixture = Fixture() let keyA = Fixture.randomKey() diff --git a/bitchatTests/Services/BridgeDropDedupStoreTests.swift b/bitchatTests/Services/BridgeDropDedupStoreTests.swift new file mode 100644 index 00000000..e1507621 --- /dev/null +++ b/bitchatTests/Services/BridgeDropDedupStoreTests.swift @@ -0,0 +1,118 @@ +// +// BridgeDropDedupStoreTests.swift +// bitchat +// +// This is free and unencumbered software released into the public domain. +// For more information, see +// + +import Foundation +import Testing +@testable import bitchat + +@Suite("Bridge drop dedup persistence") +struct BridgeDropDedupStoreTests { + + // MARK: - ExpiringIDSet + + @Test func entriesExpireAfterLifetime() { + let start = Date(timeIntervalSince1970: 1_700_000_000) + var set = ExpiringIDSet(capacity: 8, lifetime: 60) + + let inserted = set.insert("a", now: start) + #expect(inserted) + #expect(set.contains("a", now: start)) + let duplicate = set.insert("a", now: start.addingTimeInterval(30)) + #expect(!duplicate) + + // Past the lifetime the slot is free again. + let later = start.addingTimeInterval(61) + #expect(!set.contains("a", now: later)) + let reinserted = set.insert("a", now: later) + #expect(reinserted) + } + + @Test func capacityEvictsOldestFirst() { + let start = Date(timeIntervalSince1970: 1_700_000_000) + var set = ExpiringIDSet(capacity: 2, lifetime: 3600) + + set.insert("oldest", now: start) + set.insert("middle", now: start.addingTimeInterval(1)) + set.insert("newest", now: start.addingTimeInterval(2)) + + let check = start.addingTimeInterval(3) + #expect(!set.contains("oldest", now: check)) + #expect(set.contains("middle", now: check)) + #expect(set.contains("newest", now: check)) + } + + @Test func removeReleasesSlot() { + let now = Date() + var set = ExpiringIDSet(capacity: 8, lifetime: 3600) + set.insert("a", now: now) + set.remove("a") + #expect(!set.contains("a", now: now)) + let reinserted = set.insert("a", now: now) + #expect(reinserted) + } + + @Test func initPrunesExpiredPersistedEntries() { + let now = Date() + let set = ExpiringIDSet( + capacity: 8, + lifetime: 3600, + entries: [ + "stale": now.addingTimeInterval(-7200), + "fresh": now.addingTimeInterval(-60), + ], + now: now + ) + #expect(!set.contains("stale", now: now)) + #expect(set.contains("fresh", now: now)) + } + + // MARK: - Store round trip + + @Test func snapshotRoundTripsThroughDisk() { + let fileURL = FileManager.default.temporaryDirectory + .appendingPathComponent("bridge-dedup-store-\(UUID().uuidString).json") + defer { try? FileManager.default.removeItem(at: fileURL) } + let recorded = Date(timeIntervalSince1970: 1_700_000_000) + + let store = BridgeDropDedupStore(fileURL: fileURL) + store.save(BridgeDropDedupStore.Snapshot( + publishedDropKeys: ["msg-1": recorded], + seenDropEventIDs: ["event-1": recorded] + )) + + let reloaded = BridgeDropDedupStore(fileURL: fileURL).load() + #expect(reloaded.publishedDropKeys["msg-1"] == recorded) + #expect(reloaded.seenDropEventIDs["event-1"] == recorded) + } + + @Test func wipeRemovesTheRecord() { + let fileURL = FileManager.default.temporaryDirectory + .appendingPathComponent("bridge-dedup-store-\(UUID().uuidString).json") + defer { try? FileManager.default.removeItem(at: fileURL) } + + let store = BridgeDropDedupStore(fileURL: fileURL) + store.save(BridgeDropDedupStore.Snapshot( + publishedDropKeys: ["msg-1": Date()], + seenDropEventIDs: [:] + )) + store.wipe() + + let reloaded = BridgeDropDedupStore(fileURL: fileURL).load() + #expect(reloaded.publishedDropKeys.isEmpty) + #expect(reloaded.seenDropEventIDs.isEmpty) + } + + @Test func nonPersistingStoreStaysEmpty() { + let store = BridgeDropDedupStore(persistsToDisk: false) + store.save(BridgeDropDedupStore.Snapshot( + publishedDropKeys: ["msg-1": Date()], + seenDropEventIDs: [:] + )) + #expect(store.load().publishedDropKeys.isEmpty) + } +} diff --git a/bitchatTests/Services/MessageRouterTests.swift b/bitchatTests/Services/MessageRouterTests.swift index ea8ba4da..5144fba8 100644 --- a/bitchatTests/Services/MessageRouterTests.swift +++ b/bitchatTests/Services/MessageRouterTests.swift @@ -446,6 +446,34 @@ struct MessageRouterTests { #expect(transport.sentCourierMessages.count == 1) } + @Test @MainActor + func enqueueReplacementCarriesOverDepositedCourierKeys() async { + // Re-sending a queued message ID replaces the outbox entry; the + // replacement must inherit which couriers already carry the message, + // or the deposit retry re-burns the same courier slots (duplicate + // sealed copies to the same peer). + let recipient = PeerID(str: "00000000000000aa") + let recipientKey = Data(repeating: 0xBB, count: 32) + let courier = PeerID(str: "00000000000000cc") + let courierKey = Data(repeating: 0xCC, count: 32) + + let transport = MockTransport() + transport.connectedPeers.insert(courier) + transport.updatePeerSnapshots([Self.snapshot(courier, key: courierKey, verified: true)]) + + let router = MessageRouter( + transports: [transport], + courierDirectory: Self.directory(recipient: recipient, recipientKey: recipientKey) + ) + router.sendPrivate("Hello", to: recipient, recipientNickname: "Peer", messageID: "ck1") + #expect(transport.sentCourierMessages.count == 1) + + // Same message ID re-sent (e.g. a resend while still queued): the + // courier already carrying it must not receive a second copy. + router.sendPrivate("Hello", to: recipient, recipientNickname: "Peer", messageID: "ck1") + #expect(transport.sentCourierMessages.count == 1) + } + @Test @MainActor func courierBecameAvailable_ignoresTheRecipientThemselves() async { let recipient = PeerID(str: "00000000000000aa")