diff --git a/src/protocols/gossipsub/rpc.zig b/src/protocols/gossipsub/rpc.zig index 510a163..556f264 100644 --- a/src/protocols/gossipsub/rpc.zig +++ b/src/protocols/gossipsub/rpc.zig @@ -262,6 +262,7 @@ pub fn decodePublishes(allocator: std.mem.Allocator, rpc_wire: []const u8) (Erro try w.nextFieldValue(val_buf, key.wire_type); off += nv.total; if (key.field_number == 2 and key.wire_type == .length_delimited) { + if (out.items.len >= lim.max_publishes_per_rpc) return error.PayloadTooLarge; const blob = try allocator.dupe(u8, nv.value); errdefer allocator.free(blob); try out.append(allocator, blob); diff --git a/src/protocols/gossipsub/runtime.zig b/src/protocols/gossipsub/runtime.zig index 3c00114..1f237c1 100644 --- a/src/protocols/gossipsub/runtime.zig +++ b/src/protocols/gossipsub/runtime.zig @@ -114,6 +114,14 @@ pub const GossipsubConfig = struct { mesh_n_low: u8 = gs_cfg.mesh_n_low, mesh_n: u8 = gs_cfg.mesh_n, mesh_n_high: u8 = gs_cfg.mesh_n_high, + /// rust-libp2p flood_publish (default on): the originator of a published + /// message delivers it to ALL subscribed peers (remote_interest), not just + /// the current mesh — decouples first-hop coverage from mesh convergence. + flood_publish: bool = true, + /// Extra heartbeats a pruned peer stays GRAFT-ineligible past exact backoff + /// expiry (rust-libp2p backoff_slack). Prevents lockstep re-GRAFT flap. 0 = + /// exact (no slack). + backoff_slack_heartbeats: u32 = 1, gossip_lazy: u8 = gs_cfg.gossip_lazy, history_length: u8 = gs_cfg.history_length, heartbeat_interval_ms: i64 = gs_cfg.heartbeat_interval_ms, @@ -316,10 +324,23 @@ pub const ValidationResult = enum { pub const TopicValidator = *const fn (ctx: ?*anyopaque, topic: []const u8, data: []const u8) ValidationResult; /// IDONTWANT cache key (#85): the peer that asked us to skip the message id. -const IDontWantEntry = struct { +/// Key for the IDONTWANT map (rust-parity O(1) membership). Value = expires_ms. +const IDontWantKey = struct { peer: identity.PeerId, id: [20]u8, - expires_ms: i64, +}; +const IDontWantKeyContext = struct { + pub fn hash(_: IDontWantKeyContext, k: IDontWantKey) u64 { + var buf: [128]u8 = undefined; + const pb = k.peer.toBytes(&buf) catch return 0; + var h = std.hash.Wyhash.init(0); + h.update(pb); + h.update(&k.id); + return h.final(); + } + pub fn eql(_: IDontWantKeyContext, a: IDontWantKey, b: IDontWantKey) bool { + return a.peer.eql(&b.peer) and std.mem.eql(u8, &a.id, &b.id); + } }; /// Coarse spin-lock guarding all `Gossipsub` public entry points. This Zig @@ -475,7 +496,7 @@ pub const Gossipsub = struct { /// Inbound publishes dropped because the topic validator returned `ignore` (#84). inbound_dropped_validator_ignore: u64, /// Per-peer IDONTWANT cache used to suppress redundant outbound publishes / IHAVE (#85). - idontwant: std.ArrayList(IDontWantEntry), + idontwant: std.HashMap(IDontWantKey, i64, IDontWantKeyContext, std.hash_map.default_max_load_percentage), /// Outbound publishes suppressed because the destination had sent IDONTWANT (#85). suppressed_outbound_idontwant: u64, /// Always-mesh peers (libp2p gossipsub `direct peers`, #85). Never pruned, never backed-off. @@ -631,7 +652,7 @@ pub const Gossipsub = struct { .inbound_dropped_seqno_replay = 0, .inbound_dropped_validator_reject = 0, .inbound_dropped_validator_ignore = 0, - .idontwant = .empty, + .idontwant = .init(allocator), .suppressed_outbound_idontwant = 0, .direct_peers = .init(allocator), .px_dial_queue = .empty, @@ -686,7 +707,7 @@ pub const Gossipsub = struct { for (self.backoff.items) |b| self.allocator.free(b.topic); self.backoff.deinit(self.allocator); self.seqno_dedup.deinit(self.allocator); - self.idontwant.deinit(self.allocator); + self.idontwant.deinit(); self.direct_peers.deinit(); for (self.px_dial_queue.items) |b| self.allocator.free(b); self.px_dial_queue.deinit(self.allocator); @@ -745,9 +766,14 @@ pub const Gossipsub = struct { /// Drops expired back-off windows. Called from [`heartbeat`] and read paths. fn pruneBackoff(self: *Gossipsub) void { + // GC only PAST the slack window (expires + slack). Removing at exact + // expiry here — before heartbeat graft selection — would defeat the + // GRAFT-send slack (the entry would be gone, so isPeerBackedOffForGraft + // sees nothing and re-grafts in the same heartbeat the backoff lapses). + const slack = self.backoffSlackMs(); var i: usize = 0; while (i < self.backoff.items.len) { - if (self.backoff.items[i].expires_ms <= self.clock_ms) { + if (self.backoff.items[i].expires_ms + slack <= self.clock_ms) { const e = self.backoff.orderedRemove(i); self.allocator.free(e.topic); } else { @@ -756,13 +782,25 @@ pub const Gossipsub = struct { } } - /// Returns the active back-off entry's index for `(peer, topic)` or `null`. - /// Side effect: removes the entry on the fly if it has expired. + /// Extra time a `(peer,topic)` stays GRAFT-ineligible PAST its exact backoff + /// expiry (rust-libp2p `backoff_slack`, default 1 heartbeat). Ensures we never + /// re-GRAFT a peer in the same heartbeat its backoff lapses — both ends would + /// otherwise re-graft in lockstep, re-collide, and re-prune (a sustained + /// synchronized GRAFT/PRUNE flap on dense topics). The slack applies ONLY to + /// our GRAFT-SEND eligibility; GRAFT-ACCEPT + reported remaining stay exact. + fn backoffSlackMs(self: *const Gossipsub) i64 { + return @as(i64, @intCast(self.cfg.backoff_slack_heartbeats)) * self.cfg.heartbeat_interval_ms; + } + + /// Returns the back-off entry's index for `(peer, topic)` or `null`. Side + /// effect: GCs entries only once they are past expiry + slack, so the slack + /// window is observable by the GRAFT-send check. fn findActiveBackoff(self: *Gossipsub, peer: identity.PeerId, topic: []const u8) ?usize { + const slack = self.backoffSlackMs(); var i: usize = 0; while (i < self.backoff.items.len) { const b = self.backoff.items[i]; - if (b.expires_ms <= self.clock_ms) { + if (b.expires_ms + slack <= self.clock_ms) { const removed = self.backoff.orderedRemove(i); self.allocator.free(removed.topic); continue; @@ -774,13 +812,21 @@ pub const Gossipsub = struct { } /// `true` iff the local node is currently in PRUNE back-off toward `(peer, topic)`, - /// i.e. MUST NOT send GRAFT. + /// i.e. MUST NOT send GRAFT. EXACT (no slack) — used for accept-side checks + /// and the public API; the slack-aware variant is `isPeerBackedOffForGraft`. pub fn isPeerBackedOff(self: *Gossipsub, peer: identity.PeerId, topic: []const u8) bool { self.lock(); defer self.unlock(); return self.isPeerBackedOffInner(peer, topic); } fn isPeerBackedOffInner(self: *Gossipsub, peer: identity.PeerId, topic: []const u8) bool { + const idx = self.findActiveBackoff(peer, topic) orelse return false; + return self.backoff.items[idx].expires_ms > self.clock_ms; + } + + /// `true` iff we must not GRAFT `(peer,topic)` yet — backed off OR within the + /// slack window past expiry (rust `is_backoff_with_slack`). + fn isPeerBackedOffForGraft(self: *Gossipsub, peer: identity.PeerId, topic: []const u8) bool { return self.findActiveBackoff(peer, topic) != null; } @@ -824,7 +870,10 @@ pub const Gossipsub = struct { fn remainingBackoffSecondsFor(self: *Gossipsub, peer: identity.PeerId, topic: []const u8) ?u64 { const idx = self.findActiveBackoff(peer, topic) orelse return null; const remain_ms = self.backoff.items[idx].expires_ms - self.clock_ms; - if (remain_ms <= 0) return 0; + // Exact: during the slack window (remain <= 0) the peer is no longer + // backed off for ACCEPT purposes — return null so an inbound GRAFT is + // accepted (only our own GRAFT-send waits out the slack). + if (remain_ms <= 0) return null; return @intCast(@divTrunc(remain_ms + 999, 1000)); } @@ -928,37 +977,46 @@ pub const Gossipsub = struct { /// Also sweeps expired entries along the way. fn peerWantsNotPublish(self: *Gossipsub, peer: identity.PeerId, id: [20]u8) bool { if (!self.cfg.idontwant_runtime_enabled) return false; - var i: usize = 0; - var hit = false; - while (i < self.idontwant.items.len) { - const e = self.idontwant.items[i]; - if (e.expires_ms <= self.clock_ms) { - _ = self.idontwant.orderedRemove(i); - continue; - } - if (!hit and e.peer.eql(&peer) and std.mem.eql(u8, &e.id, &id)) hit = true; - i += 1; + // O(1) keyed lookup (was an O(n) scan over up to max_idontwant_entries on + // the per-forward × per-mesh-peer hot path — the owner-thread ceiling). + // Evict the specific entry if expired (O(1), self-sweeping on lookup). + const key = IDontWantKey{ .peer = peer, .id = id }; + const exp = self.idontwant.get(key) orelse return false; + if (exp <= self.clock_ms) { + _ = self.idontwant.remove(key); + return false; } - return hit; + return true; } fn rememberIDontWant(self: *Gossipsub, peer: identity.PeerId, id: [20]u8) std.mem.Allocator.Error!void { if (!self.cfg.idontwant_runtime_enabled) return; - // Already present? bump expiry only. - for (self.idontwant.items) |*e| { - if (e.peer.eql(&peer) and std.mem.eql(u8, &e.id, &id)) { - e.expires_ms = self.clock_ms + self.cfg.idontwant_ttl_ms; - return; + const key = IDontWantKey{ .peer = peer, .id = id }; + const exp = self.clock_ms + self.cfg.idontwant_ttl_ms; + if (self.idontwant.getPtr(key)) |v| { + v.* = exp; // bump expiry + return; + } + if (self.idontwant.count() >= self.cfg.max_idontwant_entries) { + self.pruneIDontWant(); + if (self.idontwant.count() >= self.cfg.max_idontwant_entries) { + // Still full of live entries — evict one arbitrary to stay bounded. + var it = self.idontwant.keyIterator(); + if (it.next()) |k| _ = self.idontwant.remove(k.*); } } - if (self.idontwant.items.len >= self.cfg.max_idontwant_entries) { - _ = self.idontwant.orderedRemove(0); + try self.idontwant.put(key, exp); + } + + /// Remove expired IDONTWANT entries (called from the heartbeat prune sweep). + fn pruneIDontWant(self: *Gossipsub) void { + var expired: std.ArrayListUnmanaged(IDontWantKey) = .empty; + defer expired.deinit(self.allocator); + var it = self.idontwant.iterator(); + while (it.next()) |e| { + if (e.value_ptr.* <= self.clock_ms) expired.append(self.allocator, e.key_ptr.*) catch break; } - try self.idontwant.append(self.allocator, .{ - .peer = peer, - .id = id, - .expires_ms = self.clock_ms + self.cfg.idontwant_ttl_ms, - }); + for (expired.items) |k| _ = self.idontwant.remove(k); } /// Mark `peer` as a direct (always-mesh) peer. Direct peers are never pruned by the @@ -1507,9 +1565,29 @@ pub const Gossipsub = struct { if (self.subs.contains(topic)) { try self.ensureTopicMesh(topic); const mp = self.mesh.getPtr(topic).?; - var pit = mp.peers.keyIterator(); - while (pit.next()) |kp| { - try self.appendPublishWire(wire, kp.*, id); + const ri: ?*TopicMesh = if (self.cfg.flood_publish) self.remote_interest.getPtr(topic) else null; + if (ri) |rip| { + // rust-libp2p flood_publish (default on): the ORIGINATOR delivers + // its own message to EVERY subscribed peer, not just the current + // mesh — so a validator's own attestation reaches all subnet + // members even when the mesh is transiently below mesh_n (mid- + // convergence / one slot post-prune). Without this, first-hop + // coverage is capped by mesh size → sparse subnets stall below + // quorum. Mesh peers not (yet) in remote_interest are still sent. + var rit = rip.peers.keyIterator(); + while (rit.next()) |kp| { + try self.appendPublishWire(wire, kp.*, id); + } + var pit = mp.peers.keyIterator(); + while (pit.next()) |kp| { + if (rip.peers.contains(kp.*)) continue; + try self.appendPublishWire(wire, kp.*, id); + } + } else { + var pit = mp.peers.keyIterator(); + while (pit.next()) |kp| { + try self.appendPublishWire(wire, kp.*, id); + } } self.allocator.free(wire); } else { @@ -1620,6 +1698,15 @@ pub const Gossipsub = struct { const sa = c.gs.peerBehaviourScore(a); const sb = c.gs.peerBehaviourScore(b); if (sa != sb) return sa < sb; + // Tie-break: per-node SALTED hash of the peer id (rust shuffles + // equal-scored peers before pruning). With scoring off all scores + // are 0, so a raw peer-id byte order would make EVERY node prune + // the same low-id peer → network-wide synchronized pruning that + // feeds the GRAFT/PRUNE flap. Salting with our own peer id + // decorrelates victims across nodes. Deterministic (testable). + const ha = c.gs.pruneTieBreak(a); + const hb = c.gs.pruneTieBreak(b); + if (ha != hb) return ha < hb; var ba: [128]u8 = undefined; var bb: [128]u8 = undefined; const ab = a.toBytes(&ba) catch return false; @@ -1630,6 +1717,19 @@ pub const Gossipsub = struct { std.mem.sort(identity.PeerId, peers, ctx, S.less); } + /// Per-node salted hash of a peer id, used as the prune tie-break so equal- + /// scored peers are pruned in a node-specific (decorrelated) order. + fn pruneTieBreak(self: *const Gossipsub, p: identity.PeerId) u64 { + var pbuf: [128]u8 = undefined; + const pb = p.toBytes(&pbuf) catch return 0; + var sbuf: [128]u8 = undefined; + const sb = self.cfg.local_peer_id.toBytes(&sbuf) catch return 0; + var h = std.hash.Wyhash.init(0); + h.update(sb); + h.update(pb); + return h.final(); + } + fn candidatesOutsideMesh(self: *Gossipsub, topic: []const u8) std.mem.Allocator.Error![]identity.PeerId { self.scratch_peers.clearRetainingCapacity(); const mp = self.mesh.getPtr(topic) orelse return &[_]identity.PeerId{}; @@ -1652,7 +1752,7 @@ pub const Gossipsub = struct { // (a peer that PRUNEs us records back-off for us too), so re-grafting a // backed-off peer just makes it re-PRUNE us -> flap. The fix for sparse- // subnet decay is to stop SPURIOUS prunes, not bypass back-off. - if (!self.direct_peers.contains(p) and self.isPeerBackedOffInner(p, topic)) continue; + if (!self.direct_peers.contains(p) and self.isPeerBackedOffForGraft(p, topic)) continue; try self.scratch_peers.append(self.allocator, p); } // Direct peers always sort to the front; otherwise score-desc, then bytes. @@ -1666,7 +1766,7 @@ pub const Gossipsub = struct { while (cit.next()) |kp| { const p = kp.*; if (p.eql(&self.cfg.local_peer_id)) continue; - if (!self.direct_peers.contains(p) and self.isPeerBackedOffInner(p, topic)) continue; + if (!self.direct_peers.contains(p) and self.isPeerBackedOffForGraft(p, topic)) continue; try self.scratch_peers.append(self.allocator, p); } self.sortPeersDirectThenScoreDescThenBytes(self.scratch_peers.items); @@ -2125,6 +2225,7 @@ pub const Gossipsub = struct { self.dup.prune(self.clock_ms); self.prunePullCache(); self.pruneRecentSeen(); + self.pruneIDontWant(); self.pruneBackoff(); self.pruneTopicUnsubscribeCooldown(); self.pruneFanout(); @@ -2329,7 +2430,7 @@ pub const Gossipsub = struct { pub fn idontwantCount(self: *const Gossipsub) usize { self.lock(); defer self.unlock(); - return self.idontwant.items.len; + return self.idontwant.count(); } /// PX dial-suggestion queue depth. pub fn dialSuggestionCount(self: *const Gossipsub) usize { @@ -3048,6 +3149,33 @@ test "remote SUBSCRIBE arriving before our own subscribe is retained and drained try std.testing.expectEqual(@as(usize, 0), g.pending_remote_subs.items.len); } +test "backoff slack: GRAFT-send stays ineligible through the slack window while ACCEPT is exact" { + const a = std.testing.allocator; + const me = try identity.PeerId.random(); + var g = try Gossipsub.init(a, .{ .local_peer_id = me, .heartbeat_interval_ms = 700, .backoff_slack_heartbeats = 1 }); + defer g.deinit(); + g.setClockMs(0); + const peer = try identity.PeerId.random(); + // expires at 30000; slack = 1 * 700 -> GRAFT-send ineligible until 30700. + try g.recordBackoff(peer, "t", 30_000); + + // Deep inside the window: both exact and graft-send say backed off. + g.setClockMs(15_000); + try std.testing.expect(g.isPeerBackedOffInner(peer, "t")); + try std.testing.expect(g.isPeerBackedOffForGraft(peer, "t")); + + // Past exact expiry but within slack: ACCEPT is exact (clear), GRAFT-send + // still ineligible — this is the window the slack protects. + g.setClockMs(30_200); + try std.testing.expect(!g.isPeerBackedOffInner(peer, "t")); + try std.testing.expect(g.isPeerBackedOffForGraft(peer, "t")); + + // Past the slack: both clear. + g.setClockMs(30_800); + try std.testing.expect(!g.isPeerBackedOffInner(peer, "t")); + try std.testing.expect(!g.isPeerBackedOffForGraft(peer, "t")); +} + test "inbound PRUNE records back-off; heartbeat refuses to re-graft inside window" { const a = std.testing.allocator; const me = try identity.PeerId.random(); diff --git a/src/protocols/gossipsub/wire_limits.zig b/src/protocols/gossipsub/wire_limits.zig index 192627a..8efa108 100644 --- a/src/protocols/gossipsub/wire_limits.zig +++ b/src/protocols/gossipsub/wire_limits.zig @@ -21,6 +21,12 @@ pub const max_message_id_bytes: usize = 256; /// Repeated message ids inside a single `IHave` / `IWant` / `IDontWant` entry. pub const max_message_ids_per_entry: usize = 8192; +/// Max publish messages decoded from a single inbound `RPC` frame. Bounds the +/// per-frame allocation + owner-thread forward loop against a peer packing a +/// huge number of tiny messages into one frame (rust caps via +/// `max_messages_per_rpc`). Generous — honest peers send a handful. +pub const max_publishes_per_rpc: usize = 8192; + /// Nested `ControlExtensions` message (`ControlMessage.extensions`, field 6). pub const max_control_extensions_blob_bytes: usize = 4096; diff --git a/src/transport/quic/conn_table.zig b/src/transport/quic/conn_table.zig index fb832b9..b0be0f0 100644 --- a/src/transport/quic/conn_table.zig +++ b/src/transport/quic/conn_table.zig @@ -73,6 +73,9 @@ pub const InboundStream = struct { stream_id: u64, raw: quic_raw_stream_io.RawAppBidiServer, handshake_done: bool = false, + /// Wall-clock ms the stream was accepted; drives the inbound reap + /// ([`inbound_request_reap_ms`]). 0 = unset (never reaped on age). + created_ms: i64 = 0, protocol_index: ?usize = null, /// channel_id once we've called `host.registerInboundReqRespChannel`. channel_id: ?u64 = null, @@ -115,6 +118,14 @@ pub const InboundStream = struct { /// RawAppStreamSlotsFull. Set well above any real RPC round-trip. pub const outbound_request_reap_ms: i64 = 15_000; +/// Deadline after which the drive loop reaps an INBOUND req/resp stream that has +/// not finished its response. Catches a peer that opens a stream and stalls +/// mid-request (never sends the body / never FINs), which otherwise leaks the +/// InboundStream + its raw-app slot forever (→ RawAppStreamSlotsFull → the node +/// silently stops serving sync). Generous (2× the outbound reaper) so a slow but +/// legitimate response is never falsely reaped. +pub const inbound_request_reap_ms: i64 = 30_000; + pub const OutboundRequest = struct { /// The peer this request is destined for. peer: identity.PeerId, diff --git a/src/transport/quic/runtime.zig b/src/transport/quic/runtime.zig index 6546bc0..b23221e 100644 --- a/src/transport/quic/runtime.zig +++ b/src/transport/quic/runtime.zig @@ -1319,6 +1319,7 @@ pub const QuicRuntime = struct { .conn = conn, .stream_id = stream_id, }, + .created_ms = self.opts.now_ms_fn(), }; try sh.inbound_streams.append(self.allocator, ist); } @@ -1358,6 +1359,7 @@ pub const QuicRuntime = struct { .client = client, }, .known_peer_id = peer_id, + .created_ms = self.opts.now_ms_fn(), }; sh.inbound_streams.append(self.allocator, ist) catch { self.allocator.destroy(ist); @@ -4072,6 +4074,30 @@ pub const QuicRuntime = struct { continue; } + // 0b. Reap a STALE inbound req/resp stream (R1): one accepted but never + // finished its response within `inbound_request_reap_ms` (peer opened + // the stream, finished multistream-select, then stalled — no request + // body, never FINs). Without this the InboundStream + its zquic + // raw-app slot leak forever → RawAppStreamSlotsFull → the node + // silently stops serving sync. + // ONLY req/resp protocols (status/blocks_by_*, indices + // proto_meshsub_last_index+1 .. proto_relay_hop) and PRE-handshake + // stalls are reapable. Persistent `/meshsub` gossip (0..3), relay + // hop/stop, and dcutr are LONG-LIVED — they never set + // response_fin_sent and must NOT be reaped on age (doing so would + // tear down every gossip stream every 30s). + const reapable = if (ist.protocol_index) |p| + (p > config.proto_meshsub_last_index and p < config.proto_relay_hop) + else + true; // never finished multistream-select → genuinely stuck + if (reapable and !ist.response_fin_sent and ist.created_ms != 0 and + self.opts.now_ms_fn() - ist.created_ms > conn_table.inbound_request_reap_ms) + { + log.warn("quic_runtime: reaping stale inbound req/resp stream (no response in {d}ms)", .{conn_table.inbound_request_reap_ms}); + self.removeInboundStreamAt(sh, i); + continue; + } + // 1. Multistream handshake: responder side. // // Bytes pulled from the raw stream live in `ms_acc`, which we own