refactor(transport): replace handshake pipeline

Author: sanity

Cargo.lock

@@ -44,6 +44,7 @@ pav_regression = "0.6.1"
 parking_lot = "0.12"
 pin-project = "1"
 rand = { features = ["small_rng"], workspace = true }
+once_cell = "1"
 redb = { optional = true, version = "3" }
 serde = { features = ["derive", "rc"], workspace = true }
 serde_json = { workspace = true }
@@ -76,12 +77,12 @@ opentelemetry_sdk = { optional = true, version = "0.31", features = ["rt-tokio"]
 
 # internal deps
 freenet-stdlib = { features = ["net"], workspace = true }
-console-subscriber = { version = "0.5.0", optional = true }
+console-subscriber = { version = "0.4.1", optional = true }
 tokio-stream = "0.1.17"
 
 [target.'cfg(windows)'.dependencies]
 winapi = { version = "0.3", features = ["sysinfoapi"] }
-wmi = "0.18.0"
+wmi = "0.17.3"
 serde = { version = "1.0", features = ["derive"] }
 
 [dev-dependencies]

crates/core/Cargo.toml

@@ -193,6 +193,7 @@ where
 
 mod sealed_msg_type {
     use super::*;
+    use crate::operations::connect::ConnectMsg;
 
     pub trait SealedTxType {
         fn tx_type_id() -> TransactionTypeId;
@@ -301,7 +302,7 @@ impl Versioned for NetMessage {
 impl Versioned for NetMessageV1 {
     fn version(&self) -> semver::Version {
         match self {
-            NetMessageV1::Connect(_) => semver::Version::new(1, 0, 0),
+            NetMessageV1::Connect(_) => semver::Version::new(1, 1, 0),
             NetMessageV1::Put(_) => semver::Version::new(1, 0, 0),
             NetMessageV1::Get(_) => semver::Version::new(1, 0, 0),
             NetMessageV1::Subscribe(_) => semver::Version::new(1, 0, 0),
@@ -363,10 +364,9 @@ pub(crate) enum NodeEvent {
         key: ContractKey,
         subscribed: bool,
     },
-    /// Send a message to a peer over the network
-    SendMessage {
-        target: PeerId,
-        msg: Box<NetMessage>,
+    /// Register expectation for an inbound connection from the given peer.
+    ExpectPeerConnection {
+        peer: PeerId,
     },
 }
 
@@ -444,8 +444,8 @@ 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())
+            NodeEvent::ExpectPeerConnection { peer } => {
+                write!(f, "ExpectPeerConnection (from {peer})")
             }
         }
     }
@@ -486,7 +486,7 @@ impl MessageStats for NetMessageV1 {
 
     fn target(&self) -> Option<PeerKeyLocation> {
         match self {
-            NetMessageV1::Connect(op) => op.target().as_ref().map(|b| b.borrow().clone()),
+            NetMessageV1::Connect(op) => op.target().cloned(),
             NetMessageV1::Put(op) => op.target().as_ref().map(|b| b.borrow().clone()),
             NetMessageV1::Get(op) => op.target().as_ref().map(|b| b.borrow().clone()),
             NetMessageV1::Subscribe(op) => op.target().as_ref().map(|b| b.borrow().clone()),

crates/core/src/message.rs

@@ -701,7 +701,7 @@ async fn process_message_v1<CB>(
                     tx_type = %msg.id().transaction_type()
                 );
                 let op_result =
-                    handle_op_request::<connect::ConnectOp, _>(&op_manager, &mut conn_manager, op)
+                    handle_op_request::<ConnectOp, _>(&op_manager, &mut conn_manager, op)
                         .instrument(span)
                         .await;
 
@@ -861,7 +861,7 @@ where
                     tx_type = %msg.id().transaction_type()
                 );
                 let op_result =
-                    handle_op_request::<connect::ConnectOp, _>(&op_manager, &mut conn_manager, op)
+                    handle_op_request::<ConnectOp, _>(&op_manager, &mut conn_manager, op)
                         .instrument(span)
                         .await;
 
@@ -879,7 +879,6 @@ where
                     }
                 }
 
-                // Pure network result processing - no client handling
                 return handle_pure_network_result(
                     tx,
                     op_result,
@@ -1153,29 +1152,28 @@ async fn handle_aborted_op(
         // is useless without connecting to the network, we will retry with exponential backoff
         // if necessary
         match op_manager.pop(&tx) {
-            // only keep attempting to connect if the node hasn't got enough connections yet
             Ok(Some(OpEnum::Connect(op)))
                 if op.has_backoff()
                     && op_manager.ring.open_connections()
                         < op_manager.ring.connection_manager.min_connections =>
             {
-                let ConnectOp {
-                    gateway, backoff, ..
-                } = *op;
+                let gateway = op.gateway().cloned();
                 if let Some(gateway) = gateway {
                     tracing::warn!("Retry connecting to gateway {}", gateway.peer);
-                    connect::join_ring_request(backoff, &gateway, op_manager).await?;
+                    connect::join_ring_request(None, &gateway, op_manager).await?;
                 }
             }
             Ok(Some(OpEnum::Connect(_))) => {
-                // if no connections were achieved just fail
                 if op_manager.ring.open_connections() == 0 && op_manager.ring.is_gateway() {
                     tracing::warn!("Retrying joining the ring with an other gateway");
                     if let Some(gateway) = gateways.iter().shuffle().next() {
                         connect::join_ring_request(None, gateway, op_manager).await?
                     }
                 }
             }
+            Ok(Some(other)) => {
+                op_manager.push(tx, other).await?;
+            }
             _ => {}
         }
     }