fix(network): validate HELLO before the simultaneous-dial index update

`update_ingress_node_id` ran before HELLO validation, so a simultaneous-dial
`Replaced` overwrote `node_id_index` to the new ingress; if HELLO then failed
validation only the new stream was killed, leaving the old session in the slab
but absent from the index — unaddressable by node id.

Validate HELLO into a candidate first and commit it (protocols, `node_db`,
`had_hello`) only once the session is accepted, so a failed HELLO never touches
the index. Surfaced via #3510.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: peilun-conflux

crates/network/src/session.rs

@@ -90,6 +90,14 @@ pub struct SessionDataWithDisconnectInfo {
     pub token_to_disconnect: Option<(StreamToken, String)>,
 }
 
+/// A HELLO validated but not yet committed; `commit_hello` applies it only
+/// after the simultaneous-dial tie-break accepts the session.
+struct HelloCandidate {
+    peer_protocols: Vec<ProtocolInfo>,
+    node_entry: NodeEntry,
+    pos_public_key: Option<(ConsensusPublicKey, ConsensusVRFPublicKey)>,
+}
+
 // id for Hello packet
 const PACKET_HELLO: u8 = 0x80;
 // id for Disconnect packet
@@ -302,11 +310,16 @@ impl Session {
             PACKET_HELLO => {
                 debug!("Read HELLO in session {:?}", self);
                 self.metadata.peer_header_version = packet.header_version;
-                // For ingress session, update the node id in `SessionManager`
+
+                // Validate before the tie-break touches the index, so a failed
+                // HELLO can't orphan a replaced session's index entry.
+                let rlp = Rlp::new(&packet.data);
+                let candidate = self.validate_hello(&rlp, host)?;
+
                 let token_to_disconnect = self.update_ingress_node_id(host)?;
 
-                // Dropped by the tie-break: stop before `Ready` so the loser
-                // never reaches `on_peer_connected`; the caller kills it.
+                // Lost the tie-break: stop before `Ready` and commit nothing;
+                // the caller kills it.
                 if token_to_disconnect.as_ref().map(|(t, _)| *t)
                     == Some(self.token())
                 {
@@ -316,9 +329,7 @@ impl Session {
                     });
                 }
 
-                // Handle Hello packet to exchange protocols
-                let rlp = Rlp::new(&packet.data);
-                let pos_public_key = self.read_hello(&rlp, host)?;
+                let pos_public_key = self.commit_hello(candidate, host);
                 Ok(SessionDataWithDisconnectInfo {
                     session_data: SessionData::Ready { pos_public_key },
                     token_to_disconnect,
@@ -406,16 +417,12 @@ impl Session {
         }
     }
 
-    /// Read Hello packet to exchange the supported protocols, and set the
-    /// `had_hello` flag to indicates that session is ready to send/receive
-    /// protocol packets.
-    ///
-    /// Besides, the node endpoint of remote peer will be added or updated in
-    /// node database, which is used to establish outgoing connections.
-    fn read_hello(
+    /// Validate a HELLO into a `HelloCandidate` without committing it. The
+    /// caller commits via `commit_hello` only after the tie-break accepts the
+    /// session, so a failed HELLO never half-updates the index.
+    fn validate_hello(
         &mut self, rlp: &Rlp, host: &NetworkServiceInner,
-    ) -> Result<Option<(ConsensusPublicKey, ConsensusVRFPublicKey)>, Error>
-    {
+    ) -> Result<HelloCandidate, Error> {
         let remote_network_id: u64 = rlp.val_at(0)?;
         if remote_network_id != host.metadata.network_id {
             debug!(
@@ -450,8 +457,7 @@ impl Session {
                 .any(|hc| hc.protocol == c.protocol && hc.version <= c.version)
         });
 
-        self.metadata.peer_protocols = peer_caps;
-        if self.metadata.peer_protocols.is_empty() {
+        if peer_caps.is_empty() {
             debug!("No common capabilities with remote peer, peer_node_id = {:?}, session = {:?}", self.metadata.id, self);
             return Err(self.send_disconnect(DisconnectReason::UselessPeer));
         }
@@ -487,14 +493,11 @@ impl Session {
                 entry, self
             );
             return Err(self.send_disconnect(DisconnectReason::IpLimited));
-        } else {
-            debug!("Received valid endpoint {:?}, session = {:?}", entry, self);
-            host.node_db.write().insert_with_token(entry, self.token());
         }
+        debug!("Received valid endpoint {:?}, session = {:?}", entry, self);
 
-        self.had_hello = Some(Instant::now());
-        match rlp.item_count()? {
-            3 => Ok(None),
+        let pos_public_key = match rlp.item_count()? {
+            3 => None,
             4 => {
                 // FIXME(lpl): Verify keys.
                 let pos_public_key_bytes: Vec<u8> = rlp.val_at(3)?;
@@ -511,13 +514,32 @@ impl Session {
                 )
                 .map_err(|e| Error::Decoder(format!("{:?}", e)))?;
 
-                Ok(Some((bls_pub_key, vrf_pub_key)))
+                Some((bls_pub_key, vrf_pub_key))
             }
-            length => Err(Error::Decoder(format!(
-                "Hello has incorrect rlp length: {:?}",
-                length
-            ))),
-        }
+            length => {
+                return Err(Error::Decoder(format!(
+                    "Hello has incorrect rlp length: {:?}",
+                    length
+                )))
+            }
+        };
+
+        Ok(HelloCandidate {
+            peer_protocols: peer_caps,
+            node_entry: entry,
+            pos_public_key,
+        })
+    }
+
+    fn commit_hello(
+        &mut self, candidate: HelloCandidate, host: &NetworkServiceInner,
+    ) -> Option<(ConsensusPublicKey, ConsensusVRFPublicKey)> {
+        self.metadata.peer_protocols = candidate.peer_protocols;
+        host.node_db
+            .write()
+            .insert_with_token(candidate.node_entry, self.token());
+        self.had_hello = Some(Instant::now());
+        candidate.pos_public_key
     }
 
     /// Assemble a packet with specified protocol id, packet id and data.