refactor(routing): add upstream_addr for connection-based routing

crates/core/src/client_events/mod.rs

@@ -422,7 +422,7 @@ async fn process_open_request(
                         let own_location = op_manager.ring.connection_manager.own_location();
                         let has_remote_peers = op_manager
                             .ring
-                            .closest_potentially_caching(&contract_key, &[own_location.peer][..])
+                            .closest_potentially_caching(&contract_key, &[own_location.peer()][..])
                             .is_some();
 
                         if !has_remote_peers {

crates/core/src/config/mod.rs

@@ -1257,8 +1257,11 @@ mod tests {
 
     #[tokio::test]
     async fn test_serde_config_args() {
+        // Use a unique ID to avoid conflicts with other tests or stale /tmp/freenet directories
+        let unique_id = format!("test-serde-{}", std::process::id());
         let args = ConfigArgs {
             mode: Some(OperationMode::Local),
+            id: Some(unique_id),
             ..Default::default()
         };
         let cfg = args.build().await.unwrap();

crates/core/src/message.rs

@@ -5,6 +5,7 @@
 use std::{
     borrow::{Borrow, Cow},
     fmt::Display,
+    net::SocketAddr,
     time::{Duration, SystemTime},
 };
 
@@ -267,6 +268,54 @@ pub(crate) trait MessageStats {
     fn requested_location(&self) -> Option<Location>;
 }
 
+/// Wrapper for inbound messages that carries the source address from the transport layer.
+/// This separates routing concerns from message content - the source address is determined by
+/// the network layer (from the packet), not embedded in the serialized message.
+///
+/// Generic over the message type so it can wrap:
+/// - `NetMessage` at the network layer (p2p_protoc.rs)
+/// - Specific operation messages (GetMsg, PutMsg, etc.) at the operation layer
+///
+/// Note: Currently unused but prepared for Phase 4 of #2164.
+/// Will be used to thread source addresses to operations for routing.
+#[allow(dead_code)]
+#[derive(Debug, Clone)]
+pub struct InboundMessage<M> {
+    /// The message content
+    pub msg: M,
+    /// The socket address this message was received from (from UDP packet source)
+    pub source_addr: SocketAddr,
+}
+
+#[allow(dead_code)]
+impl<M> InboundMessage<M> {
+    /// Create a new inbound message wrapper
+    pub fn new(msg: M, source_addr: SocketAddr) -> Self {
+        Self { msg, source_addr }
+    }
+
+    /// Transform the inner message while preserving source_addr
+    pub fn map<N>(self, f: impl FnOnce(M) -> N) -> InboundMessage<N> {
+        InboundMessage {
+            msg: f(self.msg),
+            source_addr: self.source_addr,
+        }
+    }
+
+    /// Get a reference to the inner message
+    pub fn inner(&self) -> &M {
+        &self.msg
+    }
+}
+
+#[allow(dead_code)]
+impl InboundMessage<NetMessage> {
+    /// Get the transaction ID from the wrapped network message
+    pub fn id(&self) -> &Transaction {
+        self.msg.id()
+    }
+}
+
 #[derive(Debug, Serialize, Deserialize, Clone)]
 pub(crate) enum NetMessage {
     V1(NetMessageV1),
@@ -333,8 +382,8 @@ type ConnectResult = Result<(PeerId, RemainingChecks), ()>;
 /// Internal node events emitted to the event loop.
 #[derive(Debug, Clone)]
 pub(crate) enum NodeEvent {
-    /// Drop the given peer connection.
-    DropConnection(PeerId),
+    /// Drop the given peer connection by socket address.
+    DropConnection(std::net::SocketAddr),
     // Try connecting to the given peer.
     ConnectPeer {
         peer: PeerId,

crates/core/src/node/mod.rs

@@ -61,7 +61,7 @@ use crate::operations::handle_op_request;
 pub(crate) use network_bridge::{ConnectionError, EventLoopNotificationsSender, NetworkBridge};
 
 use crate::topology::rate::Rate;
-use crate::transport::{TransportKeypair, TransportPublicKey};
+use crate::transport::{ObservedAddr, TransportKeypair, TransportPublicKey};
 pub(crate) use op_state_manager::{OpManager, OpNotAvailable};
 
 mod message_processor;
@@ -369,9 +369,12 @@ impl NodeConfig {
         let gateways: Vec<PeerKeyLocation> = self
             .gateways
             .iter()
-            .map(|node| PeerKeyLocation {
-                peer: node.peer_id.clone(),
-                location: Some(node.location),
+            .map(|node| {
+                PeerKeyLocation::with_location(
+                    node.peer_id.pub_key.clone(),
+                    node.peer_id.addr,
+                    node.location,
+                )
             })
             .collect();
 
@@ -617,6 +620,7 @@ pub(super) async fn process_message<CB>(
 #[allow(clippy::too_many_arguments)]
 pub(crate) async fn process_message_decoupled<CB>(
     msg: NetMessage,
+    source_addr: Option<ObservedAddr>,
     op_manager: Arc<OpManager>,
     conn_manager: CB,
     mut event_listener: Box<dyn NetEventRegister>,
@@ -631,6 +635,7 @@ pub(crate) async fn process_message_decoupled<CB>(
     // Pure network message processing - no client types involved
     let op_result = handle_pure_network_message(
         msg,
+        source_addr,
         op_manager.clone(),
         conn_manager,
         event_listener.as_mut(),
@@ -676,6 +681,7 @@ pub(crate) async fn process_message_decoupled<CB>(
 #[allow(clippy::too_many_arguments)]
 async fn handle_pure_network_message<CB>(
     msg: NetMessage,
+    source_addr: Option<ObservedAddr>,
     op_manager: Arc<OpManager>,
     conn_manager: CB,
     event_listener: &mut dyn NetEventRegister,
@@ -688,6 +694,7 @@ where
         NetMessage::V1(msg_v1) => {
             handle_pure_network_message_v1(
                 msg_v1,
+                source_addr,
                 op_manager,
                 conn_manager,
                 event_listener,
@@ -725,8 +732,9 @@ async fn process_message_v1<CB>(
                     transaction = %msg.id(),
                     tx_type = %msg.id().transaction_type()
                 );
+                // Legacy path - no source_addr available
                 let op_result =
-                    handle_op_request::<ConnectOp, _>(&op_manager, &mut conn_manager, op)
+                    handle_op_request::<ConnectOp, _>(&op_manager, &mut conn_manager, op, None)
                         .instrument(span)
                         .await;
 
@@ -741,8 +749,10 @@ async fn process_message_v1<CB>(
                 .await;
             }
             NetMessageV1::Put(ref op) => {
+                // Legacy path - no source_addr available
                 let op_result =
-                    handle_op_request::<put::PutOp, _>(&op_manager, &mut conn_manager, op).await;
+                    handle_op_request::<put::PutOp, _>(&op_manager, &mut conn_manager, op, None)
+                        .await;
 
                 if is_operation_completed(&op_result) {
                     if let Some(ref op_execution_callback) = pending_op_result {
@@ -767,8 +777,10 @@ async fn process_message_v1<CB>(
                 .await;
             }
             NetMessageV1::Get(ref op) => {
+                // Legacy path - no source_addr available
                 let op_result =
-                    handle_op_request::<get::GetOp, _>(&op_manager, &mut conn_manager, op).await;
+                    handle_op_request::<get::GetOp, _>(&op_manager, &mut conn_manager, op, None)
+                        .await;
                 if is_operation_completed(&op_result) {
                     if let Some(ref op_execution_callback) = pending_op_result {
                         let tx_id = *op.id();
@@ -789,10 +801,12 @@ async fn process_message_v1<CB>(
                 .await;
             }
             NetMessageV1::Subscribe(ref op) => {
+                // Legacy path - no source_addr available
                 let op_result = handle_op_request::<subscribe::SubscribeOp, _>(
                     &op_manager,
                     &mut conn_manager,
                     op,
+                    None,
                 )
                 .await;
                 if is_operation_completed(&op_result) {
@@ -815,9 +829,14 @@ async fn process_message_v1<CB>(
                 .await;
             }
             NetMessageV1::Update(ref op) => {
-                let op_result =
-                    handle_op_request::<update::UpdateOp, _>(&op_manager, &mut conn_manager, op)
-                        .await;
+                // Legacy path - no source_addr available
+                let op_result = handle_op_request::<update::UpdateOp, _>(
+                    &op_manager,
+                    &mut conn_manager,
+                    op,
+                    None,
+                )
+                .await;
                 if is_operation_completed(&op_result) {
                     if let Some(ref op_execution_callback) = pending_op_result {
                         let tx_id = *op.id();
@@ -857,6 +876,7 @@ async fn process_message_v1<CB>(
 #[allow(clippy::too_many_arguments)]
 async fn handle_pure_network_message_v1<CB>(
     msg: NetMessageV1,
+    source_addr: Option<ObservedAddr>,
     op_manager: Arc<OpManager>,
     mut conn_manager: CB,
     event_listener: &mut dyn NetEventRegister,
@@ -884,10 +904,14 @@ where
                     transaction = %msg.id(),
                     tx_type = %msg.id().transaction_type()
                 );
-                let op_result =
-                    handle_op_request::<ConnectOp, _>(&op_manager, &mut conn_manager, op)
-                        .instrument(span)
-                        .await;
+                let op_result = handle_op_request::<ConnectOp, _>(
+                    &op_manager,
+                    &mut conn_manager,
+                    op,
+                    source_addr,
+                )
+                .instrument(span)
+                .await;
 
                 if let Err(OpError::OpNotAvailable(state)) = &op_result {
                     match state {
@@ -916,8 +940,13 @@ where
                     tx = %op.id(),
                     "handle_pure_network_message_v1: Processing PUT message"
                 );
-                let op_result =
-                    handle_op_request::<put::PutOp, _>(&op_manager, &mut conn_manager, op).await;
+                let op_result = handle_op_request::<put::PutOp, _>(
+                    &op_manager,
+                    &mut conn_manager,
+                    op,
+                    source_addr,
+                )
+                .await;
                 tracing::debug!(
                     tx = %op.id(),
                     op_result_ok = op_result.is_ok(),
@@ -958,8 +987,13 @@ where
                 .await;
             }
             NetMessageV1::Get(ref op) => {
-                let op_result =
-                    handle_op_request::<get::GetOp, _>(&op_manager, &mut conn_manager, op).await;
+                let op_result = handle_op_request::<get::GetOp, _>(
+                    &op_manager,
+                    &mut conn_manager,
+                    op,
+                    source_addr,
+                )
+                .await;
 
                 // Handle pending operation results (network concern)
                 if is_operation_completed(&op_result) {
@@ -995,9 +1029,13 @@ where
                 .await;
             }
             NetMessageV1::Update(ref op) => {
-                let op_result =
-                    handle_op_request::<update::UpdateOp, _>(&op_manager, &mut conn_manager, op)
-                        .await;
+                let op_result = handle_op_request::<update::UpdateOp, _>(
+                    &op_manager,
+                    &mut conn_manager,
+                    op,
+                    source_addr,
+                )
+                .await;
 
                 if let Err(OpError::OpNotAvailable(state)) = &op_result {
                     match state {
@@ -1026,6 +1064,7 @@ where
                     &op_manager,
                     &mut conn_manager,
                     op,
+                    source_addr,
                 )
                 .await;
 
@@ -1174,7 +1213,7 @@ async fn handle_aborted_op(
                 {
                     let gateway = op.gateway().cloned();
                     if let Some(gateway) = gateway {
-                        tracing::warn!("Retry connecting to gateway {}", gateway.peer);
+                        tracing::warn!("Retry connecting to gateway {}", gateway.peer());
                         connect::join_ring_request(None, &gateway, op_manager).await?;
                     }
                 }

crates/core/src/node/network_bridge.rs

@@ -8,12 +8,12 @@
 //! See [`../../architecture.md`](../../architecture.md) for its interactions with event loops and other components.
 
 use std::future::Future;
+use std::net::SocketAddr;
 
 use either::Either;
 use serde::{Deserialize, Serialize};
 use tokio::sync::mpsc::{self, Receiver, Sender};
 
-use super::PeerId;
 use crate::message::{NetMessage, NodeEvent};
 
 mod handshake;
@@ -25,19 +25,29 @@ pub(crate) type ConnResult<T> = std::result::Result<T, ConnectionError>;
 
 /// Allows handling of connections to the network as well as sending messages
 /// to other peers in the network with whom connection has been established.
+///
+/// Connections are keyed by socket address since that's what identifies
+/// a network connection. The cryptographic identity is handled separately
+/// at the transport layer.
 pub(crate) trait NetworkBridge: Send + Sync {
-    fn drop_connection(&mut self, peer: &PeerId) -> impl Future<Output = ConnResult<()>> + Send;
-
-    fn send(&self, target: &PeerId, msg: NetMessage)
-        -> impl Future<Output = ConnResult<()>> + Send;
+    fn drop_connection(
+        &mut self,
+        peer_addr: SocketAddr,
+    ) -> impl Future<Output = ConnResult<()>> + Send;
+
+    fn send(
+        &self,
+        target_addr: SocketAddr,
+        msg: NetMessage,
+    ) -> impl Future<Output = ConnResult<()>> + Send;
 }
 
 #[derive(Debug, thiserror::Error, Serialize, Deserialize)]
 pub(crate) enum ConnectionError {
     #[error("location unknown for this node")]
     LocationUnknown,
-    #[error("unable to send message")]
-    SendNotCompleted(PeerId),
+    #[error("unable to send message to {0}")]
+    SendNotCompleted(SocketAddr),
     #[error("Unexpected connection req")]
     UnexpectedReq,
     #[error("error while de/serializing message")]
@@ -76,7 +86,7 @@ impl Clone for ConnectionError {
         match self {
             Self::LocationUnknown => Self::LocationUnknown,
             Self::Serialization(_) => Self::Serialization(None),
-            Self::SendNotCompleted(peer) => Self::SendNotCompleted(peer.clone()),
+            Self::SendNotCompleted(addr) => Self::SendNotCompleted(*addr),
             Self::IOError(err) => Self::IOError(err.clone()),
             Self::Timeout => Self::Timeout,
             Self::UnexpectedReq => Self::UnexpectedReq,