feat(peer): register accepted inbound peers in the connected pool (#3124)

crates/dig-node-core/Cargo.toml

@@ -349,7 +349,7 @@ dig-nat = "0.20"
 # certificate at all — a mutual-TLS bypass that no consumer could see from its own manifest. Its
 # publish.yml guard is unconditional and deliberate: do not weaken it, do not `--allow-dirty` past it.
 # dig_ecosystem#900 ("publish dig-gossip") was closed as superseded by #2647.
-dig-gossip = { git = "https://github.com/DIG-Network/dig-gossip", rev = "48a14ed2ed395a41c913d5c8f944e6ea0b747075", default-features = false, features = ["rustls", "relay"] }
+dig-gossip = { git = "https://github.com/DIG-Network/dig-gossip", rev = "f21f993e7c46d0f506dd200743b11b3b19f4eef6", default-features = false, features = ["rustls", "relay"] }
 # The canonical DIG network id (genesis challenge) for the gossip handshake / introducer scope.
 #
 # 0.9 is the TIP OF DIG-NODE'S CHIA LINE, and that — not 0.10 — is the correct target here (#2072).
@@ -542,7 +542,7 @@ dig-keystore = { version = "0.13", features = ["testing"] }
 # rather than on the normal dependency so the shipped binary never carries the test hook: cargo
 # unifies the feature into test builds only, since dev-dependencies are not built for `cargo build`.
 # NC-7 exception: same permanent git pin as the runtime dep above (dig_ecosystem#2647).
-dig-gossip = { git = "https://github.com/DIG-Network/dig-gossip", rev = "48a14ed2ed395a41c913d5c8f944e6ea0b747075", default-features = false, features = ["rustls", "relay", "test-util"] }
+dig-gossip = { git = "https://github.com/DIG-Network/dig-gossip", rev = "f21f993e7c46d0f506dd200743b11b3b19f4eef6", default-features = false, features = ["rustls", "relay", "test-util"] }
 # Generates the P-256 leaf key pairs the opcode-222 holdings tests sign with: `public_key_der()` IS the
 # leaf `SubjectPublicKeyInfo` DER the wire carries as `provider_spki`, and whose SHA-256 is the
 # announcing peer_id — so a test signer is a real §5.2 identity rather than a stand-in that could hide

crates/dig-node-core/src/peer.rs

@@ -2862,6 +2862,9 @@ async fn run_peer_network(node: Arc<crate::Node>) -> Result<(), String> {
         stun_server,
     );
 
+    // #3124: the listener registers every peer it ACCEPTS in this same pool, so clone the handle
+    // before the responder takes ownership of it.
+    let handle_for_pool = handle.clone();
     let mut node_responder = NodeResponder::with_pool(node, handle);
     if let Some(dht) = dht {
         node_responder = node_responder.with_dht(dht);
@@ -2881,7 +2884,7 @@ async fn run_peer_network(node: Arc<crate::Node>) -> Result<(), String> {
         );
     }
 
-    serve_peer_rpc_listener_with(listener, identity, responder, Some(pex)).await
+    serve_peer_rpc_listener_with(listener, identity, responder, Some(pex), Some(handle_for_pool)).await
 }
 
 /// Bring up the content-location DHT (#163) for a running node: build a [`crate::dht::NatDhtTransport`]
@@ -3213,19 +3216,25 @@ pub async fn serve_peer_rpc_listener(
     node: Arc<dig_nat::NodeCert>,
     responder: Arc<dyn PeerRpcResponder>,
 ) -> Result<(), String> {
-    serve_peer_rpc_listener_with(listener, node, responder, None).await
+    serve_peer_rpc_listener_with(listener, node, responder, None, None).await
 }
 
 /// Like [`serve_peer_rpc_listener`] but additionally running the node↔node **PEX** peer-sharing layer
 /// (#166) over each accepted mTLS connection when `pex` is `Some`: the node opens its outgoing PEX
 /// stream (handshake→snapshot→deltas) and serves the peer's incoming PEX stream, feeding discovered
 /// peers into the pool as dial candidates. `None` disables PEX (the FFI/base path + existing callers),
 /// leaving the serve path byte-identical to before.
+///
+/// `gossip` is the connected pool every accepted peer is registered in for the life of its serve loop
+/// (**dig_ecosystem#3124**). `None` leaves the peer unregistered — correct for the test and FFI
+/// callers below, which run no pool — and is why the 3-argument
+/// [`serve_peer_rpc_listener`] keeps its signature.
 pub async fn serve_peer_rpc_listener_with(
     listener: tokio::net::TcpListener,
     node: Arc<dig_nat::NodeCert>,
     responder: Arc<dyn PeerRpcResponder>,
     pex: Option<Arc<crate::pex::PexServing>>,
+    gossip: Option<dig_gossip::GossipHandle>,
 ) -> Result<(), String> {
     let server_config = build_server_tls_config(&node)?;
     let acceptor = tokio_rustls::TlsAcceptor::from(server_config);
@@ -3248,6 +3257,7 @@ pub async fn serve_peer_rpc_listener_with(
         let acceptor = acceptor.clone();
         let responder = responder.clone();
         let pex = pex.clone();
+        let gossip = gossip.clone();
         let spawned = spawn_with_permit(&conn_permits, async move {
             // mTLS handshake (client cert required by build_server_tls_config; a peer with no cert or
             // a failed handshake is dropped here — no unauthenticated peer traffic reaches the RPC).
@@ -3260,8 +3270,39 @@ pub async fn serve_peer_rpc_listener_with(
                     // body. `None` if (defensively) no client cert is present, which the verifier
                     // should already have rejected.
                     let caller = caller_from_tls(&tls, peer_addr);
+                    // #3124: the identity the pool slot is keyed on comes from the CERTIFICATE the
+                    // handshake just verified — the same derivation `caller_from_tls` uses — never
+                    // from the wire body and never from the DHT contact (whose `peer_id` is a
+                    // display-hex String, not this type). Read before the stream is consumed by the
+                    // mux. `None` when no client cert parsed: there is then no authenticated identity
+                    // to key a slot on, and serving uncounted is safer than inventing one.
+                    let authenticated = peer_id_from_tls(&tls);
                     let mut session = dig_nat::mux::PeerSession::server(tls);
+
+                    // COUNT the peer for as long as we serve it. `peer_addr` is its EPHEMERAL SOURCE
+                    // port, which is why this goes through the direct-INBOUND entry point — the pool
+                    // records the address for observability and never offers it as a dial target. The
+                    // tier is `Direct`: this connection arrived over TCP with no relay.
+                    let adopted = match authenticated {
+                        Some(peer_id) => {
+                            adopt_inbound_peer_in_pool(
+                                gossip.as_ref(),
+                                &peer_id,
+                                peer_addr,
+                                dig_nat::TraversalKind::Direct,
+                                &session,
+                            )
+                            .await
+                        }
+                        None => None,
+                    };
+
                     serve_peer_session_from_with(caller, &mut session, responder, pex).await;
+
+                    // The serve loop has returned, so this node is no longer serving the peer and must
+                    // stop counting it. `disconnect` only ends the ACCOUNTING — the session is ours and
+                    // is closed by dropping it here.
+                    release_inbound_pool_slot(gossip.as_ref(), adopted).await;
                 }
                 Err(e) => tracing::debug!(error = %e, "peer mTLS handshake failed; dropped"),
             }
@@ -3275,6 +3316,72 @@ pub async fn serve_peer_rpc_listener_with(
     }
 }
 
+/// Register an ACCEPTED inbound peer in the dig-gossip connected pool for as long as this node serves
+/// it, and stop counting it when the serve loop ends (**dig_ecosystem#3124**).
+///
+/// # Why this exists
+///
+/// The pool is what every subsystem reads to answer "am I connected", and until this call the node
+/// registered NOTHING it accepted — only peers it dialed. A node serving inbound peers perfectly well
+/// reported `connected_peers` as if it had none.
+///
+/// # Why it registers by HANDLE
+///
+/// The serve loop below needs `&mut PeerSession` to answer the peer's L7 RPC, and `PeerSession` is not
+/// `Clone`. Handing the session to the pool would buy the count and stop serving the peer — strictly
+/// worse than being uncounted. `adopt_direct_inbound_handle` takes a `ClosedHandle` instead, so
+/// ownership stays here and the peer is both counted and served.
+///
+/// Adoption is best-effort by design: it is ACCOUNTING, and every refusal the pool can return (the
+/// accepted-direct cap, a ban, a full pool, a peer already holding a dialable slot) is a decision this
+/// node made on purpose. None of them is a reason to refuse SERVICE to a peer whose handshake already
+/// succeeded, so a refusal is logged and the connection is served uncounted — the behaviour that
+/// shipped before this call existed.
+///
+/// Returns the `PeerId` to deregister once serving ends, or `None` when nothing was registered.
+async fn adopt_inbound_peer_in_pool(
+    gossip: Option<&dig_gossip::GossipHandle>,
+    peer_id: &dig_nat::PeerId,
+    remote: std::net::SocketAddr,
+    method: dig_nat::TraversalKind,
+    session: &dig_nat::mux::PeerSession,
+) -> Option<dig_gossip::PeerId> {
+    let gossip = gossip?;
+    // dig-nat reports the identity as its own `PeerId`; the pool keys on the gossip `PeerId` (chia
+    // `Bytes32`) over the SAME 32 bytes the mTLS handshake proved.
+    let pool_id = dig_gossip::PeerId::from(*peer_id.as_bytes());
+    let observed = dig_gossip::ObservedSession::new(session.closed_handle(), move || {
+        // A newer connection for this identity displaced the slot. This session is now obsolete to the
+        // pool, but it is still OURS to end, and the serve loop below ends it on return — so there is
+        // nothing to do here but record it. Dropping the observer silently is what #71 forbids.
+        tracing::debug!(peer_id = %pool_id, "inbound pool slot superseded by a newer connection");
+    });
+
+    match gossip
+        .adopt_direct_inbound_handle(pool_id, remote, method, observed, None)
+        .await
+    {
+        Ok(id) => Some(id),
+        Err(e) => {
+            tracing::debug!(peer_id = %pool_id, error = %e, "inbound peer not adopted into the pool; serving it uncounted");
+            None
+        }
+    }
+}
+
+/// Stop counting an inbound peer once its serve loop has ended.
+///
+/// `disconnect` only stops ACCOUNTING for a slot registered by handle — it does not close the
+/// transport, which is this side's to end and is ended by the serve loop returning.
+async fn release_inbound_pool_slot(
+    gossip: Option<&dig_gossip::GossipHandle>,
+    adopted: Option<dig_gossip::PeerId>,
+) {
+    if let (Some(gossip), Some(peer_id)) = (gossip, adopted) {
+        let _ = gossip.disconnect(&peer_id).await;
+    }
+}
+
 /// Build the authenticated caller [`dig_dht::Contact`] from an accepted mTLS server connection: read
 /// the client's leaf certificate, derive its `peer_id = SHA-256(SPKI DER)` (the SAME derivation
 /// dig-nat enforces), and pair it with the remote socket address. Returns `None` if no client cert is
@@ -3283,10 +3390,25 @@ fn caller_from_tls(
     tls: &tokio_rustls::server::TlsStream<tokio::net::TcpStream>,
     remote_addr: std::net::SocketAddr,
 ) -> Option<dig_dht::Contact> {
+    let peer_id = peer_id_from_tls(tls)?;
+    Some(crate::dht::caller_contact(&peer_id, remote_addr))
+}
+
+/// The AUTHENTICATED `peer_id = SHA-256(SPKI DER)` of an accepted mTLS peer, read from the leaf
+/// certificate rustls verified during the handshake (**dig_ecosystem#3124**).
+///
+/// Split out of [`caller_from_tls`] because the pool keys a slot on the identity itself, while the DHT
+/// contact carries it as display hex — the two are not interchangeable, and deriving both from this
+/// ONE function is what keeps them from drifting onto different sources.
+///
+/// `None` when the peer presented no certificate or it does not parse; the client-cert verifier should
+/// already have rejected such a peer.
+fn peer_id_from_tls(
+    tls: &tokio_rustls::server::TlsStream<tokio::net::TcpStream>,
+) -> Option<dig_nat::PeerId> {
     let (_io, conn) = tls.get_ref();
     let leaf = conn.peer_certificates()?.first()?;
-    let peer_id = dig_nat::peer_id_from_leaf_cert_der(leaf.as_ref())?;
-    Some(crate::dht::caller_contact(&peer_id, remote_addr))
+    dig_nat::peer_id_from_leaf_cert_der(leaf.as_ref())
 }
 
 /// Build the rustls `ServerConfig` for the mTLS peer-RPC listener from the node's CA-signed