diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 04909c7..a41a321 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -115,6 +115,10 @@ jobs: # is actually exercised on every change instead of silently rotting — it # covers exactly the long-lived /meshsub gossip + flow-control paths that # several devnet finalization bugs lived in. + # Bounded: a deadlocked soak run otherwise holds the job for the workflow + # default (observed as 90+ min hangs); the test itself finishes in ~1-2 + # min, so 10 minutes is generous while converting a hang into a fast fail. + timeout-minutes: 10 run: zig build soak-test --summary all - name: Build example executables diff --git a/src/core/connection_manager.zig b/src/core/connection_manager.zig index 1023c1c..08674c1 100644 --- a/src/core/connection_manager.zig +++ b/src/core/connection_manager.zig @@ -102,6 +102,48 @@ pub const KnownPeerDialStatus = struct { dial_inflight: bool, }; +/// A `conns` entry the reconciliation sweep flagged as orphaned (#299): still +/// present in the peer book but absent from the transport's live-connection +/// snapshot for at least [`reconcile_orphan_grace_ms`]. The caller (the +/// transport's single coordinator thread) synthesizes the close through the +/// same host path a real transport close takes, so `peer_disconnected`, +/// req/resp cleanup, peer-level teardown, and known-peer redial scheduling all +/// run exactly as if the lost lifecycle event had been delivered. +pub const OrphanedConn = struct { + conn_id: ConnectionId, + peer: identity.PeerId, +}; + +/// How long a book entry must be continuously missing from the transport's +/// live snapshot before [`reconcile`] reports it. Two effects: (a) an entry is +/// reported only after being seen orphaned on at least two sweeps (the first +/// sighting records a candidate, the report requires a later sweep past the +/// grace), so a snapshot racing an in-flight establish/close can never orphan +/// a healthy conn; and (b) the transport's own close-detection paths (which +/// run every drive lap) always win while the event pipeline is healthy — +/// reconciliation only fires when an event was genuinely lost. +pub const reconcile_orphan_grace_ms: i64 = 15_000; + +/// Snapshot of the lifecycle emit counters vs the live table sizes (#299). +/// `conns_established_total - conns_closed_total` should track +/// `tracked_conns`, and the `*_emitted_total` counters should track what the +/// embedder observed; a divergence between the two pairs is exactly the +/// "book believes peers are connected, transport disagrees" drift. +pub const LifecycleStats = struct { + conns_established_total: u64, + conns_closed_total: u64, + peer_connected_emitted_total: u64, + peer_disconnected_emitted_total: u64, + close_unknown_conn_total: u64, + close_peer_active_skew_total: u64, + reconcile_orphans_flagged_total: u64, + reconcile_stale_peers_total: u64, + /// Current size of the conn table (peer-book side). + tracked_conns: usize, + /// Current number of peers with >= 1 active connection. + connected_peers: usize, +}; + pub const ConnectionManager = struct { allocator: std.mem.Allocator, swarm: *swarm_mod.Swarm, @@ -124,6 +166,38 @@ pub const ConnectionManager = struct { /// Total trim recommendations emitted across both reason codes (#90 observability). trim_recommendations_total: u64 = 0, + // ── Lifecycle emit-count observability (#299) ─────────────────────────── + // The devnet wedge behind #299 presented as a peer book that disagreed + // with the live transport while NO lifecycle events flowed. These counters + // make that divergence measurable: compare what the transport told us + // (`conns_established_total` / `conns_closed_total`) and what we told the + // embedder (`peer_connected_emitted_total` / `peer_disconnected_emitted_total`) + // against the live table sizes via [`lifecycleStats`]. + /// Connections recorded via [`onConnectionEstablished`]. + conns_established_total: u64 = 0, + /// Connections removed via [`onConnectionClosed`] (reconciled orphans included). + conns_closed_total: u64 = 0, + /// `peer_connected` events queued to the swarm. + peer_connected_emitted_total: u64 = 0, + /// `peer_disconnected` events queued to the swarm (reconcile repairs included). + peer_disconnected_emitted_total: u64 = 0, + /// [`onConnectionClosed`] calls for a conn_id we never recorded (or already + /// removed) — the silent early-return path #299 asks to make observable. + close_unknown_conn_total: u64 = 0, + /// [`onConnectionClosed`] calls that hit `conns`/`peer_active` map skew and + /// returned early without emitting `peer_disconnected`. + close_peer_active_skew_total: u64 = 0, + /// Orphaned conn entries flagged by [`reconcile`] for a synthesized close. + reconcile_orphans_flagged_total: u64 = 0, + /// Stale `peer_active` entries repaired by [`reconcile`]. + reconcile_stale_peers_total: u64 = 0, + /// conn_id -> wall-ms it was first observed missing from the transport's + /// live snapshot ([`reconcile`] grace tracking). + orphan_candidates: std.AutoHashMap(ConnectionId, i64), + /// peer -> wall-ms it was first observed in `peer_active` with no backing + /// `conns` entry ([`reconcile`] grace tracking). + stale_peer_candidates: std.HashMap(identity.PeerId, i64, PeerIdContext, std.hash_map.default_max_load_percentage), + pub fn init(allocator: std.mem.Allocator, s: *swarm_mod.Swarm) ConnectionManager { return .{ .allocator = allocator, @@ -133,6 +207,8 @@ pub const ConnectionManager = struct { .peer_active = .init(allocator), .protected_peers = .init(allocator), .trim_recommended = .init(allocator), + .orphan_candidates = .init(allocator), + .stale_peer_candidates = .init(allocator), }; } @@ -209,6 +285,25 @@ pub const ConnectionManager = struct { self.peer_active.deinit(); self.protected_peers.deinit(); self.trim_recommended.deinit(); + self.orphan_candidates.deinit(); + self.stale_peer_candidates.deinit(); + } + + /// Emit-count instrumentation snapshot (#299). Cheap; safe to call from + /// the same single thread that applies lifecycle events. + pub fn lifecycleStats(self: *const ConnectionManager) LifecycleStats { + return .{ + .conns_established_total = self.conns_established_total, + .conns_closed_total = self.conns_closed_total, + .peer_connected_emitted_total = self.peer_connected_emitted_total, + .peer_disconnected_emitted_total = self.peer_disconnected_emitted_total, + .close_unknown_conn_total = self.close_unknown_conn_total, + .close_peer_active_skew_total = self.close_peer_active_skew_total, + .reconcile_orphans_flagged_total = self.reconcile_orphans_flagged_total, + .reconcile_stale_peers_total = self.reconcile_stale_peers_total, + .tracked_conns = self.conns.count(), + .connected_peers = self.peer_active.count(), + }; } pub fn setReqResp(self: *ConnectionManager, rr: ?*req_resp_runtime.ReqResp) void { @@ -385,6 +480,7 @@ pub const ConnectionManager = struct { const seq = self.next_seq; self.next_seq += 1; try self.conns.put(conn_id, .{ .peer = peer, .direction = direction, .seq = seq }); + self.conns_established_total += 1; const gop = try self.peer_active.getOrPut(peer); const prev = if (gop.found_existing) gop.value_ptr.* else 0; @@ -395,6 +491,7 @@ pub const ConnectionManager = struct { .direction = direction, .via_relay = opts.via_relay, } }); + self.peer_connected_emitted_total += 1; } // Per-peer cap: if this connection puts us over `max_per_peer`, recommend @@ -494,6 +591,7 @@ pub const ConnectionManager = struct { // double-close races silently and the embedder sees a wedged // publish path with no clue why no `peer_disconnected` ever // fired (the gossip-asymmetry bug observed against quinn). + self.close_unknown_conn_total += 1; log.warn( "onConnectionClosed: unknown conn_id={d} reason={s} (already closed or never registered)", .{ conn_id, @tagName(reason) }, @@ -502,9 +600,11 @@ pub const ConnectionManager = struct { }; const peer = ent.value.peer; const direction = ent.value.direction; + self.conns_closed_total += 1; _ = self.trim_recommended.remove(conn_id); const pr = self.peer_active.getPtr(peer) orelse { + self.close_peer_active_skew_total += 1; log.warn( "onConnectionClosed: conn_id={d} dir={s} but peer not in peer_active map (skew)", .{ conn_id, @tagName(direction) }, @@ -515,6 +615,7 @@ pub const ConnectionManager = struct { // Would underflow — peer_active was already 0 when we still had // a `conns` entry. Means the maps are out of sync, log and // bail rather than wrapping to u32_max. + self.close_peer_active_skew_total += 1; log.warn( "onConnectionClosed: conn_id={d} dir={s} peer_active already 0 (map skew); removing entry", .{ conn_id, @tagName(direction) }, @@ -536,6 +637,7 @@ pub const ConnectionManager = struct { .direction = direction, .reason = reason, } }); + self.peer_disconnected_emitted_total += 1; if (self.req_resp) |rr| { try rr.onPeerDisconnected(peer); @@ -568,6 +670,138 @@ pub const ConnectionManager = struct { } return count == 0; } + + /// Reconciliation sweep (#299): compare the peer book against `live` — the + /// transport's snapshot of every connection id that currently has a live + /// leg — and report entries the book believes are open but the transport + /// no longer backs. Covers lifecycle events lost in flight (e.g. dropped + /// on coordinator-queue allocation failure) and close transitions the + /// transport's own detection missed. + /// + /// Two divergence classes: + /// * **Orphaned conns** — `conns` entries missing from `live` for + /// [`reconcile_orphan_grace_ms`]. Appended to `out_orphans`; the caller + /// must route each through the normal close path + /// (`host.onConnectionClosed` with reason `.orphaned`) so + /// `peer_disconnected`, req/resp cleanup, peer-level teardown, and + /// known-peer redial scheduling all fire exactly as for a delivered + /// transport close. + /// * **Stale peers** — `peer_active` entries with NO backing `conns` + /// entry (the map-skew early returns in [`onConnectionClosed`] strand + /// these, leaving a peer "connected" forever with zero connections). + /// After the same grace the entry is removed here, a synthetic + /// `peer_disconnected` (direction `.unknown`, reason `.orphaned`) is + /// emitted, req/resp is notified, and the peer is appended to + /// `out_stale_peers` so the caller can run peer-level host teardown + /// (gossipsub / peer_protocols / kad). + /// + /// NOT thread-safe (like the rest of this struct): call from the single + /// thread that applies lifecycle events. This sweep never dials — redial + /// policy stays exactly the pre-existing known-peer backoff (see the + /// anti-churn scar tissue around `onConnectionClosed`); the learned-address + /// redial tier for inbound-only peers is a separate follow-up per #299. + pub fn reconcile( + self: *ConnectionManager, + now_ms: i64, + live: *const std.AutoHashMap(ConnectionId, void), + out_orphans: *std.ArrayList(OrphanedConn), + out_stale_peers: *std.ArrayList(identity.PeerId), + ) !void { + // Prune orphan candidates that are live again or no longer tracked, so + // a transient snapshot miss must restart the full grace window. + { + var expired: std.ArrayList(ConnectionId) = .empty; + defer expired.deinit(self.allocator); + var it = self.orphan_candidates.iterator(); + while (it.next()) |e| { + const cid = e.key_ptr.*; + if (live.contains(cid) or !self.conns.contains(cid)) { + try expired.append(self.allocator, cid); + } + } + for (expired.items) |cid| _ = self.orphan_candidates.remove(cid); + } + + // Flag book entries with no live transport leg; report after grace. + { + var it = self.conns.iterator(); + while (it.next()) |e| { + const cid = e.key_ptr.*; + if (live.contains(cid)) continue; + const gop = try self.orphan_candidates.getOrPut(cid); + if (!gop.found_existing) { + gop.value_ptr.* = now_ms; + continue; + } + if (now_ms - gop.value_ptr.* < reconcile_orphan_grace_ms) continue; + try out_orphans.append(self.allocator, .{ .conn_id = cid, .peer = e.value_ptr.peer }); + self.reconcile_orphans_flagged_total += 1; + // Leave the candidate: the caller's synthesized close removes + // the conns entry and the prune above clears it next sweep; if + // the close fails we simply re-report. + } + } + + // Stale peer_active entries: peer marked connected, zero conns entries. + var backed = std.HashMap(identity.PeerId, void, PeerIdContext, std.hash_map.default_max_load_percentage).init(self.allocator); + defer backed.deinit(); + { + var it = self.conns.valueIterator(); + while (it.next()) |ent| try backed.put(ent.peer, {}); + } + { + var expired: std.ArrayList(identity.PeerId) = .empty; + defer expired.deinit(self.allocator); + var it = self.stale_peer_candidates.iterator(); + while (it.next()) |e| { + const p = e.key_ptr.*; + if (backed.contains(p) or !self.peer_active.contains(p)) { + try expired.append(self.allocator, p); + } + } + for (expired.items) |p| _ = self.stale_peer_candidates.remove(p); + } + { + var flagged: std.ArrayList(identity.PeerId) = .empty; + defer flagged.deinit(self.allocator); + var it = self.peer_active.keyIterator(); + while (it.next()) |kp| { + const p = kp.*; + if (backed.contains(p)) continue; + const gop = try self.stale_peer_candidates.getOrPut(p); + if (!gop.found_existing) { + gop.value_ptr.* = now_ms; + continue; + } + if (now_ms - gop.value_ptr.* < reconcile_orphan_grace_ms) continue; + try flagged.append(self.allocator, p); + } + for (flagged.items) |p| { + log.warn("reconcile: repairing stale peer_active entry (peer marked connected with zero conn entries)", .{}); + _ = self.peer_active.remove(p); + _ = self.stale_peer_candidates.remove(p); + try self.swarm.queueEvent(.{ .peer_disconnected = .{ + .peer = p, + .direction = .unknown, + .reason = .orphaned, + } }); + self.peer_disconnected_emitted_total += 1; + self.reconcile_stale_peers_total += 1; + if (self.req_resp) |rr| try rr.onPeerDisconnected(p); + // Re-arm the known-peer backoff exactly like a non-local close: + // the peer is genuinely gone, and its deadline was parked at + // maxInt when the connection established — without this a + // repaired known peer would never be redialed (the same drift + // in a different coat). Standard capped backoff, no new tier. + if (self.known.getPtr(p)) |st| { + st.dial_inflight = false; + st.failure_count +|= 1; + st.next_dial_deadline_ms = now_ms + reconnectDelayMs(st.failure_count, p); + } + try out_stale_peers.append(self.allocator, p); + } + } + } }; fn peerIdFromMultiaddr(ma: *const multiaddr.Multiaddr) ?identity.PeerId { @@ -901,6 +1135,220 @@ test "connection manager notifies ReqResp on last disconnect" { try std.testing.expectEqual(@as(u32, 0), rr.inbound.count()); } +// --------------------------------------------------------------------------- +// Reconciliation + lifecycle emit-count observability (#299) +// --------------------------------------------------------------------------- + +test "reconciliation emits synthetic close for orphaned conn entry" { + if (@import("builtin").single_threaded) return error.SkipZigTest; + if (@import("builtin").os.tag == .wasi) return error.SkipZigTest; + + const a = std.testing.allocator; + var swarm = try swarm_mod.Swarm.init(a, swarm_mod.default_event_capacity); + defer swarm.deinit(); + + var cm = ConnectionManager.init(a, &swarm); + defer cm.deinit(); + + const peer = try identity.PeerId.random(); + try cm.onConnectionEstablished(7, peer, .inbound, .{}); + { + var ev = try swarm.nextEvent(100); + defer ev.deinit(a); + try std.testing.expectEqual(.peer_connected, std.meta.activeTag(ev)); + } + + // Transport truth: no live legs at all (the close event was lost). + var live = std.AutoHashMap(ConnectionId, void).init(a); + defer live.deinit(); + var orphans: std.ArrayList(OrphanedConn) = .empty; + defer orphans.deinit(a); + var stale: std.ArrayList(identity.PeerId) = .empty; + defer stale.deinit(a); + + // First sweep only records the candidate — no report before the grace. + try cm.reconcile(1_000, &live, &orphans, &stale); + try std.testing.expectEqual(@as(usize, 0), orphans.items.len); + + // Second sweep, still within the grace window: no report. + try cm.reconcile(1_000 + reconcile_orphan_grace_ms - 1, &live, &orphans, &stale); + try std.testing.expectEqual(@as(usize, 0), orphans.items.len); + + // Past the grace: the orphan is reported for a synthesized close. + try cm.reconcile(1_000 + reconcile_orphan_grace_ms, &live, &orphans, &stale); + try std.testing.expectEqual(@as(usize, 1), orphans.items.len); + try std.testing.expectEqual(@as(ConnectionId, 7), orphans.items[0].conn_id); + try std.testing.expect(orphans.items[0].peer.eql(&peer)); + try std.testing.expectEqual(@as(u64, 1), cm.reconcile_orphans_flagged_total); + try std.testing.expectEqual(@as(usize, 0), stale.items.len); + + // The caller (transport coordinator) routes the orphan through the normal + // close path; the peer's last leg -> fully disconnected, event emitted. + try std.testing.expectEqual(true, try cm.onConnectionClosed(20_000, 7, .orphaned)); + var ev = try swarm.nextEvent(100); + defer ev.deinit(a); + try std.testing.expectEqual(.peer_disconnected, std.meta.activeTag(ev)); + try std.testing.expect(ev.peer_disconnected.peer.eql(&peer)); + try std.testing.expectEqual(peer_events.DisconnectReason.orphaned, ev.peer_disconnected.reason); + try std.testing.expect(!cm.hasActiveConnection(peer)); + try std.testing.expectEqual(@as(usize, 0), cm.conns.count()); + + // Next sweep prunes the candidate for the now-closed entry. + try cm.reconcile(2_000 + reconcile_orphan_grace_ms, &live, &orphans, &stale); + try std.testing.expectEqual(@as(usize, 0), cm.orphan_candidates.count()); +} + +test "reconciliation never flags live conns; a transient miss restarts the grace" { + if (@import("builtin").single_threaded) return error.SkipZigTest; + if (@import("builtin").os.tag == .wasi) return error.SkipZigTest; + + const a = std.testing.allocator; + var swarm = try swarm_mod.Swarm.init(a, swarm_mod.default_event_capacity); + defer swarm.deinit(); + + var cm = ConnectionManager.init(a, &swarm); + defer cm.deinit(); + + const peer = try identity.PeerId.random(); + try cm.onConnectionEstablished(1, peer, .outbound, .{}); + { + var ev = try swarm.nextEvent(100); + defer ev.deinit(a); + try std.testing.expectEqual(.peer_connected, std.meta.activeTag(ev)); + } + + var live = std.AutoHashMap(ConnectionId, void).init(a); + defer live.deinit(); + var orphans: std.ArrayList(OrphanedConn) = .empty; + defer orphans.deinit(a); + var stale: std.ArrayList(identity.PeerId) = .empty; + defer stale.deinit(a); + + // Live conn is never flagged, no matter how much time passes. + try live.put(1, {}); + try cm.reconcile(0, &live, &orphans, &stale); + try cm.reconcile(10 * reconcile_orphan_grace_ms, &live, &orphans, &stale); + try std.testing.expectEqual(@as(usize, 0), orphans.items.len); + try std.testing.expectEqual(@as(usize, 0), cm.orphan_candidates.count()); + + // A transient snapshot miss (establish/close race) starts a candidate … + live.clearRetainingCapacity(); + const t0 = 20 * reconcile_orphan_grace_ms; + try cm.reconcile(t0, &live, &orphans, &stale); + try std.testing.expectEqual(@as(usize, 0), orphans.items.len); + try std.testing.expectEqual(@as(usize, 1), cm.orphan_candidates.count()); + + // … but the leg reappearing clears it, so a later miss must run the FULL + // grace again instead of inheriting the stale first-seen timestamp. + try live.put(1, {}); + try cm.reconcile(t0 + 1_000, &live, &orphans, &stale); + try std.testing.expectEqual(@as(usize, 0), cm.orphan_candidates.count()); + + live.clearRetainingCapacity(); + try cm.reconcile(t0 + 2_000, &live, &orphans, &stale); + // Past the ORIGINAL candidate's grace, but the restart means no report yet. + try cm.reconcile(t0 + reconcile_orphan_grace_ms + 1_000, &live, &orphans, &stale); + try std.testing.expectEqual(@as(usize, 0), orphans.items.len); + // Full grace from the restart: now it reports. + try cm.reconcile(t0 + 2_000 + reconcile_orphan_grace_ms, &live, &orphans, &stale); + try std.testing.expectEqual(@as(usize, 1), orphans.items.len); +} + +test "reconciliation repairs stale peer_active entry (onConnectionClosed skew path)" { + if (@import("builtin").single_threaded) return error.SkipZigTest; + if (@import("builtin").os.tag == .wasi) return error.SkipZigTest; + + const a = std.testing.allocator; + var swarm = try swarm_mod.Swarm.init(a, swarm_mod.default_event_capacity); + defer swarm.deinit(); + + var cm = ConnectionManager.init(a, &swarm); + defer cm.deinit(); + + var ma = try multiaddr.Multiaddr.fromString(a, "/ip4/127.0.0.1/udp/4001/quic-v1/p2p/12D3KooWD3eckifWpRn9wQpMG9R9hX3sD158z7EqHWmweQAJU5SA"); + defer ma.deinit(); + try cm.registerKnownPeer(&ma, null); + const peer = peerIdFromMultiaddr(&ma).?; + + try cm.onConnectionEstablished(3, peer, .inbound, .{}); + { + var ev = try swarm.nextEvent(100); + defer ev.deinit(a); + try std.testing.expectEqual(.peer_connected, std.meta.activeTag(ev)); + } + + // Simulate the drift the #299 hazards describe: the conns entry vanished + // without the peer_active bookkeeping (direct map mutation, mirroring the + // known-peer test setup pattern above). The peer now reads as "connected" + // with zero backing connections — permanently, absent reconciliation. + _ = cm.conns.remove(3); + try std.testing.expect(cm.hasActiveConnection(peer)); + + var live = std.AutoHashMap(ConnectionId, void).init(a); + defer live.deinit(); + var orphans: std.ArrayList(OrphanedConn) = .empty; + defer orphans.deinit(a); + var stale: std.ArrayList(identity.PeerId) = .empty; + defer stale.deinit(a); + + try cm.reconcile(1_000, &live, &orphans, &stale); + try std.testing.expectEqual(@as(usize, 0), stale.items.len); + + try cm.reconcile(1_000 + reconcile_orphan_grace_ms, &live, &orphans, &stale); + try std.testing.expectEqual(@as(usize, 1), stale.items.len); + try std.testing.expect(stale.items[0].eql(&peer)); + try std.testing.expect(!cm.hasActiveConnection(peer)); + try std.testing.expectEqual(@as(u64, 1), cm.reconcile_stale_peers_total); + + var ev = try swarm.nextEvent(100); + defer ev.deinit(a); + try std.testing.expectEqual(.peer_disconnected, std.meta.activeTag(ev)); + try std.testing.expect(ev.peer_disconnected.peer.eql(&peer)); + try std.testing.expectEqual(peer_events.DisconnectReason.orphaned, ev.peer_disconnected.reason); + + // Known peer: the repair re-arms the standard capped backoff (no new + // redial tier), so the dial scheduler chases the peer again. + const st = cm.knownPeerStatus(peer).?; + try std.testing.expect(st.failure_count >= 1); + try std.testing.expect(st.next_dial_deadline_ms != std.math.maxInt(i64)); + try std.testing.expect(!st.dial_inflight); +} + +test "lifecycle emit counters track established/closed and unknown-conn closes" { + if (@import("builtin").single_threaded) return error.SkipZigTest; + if (@import("builtin").os.tag == .wasi) return error.SkipZigTest; + + const a = std.testing.allocator; + var swarm = try swarm_mod.Swarm.init(a, swarm_mod.default_event_capacity); + defer swarm.deinit(); + try swarm.startBackground(); + + var cm = ConnectionManager.init(a, &swarm); + defer cm.deinit(); + + const peer = try identity.PeerId.random(); + try cm.onConnectionEstablished(1, peer, .outbound, .{}); + try cm.onConnectionEstablished(2, peer, .inbound, .{}); + + var s = cm.lifecycleStats(); + try std.testing.expectEqual(@as(u64, 2), s.conns_established_total); + try std.testing.expectEqual(@as(u64, 1), s.peer_connected_emitted_total); + try std.testing.expectEqual(@as(usize, 2), s.tracked_conns); + try std.testing.expectEqual(@as(usize, 1), s.connected_peers); + + _ = try cm.onConnectionClosed(1_000, 1, .remote_close); + _ = try cm.onConnectionClosed(1_000, 2, .remote_close); + // Unknown conn_id: the silent early-return #299 wants observable. + _ = try cm.onConnectionClosed(1_000, 99, .remote_close); + + s = cm.lifecycleStats(); + try std.testing.expectEqual(@as(u64, 2), s.conns_closed_total); + try std.testing.expectEqual(@as(u64, 1), s.peer_disconnected_emitted_total); + try std.testing.expectEqual(@as(u64, 1), s.close_unknown_conn_total); + try std.testing.expectEqual(@as(usize, 0), s.tracked_conns); + try std.testing.expectEqual(@as(usize, 0), s.connected_peers); +} + // --------------------------------------------------------------------------- // Connection trimming policy (#90) // --------------------------------------------------------------------------- diff --git a/src/core/host.zig b/src/core/host.zig index 3224708..a1a5172 100644 --- a/src/core/host.zig +++ b/src/core/host.zig @@ -698,13 +698,21 @@ pub const Host = struct { // member on every flap and the heartbeat can never re-graft them (the // candidate pool empties → coverage decays below quorum → finality stalls). if (fully_disconnected) { - self.gossipsub.onPeerDisconnected(peer); - self.peer_protocols.removePeer(peer); - if (self.kad_dht_client) |kad| { - var peer_b58: [128]u8 = undefined; - const peer_str = peer.toBase58(&peer_b58) catch return; - kad.onPeerDisconnected(peer_str); - } + self.teardownPeerState(peer); + } + } + + /// Peer-level teardown shared by the LAST-leg close path above and the + /// #299 reconciliation sweep (which repairs `peer_active` entries stranded + /// with zero backing connections and must then run the same teardown a + /// delivered last-leg close would have run). + pub fn teardownPeerState(self: *Host, peer: identity.PeerId) void { + self.gossipsub.onPeerDisconnected(peer); + self.peer_protocols.removePeer(peer); + if (self.kad_dht_client) |kad| { + var peer_b58: [128]u8 = undefined; + const peer_str = peer.toBase58(&peer_b58) catch return; + kad.onPeerDisconnected(peer_str); } } diff --git a/src/core/peer_events.zig b/src/core/peer_events.zig index ea8ea6e..9736795 100644 --- a/src/core/peer_events.zig +++ b/src/core/peer_events.zig @@ -15,6 +15,12 @@ pub const DisconnectReason = enum { remote_close, local_close, err, + /// Synthesized by the reconciliation sweep (#299): the connection manager + /// still tracked the conn but the transport no longer had a live leg for + /// it — the real close event was lost (e.g. dropped on coordinator-queue + /// allocation failure) or never detected. Distinct from `remote_close` so + /// embedders and logs can see that the peer book drifted and was repaired. + orphaned, }; /// Dial or transport handshake failure (distinct from [`DisconnectReason`] on an established conn). diff --git a/src/transport/quic/runtime.zig b/src/transport/quic/runtime.zig index 5f6ba5d..d2e03a4 100644 --- a/src/transport/quic/runtime.zig +++ b/src/transport/quic/runtime.zig @@ -60,6 +60,17 @@ pub const AutonatRuntimeOptions = config.AutonatRuntimeOptions; /// within this window is abandoned and surfaced via `onDialFailure`. const dial_handshake_timeout_ms: i64 = 20_000; +/// Cadence of the shard-0 book-vs-transport reconciliation sweep (#299). +/// Combined with [`connection_manager.reconcile_orphan_grace_ms`] an orphaned +/// entry is repaired within ~grace + one sweep (≈20 s) of the lost event — +/// prompt against the multi-hour wedge #299 describes, and slow enough that +/// the transport's own per-lap close detection always wins when healthy. +const reconcile_sweep_interval_ms: i64 = 5_000; + +/// Cadence of the lifecycle emit-counts-vs-table-sizes log line emitted from +/// the reconciliation sweep (#299 observability). +const lifecycle_stats_log_interval_ms: i64 = 60_000; + /// An outbound QUIC dial whose handshake is still in flight. The dial is /// advanced **non-blocking** every `driveLoop` tick (alongside the listener, /// `pollAccept`, and all established outbounds) until it reaches @@ -441,6 +452,35 @@ pub const QuicRuntime = struct { conn_lifecycle_queue: std.ArrayList(ConnLifecycleEvent) = .empty, conn_lifecycle_lock: conn_table.SpinLock = .{}, + // ── Lifecycle observability + reconciliation (#299) ──────────────────── + // Emit-count instrumentation: these counters, compared against the + // connection_manager's [`lifecycleStats`] and the live conn-table sizes in + // the shard-0 reconciliation sweep's periodic log line, make the "peer + // book drifted from live conns with zero lifecycle events" failure mode + // directly observable in logs instead of requiring a wedged devnet node. + /// Lifecycle events DROPPED because the coordinator-queue append failed + /// (allocation failure). Any non-zero value here is the smoking gun for + /// book-vs-transport drift; the reconciliation sweep is the backstop that + /// repairs the book afterwards. Atomic: incremented from any drive thread. + conn_lifecycle_dropped_total: std.atomic.Value(u64) = .init(0), + /// `established` lifecycle events enqueued to the coordinator. + conn_established_enqueued_total: std.atomic.Value(u64) = .init(0), + /// `closed` lifecycle events enqueued to the coordinator. + conn_closed_enqueued_total: std.atomic.Value(u64) = .init(0), + /// Outbound closes surfaced by [`detectOutboundConnectionClose`]. + outbound_closes_detected_total: std.atomic.Value(u64) = .init(0), + /// Inbound closes surfaced at DRAINING time by + /// [`detectInboundConnectionClose`] — the #299 outbound-parity fix. The + /// pre-fix inbound path only fired after zquic reaped the conn slot. + inbound_draining_closes_total: std.atomic.Value(u64) = .init(0), + /// Inbound closes surfaced by the listener reap callback + /// ([`onLifecycleClosed`] via `QuicListener.syncSeenFlags`). + inbound_reap_closes_total: std.atomic.Value(u64) = .init(0), + /// Wall-ms of the last reconciliation sweep. Shard-0 drive thread only. + last_reconcile_sweep_ms: i64 = 0, + /// Wall-ms of the last lifecycle-stats log line. Shard-0 drive thread only. + last_lifecycle_stats_log_ms: i64 = 0, + /// Identify-record coordinator queue (Phase 4). `advanceInboundStreams` runs /// on EVERY shard's drive thread, but `recordInboundIdentifyProtocols` -> /// `host.recordPeerProtocols`/`recordObservedAddr` mutate global, non- @@ -1101,8 +1141,41 @@ pub const QuicRuntime = struct { self.conn_lifecycle_lock.lock(); defer self.conn_lifecycle_lock.unlock(); self.conn_lifecycle_queue.append(self.allocator, ev) catch |err| { - log.err("quic_runtime: conn lifecycle queue append failed: {s}", .{@errorName(err)}); + // An event lost here is exactly the drift #299 describes: the + // transport moved on but the connection_manager's book never heard. + // Count it (observability) — the shard-0 reconciliation sweep is + // the backstop that repairs the book afterwards. + _ = self.conn_lifecycle_dropped_total.fetchAdd(1, .monotonic); + log.err("quic_runtime: conn lifecycle queue append failed (event DROPPED; reconciliation sweep will repair): {s}", .{@errorName(err)}); + return; }; + switch (ev) { + .established => _ = self.conn_established_enqueued_total.fetchAdd(1, .monotonic), + .closed => _ = self.conn_closed_enqueued_total.fetchAdd(1, .monotonic), + .dial_failure => {}, + } + } + + /// #299 observability: lifecycle events dropped on coordinator-queue + /// allocation failure (each one is a potential book-vs-transport drift). + pub fn lifecycleEventsDroppedCount(self: *const QuicRuntime) u64 { + return self.conn_lifecycle_dropped_total.load(.monotonic); + } + + /// #299 observability: inbound closes surfaced at DRAINING time (outbound + /// parity) rather than waiting for zquic to reap the conn slot. + pub fn inboundDrainingClosesCount(self: *const QuicRuntime) u64 { + return self.inbound_draining_closes_total.load(.monotonic); + } + + /// #299 observability: inbound closes surfaced by the listener reap callback. + pub fn inboundReapClosesCount(self: *const QuicRuntime) u64 { + return self.inbound_reap_closes_total.load(.monotonic); + } + + /// #299 observability: outbound closes surfaced by [`detectOutboundConnectionClose`]. + pub fn outboundClosesDetectedCount(self: *const QuicRuntime) u64 { + return self.outbound_closes_detected_total.load(.monotonic); } fn notifyConnEstablished( @@ -1179,6 +1252,127 @@ pub const QuicRuntime = struct { }; } + /// #299 reconciliation: compare the connection_manager's book against the + /// live transport tables and synthesize closes for entries no transport + /// leg backs (a lifecycle event lost on the coordinator queue, or a close + /// transition the per-lap detection missed). Runs ONLY on the shard-0 + /// coordinator thread — the single thread allowed to touch + /// `connection_manager` — every [`reconcile_sweep_interval_ms`]. + /// + /// Live-snapshot rules: + /// * `outbound_by_peer` / `inbound_by_peer` are read under their + /// SpinLocks (same cross-thread contract as `shardHoldsLiveLegLocked`). + /// * A mapped leg whose zquic conn is already draining/`.closed` is NOT + /// counted live: the owning shard's detect* sweeps normally surface it + /// first, but if their edge-transition detection was missed the mapped + /// entry would otherwise pin the book forever. Reading `phase` / + /// `draining` cross-thread is the same benign monotonic-flag read the + /// settled-state test probes rely on: a stale read at worst starts an + /// orphan candidate that the next sweep clears, and the two-sweep grace + /// in `reconcile` absorbs exactly that. + /// * Relay bridge/virtual conn ids are only ever written via the + /// relay/dcutr hooks, which are bound to `&shards[0]` and advanced on + /// shard 0 — this thread — so they are read without locks. + /// + /// Anti-churn: this sweep NEVER dials. Synthesized closes route through + /// `host.onConnectionClosed` with reason `.orphaned`, so redial policy is + /// exactly the pre-existing known-peer capped backoff; inbound-only peers + /// are not redialed (the learned-address redial tier is the separate #299 + /// follow-up, gated on the data this instrumentation produces). + fn sweepConnReconciliation(self: *QuicRuntime, now_ms: i64) void { + var live = std.AutoHashMap(connection_manager_mod.ConnectionId, void).init(self.allocator); + defer live.deinit(); + + var i: u8 = 0; + while (i < self.shard_count) : (i += 1) { + const s = &self.shards[i]; + + s.outbound_by_peer_lock.lock(); + var ot = s.outbound_by_peer.valueIterator(); + while (ot.next()) |v| { + const conn = &v.*.outbound.client.conn; + if (conn.draining or conn.phase == .closed) continue; + live.put(v.*.conn_id, {}) catch {}; + } + s.outbound_by_peer_lock.unlock(); + + s.inbound_by_peer_lock.lock(); + var it_in = s.inbound_by_peer.valueIterator(); + while (it_in.next()) |ref| { + if (ref.conn.draining or ref.conn.phase == .closed) continue; + const cid = s.inbound_conn_ids[ref.slot]; + if (cid == 0) continue; + live.put(cid, {}) catch {}; + } + s.inbound_by_peer_lock.unlock(); + + var rl_it = s.relayed_conn_by_peer.valueIterator(); + while (rl_it.next()) |cid| live.put(cid.*, {}) catch {}; + } + var rv_it = self.relay_live.relay_virtual.valueIterator(); + while (rv_it.next()) |vc| live.put(vc.*.conn_id, {}) catch {}; + + var orphans: std.ArrayList(connection_manager_mod.OrphanedConn) = .empty; + defer orphans.deinit(self.allocator); + var stale_peers: std.ArrayList(identity.PeerId) = .empty; + defer stale_peers.deinit(self.allocator); + + const cm = self.host.connection_manager; + cm.reconcile(now_ms, &live, &orphans, &stale_peers) catch |err| { + log.warn("quic_runtime: reconciliation sweep failed: {s}", .{@errorName(err)}); + return; + }; + + for (orphans.items) |o| { + var peer_buf: [128]u8 = undefined; + log.warn( + "quic_runtime: reconciliation closing orphaned conn entry cid={d} peer={s} — book had it open, transport has no live leg (lifecycle event lost?)", + .{ o.conn_id, peerBase58(o.peer, &peer_buf) }, + ); + self.host.onConnectionClosed(now_ms, o.conn_id, o.peer, .orphaned) catch |err| { + log.warn("quic_runtime: reconciliation close failed cid={d}: {s}", .{ o.conn_id, @errorName(err) }); + }; + } + for (stale_peers.items) |p| { + var peer_buf: [128]u8 = undefined; + log.warn( + "quic_runtime: reconciliation repaired stale peer_active entry peer={s} (marked connected with zero conn entries); running peer teardown", + .{peerBase58(p, &peer_buf)}, + ); + self.host.teardownPeerState(p); + } + + // Emit-count observability (#299): periodic counts-vs-tables line so + // the divergence the issue describes is visible in live logs, not just + // post-mortems. The per-orphan WARNs above flag active drift instantly. + if (now_ms - self.last_lifecycle_stats_log_ms >= lifecycle_stats_log_interval_ms) { + self.last_lifecycle_stats_log_ms = now_ms; + const st = cm.lifecycleStats(); + log.info( + "quic_runtime: lifecycle stats: transport_live={d} book_conns={d} book_peers={d} est_total={d} closed_total={d} connected_emitted={d} disconnected_emitted={d} unknown_conn_closes={d} skew_closes={d} orphans_flagged={d} stale_peers_repaired={d} enq_est={d} enq_closed={d} enq_dropped={d} out_closes={d} in_draining_closes={d} in_reap_closes={d}", + .{ + live.count(), + st.tracked_conns, + st.connected_peers, + st.conns_established_total, + st.conns_closed_total, + st.peer_connected_emitted_total, + st.peer_disconnected_emitted_total, + st.close_unknown_conn_total, + st.close_peer_active_skew_total, + st.reconcile_orphans_flagged_total, + st.reconcile_stale_peers_total, + self.conn_established_enqueued_total.load(.monotonic), + self.conn_closed_enqueued_total.load(.monotonic), + self.conn_lifecycle_dropped_total.load(.monotonic), + self.outbound_closes_detected_total.load(.monotonic), + self.inbound_draining_closes_total.load(.monotonic), + self.inbound_reap_closes_total.load(.monotonic), + }, + ); + } + } + // ── Identify-record coordinator (Phase 4) ────────────────────────────── // // `advanceInboundStreams` runs on EVERY shard, but the identify *record* @@ -1242,38 +1436,90 @@ pub const QuicRuntime = struct { const sh: *Shard = @ptrCast(@alignCast(ctx.?)); const self = sh.rt; if (sh.inbound_conn_notified[slot]) { - const peer = sh.inbound_conn_peer[slot] orelse identity.PeerId.random() catch return; - const cid = sh.inbound_conn_ids[slot]; - const now_ms = self.opts.now_ms_fn(); - self.notifyConnClosed(now_ms, cid, peer, .remote_close); - sh.inbound_by_peer_lock.lock(); - _ = sh.inbound_by_peer.remove(peer); - sh.inbound_by_peer_lock.unlock(); - self.clearOwner(peer, sh.index); - // Gate the persistent /meshsub teardown on LAST-leg (transport analog - // of the v0.2.45 connection_manager fix). Under sharding a peer holds - // up to 2 legs and the stream is bound to ONE (outbound-preferred). A - // flap of the OTHER leg must NOT destroy a stream living on the - // surviving leg — that silently halts the peer's attestation delivery - // (+0/-N coverage decay, never restored → finality stalls 1-2 short of - // quorum). Destroy only if the stream was on THIS (closing inbound) leg - // OR no other leg survives. If it was on this leg but the peer survives - // elsewhere, the next publish lazily reopens on the surviving leg; - // replay SUBSCRIBE so the peer re-learns our interest. (inbound_by_peer - // was already removed above, so liveLegShardForPeer excludes this leg.) - if (sh.persistent_gossip.get(peer)) |g| { - const live_leg = self.liveLegShardForPeer(peer) != null; - if (g.raw == .inbound or !live_leg) { - self.destroyPersistentGossipStream(sh, peer); - if (live_leg) self.replaySubscribeToPeer(sh, peer); - } - } + // Reached only when [`detectInboundConnectionClose`] did NOT + // already surface this close at draining time (it clears + // `inbound_conn_notified` when it does) — e.g. zquic reaped the + // slot within a single drive lap. + _ = self.inbound_reap_closes_total.fetchAdd(1, .monotonic); + self.handleInboundConnClosed(sh, slot); } sh.inbound_conn_notified[slot] = false; sh.inbound_conn_peer[slot] = null; sh.inbound_conn_ids[slot] = 0; } + /// Shared inbound-close teardown for listener slot `slot`: notify the host, + /// drop the `inbound_by_peer` leg, clear ownership, and gate the persistent + /// /meshsub teardown on LAST-leg. Invoked from the listener reap callback + /// ([`onLifecycleClosed`]) and from the draining-parity sweep + /// ([`detectInboundConnectionClose`], #299). Callers reset the per-slot + /// bookkeeping (`inbound_conn_notified`/`_peer`/`_ids`) afterwards. + fn handleInboundConnClosed(self: *QuicRuntime, sh: *Shard, slot: usize) void { + const peer = sh.inbound_conn_peer[slot] orelse identity.PeerId.random() catch return; + const cid = sh.inbound_conn_ids[slot]; + const now_ms = self.opts.now_ms_fn(); + self.notifyConnClosed(now_ms, cid, peer, .remote_close); + sh.inbound_by_peer_lock.lock(); + _ = sh.inbound_by_peer.remove(peer); + sh.inbound_by_peer_lock.unlock(); + self.clearOwner(peer, sh.index); + // Gate the persistent /meshsub teardown on LAST-leg (transport analog + // of the v0.2.45 connection_manager fix). Under sharding a peer holds + // up to 2 legs and the stream is bound to ONE (outbound-preferred). A + // flap of the OTHER leg must NOT destroy a stream living on the + // surviving leg — that silently halts the peer's attestation delivery + // (+0/-N coverage decay, never restored → finality stalls 1-2 short of + // quorum). Destroy only if the stream was on THIS (closing inbound) leg + // OR no other leg survives. If it was on this leg but the peer survives + // elsewhere, the next publish lazily reopens on the surviving leg; + // replay SUBSCRIBE so the peer re-learns our interest. (inbound_by_peer + // was already removed above, so liveLegShardForPeer excludes this leg.) + if (sh.persistent_gossip.get(peer)) |g| { + const live_leg = self.liveLegShardForPeer(peer) != null; + if (g.raw == .inbound or !live_leg) { + self.destroyPersistentGossipStream(sh, peer); + if (live_leg) self.replaySubscribeToPeer(sh, peer); + } + } + } + + /// Inbound analog of [`detectOutboundConnectionClose`] — the #299 + /// draining-parity fix. + /// + /// Outbound close detection counts `draining` as dead (see the + /// `cur_closed` rule below), but the inbound path previously fired ONLY + /// from the listener reap callback (`syncSeenFlags`), which requires zquic + /// to have freed the conn slot (`server.conns[i] == null`). zquic defers + /// that reap while the embedder still holds raw-app stream slots on the + /// conn, and otherwise until the 3×PTO draining deadline — so an inbound + /// leg's `peer_disconnected` trailed the wire-level close by hundreds of + /// milliseconds in the good case and indefinitely when the release/reap + /// chain was starved (the silent-conn-loss shape #299 describes). Sweep + /// the notified listener slots each drive lap and surface the close as + /// soon as the conn is draining or `.closed`, mirroring the outbound rule. + /// + /// The eventual reap callback for the slot is a harmless no-op afterwards + /// (`inbound_conn_notified` is already false). Conns stuck pre-`.connected` + /// need no handling here: they were never notified (no book entry) and + /// zquic reaps them on its own handshake deadline. + fn detectInboundConnectionClose(self: *QuicRuntime, sh: *Shard) void { + for (0..ZIo.MAX_CONNECTIONS) |slot| { + if (!sh.inbound_conn_notified[slot]) continue; + // Slot already reaped: the listener callback path owns that case. + const conn = sh.listener.server.conns[slot] orelse continue; + if (!(conn.draining or conn.phase == .closed)) continue; + log.warn( + "quic_runtime: inbound QUIC connection closed by remote (cid={d} slot={d} phase={s}); notifying host at draining, not waiting for reap", + .{ sh.inbound_conn_ids[slot], slot, @tagName(conn.phase) }, + ); + _ = self.inbound_draining_closes_total.fetchAdd(1, .monotonic); + self.handleInboundConnClosed(sh, slot); + sh.inbound_conn_notified[slot] = false; + sh.inbound_conn_peer[slot] = null; + sh.inbound_conn_ids[slot] = 0; + } + } + /// Register an established inbound connection with the host exactly once per /// listener slot: record the peer, fire `onConnectionEstablished(.inbound)`, /// and replay our SUBSCRIBEs so an inbound-only peer learns our topics and @@ -1497,6 +1743,7 @@ pub const QuicRuntime = struct { "quic_runtime: outbound QUIC connection closed by remote (cid={d}); notifying host", .{cid}, ); + _ = self.outbound_closes_detected_total.fetchAdd(1, .monotonic); self.notifyConnClosed(now_ms, cid, peer, .remote_close); self.clearOwner(peer, sh.index); // Remove the outbound map entry FIRST so liveLegShardForPeer reflects @@ -2030,6 +2277,12 @@ pub const QuicRuntime = struct { // that triggered the phase transition this tick. self.detectOutboundConnectionClose(sh); + // #299 parity: surface INBOUND closes at draining time too, instead + // of waiting for zquic to reap the listener slot (which can trail + // the wire-level close by hundreds of ms — or indefinitely when the + // raw-app-slot release chain is starved). + self.detectInboundConnectionClose(sh); + // Drain hook queue. The queue + the gossipsub outbox + host periodic // ticks + relay/dcutr are GLOBAL single-writer state; only shard 0 // touches them (single shard today → always taken). Per-shard @@ -2045,6 +2298,17 @@ pub const QuicRuntime = struct { // dial scheduler this iteration. self.drainConnLifecycle(); self.drainIdentifyRecords(); + + // #299: periodic book-vs-transport reconciliation. AFTER the + // lifecycle drain so the connection_manager reflects every + // delivered event before it is compared against the live + // transport tables, and on this (shard-0 coordinator) thread + // because connection_manager is single-threaded. + const rec_now = self.opts.now_ms_fn(); + if (rec_now - self.last_reconcile_sweep_ms >= reconcile_sweep_interval_ms) { + self.last_reconcile_sweep_ms = rec_now; + self.sweepConnReconciliation(rec_now); + } } // Drain THIS shard's hook sub-queue (Phase 4): directed work @@ -6279,6 +6543,134 @@ test "QuicRuntime: two instances exchange a status req/resp over UDP loopback" { try testing.expect(saw_end); } +test "QuicRuntime: inbound close surfaces at draining (outbound parity) — no peer-book drift (#299)" { + if (builtin.single_threaded) return error.SkipZigTest; + if (builtin.os.tag == .wasi) return error.SkipZigTest; + + const a = testing.allocator; + + var bundle_a = try buildTestBundle(a, "a", 0xA7); + defer bundle_a.deinit(a); + var bundle_b = try buildTestBundle(a, "b", 0xB8); + defer bundle_b.deinit(a); + + var host_a = try host_mod.Host.create(.{ + .allocator = a, + .local_peer = bundle_a.peer, + .gossipsub = .{ .local_peer_id = bundle_a.peer }, + }); + defer host_a.destroy(); + try host_a.startBackground(); + try testing.expect(host_a.waitUntilReady(5_000)); + + var rt_a = try QuicRuntime.create(.{ + .allocator = a, + .host = host_a, + .tls_pem = .{ + .pem_bytes = .{ + .cert_pem = bundle_a.cert_pem, + .key_pem = bundle_a.key_pem, + }, + }, + .listen_multiaddr = "/ip4/127.0.0.1/udp/0/quic-v1", + }); + defer rt_a.destroy(); + + var host_b = try host_mod.Host.create(.{ + .allocator = a, + .local_peer = bundle_b.peer, + .gossipsub = .{ .local_peer_id = bundle_b.peer }, + }); + defer host_b.destroy(); + try host_b.startBackground(); + try testing.expect(host_b.waitUntilReady(5_000)); + + // Destroyed explicitly mid-test (that IS the test); the guard keeps the + // teardown path leak-free when an assertion fails before that point. + var rt_b_destroyed = false; + var rt_b = try QuicRuntime.create(.{ + .allocator = a, + .host = host_b, + .tls_pem = .{ + .pem_bytes = .{ + .cert_pem = bundle_b.cert_pem, + .key_pem = bundle_b.key_pem, + }, + }, + .listen_multiaddr = "/ip4/127.0.0.1/udp/0/quic-v1", + }); + defer if (!rt_b_destroyed) rt_b.destroy(); + + try rt_a.start(); + try rt_b.start(); + + const a_port = rt_a.boundUdpPortIpv4() orelse return error.NoBoundPort; + + // B dials A, so A's ONLY leg to B is INBOUND (A never dials back). + var a_peer_b58_buf: [128]u8 = undefined; + const a_peer_b58 = try bundle_a.peer.toBase58(&a_peer_b58_buf); + const a_ma_str = try std.fmt.allocPrint(a, "/ip4/127.0.0.1/udp/{d}/quic-v1/p2p/{s}", .{ a_port, a_peer_b58 }); + defer a.free(a_ma_str); + var a_ma = try multiaddr.Multiaddr.fromString(a, a_ma_str); + defer a_ma.deinit(); + try rt_b.registerKnownPeer(&a_ma, bundle_a.peer); + + // Wait until A registered B's inbound leg and emitted peer_connected. + { + var saw_connected = false; + const deadline_ms = wall_time.milliTimestamp() + 20_000; + while (wall_time.milliTimestamp() < deadline_ms and !saw_connected) { + var ev = host_a.nextEvent(200) catch |err| switch (err) { + error.Timeout => continue, + else => return err, + }; + defer ev.deinit(a); + switch (ev) { + .peer_connected => |pc| { + try testing.expect(pc.peer.eql(&bundle_b.peer)); + try testing.expectEqual(peer_events.Direction.inbound, pc.direction); + saw_connected = true; + }, + else => {}, + } + } + try testing.expect(saw_connected); + } + try testing.expect(rtHasInboundTo(rt_a, bundle_b.peer)); + + // Tear B down. Its shard teardown sends CONNECTION_CLOSE on the outbound + // leg, which flips A's server-side conn to `draining`. zquic defers the + // slot reap (3×PTO draining deadline, longer while the embedder still + // holds raw-app stream slots) — pre-fix, A's `peer_disconnected` waited on + // that reap (`syncSeenFlags` requires `server.conns[i] == null`). + rt_b.destroy(); + rt_b_destroyed = true; + + // A must surface the close promptly and consistently. + var saw_disconnected = false; + const deadline_ms = wall_time.milliTimestamp() + 15_000; + while (wall_time.milliTimestamp() < deadline_ms and !saw_disconnected) { + var ev = host_a.nextEvent(200) catch |err| switch (err) { + error.Timeout => continue, + else => return err, + }; + defer ev.deinit(a); + switch (ev) { + .peer_disconnected => |pd| { + try testing.expect(pd.peer.eql(&bundle_b.peer)); + saw_disconnected = true; + }, + else => {}, + } + } + try testing.expect(saw_disconnected); + // The parity fix must be WHAT detected it: the close fired at draining + // time (counter below), not by waiting for the slot reap. The transport + // map and the peer book agree again afterwards (no drift). + try testing.expect(rt_a.inboundDrainingClosesCount() >= 1); + try testing.expect(!rtHasInboundTo(rt_a, bundle_b.peer)); +} + test "QuicRuntime: empty-body /status (lantern shape) still gets a response, not reaped" { if (builtin.single_threaded) return error.SkipZigTest; if (builtin.os.tag == .wasi) return error.SkipZigTest;