diff --git a/bitchat/Services/BLEService.swift b/bitchat/Services/BLEService.swift index 8a0bcdfc..124626c9 100644 --- a/bitchat/Services/BLEService.swift +++ b/bitchat/Services/BLEService.swift @@ -112,6 +112,16 @@ final class BLEService: NSObject { private var scheduledRelays: [String: DispatchWorkItem] = [:] // 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 @@ -156,6 +166,88 @@ final class BLEService: NSObject { return bleQueue.sync { (self.subscribedCentrals, self.centralToPeerID) } } } + + // 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 { + 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.. 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>() @@ -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 - for s in states where s.isConnected { - if let ch = s.characteristic { - s.peripheral.writeValue(data, for: ch, type: .withoutResponse) + // 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 } } } - // Notify all subscribed centrals + + // 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 { + writeOrEnqueue(data, to: s.peripheral, characteristic: ch) + } + } + // 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 { diff --git a/bitchat/Services/RelayController.swift b/bitchat/Services/RelayController.swift index 2fedd366..ef3087ca 100644 --- a/bitchat/Services/RelayController.swift +++ b/bitchat/Services/RelayController.swift @@ -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 = 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) } } -