From 9fb405fe1b0aa9c813a619605b34c8689a94e697 Mon Sep 17 00:00:00 2001 From: Michael Taylor Date: Mon, 7 Sep 2026 23:18:27 -0700 Subject: [PATCH 1/4] test(peer): prove a dialled and accepted peer is counted once Epic dig_ecosystem#3124 fixed an UNDER-count: `connected_peers` omitted every inbound peer. Two legs shipped -- direct inbound (dig-node#523 / PR #402) and relayed inbound (dig-node#580 / PR #579) -- each with a behavioural test asserting COUNTED + still SERVED + RELEASED over a real mTLS connection. Neither leg, and no other test in this repo, covers the cross-class case: ONE identity holding both a dialled outbound slot and an accepted inbound slot. `connected_peers_json` (crates/dig-node-core/src/peer.rs:370) maps one JSON row per `connected_pool_peers()` entry with no grouping by `peer_id`, so if the pool ever held that identity twice, `peer_count()` would read 2 for one peer and the RPC would emit two rows with the same `peer_id` and opposite `direction` -- the same defect class the epic exists for, inverted, and just as invisible to a consumer. `adopt_inbound_peer_in_pool`'s own doc (peer.rs:3658-3662) lists "a peer already holding a dialable slot" among the refusals dig-gossip can return, so the de-duplication is believed to live in dig-gossip. It has never been exercised from dig-node. This test exercises it. The test drives real connections throughout: the outbound slot is created through `adopt_nat_connection` (the single outbound adoption path, called in production from seams/dig_peer/bootstrap.rs:218 and seams/dig_peer/pex.rs:652), and the inbound slot by a real mTLS `dig_nat::connect` against `serve_peer_rpc_listener_with`. Nothing constructs a pool entry by hand. NOT YET COMPILED. This commit is a salvage checkpoint written after the authoring lane was killed mid-flight by a session rate limit, pushed so the work is durable rather than lost. The API surface it assumes (`dig_gossip::NatPeerConnection::new`, `dig_nat::PeerConnection`'s field list, `PeerSession::client`/`server`, `GossipHandle::disconnect`) is unverified against the pinned dig-gossip v0.32.0, and the sleep-then-assert ordering in steps 2-3 still needs to be replaced by a positive signal that the inbound accept path actually ran. A follow-up commit on this branch compiles it, fixes both, and records the result. Refs #3124 --- .../tests/inbound_pool_membership.rs | 186 ++++++++++++++++++ 1 file changed, 186 insertions(+) diff --git a/crates/dig-node-core/tests/inbound_pool_membership.rs b/crates/dig-node-core/tests/inbound_pool_membership.rs index 996e0e0d..646b17d9 100644 --- a/crates/dig-node-core/tests/inbound_pool_membership.rs +++ b/crates/dig-node-core/tests/inbound_pool_membership.rs @@ -493,3 +493,189 @@ async fn the_accepted_direct_cap_still_binds_after_a_supersede_and_stale_release server.abort(); service.stop().await.expect("stop"); } + +/// Build a `NatPeerConnection` over a loopback duplex with a chosen `peer_id`, remote address and +/// traversal tier -- the same pattern `peer.rs`'s own unit tests use to exercise `adopt_nat_connection` +/// (the single outbound-adoption entry point, called in production from `bootstrap.rs`/`pex.rs`) +/// without a real socket. The `peer_id` is passed in explicitly rather than derived from a TLS +/// handshake here, so the caller can make it byte-identical to a real mTLS identity used elsewhere -- +/// which is exactly how the test below gets the SAME identity into both an outbound and an inbound +/// slot. Returns the server `PeerSession` half; drop it to end the session, hold it to keep the +/// outbound slot's session alive. +fn loopback_nat_conn( + peer_id_bytes: [u8; 32], + remote: std::net::SocketAddr, + method: dig_nat::TraversalKind, +) -> (dig_gossip::NatPeerConnection, dig_nat::PeerSession) { + let (client_io, server_io) = tokio::io::duplex(64 * 1024); + let inner = dig_nat::PeerConnection { + peer_id: dig_nat::PeerId::from_bytes(peer_id_bytes), + method, + remote_addr: remote, + peer_bls_pub: None, + session: dig_nat::PeerSession::client(client_io), + }; + ( + dig_gossip::NatPeerConnection::new(inner), + dig_nat::PeerSession::server(server_io), + ) +} + +/// **dig_ecosystem#3124 -- the one unmeasured property: a peer that is BOTH dialled (outbound) and +/// accepted (inbound) is counted exactly ONCE, and both directions keep being served.** +/// +/// `adopt_inbound_peer_in_pool`'s own doc (`peer.rs:3658-3662`) lists "a peer already holding a +/// dialable slot" among dig-gossip's refusals -- meaning the de-duplication this test proves lives in +/// dig-gossip, not dig-node, and has never before been exercised FROM dig-node. Nothing here builds a +/// pool entry directly: the outbound slot is created through the real `adopt_nat_connection` adoption +/// path (the single outbound entry point), and the inbound slot is created by a real mTLS dial against +/// `serve_peer_rpc_listener_with`'s listener, exactly like every other test in this file. +#[tokio::test] +async fn a_peer_that_is_both_dialled_and_accepted_is_counted_once() { + dig_node_core::peer::install_crypto_provider(); + + let (service, gossip_a, _gdir) = running_gossip().await; + assert_eq!(gossip_a.peer_count().await, 0, "the pool starts empty"); + + let server_identity = test_identity("3124-dualslot-server"); + let server_peer_id = server_identity.peer_id(); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind"); + let listen_addr = listener.local_addr().expect("local addr"); + + let responder: Arc = Arc::new(TestResponder); + let server = tokio::spawn(serve_peer_rpc_listener_with( + listener, + server_identity, + responder, + None, + Some(gossip_a.clone()), + )); + + // Identity B -- used for BOTH the outbound and the inbound slot below. + let b_identity = test_identity("3124-dualslot-peer-b"); + let b_peer_id = b_identity.peer_id(); + let b_bytes = *b_peer_id.as_bytes(); + + // -- Step 1: B occupies an OUTBOUND (dialled) slot, via the real adoption path ----------------- + let fake_dial_addr: std::net::SocketAddr = "198.51.100.9:9444".parse().expect("addr"); + let (outbound_conn, _outbound_server_session) = + loopback_nat_conn(b_bytes, fake_dial_addr, dig_nat::TraversalKind::Direct); + let adopted = gossip_a + .adopt_nat_connection(outbound_conn) + .await + .expect("B's outbound slot is uncontested"); + assert_eq!(adopted, dig_gossip::PeerId::from(b_bytes)); + + // Precondition: exactly one DIALABLE slot for B, before the inbound leg touches anything. + assert_eq!( + gossip_a.peer_count().await, + 1, + "the outbound adoption must land before the inbound leg is driven" + ); + let pool_id_b = dig_gossip::PeerId::from(b_bytes); + let outbound_detail = gossip_a + .connected_pool_peers_detailed() + .into_iter() + .find(|p| p.peer_id == pool_id_b) + .expect("B's outbound slot exists"); + assert!( + outbound_detail.is_outbound, + "step 1 must produce a DIALLED slot, or this test measures the wrong thing" + ); + assert!( + !gossip_a.dialable_pool_peers().is_empty(), + "the outbound slot must be dialable before the inbound leg is driven" + ); + + // -- Step 2: the SAME identity B now dials A INBOUND over real mTLS ----------------------------- + let target = dig_nat::PeerTarget::with_addr(server_peer_id, listen_addr, "DIG_MAINNET"); + let config = dig_nat::NatConfig::builder() + .enabled_methods(vec![dig_nat::TraversalKind::Direct]) + .per_method_timeout(Duration::from_secs(5)) + .build(); + let mut inbound_conn = dig_nat::connect(&target, &b_identity, &config) + .await + .expect("B's transport-level connect succeeds even if the pool refuses to adopt it"); + assert_eq!( + *b_identity.peer_id().as_bytes(), + b_bytes, + "the inbound dial must authenticate as the SAME identity as the outbound slot" + ); + + // Give the server's spawned adoption a moment to run, then settle: the count must NOT go to 2. + tokio::time::sleep(Duration::from_millis(300)).await; + assert_eq!( + gossip_a.peer_count().await, + 1, + "a peer that is both dialled and accepted must be counted ONCE, not twice" + ); + + // -- Step 3: exactly ONE row for B's peer_id among connected_pool_peers() ----------------------- + // (`connected_peers_json` is `pub(crate)` inside dig-node-core and unreachable from this + // integration-test crate; `connected_pool_peers()` is its public dig-gossip source, so counting + // matching rows here proves the same property `connected_peers_json` would report.) + let matching_rows = gossip_a + .connected_pool_peers() + .into_iter() + .filter(|(peer_id, _addr, _outbound)| *peer_id == pool_id_b) + .count(); + assert_eq!( + matching_rows, 1, + "exactly one row must carry B's peer_id -- a de-duplication failure would emit two" + ); + // The surviving row is the DIALLED slot, per `adopt_inbound_peer_in_pool`'s own refusal doc + // (peer.rs:3658-3662): "a peer already holding a dialable slot" is refused, so the pre-existing + // outbound slot is kept and the inbound accept is the one that is turned away. + let surviving = gossip_a + .connected_pool_peers_detailed() + .into_iter() + .find(|p| p.peer_id == pool_id_b) + .expect("B still has exactly one slot"); + assert!( + surviving.is_outbound, + "the surviving slot must be the pre-existing DIALLED one, per the documented refusal" + ); + + // -- Step 4: still served, BOTH ways -- the discriminator against a shape that buys the count by + // dropping a connection. The dialled slot was never a live RPC session in this fixture (it is a + // bare loopback duplex with no responder on the other end), so what matters here is that the + // INBOUND session -- the one the pool refused to adopt -- is still answered. + { + let mut stream = inbound_conn.session.open_stream().await.expect("open stream"); + let req = json!({"jsonrpc":"2.0","id":21,"method":"dig.getNetworkInfo"}); + write_framed(&mut stream, &req).await.expect("write"); + let resp = read_one_frame(&mut stream).await; + assert_eq!( + resp["result"]["served_method"], "dig.getNetworkInfo", + "the un-adopted inbound peer must still be served -- refusing adoption must not refuse service" + ); + } + assert_eq!( + gossip_a.peer_count().await, + 1, + "serving the refused inbound peer must not perturb the count" + ); + + // -- Step 5: released cleanly, SESSION-scoped ------------------------------------------------------ + // Drop the inbound session first: the outbound slot must survive, because releasing it was never + // the inbound session's to release (it never held the slot). + drop(inbound_conn); + tokio::time::sleep(Duration::from_millis(300)).await; + assert_eq!( + gossip_a.peer_count().await, + 1, + "the inbound session ending must not release the outbound slot it never owned" + ); + + // Now release the outbound slot itself and confirm the count reaches zero. + gossip_a + .disconnect(&pool_id_b) + .await + .expect("release the outbound slot"); + await_peer_count(&gossip_a, 0, "after the outbound slot is released").await; + + server.abort(); + service.stop().await.expect("stop"); +} From 08d696c67fe05f2cda176b42f84f086cdb8903a7 Mon Sep 17 00:00:00 2001 From: Michael Taylor Date: Mon, 7 Sep 2026 23:55:21 -0700 Subject: [PATCH 2/4] test(peer): make the count assertion prove the accept path ran The first cut of this test slept 300ms and then asserted `peer_count() == 1`. That assertion passes just as happily when the inbound accept path has NOT RUN YET as when the inbound adoption was correctly refused -- it measured a race, not the property. On a loaded box a real double-count defect would have read green. Replace the sleep with a happens-before proof. `adopt_inbound_peer_in_pool` is called at peer.rs:3602, strictly before the accepted session starts answering RPC at peer.rs:3614, so a successful `dig.getNetworkInfo` round-trip over the inbound connection proves the server has already reached and returned from the adoption attempt. Moving that round-trip ahead of the count and row assertions makes it do double duty: it is still the "refusing adoption must not refuse service" proof, and it is now also the ordering barrier the count read needs to be meaningful. The second sleep, before the post-drop count read, is removed for a different reason: the inbound leg was refused adoption, so `release_inbound_pool_slot` ran with `adopted = None` and is a no-op. There is no pending async decrement for the drop to race against. Also corrects the surviving-slot comment against the real dig-gossip behaviour rather than inferring it from dig-node's doc. `adopt_direct_inbound_handle` REFUSES outright -- it does not supersede -- whenever the held slot's `dial_addr()` is `Some`: an accepted connection never supersedes a slot this node can dial. The assertion was already right; the reasoning behind it is now sourced. Refs #3124 --- .../tests/inbound_pool_membership.rs | 55 +++++++++++-------- 1 file changed, 33 insertions(+), 22 deletions(-) diff --git a/crates/dig-node-core/tests/inbound_pool_membership.rs b/crates/dig-node-core/tests/inbound_pool_membership.rs index 646b17d9..6d718652 100644 --- a/crates/dig-node-core/tests/inbound_pool_membership.rs +++ b/crates/dig-node-core/tests/inbound_pool_membership.rs @@ -604,8 +604,29 @@ async fn a_peer_that_is_both_dialled_and_accepted_is_counted_once() { "the inbound dial must authenticate as the SAME identity as the outbound slot" ); - // Give the server's spawned adoption a moment to run, then settle: the count must NOT go to 2. - tokio::time::sleep(Duration::from_millis(300)).await; + // -- Step 2b (moved ahead of Step 3's count/row assertions): the RPC round-trip IS the ordering + // barrier, not a courtesy check. `adopt_inbound_peer_in_pool` is called at `peer.rs:3602`, + // strictly BEFORE `serve_peer_session_from_with` starts answering RPC on the accepted session + // (peer.rs:3614) -- so a successful `dig.getNetworkInfo` response over `inbound_conn` proves the + // server has already reached and returned from the adoption attempt. A bare `sleep` before the + // count assertion below cannot make that promise: on a slow CI box the accept task may simply not + // have run yet, and `peer_count() == 1` would be trivially true for the wrong reason (the inbound + // leg never having been driven at all), not because the pool correctly refused it. This is also + // why the RPC assertion moved ahead of the "still served" comment it used to sit under -- it now + // does double duty as both the serve-path proof AND the happens-before proof for step 2/3. + { + let mut stream = inbound_conn.session.open_stream().await.expect("open stream"); + let req = json!({"jsonrpc":"2.0","id":21,"method":"dig.getNetworkInfo"}); + write_framed(&mut stream, &req).await.expect("write"); + let resp = read_one_frame(&mut stream).await; + assert_eq!( + resp["result"]["served_method"], "dig.getNetworkInfo", + "the un-adopted inbound peer must still be served -- refusing adoption must not refuse service" + ); + } + + // The RPC round-trip above already proves the adoption attempt ran and returned, so this count + // read needs no sleep to be meaningful: it must NOT go to 2. assert_eq!( gossip_a.peer_count().await, 1, @@ -625,9 +646,12 @@ async fn a_peer_that_is_both_dialled_and_accepted_is_counted_once() { matching_rows, 1, "exactly one row must carry B's peer_id -- a de-duplication failure would emit two" ); - // The surviving row is the DIALLED slot, per `adopt_inbound_peer_in_pool`'s own refusal doc - // (peer.rs:3658-3662): "a peer already holding a dialable slot" is refused, so the pre-existing - // outbound slot is kept and the inbound accept is the one that is turned away. + // The surviving row is the DIALLED slot: `adopt_direct_inbound_handle` + // (dig-gossip `service/gossip_handle.rs`, admission section) refuses outright -- it does not + // supersede -- whenever the held slot's `dial_addr()` is `Some`: "an accepted connection NEVER + // supersedes a slot this node can dial" (the #870 rule). `adopt_inbound_peer_in_pool`'s own doc + // (peer.rs:3658-3662) names the same refusal. So the pre-existing outbound slot is kept and the + // inbound accept is the one turned away -- this is a REFUSAL, not a supersede-by-newer-connection. let surviving = gossip_a .connected_pool_peers_detailed() .into_iter() @@ -637,21 +661,6 @@ async fn a_peer_that_is_both_dialled_and_accepted_is_counted_once() { surviving.is_outbound, "the surviving slot must be the pre-existing DIALLED one, per the documented refusal" ); - - // -- Step 4: still served, BOTH ways -- the discriminator against a shape that buys the count by - // dropping a connection. The dialled slot was never a live RPC session in this fixture (it is a - // bare loopback duplex with no responder on the other end), so what matters here is that the - // INBOUND session -- the one the pool refused to adopt -- is still answered. - { - let mut stream = inbound_conn.session.open_stream().await.expect("open stream"); - let req = json!({"jsonrpc":"2.0","id":21,"method":"dig.getNetworkInfo"}); - write_framed(&mut stream, &req).await.expect("write"); - let resp = read_one_frame(&mut stream).await; - assert_eq!( - resp["result"]["served_method"], "dig.getNetworkInfo", - "the un-adopted inbound peer must still be served -- refusing adoption must not refuse service" - ); - } assert_eq!( gossip_a.peer_count().await, 1, @@ -660,9 +669,11 @@ async fn a_peer_that_is_both_dialled_and_accepted_is_counted_once() { // -- Step 5: released cleanly, SESSION-scoped ------------------------------------------------------ // Drop the inbound session first: the outbound slot must survive, because releasing it was never - // the inbound session's to release (it never held the slot). + // the inbound session's to release (it never held the slot). No wait is needed here either: the + // inbound leg was REFUSED adoption (confirmed above), so `release_inbound_pool_slot` was called + // with `adopted = None` and is a documented no-op (peer.rs `release_inbound_pool_slot`) -- there + // is no pending async decrement for this drop to race against. drop(inbound_conn); - tokio::time::sleep(Duration::from_millis(300)).await; assert_eq!( gossip_a.peer_count().await, 1, From 850a077fc5af925a7a7cd965418c01abb03e96a0 Mon Sep 17 00:00:00 2001 From: Michael Taylor Date: Tue, 8 Sep 2026 04:35:29 -0700 Subject: [PATCH 3/4] test(peer): bound Step 5's non-event check to a real window Step 5 asserted the count stays 1 with a single instant read right after dropping the inbound session. A teeth check proved this cannot fail: the server's teardown does not run synchronously with the client-side drop, so the read fires before the server has acted, regardless of whether the eventual release is correct. Replace it with a poll across a 2s window that would catch an erroneous drop to 0 if it ever occurred, and explain in the comment why a window is required instead of an instant read. Also drop the same-identity assertion in Step 2 (b_bytes was derived from b_identity two lines above, so it was true by construction and checked nothing server-side); the real property is carried by peer_count() == 1. Renumber the remaining steps (2b -> 3, 3 -> 4) now that there is no gap. Refs #3124 --- .../tests/inbound_pool_membership.rs | 50 +++++++++++++------ 1 file changed, 34 insertions(+), 16 deletions(-) diff --git a/crates/dig-node-core/tests/inbound_pool_membership.rs b/crates/dig-node-core/tests/inbound_pool_membership.rs index 6d718652..1cf711f5 100644 --- a/crates/dig-node-core/tests/inbound_pool_membership.rs +++ b/crates/dig-node-core/tests/inbound_pool_membership.rs @@ -598,13 +598,13 @@ async fn a_peer_that_is_both_dialled_and_accepted_is_counted_once() { let mut inbound_conn = dig_nat::connect(&target, &b_identity, &config) .await .expect("B's transport-level connect succeeds even if the pool refuses to adopt it"); - assert_eq!( - *b_identity.peer_id().as_bytes(), - b_bytes, - "the inbound dial must authenticate as the SAME identity as the outbound slot" - ); + // (No same-identity assertion here: `b_bytes` was derived FROM `b_identity` two lines above, so + // comparing them back is true by construction and proves nothing about the server side. The + // property that matters -- that the server admitted B as the SAME identity, not a distinct one -- + // is carried by `peer_count() == 1` below: a real identity mismatch would create a SECOND slot and + // the count would read 2. That is the assertion doing the real work.) - // -- Step 2b (moved ahead of Step 3's count/row assertions): the RPC round-trip IS the ordering + // -- Step 3 (moved ahead of Step 4's count/row assertions): the RPC round-trip IS the ordering // barrier, not a courtesy check. `adopt_inbound_peer_in_pool` is called at `peer.rs:3602`, // strictly BEFORE `serve_peer_session_from_with` starts answering RPC on the accepted session // (peer.rs:3614) -- so a successful `dig.getNetworkInfo` response over `inbound_conn` proves the @@ -633,7 +633,7 @@ async fn a_peer_that_is_both_dialled_and_accepted_is_counted_once() { "a peer that is both dialled and accepted must be counted ONCE, not twice" ); - // -- Step 3: exactly ONE row for B's peer_id among connected_pool_peers() ----------------------- + // -- Step 4: exactly ONE row for B's peer_id among connected_pool_peers() ----------------------- // (`connected_peers_json` is `pub(crate)` inside dig-node-core and unreachable from this // integration-test crate; `connected_pool_peers()` is its public dig-gossip source, so counting // matching rows here proves the same property `connected_peers_json` would report.) @@ -669,16 +669,34 @@ async fn a_peer_that_is_both_dialled_and_accepted_is_counted_once() { // -- Step 5: released cleanly, SESSION-scoped ------------------------------------------------------ // Drop the inbound session first: the outbound slot must survive, because releasing it was never - // the inbound session's to release (it never held the slot). No wait is needed here either: the - // inbound leg was REFUSED adoption (confirmed above), so `release_inbound_pool_slot` was called - // with `adopted = None` and is a documented no-op (peer.rs `release_inbound_pool_slot`) -- there - // is no pending async decrement for this drop to race against. + // the inbound session's to release (it never held the slot). + // + // This checks a NON-event -- that no release happens -- so a single instant read right after + // `drop` cannot prove it: dropping the CLIENT side does not synchronously run the SERVER's + // teardown. The server must first observe the closed transport in its own accept/serve task and + // only then run `release_inbound_pool_slot`; an instant read fires before the server has had a + // chance to act, which passes just as well under the defect this step exists to catch (an + // erroneous release of the outbound slot) as it does under correct behaviour -- it cannot tell + // the two apart. Poll across a bounded window instead: if the inbound leg's teardown incorrectly + // released the OUTBOUND slot it never owned, the count drops to 0 at some point inside the + // window and this loop catches it; if the release is correctly a no-op, the count simply stays + // at 1 for the whole window. + // + // There is no cheap positive signal available here that the server has specifically finished + // processing THIS disconnect (the RPC-round-trip trick Step 3 uses needs a live stream, which + // `drop` just closed) -- so the window is the whole proof, not a supplement to one. drop(inbound_conn); - assert_eq!( - gossip_a.peer_count().await, - 1, - "the inbound session ending must not release the outbound slot it never owned" - ); + let deadline = std::time::Instant::now() + Duration::from_secs(2); + while std::time::Instant::now() < deadline { + assert_eq!( + gossip_a.peer_count().await, + 1, + "the inbound session ending released the outbound slot it never owned -- a \ + session-scoped release defect: the refused inbound leg tore down B's dialled slot when \ + its own transport closed" + ); + tokio::time::sleep(Duration::from_millis(50)).await; + } // Now release the outbound slot itself and confirm the count reaches zero. gossip_a From 4ae4ad97edce5fea880178e2500bfba4e6f02166 Mon Sep 17 00:00:00 2001 From: Michael Taylor Date: Tue, 8 Sep 2026 05:08:21 -0700 Subject: [PATCH 4/4] style(peer): rustfmt the inbound pool membership test --- crates/dig-node-core/tests/inbound_pool_membership.rs | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/crates/dig-node-core/tests/inbound_pool_membership.rs b/crates/dig-node-core/tests/inbound_pool_membership.rs index 1cf711f5..523b6959 100644 --- a/crates/dig-node-core/tests/inbound_pool_membership.rs +++ b/crates/dig-node-core/tests/inbound_pool_membership.rs @@ -615,7 +615,11 @@ async fn a_peer_that_is_both_dialled_and_accepted_is_counted_once() { // why the RPC assertion moved ahead of the "still served" comment it used to sit under -- it now // does double duty as both the serve-path proof AND the happens-before proof for step 2/3. { - let mut stream = inbound_conn.session.open_stream().await.expect("open stream"); + let mut stream = inbound_conn + .session + .open_stream() + .await + .expect("open stream"); let req = json!({"jsonrpc":"2.0","id":21,"method":"dig.getNetworkInfo"}); write_framed(&mut stream, &req).await.expect("write"); let resp = read_one_frame(&mut stream).await;