diff --git a/bitchat/Models/RequestSyncPacket.swift b/bitchat/Models/RequestSyncPacket.swift index b34dd925..0c6cb3cd 100644 --- a/bitchat/Models/RequestSyncPacket.swift +++ b/bitchat/Models/RequestSyncPacket.swift @@ -1,19 +1,21 @@ +import BitFoundation import Foundation // REQUEST_SYNC payload TLV (type, length16, value) // - 0x01: P (uint8) — Golomb-Rice parameter // - 0x02: M (uint32, big-endian) — hash range (N * 2^P) // - 0x03: data (opaque) — GR bitstream bytes (MSB-first) -// - 0x04: types (bitfield) — SyncTypeFlags of covered message types -// - 0x05: sinceTimestamp (uint64, big-endian) — oldest ts the filter covers -// - 0x06: fragmentIdFilter (utf8) — reserved -// -// TODO(v2): fragmentIdFilter (0x06) is parsed and re-serialized but never -// populated or honored — it's the reserved surface for incremental fragment -// sync (request the missing fragments of one file by ID instead of diffing the -// whole fragment set). Either wire it into buildGcsPayload/_handleRequestSync -// or drop the field; don't leave it as silent dead protocol surface. +// - 0x04: types (SyncTypeFlags) — packet types the filter covers +// - 0x05: sinceTimestamp (uint64, big-endian) — filter coverage cursor +// - 0x06: fragmentIdFilter (UTF-8) — comma-separated 16-hex-char (8-byte) +// fragment stream IDs; restricts the fragment diff to exactly those +// streams (targeted resync for stalled reassemblies) struct RequestSyncPacket { + /// Maximum fragment IDs one 0x06 filter may carry. Each ID encodes as + /// 16 hex chars plus a comma separator, so the largest encoded value is + /// 60 * 17 - 1 = 1019 bytes, which fits the 1024-byte decoder cap. + static let maxFragmentIdFilterCount = 60 + let p: Int let m: UInt32 let data: Data @@ -21,6 +23,29 @@ struct RequestSyncPacket { let sinceTimestamp: UInt64? let fragmentIdFilter: String? + /// Encodes 8-byte fragment stream IDs as the 0x06 filter string, + /// dropping malformed IDs and capping at `maxFragmentIdFilterCount`. + static func encodeFragmentIdFilter(_ fragmentIDs: [Data]) -> String? { + let tokens = fragmentIDs + .filter { $0.count == 8 } + .prefix(maxFragmentIdFilterCount) + .map { $0.hexEncodedString() } + guard !tokens.isEmpty else { return nil } + return tokens.joined(separator: ",") + } + + /// Decodes a 0x06 filter string back into 8-byte fragment stream IDs, + /// ignoring malformed tokens and capping at `maxFragmentIdFilterCount`. + static func decodeFragmentIdFilter(_ filter: String?) -> Set? { + guard let filter else { return nil } + var ids: Set = [] + for token in filter.split(separator: ",").prefix(maxFragmentIdFilterCount) { + guard token.count == 16, let id = Data(hexString: String(token)) else { continue } + ids.insert(id) + } + return ids.isEmpty ? nil : ids + } + init(p: Int, m: UInt32, data: Data, types: SyncTypeFlags? = nil, sinceTimestamp: UInt64? = nil, fragmentIdFilter: String? = nil) { self.p = p self.m = m @@ -97,7 +122,9 @@ struct RequestSyncPacket { sinceTimestamp = ts } case 0x06: - if let fid = String(data: v, encoding: .utf8) { + // Same acceptance cap as the GCS payload; an oversized filter + // is ignored rather than failing the whole request. + if v.count <= maxAcceptBytes, let fid = String(data: v, encoding: .utf8) { fragmentIdFilter = fid } default: diff --git a/bitchat/Services/BLE/BLEFragmentAssemblyBuffer.swift b/bitchat/Services/BLE/BLEFragmentAssemblyBuffer.swift index 4d41b2d3..83311584 100644 --- a/bitchat/Services/BLE/BLEFragmentAssemblyBuffer.swift +++ b/bitchat/Services/BLE/BLEFragmentAssemblyBuffer.swift @@ -64,6 +64,9 @@ struct BLEFragmentAssemblyBuffer { let type: UInt8 let total: Int let timestamp: Date + let isBroadcast: Bool + var lastFragmentAt: Date + var lastResyncRequestAt: Date? } private var fragmentsByKey: [BLEFragmentKey: [Int: Data]] = [:] @@ -105,7 +108,15 @@ struct BLEFragmentAssemblyBuffer { return .oversized(header: header, projectedSize: projectedSize, limit: limit, started: started) } + // Only actual progress resets the stall clock: fragment packets + // bypass the packet deduplicator, so relayed duplicates of an + // already-held index must not keep suppressing the targeted + // REQUEST_SYNC for a stalled stream. + let isNewIndex = fragmentsByKey[header.key]?[header.index] == nil fragmentsByKey[header.key]?[header.index] = header.fragmentData + if isNewIndex { + metadataByKey[header.key]?.lastFragmentAt = now + } guard let fragments = fragmentsByKey[header.key], fragments.count == header.total else { @@ -138,10 +149,59 @@ struct BLEFragmentAssemblyBuffer { } fragmentsByKey[header.key] = [:] - metadataByKey[header.key] = Metadata(type: header.originalType, total: header.total, timestamp: now) + metadataByKey[header.key] = Metadata( + type: header.originalType, + total: header.total, + timestamp: now, + isBroadcast: header.isBroadcastFragment, + lastFragmentAt: now + ) return true } + /// Fragment stream IDs (8-byte, big-endian) of incomplete broadcast + /// reassemblies that have not seen a new fragment for `stalledAfter` + /// seconds — candidates for a targeted REQUEST_SYNC. Each returned + /// stream is marked so it is not re-requested within `retryAfter`. + /// At most `RequestSyncPacket.maxFragmentIdFilterCount` streams are + /// returned per pass — the wire filter cannot carry more — selected + /// oldest-stall first; overflow streams stay unmarked and eligible for + /// the next pass. Directed reassemblies are excluded: peers only archive + /// broadcast fragments for gossip sync, so a targeted request cannot + /// recover them. + mutating func stalledBroadcastFragmentIDs( + stalledAfter: TimeInterval, + retryAfter: TimeInterval, + now: Date = Date() + ) -> [Data] { + var candidates: [(key: BLEFragmentKey, lastFragmentAt: Date)] = [] + for (key, metadata) in metadataByKey { + guard metadata.isBroadcast, + let fragments = fragmentsByKey[key], + fragments.count < metadata.total, + now.timeIntervalSince(metadata.lastFragmentAt) >= stalledAfter else { continue } + if let lastRequest = metadata.lastResyncRequestAt, + now.timeIntervalSince(lastRequest) < retryAfter { continue } + candidates.append((key: key, lastFragmentAt: metadata.lastFragmentAt)) + } + + // Mark only the streams that will actually go on the wire, so the + // overflow is not silently suppressed for `retryAfter`. + let selected = candidates + .sorted { + if $0.lastFragmentAt != $1.lastFragmentAt { + return $0.lastFragmentAt < $1.lastFragmentAt + } + return ($0.key.sender, $0.key.id) < ($1.key.sender, $1.key.id) + } + .prefix(RequestSyncPacket.maxFragmentIdFilterCount) + + return selected.map { candidate in + metadataByKey[candidate.key]?.lastResyncRequestAt = now + return withUnsafeBytes(of: candidate.key.id.bigEndian) { Data($0) } + } + } + private static func assemblyLimit(for originalType: UInt8) -> Int { if originalType == MessageType.fileTransfer.rawValue { // Allow headroom for TLV metadata and binary framing overhead. diff --git a/bitchat/Services/BLE/BLEService.swift b/bitchat/Services/BLE/BLEService.swift index 598f9821..db134e41 100644 --- a/bitchat/Services/BLE/BLEService.swift +++ b/bitchat/Services/BLE/BLEService.swift @@ -69,7 +69,9 @@ final class BLEService: NSObject { #endif private var selfBroadcastTracker = BLESelfBroadcastTracker() private let meshTopology = MeshTopologyTracker() - + // Route health for originated source routes; guarded by collectionsQueue. + private var sourceRouteFailures = BLESourceRouteFailureCache() + // 5. Fragment Reassembly (necessary for messages > MTU) private var fragmentAssemblyBuffer = BLEFragmentAssemblyBuffer() private var outboundFragmentTransfers = BLEOutboundFragmentTransferScheduler() @@ -586,6 +588,7 @@ final class BLEService: NSObject { let entries = outboundFragmentTransfers.removeAll().map { ($0.id, $0.workItems) } peerRegistry.removeAll() fragmentAssemblyBuffer.removeAll() + sourceRouteFailures = BLESourceRouteFailureCache() // Also clear pending message queues to avoid stale state across sessions pendingNoiseSessionQueues.removeAll() pendingDirectedRelays.removeAll() @@ -2392,13 +2395,31 @@ extension BLEService { } private func computeRoute(to peerID: PeerID) -> [Data]? { - meshTopology.computeRoute(from: myPeerIDData, to: routingData(for: peerID)) + // Version-gated: every hop and the recipient must have been observed + // speaking v2, since a v1-only node drops v2 frames on decode. + meshTopology.computeRoute( + from: myPeerIDData, + to: routingData(for: peerID), + maxHops: TransportConfig.bleSourceRouteMaxIntermediateHops, + requiringVersion: 2 + ) } private func applyRouteIfAvailable(_ packet: BitchatPacket, to recipient: PeerID) -> BitchatPacket { - guard let route = computeRoute(to: recipient), route.count >= 1 else { - return packet - } + let now = Date() + let route = BLESourceRouteOriginationPolicy.route( + for: packet, + to: recipient, + localPeerIDData: myPeerIDData, + isRecipientConnected: { self.isPeerConnected($0) }, + shouldAttemptRoute: { peer in + self.collectionsQueue.sync(flags: .barrier) { + self.sourceRouteFailures.shouldAttemptRoute(to: peer, now: now) + } + }, + computeRoute: { self.computeRoute(to: $0) } + ) + guard let route else { return packet } // Create new packet with route applied and version upgraded to 2 let routedPacket = BitchatPacket( type: packet.type, @@ -2416,6 +2437,9 @@ extension BLEService { SecureLogger.error("❌ Failed to re-sign packet with route", category: .security) return packet // Return original packet if signing fails } + collectionsQueue.sync(flags: .barrier) { + sourceRouteFailures.noteRoutedSend(to: recipient, now: now) + } return signedPacket } @@ -3187,13 +3211,23 @@ extension BLEService { // Update peer info without verbose logging - update the peer we received from, not the original sender updatePeerLastSeen(peerID) - // Track recent traffic timestamps for adaptive behavior + // Track recent traffic timestamps for adaptive behavior; the same + // barrier hop confirms route health for the packet's originator. collectionsQueue.async(flags: .barrier) { [weak self] in guard let self = self else { return } self.recentTrafficTracker.recordPacket(at: Date()) + self.sourceRouteFailures.noteInboundActivity(from: senderID) + } + + // Per-peer protocol version: originated source routes only use hops + // observed speaking v2 (a v1-only node cannot decode v2 frames). + if packet.version >= 2 { + meshTopology.recordObservedVersion(packet.version, for: packet.senderID) + if peerID != senderID { + meshTopology.recordObservedVersion(packet.version, for: routingData(for: peerID)) + } } - // Process by type switch context.messageType { case .announce: @@ -3760,10 +3794,21 @@ extension BLEService { // Clean old processed messages efficiently messageDeduplicator.cleanup() - // Clean old fragments (> configured seconds old) - collectionsQueue.sync(flags: .barrier) { + // Clean old fragments (> configured seconds old), then ask peers for + // the specific fragment streams whose reassembly has stalled instead + // of waiting for the next periodic GCS fragment round. + let stalledFragmentIDs = collectionsQueue.sync(flags: .barrier) { () -> [Data] in let cutoff = now.addingTimeInterval(-TransportConfig.bleFragmentLifetimeSeconds) fragmentAssemblyBuffer.removeExpired(before: cutoff) + sourceRouteFailures.prune(now: now) + return fragmentAssemblyBuffer.stalledBroadcastFragmentIDs( + stalledAfter: TransportConfig.bleFragmentResyncStallSeconds, + retryAfter: TransportConfig.bleFragmentResyncRetrySeconds, + now: now + ) + } + if !stalledFragmentIDs.isEmpty { + gossipSyncManager?.requestMissingFragments(fragmentIDs: stalledFragmentIDs) } // Clean old connection timeout backoff entries (> window) diff --git a/bitchat/Services/BLE/BLESourceRouteFailureCache.swift b/bitchat/Services/BLE/BLESourceRouteFailureCache.swift new file mode 100644 index 00000000..fcf23d2c --- /dev/null +++ b/bitchat/Services/BLE/BLESourceRouteFailureCache.swift @@ -0,0 +1,95 @@ +import BitFoundation +import Foundation + +/// Tracks whether source-routed sends to a recipient appear to be working. +/// +/// A routed unicast rides exactly one path, so a broken hop silently loses the +/// packet where a flood would have healed around it. Rather than building a +/// retransmission machine (MessageRouter already retries at a higher layer), +/// this cache degrades: a routed send that sees no inbound traffic from the +/// recipient within the confirmation window marks the route as failed, and +/// subsequent sends fall back to flooding until the suppression TTL lapses. +struct BLESourceRouteFailureCache { + struct Config { + /// How long a routed send may go unconfirmed before it counts as a + /// route failure. + var confirmationWindowSeconds: TimeInterval = TransportConfig.bleSourceRouteConfirmationWindowSeconds + /// How long to flood instead of routing after a failure. + var suppressionSeconds: TimeInterval = TransportConfig.bleSourceRouteSuppressionSeconds + } + + private struct State { + var pendingSince: Date? + var suppressedUntil: Date? + } + + private let config: Config + private var states: [PeerID: State] = [:] + + init(config: Config = Config()) { + self.config = config + } + + /// Whether the next directed send to `recipient` may carry a source + /// route. Flips the recipient into suppression when the last routed send + /// went unconfirmed past the confirmation window. + mutating func shouldAttemptRoute(to recipient: PeerID, now: Date = Date()) -> Bool { + guard var state = states[recipient] else { return true } + + if let until = state.suppressedUntil { + guard now >= until else { return false } + state.suppressedUntil = nil + } + + if let pending = state.pendingSince, + now.timeIntervalSince(pending) > config.confirmationWindowSeconds { + // The routed send was never confirmed: treat the route as broken + // and flood until the suppression window lapses. + state.pendingSince = nil + state.suppressedUntil = now.addingTimeInterval(config.suppressionSeconds) + states[recipient] = state + return false + } + + states[recipient] = state + return true + } + + /// Records that a source-routed packet was sent to `recipient`. Keeps the + /// earliest unconfirmed send so back-to-back packets share one deadline. + mutating func noteRoutedSend(to recipient: PeerID, now: Date = Date()) { + var state = states[recipient] ?? State() + if state.pendingSince == nil { + state.pendingSince = now + } + states[recipient] = state + } + + /// Any inbound packet authored by `peer` confirms the pending routed send + /// (delivery acks and replies arrive this way). Deliberately does not + /// lift an active suppression: that traffic may have arrived via flood. + mutating func noteInboundActivity(from peer: PeerID) { + guard var state = states[peer] else { return } + state.pendingSince = nil + if state.suppressedUntil == nil { + states.removeValue(forKey: peer) + } else { + states[peer] = state + } + } + + /// Drops entries that can no longer influence a routing decision. An + /// expired-but-unconverted pending entry is kept for as long as the + /// suppression it would trigger could still be active. + mutating func prune(now: Date = Date()) { + let pendingRetention = config.confirmationWindowSeconds + config.suppressionSeconds + states = states.filter { _, state in + if let until = state.suppressedUntil, now < until { return true } + if let pending = state.pendingSince, + now.timeIntervalSince(pending) <= pendingRetention { + return true + } + return false + } + } +} diff --git a/bitchat/Services/BLE/BLESourceRouteOriginationPolicy.swift b/bitchat/Services/BLE/BLESourceRouteOriginationPolicy.swift new file mode 100644 index 00000000..5cc8bf9c --- /dev/null +++ b/bitchat/Services/BLE/BLESourceRouteOriginationPolicy.swift @@ -0,0 +1,40 @@ +import BitFoundation +import Foundation + +/// Decides whether an outbound directed packet should carry a v2 source +/// route. Pure gating logic so BLEService's hot send path stays a thin wire. +enum BLESourceRouteOriginationPolicy { + /// Returns the intermediate-hop route to attach, or nil to keep the + /// current flood/direct-write behavior unchanged. + /// + /// Routes are only originated when every gate passes: + /// - we authored the packet (relays must not rewrite and re-sign someone + /// else's packet; route-following for in-flight routed packets lives in + /// `BLERouteForwardingPolicy`), + /// - the packet is directed at a single peer (not broadcast), + /// - the packet has TTL headroom to traverse hops (link-local TTL-0 + /// packets like REQUEST_SYNC never route), + /// - the recipient is not directly connected (a direct write already + /// delivers in one hop), + /// - routing to the recipient is not suppressed by a recent unconfirmed + /// routed send, and + /// - the topology yields a complete path. + static func route( + for packet: BitchatPacket, + to recipient: PeerID, + localPeerIDData: Data, + isRecipientConnected: (PeerID) -> Bool, + shouldAttemptRoute: (PeerID) -> Bool, + computeRoute: (PeerID) -> [Data]? + ) -> [Data]? { + guard packet.senderID == localPeerIDData else { return nil } + guard let recipientData = packet.recipientID, + recipientData.count == 8, + !recipientData.allSatisfy({ $0 == 0xFF }) else { return nil } + guard packet.ttl > 1 else { return nil } + guard !isRecipientConnected(recipient) else { return nil } + guard shouldAttemptRoute(recipient) else { return nil } + guard let route = computeRoute(recipient), !route.isEmpty else { return nil } + return route + } +} diff --git a/bitchat/Services/MeshTopologyTracker.swift b/bitchat/Services/MeshTopologyTracker.swift index fdd8749f..eb4e0c3c 100644 --- a/bitchat/Services/MeshTopologyTracker.swift +++ b/bitchat/Services/MeshTopologyTracker.swift @@ -10,6 +10,10 @@ final class MeshTopologyTracker { private var claims: [RoutingID: Set] = [:] // Last time we received an update from a node private var lastSeen: [RoutingID: Date] = [:] + // Highest protocol version observed from each node's decoded packets. + // Nodes absent from this map are assumed v1-only and are never used as + // hops (or targets) for version-gated routes. + private var observedVersions: [RoutingID: (version: UInt8, seenAt: Date)] = [:] // Maximum age for topology claims to be considered fresh for routing // Routes computed using stale topology can fail when the network has changed @@ -19,18 +23,29 @@ final class MeshTopologyTracker { queue.sync(flags: .barrier) { self.claims.removeAll() self.lastSeen.removeAll() + self.observedVersions.removeAll() } } /// Update the topology with a node's self-reported neighbor list - func updateNeighbors(for sourceData: Data?, neighbors: [Data]) { + func updateNeighbors(for sourceData: Data?, neighbors: [Data], at now: Date = Date()) { guard let source = sanitize(sourceData) else { return } // Sanitize neighbors and exclude self-loops let validNeighbors = Set(neighbors.compactMap { sanitize($0) }).subtracting([source]) - + queue.sync(flags: .barrier) { self.claims[source] = validNeighbors - self.lastSeen[source] = Date() + self.lastSeen[source] = now + } + } + + /// Record the protocol version observed on a decoded packet from a node. + /// Only versions above the v1 baseline are stored; the highest wins. + func recordObservedVersion(_ version: UInt8, for peerData: Data?, at now: Date = Date()) { + guard version > 1, let peer = sanitize(peerData) else { return } + queue.sync(flags: .barrier) { + let current = self.observedVersions[peer]?.version ?? 1 + self.observedVersions[peer] = (version: max(version, current), seenAt: now) } } @@ -39,28 +54,37 @@ final class MeshTopologyTracker { queue.sync(flags: .barrier) { self.claims.removeValue(forKey: peer) self.lastSeen.removeValue(forKey: peer) + self.observedVersions.removeValue(forKey: peer) } } - + /// Prune nodes that haven't updated their topology in `age` seconds - func prune(olderThan age: TimeInterval) { - let deadline = Date().addingTimeInterval(-age) + func prune(olderThan age: TimeInterval, now: Date = Date()) { + let deadline = now.addingTimeInterval(-age) queue.sync(flags: .barrier) { let stale = self.lastSeen.filter { $0.value < deadline } for (peer, _) in stale { self.claims.removeValue(forKey: peer) self.lastSeen.removeValue(forKey: peer) } + self.observedVersions = self.observedVersions.filter { $0.value.seenAt >= deadline } } } - func computeRoute(from start: Data?, to goal: Data?, maxHops: Int = 10) -> [Data]? { + /// BFS over confirmed, fresh edges. When `requiringVersion` is set, every + /// node on the path except the source (i.e. all intermediate hops and the + /// target) must have been observed speaking at least that protocol + /// version — a v1-only hop cannot decode a v2 routed packet. + func computeRoute(from start: Data?, to goal: Data?, maxHops: Int = 10, requiringVersion: UInt8? = nil, now: Date = Date()) -> [Data]? { guard let source = sanitize(start), let target = sanitize(goal) else { return nil } if source == target { return [] } // Direct connection, no intermediate hops return queue.sync { - let now = Date() let freshnessDeadline = now.addingTimeInterval(-Self.routeFreshnessThreshold) + func meetsRequiredVersion(_ peer: RoutingID) -> Bool { + guard let requiringVersion else { return true } + return (observedVersions[peer]?.version ?? 1) >= requiringVersion + } // BFS var visited: Set = [source] @@ -86,6 +110,10 @@ final class MeshTopologyTracker { for neighbor in neighbors { if visited.contains(neighbor) { continue } + // Version gate: skip nodes not known to speak the + // required protocol version. + guard meetsRequiredVersion(neighbor) else { continue } + // CONFIRMED EDGE CHECK: // 'last' claims 'neighbor' (checked above) // Does 'neighbor' claim 'last'? diff --git a/bitchat/Services/TransportConfig.swift b/bitchat/Services/TransportConfig.swift index 59d15c55..649a8be9 100644 --- a/bitchat/Services/TransportConfig.swift +++ b/bitchat/Services/TransportConfig.swift @@ -208,6 +208,25 @@ enum TransportConfig { static let bleSubscriptionRateLimitWindowSeconds: TimeInterval = 60.0 // Window for tracking subscription attempts static let bleSubscriptionRateLimitMaxAttempts: Int = 5 // Max attempts before extended cooldown + // Source routing (v2 directed packets) + // Longest path we will originate, in intermediate hops between us and the + // recipient. Keep small: every hop must be a fresh, confirmed, v2-capable + // node, and long stale paths fail more often than floods. + static let bleSourceRouteMaxIntermediateHops: Int = 4 + // A routed send with no inbound traffic from the recipient within this + // window counts as a route failure. + static let bleSourceRouteConfirmationWindowSeconds: TimeInterval = 10.0 + // After a route failure, directed sends to that recipient flood instead + // of routing until this lapses. + static let bleSourceRouteSuppressionSeconds: TimeInterval = 60.0 + + // Targeted fragment resync (REQUEST_SYNC fragmentIdFilter) + // A broadcast reassembly with no new fragment for this long is stalled + // and triggers a targeted REQUEST_SYNC naming its fragment stream. + static let bleFragmentResyncStallSeconds: TimeInterval = 5.0 + // Minimum spacing between targeted resync requests for the same stream. + static let bleFragmentResyncRetrySeconds: TimeInterval = 10.0 + // Store-and-forward for directed packets at relays. Spooled packets retry // on each maintenance flush until the window lapses; a longer window lets // brief link gaps (walking between rooms, reconnect churn) heal themselves. diff --git a/bitchat/Sync/GossipSyncManager.swift b/bitchat/Sync/GossipSyncManager.swift index 2d7118ab..5edf1c4c 100644 --- a/bitchat/Sync/GossipSyncManager.swift +++ b/bitchat/Sync/GossipSyncManager.swift @@ -274,11 +274,29 @@ final class GossipSyncManager { delegate?.sendPacket(signed) } - private func sendRequestSync(to peerID: PeerID, types: SyncTypeFlags) { + /// Targeted fragment recovery: ask connected peers for the specific + /// fragment streams whose reassembly has stalled, instead of waiting on + /// the next periodic GCS fragment round to cover them. + func requestMissingFragments(fragmentIDs: [Data]) { + queue.async { [weak self] in + self?._requestMissingFragments(fragmentIDs) + } + } + + private func _requestMissingFragments(_ fragmentIDs: [Data]) { + guard let filter = RequestSyncPacket.encodeFragmentIdFilter(fragmentIDs) else { return } + guard let connectedPeers = delegate?.getConnectedPeers(), !connectedPeers.isEmpty else { return } + SecureLogger.debug("Requesting \(fragmentIDs.count) stalled fragment stream(s) from \(connectedPeers.count) peer(s)", category: .sync) + for peerID in connectedPeers { + sendRequestSync(to: peerID, types: .fragment, fragmentIdFilter: filter) + } + } + + private func sendRequestSync(to peerID: PeerID, types: SyncTypeFlags, fragmentIdFilter: String? = nil) { // Register the request for RSR validation requestSyncManager.registerRequest(to: peerID) - - let payload = buildGcsPayload(for: types) + + let payload = buildGcsPayload(for: types, fragmentIdFilter: fragmentIdFilter) var recipient = Data() var temp = peerID.id while temp.count >= 2 && recipient.count < 8 { @@ -355,9 +373,19 @@ final class GossipSyncManager { } if requestedTypes.contains(.fragment) { + // A fragment-ID filter narrows the diff to exactly the named + // fragment streams (targeted resync for stalled reassemblies) + // and bypasses the since-cursor for them; the GCS filter still + // excludes the pieces the requester already holds. Fragment + // payloads start with the 8-byte stream ID. + let fragmentIdFilter = RequestSyncPacket.decodeFragmentIdFilter(request.fragmentIdFilter) let frags = fragments.allPackets(isFresh: isPacketFresh) for pkt in frags { - if let since, pkt.timestamp < since { continue } + if let fragmentIdFilter { + guard fragmentIdFilter.contains(Data(pkt.payload.prefix(8))) else { continue } + } else if let since, pkt.timestamp < since { + continue + } let idBytes = PacketIdUtil.computeId(pkt) if !mightContain(idBytes) { var toSend = pkt @@ -400,7 +428,7 @@ final class GossipSyncManager { } // Build REQUEST_SYNC payload using current candidates and GCS params - private func buildGcsPayload(for types: SyncTypeFlags) -> Data { + private func buildGcsPayload(for types: SyncTypeFlags, fragmentIdFilter: String? = nil) -> Data { var candidates: [BitchatPacket] = [] if types.contains(.announce) { for (_, pkt) in latestAnnouncementByPeer where isPacketFresh(pkt) { @@ -421,7 +449,7 @@ final class GossipSyncManager { } if candidates.isEmpty { let p = GCSFilter.deriveP(targetFpr: config.gcsTargetFpr) - let req = RequestSyncPacket(p: p, m: 1, data: Data(), types: types) + let req = RequestSyncPacket(p: p, m: 1, data: Data(), types: types, fragmentIdFilter: fragmentIdFilter) return req.encode() } @@ -442,7 +470,7 @@ final class GossipSyncManager { } let takeN = min(candidates.count, min(nMax, cap)) if takeN <= 0 { - let req = RequestSyncPacket(p: p, m: 1, data: Data(), types: types) + let req = RequestSyncPacket(p: p, m: 1, data: Data(), types: types, fragmentIdFilter: fragmentIdFilter) return req.encode() } let included = Array(candidates.prefix(takeN)) @@ -460,7 +488,7 @@ final class GossipSyncManager { let sinceTimestamp: UInt64? = (covered < candidates.count && covered > 0) ? included[covered - 1].timestamp : nil - let req = RequestSyncPacket(p: params.p, m: params.m, data: params.data, types: types, sinceTimestamp: sinceTimestamp) + let req = RequestSyncPacket(p: params.p, m: params.m, data: params.data, types: types, sinceTimestamp: sinceTimestamp, fragmentIdFilter: fragmentIdFilter) return req.encode() } diff --git a/bitchatTests/GossipSyncManagerTests.swift b/bitchatTests/GossipSyncManagerTests.swift index 0e7b35eb..81516543 100644 --- a/bitchatTests/GossipSyncManagerTests.swift +++ b/bitchatTests/GossipSyncManagerTests.swift @@ -507,6 +507,98 @@ struct GossipSyncManagerTests { #expect(sentPackets[0].type == MessageType.fragment.rawValue) } + // MARK: - Fragment-ID filter (targeted resync) + + private func makeFragmentPacket(sender: Data, fragmentID: Data, index: UInt16, timestamp: UInt64) -> BitchatPacket { + // Fragment payload: 8-byte stream ID + index + total + original type. + var payload = fragmentID + payload.append(contentsOf: withUnsafeBytes(of: index.bigEndian) { Data($0) }) + payload.append(contentsOf: withUnsafeBytes(of: UInt16(4).bigEndian) { Data($0) }) + payload.append(MessageType.fileTransfer.rawValue) + payload.append(Data([0xEE])) + return BitchatPacket( + type: MessageType.fragment.rawValue, + senderID: sender, + recipientID: nil, + timestamp: timestamp, + payload: payload, + signature: nil, + ttl: 1 + ) + } + + @Test func handleRequestSyncHonorsFragmentIdFilter() async throws { + var config = GossipSyncManager.Config() + config.fragmentCapacity = 10 + config.messageSyncIntervalSeconds = 0 + config.fragmentSyncIntervalSeconds = 0 + config.fileTransferSyncIntervalSeconds = 0 + + let requestSyncManager = RequestSyncManager() + let manager = GossipSyncManager(myPeerID: myPeerID, config: config, requestSyncManager: requestSyncManager) + let delegate = RecordingDelegate() + manager.delegate = delegate + + let sender = try #require(Data(hexString: "aabbccddeeff0011")) + let wantedID = try #require(Data(hexString: "0102030405060708")) + let otherID = try #require(Data(hexString: "1112131415161718")) + let nowMs = UInt64(Date().timeIntervalSince1970 * 1000) + + let wanted = makeFragmentPacket(sender: sender, fragmentID: wantedID, index: 1, timestamp: nowMs - 60_000) + let other = makeFragmentPacket(sender: sender, fragmentID: otherID, index: 2, timestamp: nowMs) + manager.onPublicPacketSeen(wanted) + manager.onPublicPacketSeen(other) + + // The since-cursor sits after both fragments; without the filter the + // responder would send nothing for `wanted`. The filter both bypasses + // the cursor and restricts the diff to exactly the named stream. + let request = RequestSyncPacket( + p: 7, + m: 1, + data: Data(), + types: .fragment, + sinceTimestamp: nowMs + 1, + fragmentIdFilter: RequestSyncPacket.encodeFragmentIdFilter([wantedID]) + ) + manager.handleRequestSync(from: PeerID(str: "FFFFFFFFFFFFFFFF"), request: request) + + try await TestHelpers.waitFor({ delegate.packets.count == 1 }, timeout: TestConstants.shortTimeout) + // Barrier: flush the sync queue so a late second packet would be visible. + manager._performMaintenanceSynchronously(now: Date()) + let sentPackets = delegate.packets + #expect(sentPackets.count == 1) + let sent = try #require(sentPackets.first) + #expect(sent.type == MessageType.fragment.rawValue) + #expect(sent.payload.prefix(8) == wantedID) + #expect(sent.ttl == 0) + #expect(sent.isRSR) + } + + @Test func requestMissingFragmentsSendsFilteredRequestToConnectedPeers() async throws { + var config = GossipSyncManager.Config() + config.messageSyncIntervalSeconds = 0 + config.fragmentSyncIntervalSeconds = 0 + config.fileTransferSyncIntervalSeconds = 0 + + let requestSyncManager = RequestSyncManager() + let manager = GossipSyncManager(myPeerID: myPeerID, config: config, requestSyncManager: requestSyncManager) + let delegate = RecordingDelegate() + delegate.connectedPeers = [PeerID(str: "FFFFFFFFFFFFFFFF")] + manager.delegate = delegate + + let stalledID = try #require(Data(hexString: "0102030405060708")) + manager.requestMissingFragments(fragmentIDs: [stalledID]) + + try await TestHelpers.waitFor({ delegate.packets.count == 1 }, timeout: TestConstants.shortTimeout) + let sent = try #require(delegate.packets.first) + #expect(sent.type == MessageType.requestSync.rawValue) + #expect(sent.ttl == 0) + let request = try #require(RequestSyncPacket.decode(from: sent.payload)) + #expect(request.types == .fragment) + let ids = try #require(RequestSyncPacket.decodeFragmentIdFilter(request.fragmentIdFilter)) + #expect(ids == Set([stalledID])) + } + // MARK: - Archive persistence @Test func publicMessagesRestoreFromArchiveAcrossRestart() async throws { @@ -583,6 +675,7 @@ struct GossipSyncManagerTests { private final class RecordingDelegate: GossipSyncManager.Delegate { var onSend: (() -> Void)? + var connectedPeers: [PeerID] = [] private(set) var lastPacket: BitchatPacket? private(set) var packets: [BitchatPacket] = [] private let lock = NSLock() @@ -604,6 +697,6 @@ private final class RecordingDelegate: GossipSyncManager.Delegate { } func getConnectedPeers() -> [PeerID] { - return [] + return connectedPeers } } diff --git a/bitchatTests/Services/BLEFragmentAssemblyBufferTests.swift b/bitchatTests/Services/BLEFragmentAssemblyBufferTests.swift index ed7c66a1..a4423ec9 100644 --- a/bitchatTests/Services/BLEFragmentAssemblyBufferTests.swift +++ b/bitchatTests/Services/BLEFragmentAssemblyBufferTests.swift @@ -135,6 +135,140 @@ struct BLEFragmentAssemblyBufferTests { } } + @Test + func stalledBroadcastAssemblyReportsFragmentIDOnceUntilRetryLapses() throws { + var buffer = BLEFragmentAssemblyBuffer() + let fragmentID = Data((1...8).map { UInt8($0) }) + let packet = makePacket(payload: makePayload(count: 256)) + let fragments = try makeFragments(for: packet, chunkSize: 128, fragmentID: fragmentID) + let first = try #require(BLEFragmentHeader(packet: fragments[0])) + + let t0 = Date(timeIntervalSince1970: 100) + _ = buffer.append(first, maxInFlightAssemblies: 8, now: t0) + + // Not yet stalled. + let early = buffer.stalledBroadcastFragmentIDs(stalledAfter: 5, retryAfter: 10, now: t0.addingTimeInterval(4)) + #expect(early.isEmpty) + + // Stalled: reported once, big-endian stream ID. + let stalled = buffer.stalledBroadcastFragmentIDs(stalledAfter: 5, retryAfter: 10, now: t0.addingTimeInterval(6)) + #expect(stalled == [fragmentID]) + + // Within the retry window: not re-reported. + let repeated = buffer.stalledBroadcastFragmentIDs(stalledAfter: 5, retryAfter: 10, now: t0.addingTimeInterval(8)) + #expect(repeated.isEmpty) + + // After the retry window it is requested again. + let retried = buffer.stalledBroadcastFragmentIDs(stalledAfter: 5, retryAfter: 10, now: t0.addingTimeInterval(17)) + #expect(retried == [fragmentID]) + } + + @Test + func newFragmentResetsStallClockAndCompletionStopsRequests() throws { + var buffer = BLEFragmentAssemblyBuffer() + let fragmentID = Data((10...17).map { UInt8($0) }) + let packet = makePacket(payload: makePayload(count: 384)) + let fragments = try makeFragments(for: packet, chunkSize: 128, fragmentID: fragmentID) + let headers = try fragments.map { try #require(BLEFragmentHeader(packet: $0)) } + #expect(headers.count >= 3) + + let t0 = Date(timeIntervalSince1970: 100) + _ = buffer.append(headers[0], maxInFlightAssemblies: 8, now: t0) + // A fragment arriving at t0+4 resets the stall clock. + _ = buffer.append(headers[1], maxInFlightAssemblies: 8, now: t0.addingTimeInterval(4)) + let afterProgress = buffer.stalledBroadcastFragmentIDs(stalledAfter: 5, retryAfter: 10, now: t0.addingTimeInterval(6)) + #expect(afterProgress.isEmpty) + + // Completion removes the assembly entirely. + var result: BLEFragmentAssemblyBuffer.AppendResult? + for header in headers.dropFirst(2) { + result = buffer.append(header, maxInFlightAssemblies: 8, now: t0.addingTimeInterval(5)) + } + guard case .complete = result else { + Issue.record("Expected assembly to complete") + return + } + let afterCompletion = buffer.stalledBroadcastFragmentIDs(stalledAfter: 5, retryAfter: 10, now: t0.addingTimeInterval(60)) + #expect(afterCompletion.isEmpty) + } + + @Test + func duplicateFragmentsDoNotResetStallClock() throws { + var buffer = BLEFragmentAssemblyBuffer() + let fragmentID = Data((20...27).map { UInt8($0) }) + let packet = makePacket(payload: makePayload(count: 256)) + let fragments = try makeFragments(for: packet, chunkSize: 128, fragmentID: fragmentID) + let first = try #require(BLEFragmentHeader(packet: fragments[0])) + + let t0 = Date(timeIntervalSince1970: 100) + _ = buffer.append(first, maxInFlightAssemblies: 8, now: t0) + + // Relay duplicates of the same index arrive every few seconds; they + // bring no new data, so they must not keep the stream "fresh". + _ = buffer.append(first, maxInFlightAssemblies: 8, now: t0.addingTimeInterval(3)) + _ = buffer.append(first, maxInFlightAssemblies: 8, now: t0.addingTimeInterval(5)) + + let stalled = buffer.stalledBroadcastFragmentIDs(stalledAfter: 5, retryAfter: 10, now: t0.addingTimeInterval(6)) + #expect(stalled == [fragmentID]) + } + + @Test + func overflowStalledStreamsRotateAcrossPasses() throws { + var buffer = BLEFragmentAssemblyBuffer() + let cap = RequestSyncPacket.maxFragmentIdFilterCount + let streamCount = cap + 10 + let t0 = Date(timeIntervalSince1970: 100) + + // Incomplete broadcast assemblies with staggered last-fragment times + // (stream 0 is the oldest stall). + var ids: [Data] = [] + for i in 0..> 8), UInt8(i & 0xFF)]) + ids.append(fragmentID) + let header = try #require(BLEFragmentHeader(packet: makeFragmentPacket( + fragmentID: fragmentID, + index: 0, + total: 2, + originalType: MessageType.message.rawValue, + fragmentData: Data([0x01]) + ))) + _ = buffer.append(header, maxInFlightAssemblies: streamCount, now: t0.addingTimeInterval(Double(i))) + } + + // All streams are stalled; only the cap's worth (oldest first) is + // requested and rate-limited, the overflow stays eligible. + let firstPassAt = t0.addingTimeInterval(Double(streamCount) + 5) + let firstPass = buffer.stalledBroadcastFragmentIDs(stalledAfter: 5, retryAfter: 60, now: firstPassAt) + #expect(firstPass == Array(ids.prefix(cap))) + + // Next pass picks up exactly the overflow streams. + let secondPass = buffer.stalledBroadcastFragmentIDs(stalledAfter: 5, retryAfter: 60, now: firstPassAt.addingTimeInterval(1)) + #expect(secondPass == Array(ids.suffix(streamCount - cap))) + + // Nothing left until a retry window lapses. + let thirdPass = buffer.stalledBroadcastFragmentIDs(stalledAfter: 5, retryAfter: 60, now: firstPassAt.addingTimeInterval(2)) + #expect(thirdPass.isEmpty) + } + + @Test + func directedAssembliesAreNeverReportedAsStalled() throws { + var buffer = BLEFragmentAssemblyBuffer() + let fragment = makeFragmentPacket( + fragmentID: Data(repeating: 0x0A, count: 8), + index: 0, + total: 2, + originalType: MessageType.message.rawValue, + fragmentData: Data([0x01]), + recipientID: Data(hexString: "0102030405060708") + ) + let header = try #require(BLEFragmentHeader(packet: fragment)) + + let t0 = Date(timeIntervalSince1970: 100) + _ = buffer.append(header, maxInFlightAssemblies: 8, now: t0) + let stalled = buffer.stalledBroadcastFragmentIDs(stalledAfter: 5, retryAfter: 10, now: t0.addingTimeInterval(60)) + #expect(stalled.isEmpty) + } + private func makePacket(payload: Data, timestamp: UInt64 = 0x0102030405) -> BitchatPacket { BitchatPacket( type: MessageType.message.rawValue, diff --git a/bitchatTests/Services/BLESourceRouteFailureCacheTests.swift b/bitchatTests/Services/BLESourceRouteFailureCacheTests.swift new file mode 100644 index 00000000..5d3315f6 --- /dev/null +++ b/bitchatTests/Services/BLESourceRouteFailureCacheTests.swift @@ -0,0 +1,98 @@ +// +// BLESourceRouteFailureCacheTests.swift +// bitchatTests +// +// This is free and unencumbered software released into the public domain. +// For more information, see +// + +import Testing +import Foundation +import BitFoundation +@testable import bitchat + +struct BLESourceRouteFailureCacheTests { + private let recipient = PeerID(str: "0102030405060708") + private let config = BLESourceRouteFailureCache.Config( + confirmationWindowSeconds: 10, + suppressionSeconds: 60 + ) + + private func attempts(_ cache: inout BLESourceRouteFailureCache, at date: Date) -> Bool { + cache.shouldAttemptRoute(to: recipient, now: date) + } + + @Test func allowsRoutingByDefault() { + var cache = BLESourceRouteFailureCache(config: config) + #expect(attempts(&cache, at: Date())) + } + + @Test func unconfirmedRoutedSendSuppressesRouting() { + var cache = BLESourceRouteFailureCache(config: config) + let t0 = Date() + + cache.noteRoutedSend(to: recipient, now: t0) + // Inside the confirmation window: keep routing. + #expect(attempts(&cache, at: t0.addingTimeInterval(5))) + // Past the window with no inbound traffic: route failed, flood. + #expect(!attempts(&cache, at: t0.addingTimeInterval(11))) + // Still suppressed for the suppression TTL. + #expect(!attempts(&cache, at: t0.addingTimeInterval(40))) + // Suppression lapses: routing may be attempted again. + #expect(attempts(&cache, at: t0.addingTimeInterval(11 + 61))) + } + + @Test func inboundActivityConfirmsPendingSend() { + var cache = BLESourceRouteFailureCache(config: config) + let t0 = Date() + + cache.noteRoutedSend(to: recipient, now: t0) + cache.noteInboundActivity(from: recipient) + // Confirmed: no suppression even long after the window. + #expect(attempts(&cache, at: t0.addingTimeInterval(30))) + } + + @Test func inboundActivityDoesNotLiftActiveSuppression() { + var cache = BLESourceRouteFailureCache(config: config) + let t0 = Date() + + cache.noteRoutedSend(to: recipient, now: t0) + // Trip the failure → suppression starts at t0+15. + #expect(!attempts(&cache, at: t0.addingTimeInterval(15))) + // Inbound traffic may have arrived via flood; suppression holds. + cache.noteInboundActivity(from: recipient) + #expect(!attempts(&cache, at: t0.addingTimeInterval(20))) + #expect(attempts(&cache, at: t0.addingTimeInterval(15 + 61))) + } + + @Test func backToBackSendsShareOneDeadline() { + var cache = BLESourceRouteFailureCache(config: config) + let t0 = Date() + + cache.noteRoutedSend(to: recipient, now: t0) + cache.noteRoutedSend(to: recipient, now: t0.addingTimeInterval(8)) + // Deadline runs from the first unconfirmed send. + #expect(!attempts(&cache, at: t0.addingTimeInterval(11))) + } + + @Test func pruneDropsExpiredEntries() { + var cache = BLESourceRouteFailureCache(config: config) + let t0 = Date() + + cache.noteRoutedSend(to: recipient, now: t0) + // Past confirmation + suppression: the entry can no longer matter. + cache.prune(now: t0.addingTimeInterval(75)) + #expect(attempts(&cache, at: t0.addingTimeInterval(76))) + } + + @Test func pruneKeepsEntriesThatStillMatter() { + var cache = BLESourceRouteFailureCache(config: config) + let t0 = Date() + + cache.noteRoutedSend(to: recipient, now: t0) + cache.prune(now: t0.addingTimeInterval(30)) + // The unconverted pending entry survives pruning and still converts + // into a suppression on the next routing decision. + #expect(!attempts(&cache, at: t0.addingTimeInterval(31))) + } +} diff --git a/bitchatTests/Services/BLESourceRouteOriginationPolicyTests.swift b/bitchatTests/Services/BLESourceRouteOriginationPolicyTests.swift new file mode 100644 index 00000000..c762a0c7 --- /dev/null +++ b/bitchatTests/Services/BLESourceRouteOriginationPolicyTests.swift @@ -0,0 +1,98 @@ +// +// BLESourceRouteOriginationPolicyTests.swift +// bitchatTests +// +// This is free and unencumbered software released into the public domain. +// For more information, see +// + +import Testing +import Foundation +import BitFoundation +@testable import bitchat + +struct BLESourceRouteOriginationPolicyTests { + private let localPeerIDData = Data(hexString: "0102030405060708")! + private let recipient = PeerID(str: "1112131415161718") + private let hop = Data(hexString: "2122232425262728")! + + private func makePacket( + senderID: Data? = nil, + recipientID: Data? = Data(hexString: "1112131415161718"), + ttl: UInt8 = 7 + ) -> BitchatPacket { + BitchatPacket( + type: MessageType.noiseEncrypted.rawValue, + senderID: senderID ?? localPeerIDData, + recipientID: recipientID, + timestamp: UInt64(Date().timeIntervalSince1970 * 1000), + payload: Data([0x01]), + signature: nil, + ttl: ttl + ) + } + + private func route( + packet: BitchatPacket, + isRecipientConnected: Bool = false, + shouldAttemptRoute: Bool = true, + computedRoute: [Data]? = nil + ) -> [Data]? { + BLESourceRouteOriginationPolicy.route( + for: packet, + to: recipient, + localPeerIDData: localPeerIDData, + isRecipientConnected: { _ in isRecipientConnected }, + shouldAttemptRoute: { _ in shouldAttemptRoute }, + computeRoute: { _ in computedRoute ?? [self.hop] } + ) + } + + @Test func routesWhenAllGatesPass() { + #expect(route(packet: makePacket()) == [hop]) + } + + @Test func relayedPacketNeverGetsRoute() { + let relayed = makePacket(senderID: Data(hexString: "aabbccddeeff0011")) + #expect(route(packet: relayed) == nil) + } + + @Test func broadcastRecipientNeverGetsRoute() { + let broadcast = makePacket(recipientID: Data(repeating: 0xFF, count: 8)) + #expect(route(packet: broadcast) == nil) + let noRecipient = makePacket(recipientID: nil) + #expect(route(packet: noRecipient) == nil) + } + + @Test func linkLocalTTLNeverGetsRoute() { + // TTL 0/1 packets (e.g. REQUEST_SYNC) cannot traverse hops. + #expect(route(packet: makePacket(ttl: 0)) == nil) + #expect(route(packet: makePacket(ttl: 1)) == nil) + } + + @Test func directlyConnectedRecipientNeverGetsRoute() { + #expect(route(packet: makePacket(), isRecipientConnected: true) == nil) + } + + @Test func suppressedRecipientFallsBackToFlood() { + #expect(route(packet: makePacket(), shouldAttemptRoute: false) == nil) + } + + @Test func missingOrEmptyRouteFallsBackToFlood() { + var sawComputeRoute = false + let result = BLESourceRouteOriginationPolicy.route( + for: makePacket(), + to: recipient, + localPeerIDData: localPeerIDData, + isRecipientConnected: { _ in false }, + shouldAttemptRoute: { _ in true }, + computeRoute: { _ in + sawComputeRoute = true + return nil + } + ) + #expect(result == nil) + #expect(sawComputeRoute) + #expect(route(packet: makePacket(), computedRoute: []) == nil) + } +} diff --git a/bitchatTests/Services/MeshTopologyTrackerTests.swift b/bitchatTests/Services/MeshTopologyTrackerTests.swift index 1ae0f5f5..fa5d2a4d 100644 --- a/bitchatTests/Services/MeshTopologyTrackerTests.swift +++ b/bitchatTests/Services/MeshTopologyTrackerTests.swift @@ -94,10 +94,132 @@ struct MeshTopologyTrackerTests { tracker.updateNeighbors(for: a, neighbors: [b]) tracker.updateNeighbors(for: b, neighbors: [a]) - + // When start == end, route should be empty (no intermediate hops needed) let route = try #require(tracker.computeRoute(from: a, to: a)) #expect(route == []) } + @Test func noPathReturnsNil() throws { + let tracker = MeshTopologyTracker() + let a = try hex("0101010101010101") + let b = try hex("0202020202020202") + let c = try hex("0303030303030303") + let d = try hex("0404040404040404") + + // Two disconnected islands: A-B and C-D + tracker.updateNeighbors(for: a, neighbors: [b]) + tracker.updateNeighbors(for: b, neighbors: [a]) + tracker.updateNeighbors(for: c, neighbors: [d]) + tracker.updateNeighbors(for: d, neighbors: [c]) + + #expect(tracker.computeRoute(from: a, to: d) == nil) + } + + /// Build a confirmed line topology n0 - n1 - ... - n(count-1). + private func makeLine(_ tracker: MeshTopologyTracker, count: Int) throws -> [Data] { + let nodes = try (0.. 0 { neighbors.append(nodes[i - 1]) } + if i < count - 1 { neighbors.append(nodes[i + 1]) } + tracker.updateNeighbors(for: nodes[i], neighbors: neighbors) + } + return nodes + } + + @Test func maxHopsCapsIntermediateHopCount() throws { + let tracker = MeshTopologyTracker() + // 7 nodes: source + 5 intermediates + target + let nodes = try makeLine(tracker, count: 7) + + // 5 intermediates exceed a 4-hop cap + #expect(tracker.computeRoute(from: nodes[0], to: nodes[6], maxHops: 4) == nil) + // 4 intermediates fit exactly + let route = try #require(tracker.computeRoute(from: nodes[0], to: nodes[5], maxHops: 4)) + #expect(route == Array(nodes[1...4])) + } + + @Test func staleNeighborBlocksRoute() throws { + let tracker = MeshTopologyTracker() + let a = try hex("0101010101010101") + let b = try hex("0202020202020202") + let c = try hex("0303030303030303") + + let staleDate = Date().addingTimeInterval(-120) // past 60s freshness + tracker.updateNeighbors(for: a, neighbors: [b]) + tracker.updateNeighbors(for: b, neighbors: [a, c], at: staleDate) + tracker.updateNeighbors(for: c, neighbors: [b]) + + #expect(tracker.computeRoute(from: a, to: c) == nil) + + // Refreshing B restores the route. + tracker.updateNeighbors(for: b, neighbors: [a, c]) + let route = try #require(tracker.computeRoute(from: a, to: c)) + #expect(route == [b]) + } + + @Test func versionGateBlocksV1AndUnknownHops() throws { + let tracker = MeshTopologyTracker() + let a = try hex("0101010101010101") + let b = try hex("0202020202020202") + let c = try hex("0303030303030303") + + tracker.updateNeighbors(for: a, neighbors: [b]) + tracker.updateNeighbors(for: b, neighbors: [a, c]) + tracker.updateNeighbors(for: c, neighbors: [b]) + + // Without the gate the route exists. + #expect(tracker.computeRoute(from: a, to: c) == [b]) + // Version-unknown hops are assumed v1-only and block gated routes. + #expect(tracker.computeRoute(from: a, to: c, requiringVersion: 2) == nil) + + // A v1 observation does not unlock the gate. + tracker.recordObservedVersion(1, for: b) + tracker.recordObservedVersion(2, for: c) + #expect(tracker.computeRoute(from: a, to: c, requiringVersion: 2) == nil) + + // Once the hop is observed speaking v2 the route opens. + tracker.recordObservedVersion(2, for: b) + let route = try #require(tracker.computeRoute(from: a, to: c, requiringVersion: 2)) + #expect(route == [b]) + } + + @Test func versionGateRequiresV2Target() throws { + let tracker = MeshTopologyTracker() + let a = try hex("0101010101010101") + let b = try hex("0202020202020202") + let c = try hex("0303030303030303") + + tracker.updateNeighbors(for: a, neighbors: [b]) + tracker.updateNeighbors(for: b, neighbors: [a, c]) + tracker.updateNeighbors(for: c, neighbors: [b]) + tracker.recordObservedVersion(2, for: b) + + // The recipient must decode the v2 frame too. + #expect(tracker.computeRoute(from: a, to: c, requiringVersion: 2) == nil) + + tracker.recordObservedVersion(2, for: c) + #expect(tracker.computeRoute(from: a, to: c, requiringVersion: 2) == [b]) + } + + @Test func pruneDropsStaleObservedVersions() throws { + let tracker = MeshTopologyTracker() + let a = try hex("0101010101010101") + let b = try hex("0202020202020202") + let c = try hex("0303030303030303") + + tracker.updateNeighbors(for: a, neighbors: [b]) + tracker.updateNeighbors(for: b, neighbors: [a, c]) + tracker.updateNeighbors(for: c, neighbors: [b]) + let old = Date().addingTimeInterval(-120) + tracker.recordObservedVersion(2, for: b, at: old) + tracker.recordObservedVersion(2, for: c, at: old) + + #expect(tracker.computeRoute(from: a, to: c, requiringVersion: 2) == [b]) + tracker.prune(olderThan: 60) + // Claims are fresh but the version observations aged out. + #expect(tracker.computeRoute(from: a, to: c, requiringVersion: 2) == nil) + } + } diff --git a/bitchatTests/Sync/RequestSyncPacketFragmentFilterTests.swift b/bitchatTests/Sync/RequestSyncPacketFragmentFilterTests.swift new file mode 100644 index 00000000..d055bde6 --- /dev/null +++ b/bitchatTests/Sync/RequestSyncPacketFragmentFilterTests.swift @@ -0,0 +1,76 @@ +// +// RequestSyncPacketFragmentFilterTests.swift +// bitchatTests +// +// This is free and unencumbered software released into the public domain. +// For more information, see +// + +import Testing +import Foundation +import BitFoundation +@testable import bitchat + +struct RequestSyncPacketFragmentFilterTests { + + @Test func fragmentIdFilterRoundTripsThroughWireEncoding() throws { + let id1 = try #require(Data(hexString: "00112233445566aa")) + let id2 = try #require(Data(hexString: "ffeeddccbbaa9988")) + let filter = try #require(RequestSyncPacket.encodeFragmentIdFilter([id1, id2])) + + let packet = RequestSyncPacket(p: 7, m: 128, data: Data([0x01]), types: .fragment, fragmentIdFilter: filter) + let decoded = try #require(RequestSyncPacket.decode(from: packet.encode())) + + #expect(decoded.fragmentIdFilter == filter) + let ids = try #require(RequestSyncPacket.decodeFragmentIdFilter(decoded.fragmentIdFilter)) + #expect(ids == Set([id1, id2])) + } + + @Test func encodeCapsFilterAtMaxCountWithinDecoderBudget() throws { + let ids = (0..<100).map { i -> Data in + var id = Data(repeating: 0, count: 8) + id[7] = UInt8(i) + return id + } + let filter = try #require(RequestSyncPacket.encodeFragmentIdFilter(ids)) + + let tokens = filter.split(separator: ",") + #expect(tokens.count == RequestSyncPacket.maxFragmentIdFilterCount) + // 60 IDs * 17 bytes ("<16 hex>,") - 1 = 1019 ≤ the 1024-byte cap. + #expect(filter.utf8.count == 1019) + #expect(filter.utf8.count <= 1024) + } + + @Test func encodeDropsMalformedIDs() throws { + let good = try #require(Data(hexString: "0011223344556677")) + let short = Data([0x01, 0x02]) + let filter = try #require(RequestSyncPacket.encodeFragmentIdFilter([short, good])) + #expect(filter == good.hexEncodedString()) + #expect(RequestSyncPacket.encodeFragmentIdFilter([short]) == nil) + #expect(RequestSyncPacket.encodeFragmentIdFilter([]) == nil) + } + + @Test func decodeIgnoresMalformedTokens() throws { + let good = try #require(Data(hexString: "0011223344556677")) + let ids = try #require( + RequestSyncPacket.decodeFragmentIdFilter("zzzz,0011,0011223344556677,") + ) + #expect(ids == Set([good])) + #expect(RequestSyncPacket.decodeFragmentIdFilter(nil) == nil) + #expect(RequestSyncPacket.decodeFragmentIdFilter("not-hex") == nil) + } + + @Test func decoderIgnoresOversizedFilterValue() throws { + // Hand-roll a payload whose 0x06 TLV exceeds the acceptance cap; the + // request must still decode, with the filter dropped. + var payload = RequestSyncPacket(p: 7, m: 128, data: Data([0x01])).encode() + let oversized = Data(repeating: UInt8(ascii: "a"), count: 1025) + payload.append(0x06) + payload.append(UInt8((oversized.count >> 8) & 0xFF)) + payload.append(UInt8(oversized.count & 0xFF)) + payload.append(oversized) + + let decoded = try #require(RequestSyncPacket.decode(from: payload)) + #expect(decoded.fragmentIdFilter == nil) + } +} diff --git a/docs/REQUEST_SYNC_MANAGER.md b/docs/REQUEST_SYNC_MANAGER.md index b304508e..a885f159 100644 --- a/docs/REQUEST_SYNC_MANAGER.md +++ b/docs/REQUEST_SYNC_MANAGER.md @@ -20,9 +20,11 @@ The new implementation introduces a **RequestSyncManager** to track outgoing syn ### Request Sync Payload The `REQUEST_SYNC` packet payload (TLV encoded) has been updated to include: -* **Future Filters**: - * `sinceTimestamp` (Type 0x05): To request packets since a certain time (UInt64 big-endian). - * `fragmentIdFilter` (Type 0x06): To request specific fragments (UTF-8 string). +* `sinceTimestamp` (Type 0x05): filter-coverage cursor (UInt64 big-endian). The requester's GCS filter only covers packets at or after this timestamp; the responder skips older packets instead of re-sending them every round. +* `fragmentIdFilter` (Type 0x06): targeted fragment resync (UTF-8 string). Comma-separated 16-hex-char (8-byte) fragment **stream IDs** — the ID that prefixes every fragment payload. + * **Requester**: when a broadcast reassembly stalls (no new fragment for 5 s), the fragment assembler reports the stream ID and a `REQUEST_SYNC` with `types = fragment` and this filter goes to each connected peer (re-requested at most every 10 s per stream). Directed reassemblies are excluded — peers only archive broadcast fragments for sync. + * **Responder**: when the filter is present, the fragment diff is restricted to exactly the named streams and the `sinceTimestamp` cursor is bypassed for them; the GCS filter still excludes pieces the requester already holds. Responses keep RSR marking, TTL 0, per-peer response rate limiting (8/30 s), and `REQUEST_SYNC` itself remains link-local (TTL 0, never relayed). + * **Bounds**: at most 60 IDs per request. Each ID encodes as 16 hex chars plus a comma separator, so the largest value is 60 × 17 − 1 = 1019 bytes, within the decoder's 1024-byte acceptance cap; oversized filter values are ignored (the rest of the request still decodes). ## Architecture diff --git a/docs/SOURCE_ROUTING.md b/docs/SOURCE_ROUTING.md index 6832a16e..55d00232 100644 --- a/docs/SOURCE_ROUTING.md +++ b/docs/SOURCE_ROUTING.md @@ -2,7 +2,7 @@ This document specifies the Source-Based Routing extension (v2) for the BitChat protocol. This upgrade enables efficient unicast routing across the mesh by allowing senders to specify an explicit path of intermediate relays. -**Status:** Implemented in Android and iOS. Backward compatible (v1 clients ignore routing data). +**Status:** Implemented in Android and iOS: both decode routed packets, forward along routes, and originate routes. iOS origination is policy-gated (see §8). Backward compatible (v1 clients never receive routed frames from iOS: routes are only originated when every node on the path has been observed speaking v2). --- @@ -144,3 +144,44 @@ When a node receives a packet **not** addressed to itself: * **Fallback:** If the Next Hop is unreachable, **fall back to broadcast/flood** to ensure delivery. 3. **If NO (Standard):** * Flood the packet to all connected neighbors (subject to TTL and probability rules). + +--- + +## 8. iOS Origination Policy + +iOS attaches a route (upgrading the packet to v2 and re-signing it) only when +**all** of the following hold at send time (`BLESourceRouteOriginationPolicy`): + +1. **Authored locally.** The packet's `SenderID` is our own peer ID. Relays + never rewrite someone else's packet — adding a route would force a + re-sign under the wrong key. Relays only *follow* existing routes + (`BLERouteForwardingPolicy`). +2. **Directed.** The packet has a single-peer `RecipientID` (not the + broadcast ID). In practice this covers Noise-encrypted private traffic, + private file transfers, and their fragments (fragments inherit the + parent's route and version, per §5). +3. **TTL headroom.** `TTL > 1`. Link-local packets (e.g. `REQUEST_SYNC`, + always TTL 0) never carry routes. +4. **Recipient not directly connected.** A direct write already delivers in + one hop; a route would only add bytes. +5. **Complete v2 path exists.** BFS over the confirmed-edge mesh graph + (`MeshTopologyTracker`, built from verified announce `directNeighbors` + claims, entries expiring after 60 s) finds a path with **at most 4 + intermediate hops** where every intermediate hop **and the recipient** + has been observed originating or relaying a v2 packet. Nodes never seen + speaking v2 are assumed v1-only and are excluded — a v1 client cannot + decode a v2 frame, so routing through it would silently drop the packet. +6. **No recent route failure.** See below. + +If any gate fails, behavior is exactly the pre-routing flood/direct-write +path — v1 peers observe no change. + +### Failure Fallback + +A routed unicast rides one path; a broken hop loses the packet where a flood +would heal around it. iOS keeps a small per-recipient health cache +(`BLESourceRouteFailureCache`): a routed send that sees no inbound packet +authored by the recipient within 10 s counts as a route failure, and directed +sends to that recipient fall back to flooding for the next 60 s before +routing is attempted again. Retransmission of the payload itself stays where +it always was (MessageRouter and higher layers). diff --git a/localPackages/BitFoundation/Sources/BitFoundation/BitchatPacket.swift b/localPackages/BitFoundation/Sources/BitFoundation/BitchatPacket.swift index b46a552d..ca29a2b2 100644 --- a/localPackages/BitFoundation/Sources/BitFoundation/BitchatPacket.swift +++ b/localPackages/BitFoundation/Sources/BitFoundation/BitchatPacket.swift @@ -14,7 +14,7 @@ import struct Foundation.Date /// including TTL for hop limiting and optional encryption. /// - Note: Packets larger than BLE MTU (512 bytes) are automatically fragmented public struct BitchatPacket: Codable { - let version: UInt8 + public let version: UInt8 public let type: UInt8 public let senderID: Data public let recipientID: Data?