mirror of
https://github.com/permissionlesstech/bitchat.git
synced 2026-07-24 23:25:19 +00:00
Originate v2 source routes and wire fragmentIdFilter targeted resync
Part A — source-route origination policy: - Gate route application (BLESourceRouteOriginationPolicy): only packets we author, directed at a single peer, with TTL headroom, whose recipient is not directly connected. Relays no longer attach routes to (and re-sign) packets they merely forward. - Version-gate paths: MeshTopologyTracker records the highest protocol version observed per peer; BFS routes require every intermediate hop and the recipient to be v2-observed, capped at 4 intermediate hops. - Degrade on failure: BLESourceRouteFailureCache marks a routed send that sees no inbound traffic from the recipient within 10s as failed and floods for 60s before retrying routes. Part B — REQUEST_SYNC fragmentIdFilter (TLV 0x06): - Requester: BLEFragmentAssemblyBuffer reports stalled broadcast reassemblies (no new fragment for 5s, retried at most every 10s); the maintenance pass sends a types=fragment REQUEST_SYNC naming the stalled 8-byte fragment stream IDs to each connected peer. - Responder: GossipSyncManager restricts the fragment diff to exactly the named streams, bypassing the since-cursor while the GCS filter still excludes pieces the requester holds; RSR/TTL-0/rate-limit semantics unchanged and REQUEST_SYNC stays link-local. - Bounds: at most 60 IDs per request (60*17-1 = 1019 bytes <= the 1024-byte decoder cap); oversized 0x06 values are ignored, not fatal. Docs: SOURCE_ROUTING.md gains the iOS origination policy (§8); REQUEST_SYNC_MANAGER.md documents 0x05/0x06 as implemented. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
jack
@@ -1,10 +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 (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
|
||||
@@ -12,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<Data>? {
|
||||
guard let filter else { return nil }
|
||||
var ids: Set<Data> = []
|
||||
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
|
||||
@@ -88,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:
|
||||
|
||||
@@ -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]] = [:]
|
||||
@@ -106,6 +109,7 @@ struct BLEFragmentAssemblyBuffer {
|
||||
}
|
||||
|
||||
fragmentsByKey[header.key]?[header.index] = header.fragmentData
|
||||
metadataByKey[header.key]?.lastFragmentAt = now
|
||||
|
||||
guard let fragments = fragmentsByKey[header.key],
|
||||
fragments.count == header.total else {
|
||||
@@ -138,10 +142,41 @@ 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`.
|
||||
/// 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 stalled: [Data] = []
|
||||
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 }
|
||||
metadataByKey[key]?.lastResyncRequestAt = now
|
||||
stalled.append(withUnsafeBytes(of: key.id.bigEndian) { Data($0) })
|
||||
}
|
||||
return stalled
|
||||
}
|
||||
|
||||
private static func assemblyLimit(for originalType: UInt8) -> Int {
|
||||
if originalType == MessageType.fileTransfer.rawValue {
|
||||
// Allow headroom for TLV metadata and binary framing overhead.
|
||||
|
||||
@@ -67,7 +67,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()
|
||||
@@ -576,6 +578,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()
|
||||
@@ -2375,13 +2378,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,
|
||||
@@ -2399,6 +2420,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
|
||||
}
|
||||
|
||||
@@ -3170,13 +3194,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:
|
||||
@@ -3681,10 +3715,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)
|
||||
|
||||
@@ -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
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
}
|
||||
@@ -10,6 +10,10 @@ final class MeshTopologyTracker {
|
||||
private var claims: [RoutingID: Set<RoutingID>] = [:]
|
||||
// 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<RoutingID> = [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'?
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -258,11 +258,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 {
|
||||
@@ -340,9 +358,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
|
||||
@@ -369,7 +397,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 (_, pair) in latestAnnouncementByPeer where isPacketFresh(pair.packet) {
|
||||
@@ -387,7 +415,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()
|
||||
}
|
||||
|
||||
@@ -406,7 +434,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))
|
||||
@@ -424,7 +452,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()
|
||||
}
|
||||
|
||||
|
||||
@@ -469,6 +469,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 {
|
||||
@@ -545,6 +637,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()
|
||||
@@ -566,6 +659,6 @@ private final class RecordingDelegate: GossipSyncManager.Delegate {
|
||||
}
|
||||
|
||||
func getConnectedPeers() -> [PeerID] {
|
||||
return []
|
||||
return connectedPeers
|
||||
}
|
||||
}
|
||||
|
||||
@@ -135,6 +135,82 @@ 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 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,
|
||||
|
||||
@@ -0,0 +1,98 @@
|
||||
//
|
||||
// BLESourceRouteFailureCacheTests.swift
|
||||
// bitchatTests
|
||||
//
|
||||
// This is free and unencumbered software released into the public domain.
|
||||
// For more information, see <https://unlicense.org>
|
||||
//
|
||||
|
||||
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)))
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,98 @@
|
||||
//
|
||||
// BLESourceRouteOriginationPolicyTests.swift
|
||||
// bitchatTests
|
||||
//
|
||||
// This is free and unencumbered software released into the public domain.
|
||||
// For more information, see <https://unlicense.org>
|
||||
//
|
||||
|
||||
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)
|
||||
}
|
||||
}
|
||||
@@ -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..<count).map { try hex(String(format: "%016x", $0 + 1)) }
|
||||
for i in 0..<count {
|
||||
var neighbors: [Data] = []
|
||||
if i > 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)
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -0,0 +1,76 @@
|
||||
//
|
||||
// RequestSyncPacketFragmentFilterTests.swift
|
||||
// bitchatTests
|
||||
//
|
||||
// This is free and unencumbered software released into the public domain.
|
||||
// For more information, see <https://unlicense.org>
|
||||
//
|
||||
|
||||
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)
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
|
||||
|
||||
+42
-1
@@ -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).
|
||||
|
||||
@@ -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?
|
||||
|
||||
Reference in New Issue
Block a user