fix: immediate peer discovery by sending FindOptimalPeer over network (#1926)

sanity · claude · github-actions[bot] · web-flow · commit e858f850f841 · 2025-10-09T01:25:14.000Z
Co-authored-by: Claude &lt;noreply@anthropic.com&gt;
Co-authored-by: claude[bot] &lt;41898282+claude[bot]@users.noreply.github.com&gt;
Co-authored-by: nacho.d.g &lt;iduartgomez@users.noreply.github.com&gt;
diff --git a/crates/core/src/message.rs b/crates/core/src/message.rs
@@ -334,6 +334,11 @@ pub(crate) enum NodeEvent {
         key: ContractKey,
         subscribed: bool,
     },
+    /// Send a message to a peer over the network
+    SendMessage {
+        target: PeerId,
+        msg: Box<NetMessage>,
+    },
 }
 
 #[derive(Debug, Clone)]
@@ -410,6 +415,9 @@ impl Display for NodeEvent {
                     "Local subscribe complete (tx: {tx}, key: {key}, subscribed: {subscribed})"
                 )
             }
+            NodeEvent::SendMessage { target, msg } => {
+                write!(f, "SendMessage (to {target}, tx: {})", msg.id())
+            }
         }
     }
 }
diff --git a/crates/core/src/node/network_bridge/p2p_protoc.rs b/crates/core/src/node/network_bridge/p2p_protoc.rs
@@ -308,12 +308,13 @@ impl P2pConnManager {
                                 ChannelCloseReason::Handshake
                                 | ChannelCloseReason::Bridge
                                 | ChannelCloseReason::Controller => {
-                                    // Critical internal channels closed - perform cleanup and shutdown gracefully
+                                    // All ClosedChannel events are critical - the transport is unable to establish
+                                    // more connections, rendering this peer useless. Perform cleanup and shutdown.
                                     tracing::error!(
                                         ?reason,
                                         is_gateway = self.bridge.op_manager.ring.is_gateway(),
                                         num_connections = self.connections.len(),
-                                        "🔴 CRITICAL CHANNEL CLOSED - performing cleanup and shutting down"
+                                        "Critical channel closed - performing cleanup and shutting down"
                                     );
 
                                     // Clean up all active connections
@@ -401,6 +402,15 @@ impl P2pConnManager {
                                 )
                                 .await?;
                             }
+                            NodeEvent::SendMessage { target, msg } => {
+                                // Send the message to the target peer over the network
+                                tracing::debug!(
+                                    tx = %msg.id(),
+                                    %target,
+                                    "SendMessage event: sending message to peer via network bridge"
+                                );
+                                self.bridge.send(&target, *msg).await?;
+                            }
                             NodeEvent::QueryConnections { callback } => {
                                 let connections = self.connections.keys().cloned().collect();
                                 timeout(
@@ -718,14 +728,9 @@ impl P2pConnManager {
                         self.handle_handshake_action(event, state, handshake_handler_msg).await?;
                         Ok(EventResult::Continue)
                     }
-                    Err(HandshakeError::ChannelClosed) => {
-                        tracing::error!(
-                            "🔴 HANDSHAKE CHANNEL CLOSED - handshake handler's channel has closed"
-                        );
-                        Ok(EventResult::Event(
-                            ConnEvent::ClosedChannel(ChannelCloseReason::Handshake).into(),
-                        ))
-                    }
+                    Err(HandshakeError::ChannelClosed) => Ok(EventResult::Event(
+                        ConnEvent::ClosedChannel(ChannelCloseReason::Handshake).into(),
+                    )),
                     Err(e) => {
                         tracing::warn!("Handshake error: {:?}", e);
                         Ok(EventResult::Continue)
@@ -1161,20 +1166,14 @@ impl P2pConnManager {
         match msg {
             Some(Left((_, msg))) => EventResult::Event(ConnEvent::OutboundMessage(*msg).into()),
             Some(Right(action)) => EventResult::Event(ConnEvent::NodeAction(action).into()),
-            None => {
-                tracing::error!("🔴 BRIDGE CHANNEL CLOSED - P2P bridge channel has closed");
-                EventResult::Event(ConnEvent::ClosedChannel(ChannelCloseReason::Bridge).into())
-            }
+            None => EventResult::Event(ConnEvent::ClosedChannel(ChannelCloseReason::Bridge).into()),
         }
     }
 
     fn handle_node_controller_msg(&self, msg: Option<NodeEvent>) -> EventResult {
         match msg {
             Some(msg) => EventResult::Event(ConnEvent::NodeAction(msg).into()),
             None => {
-                tracing::error!(
-                    "🔴 CONTROLLER CHANNEL CLOSED - node controller channel has closed"
-                );
                 EventResult::Event(ConnEvent::ClosedChannel(ChannelCloseReason::Controller).into())
             }
         }
diff --git a/crates/core/src/node/testing_impl.rs b/crates/core/src/node/testing_impl.rs
@@ -935,6 +935,11 @@ where
                 NodeEvent::QueryNodeDiagnostics { .. } => {
                     unimplemented!()
                 }
+                NodeEvent::SendMessage { target, msg } => {
+                    tracing::debug!(tx = %msg.id(), %target, "SendMessage event in testing_impl");
+                    conn_manager.send(&target, *msg).await?;
+                    continue;
+                }
             },
             Err(err) => {
                 super::report_result(
diff --git a/crates/core/src/operations/connect.rs b/crates/core/src/operations/connect.rs
@@ -180,7 +180,11 @@ impl Operation for ConnectOp {
                     };
                     let mut skip_connections = skip_connections.clone();
                     let mut skip_forwards = skip_forwards.clone();
-                    skip_connections.extend([this_peer.clone(), query_target.peer.clone()]);
+                    skip_connections.extend([
+                        this_peer.clone(),
+                        query_target.peer.clone(),
+                        joiner.peer.clone(),
+                    ]);
                     skip_forwards.extend([this_peer.clone(), query_target.peer.clone()]);
                     if this_peer == &query_target.peer {
                         // this peer should be the original target queries
@@ -222,9 +226,21 @@ impl Operation for ConnectOp {
                                 tx = %id,
                                 query_target = %query_target.peer,
                                 joiner = %joiner.peer,
-                                "Gateway has no desirable peer to offer to joiner",
+                                "Gateway found no suitable peers to forward CheckConnectivity request",
                             );
-                            return_msg = None;
+                            // Send a negative response back to the joiner to inform them
+                            // that no suitable peers are currently available
+                            let response = ConnectResponse::AcceptedBy {
+                                accepted: false,
+                                acceptor: own_loc.clone(),
+                                joiner: joiner.peer.clone(),
+                            };
+                            return_msg = Some(ConnectMsg::Response {
+                                id: *id,
+                                sender: own_loc.clone(),
+                                target: joiner.clone(),
+                                msg: response,
+                            });
                             new_state = None;
                         }
                     } else {
@@ -269,17 +285,27 @@ impl Operation for ConnectOp {
                         },
                     ..
                 } => {
+                    let this_peer = op_manager.ring.connection_manager.own_location();
                     if sender.peer == joiner.peer {
                         tracing::error!(
                             tx = %id,
                             sender = %sender.peer,
                             joiner = %joiner.peer,
-                            at = %op_manager.ring.connection_manager.own_location().peer,
-                            "Connectivity check from self, aborting"
+                            at = %this_peer.peer,
+                            "Connectivity check from self (sender == joiner), rejecting operation"
                         );
-                        std::process::exit(1);
+                        return Err(OpError::UnexpectedOpState);
+                    }
+                    if this_peer.peer == joiner.peer {
+                        tracing::error!(
+                            tx = %id,
+                            this_peer = %this_peer.peer,
+                            joiner = %joiner.peer,
+                            sender = %sender.peer,
+                            "Received CheckConnectivity where this peer is the joiner (self-connection attempt), rejecting operation"
+                        );
+                        return Err(OpError::UnexpectedOpState);
                     }
-                    let this_peer = op_manager.ring.connection_manager.own_location();
                     let joiner_loc = joiner
                         .location
                         .expect("should be already set at the p2p bridge level");
@@ -978,16 +1004,19 @@ async fn connect_request(
                 None,
             );
 
-            // Push the new operation and send the message
+            // Push the new operation
             op_manager
                 .push(new_tx_id, OpEnum::Connect(Box::new(new_op)))
                 .await?;
 
+            // Send the FindOptimalPeer message to the gateway over the network
+            // We use notify_node_event with a SendMessage event to ensure it goes through
+            // the proper network channel, not just local processing
             op_manager
-                .notify_op_change(
-                    NetMessage::from(msg),
-                    OpEnum::Connect(Box::new(ConnectOp::new(new_tx_id, None, None, None))),
-                )
+                .notify_node_event(NodeEvent::SendMessage {
+                    target: gateway.peer.clone(),
+                    msg: Box::new(NetMessage::from(msg)),
+                })
                 .await?;
             Ok(())
         }
@@ -1171,13 +1200,17 @@ fn select_forward_target(
     left_htl: usize,
     skip_forwards: &HashSet<PeerId>,
 ) -> Option<PeerKeyLocation> {
+    // Create an extended skip list that includes the joiner to prevent forwarding to the joiner
+    let mut extended_skip = skip_forwards.clone();
+    extended_skip.insert(joiner.peer.clone());
+
     if left_htl >= connection_manager.rnd_if_htl_above {
         tracing::debug!(
             tx = %id,
             joiner = %joiner.peer,
             "Randomly selecting peer to forward connect request",
         );
-        connection_manager.random_peer(|p| !skip_forwards.contains(p))
+        connection_manager.random_peer(|p| !extended_skip.contains(p))
     } else {
         tracing::debug!(
             tx = %id,
@@ -1188,7 +1221,7 @@ fn select_forward_target(
             .routing(
                 joiner.location.unwrap(),
                 Some(&request_peer.peer),
-                skip_forwards,
+                &extended_skip,
                 router,
             )
             .and_then(|pkl| (pkl.peer != joiner.peer).then_some(pkl))
diff --git a/crates/core/src/operations/put.rs b/crates/core/src/operations/put.rs
@@ -938,6 +938,14 @@ pub(crate) async fn request_put(op_manager: &OpManager, mut put_op: PutOp) -> Re
         .ring
         .closest_potentially_caching(&key, [&own_location.peer].as_slice());
 
+    tracing::debug!(
+        tx = %id,
+        %key,
+        target_found = target.is_some(),
+        target_peer = ?target.as_ref().map(|t| t.peer.to_string()),
+        "Determined PUT routing target"
+    );
+
     // No other peers found - handle locally
     if target.is_none() {
         tracing::debug!(tx = %id, %key, "No other peers available, handling put operation locally");
@@ -1017,6 +1025,16 @@ pub(crate) async fn request_put(op_manager: &OpManager, mut put_op: PutOp) -> Re
     }
 
     // At least one peer found - forward to network
+    let target_peer = target.unwrap();
+
+    tracing::debug!(
+        tx = %id,
+        %key,
+        target_peer = %target_peer.peer,
+        target_location = ?target_peer.location,
+        "Forwarding PUT to target peer"
+    );
+
     put_op.state = Some(PutState::AwaitingResponse {
         key,
         upstream: None,
@@ -1032,9 +1050,15 @@ pub(crate) async fn request_put(op_manager: &OpManager, mut put_op: PutOp) -> Re
         related_contracts,
         value,
         htl,
-        target: target.unwrap(),
+        target: target_peer,
     };
 
+    tracing::debug!(
+        tx = %id,
+        %key,
+        "Calling notify_op_change to send PUT message to network"
+    );
+
     // Use notify_op_change to trigger the operation processing
     // This will cause the operation to be processed through process_message for network propagation
     op_manager
diff --git a/crates/core/tests/connectivity.rs b/crates/core/tests/connectivity.rs
@@ -493,6 +493,9 @@ async fn test_three_node_network_connectivity() -> TestResult {
     let peer1_ws_port = peer1_ws_socket.local_addr()?.port();
     let peer2_ws_port = peer2_ws_socket.local_addr()?.port();
 
+    // Generate a single consistent location for the gateway
+    let gateway_location = RNG.lock().unwrap().random();
+
     let gateway_config = ConfigArgs {
         ws_api: WebsocketApiArgs {
             address: Some(Ipv4Addr::LOCALHOST.into()),
@@ -504,7 +507,7 @@ async fn test_three_node_network_connectivity() -> TestResult {
             is_gateway: true,
             skip_load_from_network: true,
             gateways: Some(vec![]),
-            location: Some(RNG.lock().unwrap().random()),
+            location: Some(gateway_location),
             ignore_protocol_checking: true,
             address: Some(Ipv4Addr::LOCALHOST.into()),
             network_port: Some(gateway_port),
@@ -522,10 +525,10 @@ async fn test_three_node_network_connectivity() -> TestResult {
         ..Default::default()
     };
 
-    // Gateway info for peers
+    // Gateway info for peers - use the SAME location as gateway config
     let gateway_info = InlineGwConfig {
         address: (Ipv4Addr::LOCALHOST, gateway_port).into(),
-        location: Some(RNG.lock().unwrap().random()),
+        location: Some(gateway_location),
         public_key_path: temp_dir_gw.path().join("public.pem"),
     };
 
diff --git a/crates/core/tests/ubertest.rs b/crates/core/tests/ubertest.rs

Original file line number	Diff line number	Diff line change
`@@ -334,6 +334,11 @@ pub(crate) enum NodeEvent {`
`334`	`334`	`key: ContractKey,`
`335`	`335`	`subscribed: bool,`
`336`	`336`	`},`
	`337`	`+ /// Send a message to a peer over the network`
	`338`	`+ SendMessage {`
	`339`	`+ target: PeerId,`
	`340`	`+ msg: Box<NetMessage>,`
	`341`	`+ },`
`337`	`342`	`}`
`338`	`343`
`339`	`344`	`#[derive(Debug, Clone)]`
`@@ -410,6 +415,9 @@ impl Display for NodeEvent {`
`410`	`415`	`"Local subscribe complete (tx: {tx}, key: {key}, subscribed: {subscribed})"`
`411`	`416`	`)`
`412`	`417`	`}`
	`418`	`+ NodeEvent::SendMessage { target, msg } => {`
	`419`	`+ write!(f, "SendMessage (to {target}, tx: {})", msg.id())`
	`420`	`+ }`
`413`	`421`	`}`
`414`	`422`	`}`
`415`	`423`	`}`