Improve BLE mesh relay and flooding

Add last-hop suppression using ingress-link tracking to prevent echo. Implement deterministic always-relay for handshakes and directed encrypted/fragments; widen jitter for broadcasts; keep TTL cap only for broadcast. Add deterministic K-of-N broadcast fanout to reduce amplification in dense topologies. Introduce backpressure-aware writes using canSendWriteWithoutResponse with per-peripheral queues and draining on peripheralIsReady. Minor helpers for messageID, deterministic selection, and maintenance cleanup.
2026-07-25 04:05:20 +00:00 · 2025-08-25 11:03:07 +02:00
parent acc11101ad
commit fcfe32b754
2 changed files with 198 additions and 19 deletions
@@ -113,6 +113,16 @@ final class BLEService: NSObject {
    // Track short-lived traffic bursts to adapt announces/scanning under load
    private var recentPacketTimestamps: [Date] = []

+    // Ingress link tracking for last-hop suppression
+    private enum LinkID: Hashable {
+        case peripheral(String)
+        case central(String)
+    }
+    private var ingressByMessageID: [String: (link: LinkID, timestamp: Date)] = [:]
+
+    // Backpressure-aware write queue per peripheral
+    private var pendingPeripheralWrites: [String: [Data]] = [:]
+    
    // MARK: - Maintenance Timer
    
    private var maintenanceTimer: DispatchSourceTimer?  // Single timer for all maintenance tasks
@@ -157,6 +167,88 @@ final class BLEService: NSObject {
        }
    }

+    // MARK: - Helpers: IDs, selection, and write backpressure
+    private func makeMessageID(for packet: BitchatPacket) -> String {
+        let senderID = packet.senderID.hexEncodedString()
+        return "\(senderID)-\(packet.timestamp)-\(packet.type)"
+    }
+
+    private func subsetSizeForFanout(_ n: Int) -> Int {
+        guard n > 0 else { return 0 }
+        if n <= 2 { return n }
+        // approx ceil(log2(n)) + 1 without floating point
+        var v = n - 1
+        var bits = 0
+        while v > 0 { v >>= 1; bits += 1 }
+        return min(n, max(1, bits + 1))
+    }
+
+    private func selectDeterministicSubset(ids: [String], k: Int, seed: String) -> Set<String> {
+        guard k > 0 && ids.count > k else { return Set(ids) }
+        // Stable order by SHA256(seed || "::" || id)
+        var scored: [(score: [UInt8], id: String)] = []
+        for id in ids {
+            let msg = (seed + "::" + id).data(using: .utf8) ?? Data()
+            let digest = Array(SHA256.hash(data: msg))
+            scored.append((digest, id))
+        }
+        scored.sort { a, b in
+            for i in 0..<min(a.score.count, b.score.count) {
+                if a.score[i] != b.score[i] { return a.score[i] < b.score[i] }
+            }
+            return a.id < b.id
+        }
+        return Set(scored.prefix(k).map { $0.id })
+    }
+
+    private func writeOrEnqueue(_ data: Data, to peripheral: CBPeripheral, characteristic: CBCharacteristic) {
+        // BLE operations run on bleQueue; keep queue affinity
+        bleQueue.async { [weak self] in
+            guard let self = self else { return }
+            let uuid = peripheral.identifier.uuidString
+            if peripheral.canSendWriteWithoutResponse {
+                peripheral.writeValue(data, for: characteristic, type: .withoutResponse)
+            } else {
+                self.collectionsQueue.async(flags: .barrier) {
+                    self.pendingPeripheralWrites[uuid, default: []].append(data)
+                }
+            }
+        }
+    }
+
+    private func drainPendingWrites(for peripheral: CBPeripheral) {
+        let uuid = peripheral.identifier.uuidString
+        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: [Data] = []
+            self.collectionsQueue.sync {
+                queueCopy = self.pendingPeripheralWrites[uuid] ?? []
+            }
+            guard !queueCopy.isEmpty else { return }
+            var sent = 0
+            for item in queueCopy {
+                if peripheral.canSendWriteWithoutResponse {
+                    peripheral.writeValue(item, for: ch, type: .withoutResponse)
+                    sent += 1
+                } else {
+                    break
+                }
+            }
+            if sent > 0 {
+                self.collectionsQueue.async(flags: .barrier) {
+                    var q = self.pendingPeripheralWrites[uuid] ?? []
+                    if sent <= q.count {
+                        q.removeFirst(sent)
+                    } else {
+                        q.removeAll()
+                    }
+                    self.pendingPeripheralWrites[uuid] = q.isEmpty ? nil : q
+                }
+            }
+        }
+    }
+    
    // MARK: - Peer snapshots publisher (non-UI convenience)
    private let peerSnapshotSubject = PassthroughSubject<[TransportPeerSnapshot], Never>()
    var peerSnapshotPublisher: AnyPublisher<[TransportPeerSnapshot], Never> {
@@ -369,7 +461,7 @@ final class BLEService: NSObject {
            // Send to peripherals we're connected to as central
            for state in peripherals.values where state.isConnected {
                if let characteristic = state.characteristic {
-                    state.peripheral.writeValue(data, for: characteristic, type: .withoutResponse)
+                    writeOrEnqueue(data, to: state.peripheral, characteristic: characteristic)
                }
            }
            
@@ -877,7 +969,7 @@ final class BLEService: NSObject {
           let state = (DispatchQueue.getSpecific(key: bleQueueKey) != nil) ? peripherals[peripheralUUID] : bleQueue.sync(execute: { peripherals[peripheralUUID] }),
           state.isConnected,
           let characteristic = state.characteristic {
-            state.peripheral.writeValue(data, for: characteristic, type: .withoutResponse)
+            writeOrEnqueue(data, to: state.peripheral, characteristic: characteristic)
            sentEncrypted = true
        }

@@ -908,6 +1000,10 @@ final class BLEService: NSObject {
    }

    private func sendOnAllLinks(packet: BitchatPacket, data: Data, pad: Bool, directedOnlyPeer: String?) {
+        // Determine last-hop link for this message to avoid echoing back
+        let messageID = makeMessageID(for: packet)
+        let ingressLink: LinkID? = collectionsQueue.sync { ingressByMessageID[messageID]?.link }
+
        let states = snapshotPeripheralStates()
        var minCentralWriteLen: Int?
        for s in states where s.isConnected {
@@ -932,15 +1028,54 @@ final class BLEService: NSObject {
            sendFragmentedPacket(packet, pad: pad, maxChunk: chunk, directedOnlyPeer: directedOnlyPeer)
            return
        }
-        // Writes to connected peripherals
+        // Build link lists and apply K-of-N fanout for broadcasts; always exclude ingress link
+        let connectedPeripheralIDs: [String] = states.filter { $0.isConnected }.map { $0.peripheral.identifier.uuidString }
+        let subscribedCentrals: [CBCentral]
+        var centralIDs: [String] = []
+        if let _ = characteristic {
+            let (centrals, _) = snapshotSubscribedCentrals()
+            subscribedCentrals = centrals
+            centralIDs = centrals.map { $0.identifier.uuidString }
+        } else {
+            subscribedCentrals = []
+        }
+
+        // Exclude ingress link
+        var allowedPeripheralIDs = connectedPeripheralIDs
+        var allowedCentralIDs = centralIDs
+        if let ingress = ingressLink {
+            switch ingress {
+            case .peripheral(let id):
+                allowedPeripheralIDs.removeAll { $0 == id }
+            case .central(let id):
+                allowedCentralIDs.removeAll { $0 == id }
+            }
+        }
+
+        // For broadcast (no directed peer) and non-fragment, choose a subset deterministically
+        var selectedPeripheralIDs = Set(allowedPeripheralIDs)
+        var selectedCentralIDs = Set(allowedCentralIDs)
+        if directedOnlyPeer == nil && packet.type != MessageType.fragment.rawValue {
+            let kp = subsetSizeForFanout(allowedPeripheralIDs.count)
+            let kc = subsetSizeForFanout(allowedCentralIDs.count)
+            selectedPeripheralIDs = selectDeterministicSubset(ids: allowedPeripheralIDs, k: kp, seed: messageID)
+            selectedCentralIDs = selectDeterministicSubset(ids: allowedCentralIDs, k: kc, seed: messageID)
+        }
+
+        // Writes to selected connected peripherals
        for s in states where s.isConnected {
+            let pid = s.peripheral.identifier.uuidString
+            guard selectedPeripheralIDs.contains(pid) else { continue }
            if let ch = s.characteristic {
-                s.peripheral.writeValue(data, for: ch, type: .withoutResponse)
+                writeOrEnqueue(data, to: s.peripheral, characteristic: ch)
            }
        }
-        // Notify all subscribed centrals
+        // Notify selected subscribed centrals
        if let ch = characteristic {
-            _ = peripheralManager?.updateValue(data, for: ch, onSubscribedCentrals: nil)
+            let targets = subscribedCentrals.filter { selectedCentralIDs.contains($0.identifier.uuidString) }
+            if !targets.isEmpty {
+                _ = peripheralManager?.updateValue(data, for: ch, onSubscribedCentrals: targets)
+            }
        }
    }
    
@@ -954,7 +1089,7 @@ final class BLEService: NSObject {
        
        // Fire-and-forget principle: always use .withoutResponse for speed
        // CoreBluetooth will handle fragmentation at L2CAP layer
-        peripheral.writeValue(data, for: characteristic, type: .withoutResponse)
+        writeOrEnqueue(data, to: peripheral, characteristic: characteristic)
    }
    
    // MARK: - Fragmentation (Required for messages > BLE MTU)
@@ -1162,6 +1297,7 @@ final class BLEService: NSObject {
                ttl: packet.ttl,
                senderIsSelf: senderID == myPeerID,
                isEncrypted: packet.type == MessageType.noiseEncrypted.rawValue,
+                isDirectedEncrypted: (packet.type == MessageType.noiseEncrypted.rawValue) && (packet.recipientID != nil),
                isDirectedFragment: packet.type == MessageType.fragment.rawValue && packet.recipientID != nil,
                isHandshake: packet.type == MessageType.noiseHandshake.rawValue,
                degree: degree,
@@ -1727,6 +1863,15 @@ final class BLEService: NSObject {
                }
            }
        }
+
+        // Clean ingress link records older than 3 seconds
+        collectionsQueue.async(flags: .barrier) { [weak self] in
+            guard let self = self else { return }
+            let cutoff = now.addingTimeInterval(-3)
+            if !self.ingressByMessageID.isEmpty {
+                self.ingressByMessageID = self.ingressByMessageID.filter { $0.value.timestamp >= cutoff }
+            }
+        }
    }

    private func updateScanningDutyCycle(connectedCount: Int) {
@@ -2245,12 +2390,22 @@ extension BLEService: CBPeripheralDelegate {
                peerToPeripheralUUID[senderID] = peripheralUUID
                // Mapping update - direct announce from peer
            }
+            // Record ingress link for last-hop suppression and process
+            let msgID = makeMessageID(for: packet)
+            collectionsQueue.async(flags: .barrier) { [weak self] in
+                self?.ingressByMessageID[msgID] = (.peripheral(peripheralUUID), Date())
+            }
            // Process the announce packet regardless of whether we updated the mapping
            handleReceivedPacket(packet, from: senderID)
        } else {
            // For non-announce packets, DO NOT update mappings
            // These could be relayed packets from other peers
            // Always use the packet's original senderID
+            // Record ingress link for last-hop suppression and process
+            let msgID = makeMessageID(for: packet)
+            collectionsQueue.async(flags: .barrier) { [weak self] in
+                self?.ingressByMessageID[msgID] = (.peripheral(peripheralUUID), Date())
+            }
            handleReceivedPacket(packet, from: senderID)
        }
    }
@@ -2265,7 +2420,8 @@ extension BLEService: CBPeripheralDelegate {
    }
    
    func peripheralIsReady(toSendWriteWithoutResponse peripheral: CBPeripheral) {
-        // Suppress verbose ready logs
+        // Resume queued writes for this peripheral
+        drainPendingWrites(for: peripheral)
    }
    
    func peripheral(_ peripheral: CBPeripheral, didModifyServices invalidatedServices: [CBService]) {
@@ -2489,8 +2645,18 @@ extension BLEService: CBPeripheralManagerDelegate {
                }
                if packet.type == MessageType.announce.rawValue {
                    if packet.ttl == messageTTL { centralToPeerID[centralUUID] = senderID }
+                    // Record ingress link for last-hop suppression then process
+                    let msgID = makeMessageID(for: packet)
+                    collectionsQueue.async(flags: .barrier) { [weak self] in
+                        self?.ingressByMessageID[msgID] = (.central(centralUUID), Date())
+                    }
                    handleReceivedPacket(packet, from: senderID)
                } else {
+                    // Record ingress link for last-hop suppression then process
+                    let msgID = makeMessageID(for: packet)
+                    collectionsQueue.async(flags: .barrier) { [weak self] in
+                        self?.ingressByMessageID[msgID] = (.central(centralUUID), Date())
+                    }
                    handleReceivedPacket(packet, from: senderID)
                }
            } else {
@@ -12,6 +12,7 @@ struct RelayController {
    static func decide(ttl: UInt8,
                       senderIsSelf: Bool,
                       isEncrypted: Bool,
+                       isDirectedEncrypted: Bool,
                       isDirectedFragment: Bool,
                       isHandshake: Bool,
                       degree: Int,
@@ -19,7 +20,17 @@ struct RelayController {
        // Suppress obvious non-relays
        if ttl <= 1 || senderIsSelf { return RelayDecision(shouldRelay: false, newTTL: ttl, delayMs: 0) }

-        // Degree-aware probability to reduce floods in dense graphs
+        // For session-critical or directed traffic, be deterministic and reliable
+        if isHandshake || isDirectedFragment || isDirectedEncrypted {
+            // Always relay with no TTL cap for these types
+            let newTTL = (ttl &- 1)
+            // Slight jitter to desynchronize without adding too much latency
+            let delayRange: ClosedRange<Int> = isHandshake ? 20...60 : 40...120
+            let delayMs = Int.random(in: delayRange)
+            return RelayDecision(shouldRelay: true, newTTL: newTTL, delayMs: delayMs)
+        }
+
+        // Degree-aware probability to reduce floods in dense graphs (broadcast/public)
        let baseProb: Double
        switch degree {
        case 0...2: baseProb = 1.0
@@ -28,20 +39,22 @@ struct RelayController {
        case 7...9: baseProb = 0.55
        default:    baseProb = 0.45
        }
-        var prob = baseProb
-        if isHandshake { prob = max(0.3, baseProb - 0.2) }
-
-        // Sample a forwarding decision
+        let prob = baseProb
        let shouldRelay = Double.random(in: 0...1) <= prob

-        // TTL clamping in dense graphs
+        // TTL clamping in dense graphs (only for broadcast)
        let ttlCap: UInt8 = degree >= highDegreeThreshold ? 3 : 5
        let clamped = max(1, min(ttl, ttlCap))
        let newTTL = clamped &- 1

-        // Short jitter to desynchronize rebroadcasts
-        let delayMs = Int.random(in: 20...80)
+        // Wider jitter window to allow duplicate suppression to win more often
+        let delayMs: Int
+        switch degree {
+        case 0...2: delayMs = Int.random(in: 40...100)
+        case 3...5: delayMs = Int.random(in: 60...150)
+        case 6...9: delayMs = Int.random(in: 80...180)
+        default:    delayMs = Int.random(in: 100...220)
+        }
        return RelayDecision(shouldRelay: shouldRelay, newTTL: newTTL, delayMs: delayMs)
    }
 }
-