diff --git a/bitchat/Services/BLE/BLEService.swift b/bitchat/Services/BLE/BLEService.swift index cf3cf67d..64f71aaf 100644 --- a/bitchat/Services/BLE/BLEService.swift +++ b/bitchat/Services/BLE/BLEService.swift @@ -507,45 +507,52 @@ final class BLEService: NSObject { signature: nil, 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) } } - + // Give leave message a moment to send (cooperative delay allows BLE callbacks to fire) let deadline = Date().addingTimeInterval(TransportConfig.bleThreadSleepWriteShortDelaySeconds) while Date() < deadline { RunLoop.current.run(until: Date().addingTimeInterval(0.01)) } - + // Clear pending notifications collectionsQueue.sync(flags: .barrier) { pendingNotifications.removeAll() } - + // Stop timer maintenanceTimer?.cancel() maintenanceTimer = nil scanDutyTimer?.cancel() scanDutyTimer = nil - + 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,15 +582,15 @@ final class BLEService: NSObject { // Clear processed messages messageDeduplicator.reset() - // Clear peripheral references - peripherals.removeAll() - peerToPeripheralUUID.removeAll() - subscribedCentrals.removeAll() - centralToPeerID.removeAll() + // Clear peripheral references (synchronized access to avoid races with BLE callbacks) + bleQueue.sync { + peripherals.removeAll() + peerToPeripheralUUID.removeAll() + subscribedCentrals.removeAll() + centralToPeerID.removeAll() + centralSubscriptionRateLimits.removeAll() + } meshTopology.reset() - - // BCH-01-004: Clear rate-limit state - centralSubscriptionRateLimits.removeAll() } // 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) } } @@ -1726,15 +1784,22 @@ extension BLEService: CBCentralManagerDelegate { guard let self = self, 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) - self.peripherals[peripheralID] = nil - self.recentConnectTimeouts[peripheralID] = Date() - self.failureCounts[peripheralID, default: 0] += 1 - // Try next candidate if any - self.tryConnectFromQueue() + central.cancelPeripheralConnection(peripheral) + self.peripherals[peripheralID] = nil + self.recentConnectTimeouts[peripheralID] = Date() + self.failureCounts[peripheralID, default: 0] += 1 + // Try next candidate if any + self.tryConnectFromQueue() } } @@ -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 @@ -2150,11 +2217,12 @@ extension BLEService: CBPeripheralDelegate { 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() - + // Create characteristic characteristic = CBMutableCharacteristic( type: BLEService.characteristicUUID, @@ -2162,14 +2230,54 @@ extension BLEService: CBPeripheralManagerDelegate { value: nil, permissions: [.readable, .writeable] ) - + // Create service let service = CBMutableService(type: BLEService.serviceUUID, primary: true) service.characteristics = [characteristic!] - + // 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,28 +2455,38 @@ 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,23 +2969,66 @@ 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) - + #if os(iOS) @objc private func appDidBecomeActive() { isAppActive = true @@ -2982,17 +3157,20 @@ 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) return messages } - + guard let messages = pendingMessages, !messages.isEmpty else { return } - + 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) } } @@ -4066,7 +4263,12 @@ extension BLEService { if maintenanceCounter % 2 == 1 { 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) diff --git a/bitchat/Services/MeshTopologyTracker.swift b/bitchat/Services/MeshTopologyTracker.swift index de9e6085..fdd8749f 100644 --- a/bitchat/Services/MeshTopologyTracker.swift +++ b/bitchat/Services/MeshTopologyTracker.swift @@ -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,25 +59,33 @@ 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 = [source] // Queue stores paths: [Start, Hop1, Hop2, ..., Current] var queuePaths: [[RoutingID]] = [[source]] - + while !queuePaths.isEmpty { let path = queuePaths.removeFirst() // Limit path length (path contains source + maxHops + target) -> maxHops intermediate // If maxHops = 10, max edges = 11, max nodes = 12. if path.count > maxHops + 1 { continue } - + guard let last = path.last else { continue } - + // 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 } - + // CONFIRMED EDGE CHECK: // 'last' claims 'neighbor' (checked above) // Does 'neighbor' claim 'last'? @@ -81,16 +93,21 @@ final class MeshTopologyTracker { neighborClaims.contains(last) else { 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) - + if neighbor == target { // Return only intermediate hops // Path: [Source, I1, I2, Target] -> [I1, I2] return Array(nextPath.dropFirst().dropLast()) } - + visited.insert(neighbor) queuePaths.append(nextPath) } diff --git a/bitchat/Services/MessageRouter.swift b/bitchat/Services/MessageRouter.swift index c231f22c..cb771487 100644 --- a/bitchat/Services/MessageRouter.swift +++ b/bitchat/Services/MessageRouter.swift @@ -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) + } + } + } } diff --git a/bitchat/Services/NoiseEncryptionService.swift b/bitchat/Services/NoiseEncryptionService.swift index 9b5ff0cb..86caaecb 100644 --- a/bitchat/Services/NoiseEncryptionService.swift +++ b/bitchat/Services/NoiseEncryptionService.swift @@ -616,6 +616,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 diff --git a/bitchat/Services/TransportConfig.swift b/bitchat/Services/TransportConfig.swift index 49d01928..d544c6df 100644 --- a/bitchat/Services/TransportConfig.swift +++ b/bitchat/Services/TransportConfig.swift @@ -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 diff --git a/bitchatTests/BLEServiceTests.swift b/bitchatTests/BLEServiceTests.swift index af6eefdd..51e833c6 100644 --- a/bitchatTests/BLEServiceTests.swift +++ b/bitchatTests/BLEServiceTests.swift @@ -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) } } }