mirror of
https://github.com/permissionlesstech/bitchat.git
synced 2026-07-27 02:25:19 +00:00
Originate v2 source routes and wire fragmentIdFilter targeted resync (#1378)
* 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> * Fix stall-clock refresh on duplicates and overflow suppression in fragment resync Two fixes to stalledBroadcastFragmentIDs bookkeeping in BLEFragmentAssemblyBuffer: - Duplicate fragments no longer reset the stall clock. Fragment packets bypass the packet deduplicator, so relayed duplicates of an already-held index arriving every few seconds kept lastFragmentAt fresh and suppressed the targeted REQUEST_SYNC indefinitely. Now lastFragmentAt only updates when the index is new (actual progress). - Only the streams that will actually be encoded on the wire are rate-limited. Previously every stalled candidate got lastResyncRequestAt set, but encodeFragmentIdFilter serializes at most RequestSyncPacket.maxFragmentIdFilterCount (60) IDs, so overflow streams were suppressed for retryAfter without ever being requested. Selection now caps at that shared constant, oldest stall first, so overflow stays eligible and rotates fairly on the next pass. Tests: duplicates arriving periodically still trigger the stall report; 70 stalled streams yield the 60 oldest on the first pass and the remaining 10 on the next. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> --------- Co-authored-by: jack <jackjackbits@users.noreply.github.com> Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
jack
Claude Fable 5
parent
cee2bcd535
commit
70229f0be1
@@ -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()
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user