Improve BLE mesh reliability for large transfers (#964)

* Improve BLE mesh reliability for large transfers Major reliability improvements for fragment-based transfers (photos, files): **Notification Queue Fixes** - Fix silent packet loss when notification queue is full - now queues for retry - Fix retry queue bug where remaining items were lost when one retry failed - Add periodic drain mechanism as backup (every 5 seconds) **Write Queue Fixes** - Fix drainPendingWrites to use atomic take-send-requeue pattern - Add logging when peripheral is ready for more writes - Add periodic drain for pending writes as backup **Fragment Pacing** - Increase fragment spacing from 4-5ms to 25-30ms to prevent buffer overflow - Conservative pacing prevents packet loss on congested BLE connections **Thread Safety** - Fix race conditions in stopServices() and emergencyDisconnectAll() - Synchronize access to peripherals/centrals dictionaries during cleanup - Clear pending message queues in emergencyDisconnectAll() **Error Recovery** - Add handlers for all BLE state transitions (poweredOff, unauthorized, etc.) - Clear Noise session and re-initiate handshake on decryption failure - Queue ACKs/receipts for delivery after handshake instead of dropping - Re-queue failed pending messages for retry **Other Improvements** - Add route freshness validation in MeshTopologyTracker (60s threshold) - Add TTL (24h) and size limits (100 per peer) for MessageRouter outbox - Fix connection timeout to check peripheral.state before canceling - Add NoiseEncryptionService.clearSession(for:) method Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com> * Fix race condition and increase test timeouts - Fix race in sendNoisePayload: use sync barrier instead of async to ensure payload is queued before initiating handshake (addresses Codex review feedback) - Increase BLEServiceTests sleep from 0.5s to 1.0s for CI reliability Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com> --------- Co-authored-by: jack <jackjackbits@users.noreply.github.com> Co-authored-by: Claude Opus 4.5 <noreply@anthropic.com>
2026-07-24 22:45:19 +00:00 · 2026-01-17 07:39:31 -10:00
parent 806c420313
commit 9964710de2
6 changed files with 366 additions and 94 deletions
@@ -508,19 +508,25 @@ final class BLEService: NSObject {
            ttl: messageTTL
        )

-        // Send immediately to all connected peers
+        // Send immediately to all connected peers (synchronized access to BLE state)
        if let data = leavePacket.toBinaryData(padding: false) {
            let leavePriority = priority(for: leavePacket, data: data)
+
+            // Snapshot BLE state under bleQueue to avoid races with delegate callbacks
+            let (peripheralStates, centralsCount, char) = bleQueue.sync {
+                (Array(peripherals.values), subscribedCentrals.count, characteristic)
+            }
+
            // Send to peripherals we're connected to as central
-            for state in peripherals.values where state.isConnected {
+            for state in peripheralStates where state.isConnected {
                if let characteristic = state.characteristic {
                    writeOrEnqueue(data, to: state.peripheral, characteristic: characteristic, priority: leavePriority)
                }
            }

            // Send to centrals subscribed to us as peripheral
-            if subscribedCentrals.count > 0 && characteristic != nil {
-                peripheralManager?.updateValue(data, for: characteristic!, onSubscribedCentrals: nil)
+            if centralsCount > 0, let ch = char {
+                peripheralManager?.updateValue(data, for: ch, onSubscribedCentrals: nil)
            }
        }

@@ -544,8 +550,9 @@ final class BLEService: NSObject {
        centralManager?.stopScan()
        peripheralManager?.stopAdvertising()

-        // Disconnect all peripherals
-        for state in peripherals.values {
+        // Disconnect all peripherals (synchronized access)
+        let peripheralsToDisconnect = bleQueue.sync { Array(peripherals.values) }
+        for state in peripheralsToDisconnect {
            centralManager?.cancelPeripheralConnection(state.peripheral)
        }
    }
@@ -560,6 +567,10 @@ final class BLEService: NSObject {
            incomingFragments.removeAll()
            fragmentMetadata.removeAll()
            activeTransfers.removeAll()
+            // Also clear pending message queues to avoid stale state across sessions
+            pendingMessagesAfterHandshake.removeAll()
+            pendingNoisePayloadsAfterHandshake.removeAll()
+            pendingDirectedRelays.removeAll()
            return entries
        }

@@ -571,16 +582,16 @@ final class BLEService: NSObject {
        // Clear processed messages
        messageDeduplicator.reset()

-        // Clear peripheral references
+        // Clear peripheral references (synchronized access to avoid races with BLE callbacks)
+        bleQueue.sync {
            peripherals.removeAll()
            peerToPeripheralUUID.removeAll()
            subscribedCentrals.removeAll()
            centralToPeerID.removeAll()
-        meshTopology.reset()
-
-        // BCH-01-004: Clear rate-limit state
            centralSubscriptionRateLimits.removeAll()
        }
+        meshTopology.reset()
+    }
    
    // MARK: Connectivity and peers
    
@@ -999,7 +1010,13 @@ final class BLEService: NSObject {
        if let ch = characteristic {
            let targets = subscribedCentrals.filter { selectedCentralIDs.contains($0.identifier.uuidString) }
            if !targets.isEmpty {
-                _ = peripheralManager?.updateValue(data, for: ch, onSubscribedCentrals: targets)
+                let success = peripheralManager?.updateValue(data, for: ch, onSubscribedCentrals: targets) ?? false
+                if !success {
+                    // Notification queue full - queue for retry to prevent silent packet loss
+                    // This is critical for fragment delivery reliability
+                    let context = packet.type == MessageType.fragment.rawValue ? "fragment" : "broadcast"
+                    enqueuePendingNotification(data: data, centrals: targets, context: context)
+                }
            }
        }
    }
@@ -1583,9 +1600,50 @@ extension BLEService: CBCentralManagerDelegate {
            self.delegate?.didUpdateBluetoothState(central.state)
        }

-        if central.state == .poweredOn {
+        switch central.state {
+        case .poweredOn:
            // Start scanning - use allow duplicates for faster discovery when active
            startScanning()
+
+        case .poweredOff:
+            // Bluetooth was turned off - stop scanning and clean up connection state
+            SecureLogger.info("📴 Bluetooth powered off - cleaning up central state", category: .session)
+            central.stopScan()
+            // Mark all peripheral connections as disconnected (they are now invalid)
+            let peerIDs: [PeerID] = peripherals.compactMap { $0.value.peerID }
+            for state in peripherals.values {
+                central.cancelPeripheralConnection(state.peripheral)
+            }
+            peripherals.removeAll()
+            peerToPeripheralUUID.removeAll()
+            // Notify UI of disconnections
+            for peerID in peerIDs {
+                notifyUI { [weak self] in
+                    self?.notifyPeerDisconnectedDebounced(peerID)
+                }
+            }
+
+        case .unauthorized:
+            // User denied Bluetooth permission
+            SecureLogger.warning("🚫 Bluetooth unauthorized - user denied permission", category: .session)
+            central.stopScan()
+            peripherals.removeAll()
+            peerToPeripheralUUID.removeAll()
+
+        case .unsupported:
+            // Device doesn't support BLE
+            SecureLogger.error("❌ Bluetooth LE not supported on this device", category: .session)
+
+        case .resetting:
+            // Bluetooth stack is resetting - will get another state update when done
+            SecureLogger.info("🔄 Bluetooth stack resetting...", category: .session)
+
+        case .unknown:
+            // Initial state before we know the actual state
+            SecureLogger.debug("❓ Bluetooth state unknown (initializing)", category: .session)
+
+        @unknown default:
+            SecureLogger.warning("⚠️ Unknown Bluetooth state: \(central.state.rawValue)", category: .session)
        }
    }
    
@@ -1727,6 +1785,13 @@ extension BLEService: CBCentralManagerDelegate {
                  let state = self.peripherals[peripheralID],
                  state.isConnecting && !state.isConnected else { return }

+            // Double-check actual CBPeripheral state to avoid canceling a just-connected peripheral
+            // This prevents a race where connection completes just as timeout fires
+            guard peripheral.state != .connected else {
+                SecureLogger.debug("⏱️ Timeout fired but peripheral already connected: \(advertisedName)", category: .session)
+                return
+            }
+
            // Connection timed out - cancel it
            SecureLogger.debug("⏱️ Timeout: \(advertisedName)", category: .session)
            central.cancelPeripheralConnection(peripheral)
@@ -2110,7 +2175,8 @@ extension BLEService: CBPeripheralDelegate {
    }
    
    func peripheralIsReady(toSendWriteWithoutResponse peripheral: CBPeripheral) {
-        // Resume queued writes for this peripheral
+        // Resume queued writes for this peripheral - called when canSendWriteWithoutResponse becomes true again
+        SecureLogger.debug("📤 Peripheral \(peripheral.name ?? peripheral.identifier.uuidString.prefix(8).description) ready for more writes", category: .session)
        drainPendingWrites(for: peripheral)
    }
    
@@ -2143,6 +2209,7 @@ extension BLEService: CBPeripheralDelegate {
            }
        }
    }
+
 }

 // MARK: - CBPeripheralManagerDelegate
@@ -2151,7 +2218,8 @@ extension BLEService: CBPeripheralManagerDelegate {
    func peripheralManagerDidUpdateState(_ peripheral: CBPeripheralManager) {
        SecureLogger.debug("📡 Peripheral manager state: \(peripheral.state.rawValue)", category: .session)

-        if peripheral.state == .poweredOn {
+        switch peripheral.state {
+        case .poweredOn:
            // Remove all services first to ensure clean state
            peripheral.removeAllServices()

@@ -2170,6 +2238,46 @@ extension BLEService: CBPeripheralManagerDelegate {
            // Add service (advertising will start in didAdd delegate)
            SecureLogger.debug("🔧 Adding BLE service...", category: .session)
            peripheral.add(service)
+
+        case .poweredOff:
+            // Bluetooth was turned off - clean up peripheral state
+            SecureLogger.info("📴 Bluetooth powered off - cleaning up peripheral state", category: .session)
+            peripheral.stopAdvertising()
+            // Clear subscribed centrals (they are now invalid)
+            let centralPeerIDs = centralToPeerID.values.map { $0 }
+            subscribedCentrals.removeAll()
+            centralToPeerID.removeAll()
+            centralSubscriptionRateLimits.removeAll()
+            characteristic = nil
+            // Notify UI of disconnections
+            for peerID in centralPeerIDs {
+                notifyUI { [weak self] in
+                    self?.notifyPeerDisconnectedDebounced(peerID)
+                }
+            }
+
+        case .unauthorized:
+            // User denied Bluetooth permission
+            SecureLogger.warning("🚫 Bluetooth unauthorized for peripheral role", category: .session)
+            peripheral.stopAdvertising()
+            subscribedCentrals.removeAll()
+            centralToPeerID.removeAll()
+            centralSubscriptionRateLimits.removeAll()
+            characteristic = nil
+
+        case .unsupported:
+            // Device doesn't support BLE peripheral role
+            SecureLogger.error("❌ Bluetooth LE peripheral role not supported", category: .session)
+
+        case .resetting:
+            // Bluetooth stack is resetting
+            SecureLogger.info("🔄 Bluetooth peripheral stack resetting...", category: .session)
+
+        case .unknown:
+            SecureLogger.debug("❓ Peripheral Bluetooth state unknown (initializing)", category: .session)
+
+        @unknown default:
+            SecureLogger.warning("⚠️ Unknown peripheral Bluetooth state: \(peripheral.state.rawValue)", category: .session)
        }
    }
    
@@ -2347,29 +2455,39 @@ extension BLEService: CBPeripheralManagerDelegate {
            self.pendingNotifications.removeAll()
            
            // Try to send pending notifications
-            for (data, centrals) in pending {
+            var sentCount = 0
+            for (index, (data, centrals)) in pending.enumerated() {
                if let centrals = centrals {
                    // Send to specific centrals
                    let success = self.peripheralManager?.updateValue(data, for: characteristic, onSubscribedCentrals: centrals) ?? false
                    if !success {
-                        // Still full, re-queue
-                        self.pendingNotifications.append((data: data, centrals: centrals))
-                        SecureLogger.debug("⚠️ Notification queue still full, re-queuing", category: .session)
+                        // Still full, re-queue this and all remaining items
+                        let remaining = pending.dropFirst(index)
+                        self.pendingNotifications.append(contentsOf: remaining)
+                        SecureLogger.debug("⚠️ Notification queue still full after \(sentCount) sent, re-queuing \(remaining.count) items", category: .session)
                        break  // Stop trying, wait for next ready callback
                    } else {
-                        SecureLogger.debug("✅ Sent pending notification from retry queue", category: .session)
+                        sentCount += 1
                    }
                } else {
                    // Broadcast to all
                    let success = self.peripheralManager?.updateValue(data, for: characteristic, onSubscribedCentrals: nil) ?? false
                    if !success {
-                        // Still full, re-queue
-                        self.pendingNotifications.append((data: data, centrals: nil))
+                        // Still full, re-queue this and all remaining items
+                        let remaining = pending.dropFirst(index)
+                        self.pendingNotifications.append(contentsOf: remaining)
+                        SecureLogger.debug("⚠️ Notification queue still full after \(sentCount) sent, re-queuing \(remaining.count) items", category: .session)
                        break
+                    } else {
+                        sentCount += 1
                    }
                }
            }

+            if sentCount > 0 {
+                SecureLogger.debug("✅ Sent \(sentCount) pending notifications from retry queue", category: .session)
+            }
+            
            if !self.pendingNotifications.isEmpty {
                SecureLogger.debug("📋 Still have \(self.pendingNotifications.count) pending notifications", category: .session)
            }
@@ -2685,7 +2803,15 @@ extension BLEService {
    
    private func sendNoisePayload(_ typedPayload: Data, to peerID: PeerID) {
        guard noiseService.hasSession(with: peerID) else {
-            // Lazy-handshake path: queue? For now, initiate handshake and drop
+            // No session yet - queue the payload SYNCHRONOUSLY before initiating handshake
+            // to prevent race where fast handshake completion drains empty queue
+            collectionsQueue.sync(flags: .barrier) {
+                if self.pendingNoisePayloadsAfterHandshake[peerID] == nil {
+                    self.pendingNoisePayloadsAfterHandshake[peerID] = []
+                }
+                self.pendingNoisePayloadsAfterHandshake[peerID]?.append(typedPayload)
+                SecureLogger.debug("📥 Queued noise payload for \(peerID) pending handshake", category: .session)
+            }
            initiateNoiseHandshake(with: peerID)
            return
        }
@@ -2823,13 +2949,19 @@ extension BLEService {
        bleQueue.async { [weak self] in
            guard let self = self else { return }
            guard let state = self.peripherals[uuid], let ch = state.characteristic else { return }
-            var queueCopy: [PendingWrite] = []
-            self.collectionsQueue.sync {
-                queueCopy = self.pendingPeripheralWrites[uuid] ?? []
+
+            // Atomically take all pending items from the queue to avoid race conditions
+            // where new items could be enqueued between read and update
+            let itemsToSend: [PendingWrite] = self.collectionsQueue.sync(flags: .barrier) {
+                let items = self.pendingPeripheralWrites[uuid] ?? []
+                self.pendingPeripheralWrites[uuid] = nil
+                return items
            }
-            guard !queueCopy.isEmpty else { return }
+            guard !itemsToSend.isEmpty else { return }
+
+            // Send as many as possible
            var sent = 0
-            for item in queueCopy {
+            for item in itemsToSend {
                if peripheral.canSendWriteWithoutResponse {
                    peripheral.writeValue(item.data, for: ch, type: .withoutResponse)
                    sent += 1
@@ -2837,19 +2969,62 @@ extension BLEService {
                    break
                }
            }
-            if sent > 0 {
+
+            // Re-enqueue any items that couldn't be sent (maintaining order)
+            let unsent = Array(itemsToSend.dropFirst(sent))
+            if !unsent.isEmpty {
                self.collectionsQueue.async(flags: .barrier) {
-                    var q = self.pendingPeripheralWrites[uuid] ?? []
-                    if sent > 0 {
-                        let toRemove = min(sent, q.count)
-                        if toRemove > 0 {
-                            q.removeFirst(toRemove)
-                        }
-                        self.pendingPeripheralWrites[uuid] = q.isEmpty ? nil : q
+                    var existing = self.pendingPeripheralWrites[uuid] ?? []
+                    // Prepend unsent items to maintain priority order
+                    existing.insert(contentsOf: unsent, at: 0)
+                    self.pendingPeripheralWrites[uuid] = existing
                }
            }
        }
    }
+
+    /// Periodically try to drain pending notifications as a backup mechanism
+    private func drainPendingNotificationsIfPossible() {
+        collectionsQueue.async(flags: .barrier) { [weak self] in
+            guard let self = self,
+                  let characteristic = self.characteristic,
+                  !self.pendingNotifications.isEmpty else { return }
+
+            let pending = self.pendingNotifications
+            self.pendingNotifications.removeAll()
+
+            var sentCount = 0
+            for (index, (data, centrals)) in pending.enumerated() {
+                let success: Bool
+                if let centrals = centrals {
+                    success = self.peripheralManager?.updateValue(data, for: characteristic, onSubscribedCentrals: centrals) ?? false
+                } else {
+                    success = self.peripheralManager?.updateValue(data, for: characteristic, onSubscribedCentrals: nil) ?? false
+                }
+
+                if !success {
+                    // Re-queue this and all remaining items
+                    let remaining = pending.dropFirst(index)
+                    self.pendingNotifications.append(contentsOf: remaining)
+                    break
+                } else {
+                    sentCount += 1
+                }
+            }
+
+            if sentCount > 0 {
+                SecureLogger.debug("🔄 Periodic drain: sent \(sentCount) pending notifications", category: .session)
+            }
+        }
+    }
+
+    /// Periodically try to drain pending writes for all connected peripherals
+    private func drainAllPendingWrites() {
+        let uuids = collectionsQueue.sync { Array(pendingPeripheralWrites.keys) }
+        for uuid in uuids {
+            guard let state = peripherals[uuid], state.isConnected else { continue }
+            drainPendingWrites(for: state.peripheral)
+        }
    }

    // MARK: Application State Handlers (iOS)
@@ -2982,7 +3157,7 @@ extension BLEService {
    }
    
    private func sendPendingMessagesAfterHandshake(for peerID: PeerID) {
-        // Get and clear pending messages for this peer
+        // Atomically take all pending messages to process (prevents concurrent modification)
        let pendingMessages = collectionsQueue.sync(flags: .barrier) { () -> [(content: String, messageID: String)]? in
            let messages = pendingMessagesAfterHandshake[peerID]
            pendingMessagesAfterHandshake.removeValue(forKey: peerID)
@@ -2993,6 +3168,9 @@ extension BLEService {

        SecureLogger.debug("📤 Sending \(messages.count) pending messages after handshake to \(peerID)", category: .session)

+        // Track failed messages for re-queuing
+        var failedMessages: [(content: String, messageID: String)] = []
+
        // Send each pending message directly (we know session is established)
        for (content, messageID) in messages {
            do {
@@ -3000,6 +3178,7 @@ extension BLEService {
                let privateMessage = PrivateMessagePacket(messageID: messageID, content: content)
                guard let tlvData = privateMessage.encode() else {
                    SecureLogger.error("Failed to encode pending private message TLV")
+                    failedMessages.append((content, messageID))
                    continue
                }

@@ -3029,6 +3208,7 @@ extension BLEService {
                SecureLogger.debug("✅ Sent pending message \(messageID) to \(peerID) after handshake", category: .session)
            } catch {
                SecureLogger.error("Failed to send pending message after handshake: \(error)")
+                failedMessages.append((content, messageID))

                // Notify delegate of failure
                notifyUI { [weak self] in
@@ -3036,6 +3216,19 @@ extension BLEService {
                }
            }
        }
+
+        // Re-queue any failed messages for retry on next handshake
+        if !failedMessages.isEmpty {
+            collectionsQueue.async(flags: .barrier) { [weak self] in
+                guard let self = self else { return }
+                if self.pendingMessagesAfterHandshake[peerID] == nil {
+                    self.pendingMessagesAfterHandshake[peerID] = []
+                }
+                // Prepend failed messages to maintain order
+                self.pendingMessagesAfterHandshake[peerID]?.insert(contentsOf: failedMessages, at: 0)
+                SecureLogger.warning("⚠️ Re-queued \(failedMessages.count) failed messages for \(peerID)", category: .session)
+            }
+        }
    }
    
    // MARK: Fragmentation (Required for messages > BLE MTU)
@@ -3925,7 +4118,11 @@ extension BLEService {
                initiateNoiseHandshake(with: peerID)
            }
        } catch {
-            SecureLogger.error("❌ Failed to decrypt message from \(peerID): \(error)")
+            // Decryption failed - clear the corrupted session and re-initiate handshake
+            // This handles cases where session state got out of sync (nonce mismatch, etc.)
+            SecureLogger.error("❌ Failed to decrypt message from \(peerID): \(error) - clearing session and re-initiating handshake")
+            noiseService.clearSession(for: peerID)
+            initiateNoiseHandshake(with: peerID)
        }
    }

@@ -4067,6 +4264,11 @@ extension BLEService {
            flushDirectedSpool()
        }

+        // Periodically attempt to drain pending notifications and writes as backup
+        // in case callbacks are missed or delayed (every maintenance cycle = 5 seconds)
+        drainPendingNotificationsIfPossible()
+        drainAllPendingWrites()
+
        // No rotating alias: nothing to refresh
        
        // Reset counter to prevent overflow (every 60 seconds)
@@ -11,6 +11,10 @@ final class MeshTopologyTracker {
    // Last time we received an update from a node
    private var lastSeen: [RoutingID: Date] = [:]

+    // Maximum age for topology claims to be considered fresh for routing
+    // Routes computed using stale topology can fail when the network has changed
+    private static let routeFreshnessThreshold: TimeInterval = 60 // 60 seconds
+
    func reset() {
        queue.sync(flags: .barrier) {
            self.claims.removeAll()
@@ -55,6 +59,9 @@ final class MeshTopologyTracker {
        if source == target { return [] } // Direct connection, no intermediate hops

        return queue.sync {
+            let now = Date()
+            let freshnessDeadline = now.addingTimeInterval(-Self.routeFreshnessThreshold)
+
            // BFS
            var visited: Set<RoutingID> = [source]
            // Queue stores paths: [Start, Hop1, Hop2, ..., Current]
@@ -71,6 +78,11 @@ final class MeshTopologyTracker {
                // Get neighbors that 'last' claims to see
                guard let neighbors = claims[last] else { continue }

+                // Check if 'last' node's topology info is fresh
+                guard let lastSeenTime = lastSeen[last], lastSeenTime > freshnessDeadline else {
+                    continue // Skip stale nodes
+                }
+
                for neighbor in neighbors {
                    if visited.contains(neighbor) { continue }

@@ -82,6 +94,11 @@ final class MeshTopologyTracker {
                        continue
                    }

+                    // Check if 'neighbor' node's topology info is fresh
+                    guard let neighborSeenTime = lastSeen[neighbor], neighborSeenTime > freshnessDeadline else {
+                        continue // Skip edges to stale nodes
+                    }
+
                    var nextPath = path
                    nextPath.append(neighbor)

@@ -5,7 +5,20 @@ import Foundation
@MainActor
 final class MessageRouter {
    private let transports: [Transport]
-    private var outbox: [PeerID: [(content: String, nickname: String, messageID: String)]] = [:] // peerID -> queued messages
+
+    // Outbox entry with timestamp for TTL-based eviction
+    private struct QueuedMessage {
+        let content: String
+        let nickname: String
+        let messageID: String
+        let timestamp: Date
+    }
+
+    private var outbox: [PeerID: [QueuedMessage]] = [:]
+
+    // Outbox limits to prevent unbounded memory growth
+    private static let maxMessagesPerPeer = 100
+    private static let messageTTLSeconds: TimeInterval = 24 * 60 * 60 // 24 hours

    init(transports: [Transport]) {
        self.transports = transports
@@ -51,10 +64,19 @@ final class MessageRouter {
            SecureLogger.debug("Routing PM via \(type(of: transport)) to \(peerID.id.prefix(8))… id=\(messageID.prefix(8))…", category: .session)
            transport.sendPrivateMessage(content, to: peerID, recipientNickname: recipientNickname, messageID: messageID)
        } else {
-            // Queue for later
+            // Queue for later with timestamp for TTL tracking
            if outbox[peerID] == nil { outbox[peerID] = [] }
-            outbox[peerID]?.append((content, recipientNickname, messageID))
-            SecureLogger.debug("Queued PM for \(peerID.id.prefix(8))… (no reachable transport) id=\(messageID.prefix(8))…", category: .session)
+
+            let message = QueuedMessage(content: content, nickname: recipientNickname, messageID: messageID, timestamp: Date())
+            outbox[peerID]?.append(message)
+
+            // Enforce per-peer size limit with FIFO eviction
+            if let count = outbox[peerID]?.count, count > Self.maxMessagesPerPeer {
+                let evicted = outbox[peerID]?.removeFirst()
+                SecureLogger.warning("📤 Outbox overflow for \(peerID.id.prefix(8))… - evicted oldest message: \(evicted?.messageID.prefix(8) ?? "?")…", category: .session)
+            }
+
+            SecureLogger.debug("Queued PM for \(peerID.id.prefix(8))… (no reachable transport) id=\(messageID.prefix(8))… queue=\(outbox[peerID]?.count ?? 0)", category: .session)
        }
    }

@@ -87,14 +109,22 @@ final class MessageRouter {
    func flushOutbox(for peerID: PeerID) {
        guard let queued = outbox[peerID], !queued.isEmpty else { return }
        SecureLogger.debug("Flushing outbox for \(peerID.id.prefix(8))… count=\(queued.count)", category: .session)
-        var remaining: [(content: String, nickname: String, messageID: String)] = []

-        for (content, nickname, messageID) in queued {
+        let now = Date()
+        var remaining: [QueuedMessage] = []
+
+        for message in queued {
+            // Skip expired messages (TTL exceeded)
+            if now.timeIntervalSince(message.timestamp) > Self.messageTTLSeconds {
+                SecureLogger.debug("⏰ Expired queued message for \(peerID.id.prefix(8))… id=\(message.messageID.prefix(8))… (age: \(Int(now.timeIntervalSince(message.timestamp)))s)", category: .session)
+                continue
+            }
+
            if let transport = reachableTransport(for: peerID) {
-                SecureLogger.debug("Outbox -> \(type(of: transport)) for \(peerID.id.prefix(8))… id=\(messageID.prefix(8))…", category: .session)
-                transport.sendPrivateMessage(content, to: peerID, recipientNickname: nickname, messageID: messageID)
+                SecureLogger.debug("Outbox -> \(type(of: transport)) for \(peerID.id.prefix(8))… id=\(message.messageID.prefix(8))…", category: .session)
+                transport.sendPrivateMessage(message.content, to: peerID, recipientNickname: message.nickname, messageID: message.messageID)
            } else {
-                remaining.append((content, nickname, messageID))
+                remaining.append(message)
            }
        }

@@ -108,4 +138,15 @@ final class MessageRouter {
    func flushAllOutbox() {
        for key in Array(outbox.keys) { flushOutbox(for: key) }
    }
+
+    /// Periodically clean up expired messages from all outboxes
+    func cleanupExpiredMessages() {
+        let now = Date()
+        for peerID in Array(outbox.keys) {
+            outbox[peerID]?.removeAll { now.timeIntervalSince($0.timestamp) > Self.messageTTLSeconds }
+            if outbox[peerID]?.isEmpty == true {
+                outbox.removeValue(forKey: peerID)
+            }
+        }
+    }
 }
@@ -617,6 +617,17 @@ final class NoiseEncryptionService {
        rateLimiter.resetAll()
    }

+    /// Clear session for a specific peer (e.g., on decryption failure to allow re-handshake)
+    func clearSession(for peerID: PeerID) {
+        sessionManager.removeSession(for: peerID)
+        serviceQueue.sync(flags: .barrier) {
+            if let fingerprint = peerFingerprints.removeValue(forKey: peerID) {
+                fingerprintToPeerID.removeValue(forKey: fingerprint)
+            }
+        }
+        SecureLogger.debug("🔓 Cleared Noise session for \(peerID)", category: .session)
+    }
+    
    // MARK: - Private Helpers
    
    private func handleSessionEstablished(peerID: PeerID, remoteStaticKey: Curve25519.KeyAgreement.PublicKey) {
@@ -96,11 +96,12 @@ enum TransportConfig {
    // Keep scanning fully ON when we saw traffic very recently
    static let bleRecentTrafficForceScanSeconds: TimeInterval = 10.0
    static let bleThreadSleepWriteShortDelaySeconds: TimeInterval = 0.05
-    static let bleExpectedWritePerFragmentMs: Int = 8
-    static let bleExpectedWriteMaxMs: Int = 2000
-    // Faster fragment pacing; use slightly tighter spacing for directed trains
-    static let bleFragmentSpacingMs: Int = 5
-    static let bleFragmentSpacingDirectedMs: Int = 4
+    static let bleExpectedWritePerFragmentMs: Int = 20
+    static let bleExpectedWriteMaxMs: Int = 5000
+    // Fragment pacing: Conservative spacing to prevent BLE buffer overflow
+    // Aggressive pacing causes packet loss; needs 25-30ms between fragments for reliable delivery
+    static let bleFragmentSpacingMs: Int = 30
+    static let bleFragmentSpacingDirectedMs: Int = 25
    static let bleAnnounceIntervalSeconds: TimeInterval = 4.0
    static let bleDutyOnDurationDense: TimeInterval = 3.0
    static let bleDutyOffDurationDense: TimeInterval = 15.0
@@ -73,7 +73,7 @@ struct BLEServiceTests {
            service.sendMessage("Hello, world!")
            
            // Allow async processing
-            try await sleep(0.5)
+            try await sleep(1.0)
        }
        #expect(service.sentMessages.count == 1)
    }
@@ -97,7 +97,7 @@ struct BLEServiceTests {
            )
            
            // Allow async processing
-            try await sleep(0.5)
+            try await sleep(1.0)
        }
        #expect(service.sentMessages.count == 1)
    }
@@ -113,7 +113,7 @@ struct BLEServiceTests {
            service.sendMessage("@alice @bob check this out", mentions: ["alice", "bob"])
            
            // Allow async processing
-            try await sleep(0.5)
+            try await sleep(1.0)
        }
    }
    
@@ -146,7 +146,7 @@ struct BLEServiceTests {
            service.simulateIncomingMessage(incomingMessage)
            
            // Allow async processing
-            try await sleep(0.5)
+            try await sleep(1.0)
        }
    }
    
@@ -189,7 +189,7 @@ struct BLEServiceTests {
            service.simulateIncomingPacket(packet)
            
            // Allow async processing
-            try await sleep(0.5)
+            try await sleep(1.0)
        }
    }
    
@@ -231,7 +231,7 @@ struct BLEServiceTests {
            service.sendMessage("Test delivery")
            
            // Allow async processing
-            try await sleep(0.5)
+            try await sleep(1.0)
        }
    }
    
@@ -273,7 +273,7 @@ struct BLEServiceTests {
            service.simulateIncomingPacket(packet)
            
            // Allow async processing
-            try await sleep(0.5)
+            try await sleep(1.0)
        }
    }
 }