fix: use ObservedAddr newtype for NAT routing in seeding

Apply ObservedAddr newtype for consistent NAT routing:
- Use ObservedAddr in seeding subscriber tracking
- Update subscribe/update operations for address handling

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <[email protected]>

Author: sanity

crates/core/src/node/testing_impl/in_memory.rs

@@ -125,12 +125,12 @@ where
                 self.op_manager.ring.seed_contract(key);
             }
             if let Some(subscribers) = contract_subscribers.get(&key) {
-                // add contract subscribers
+                // add contract subscribers (test setup - no upstream_addr)
                 for subscriber in subscribers {
                     if self
                         .op_manager
                         .ring
-                        .add_subscriber(&key, subscriber.clone())
+                        .add_subscriber(&key, subscriber.clone(), None)
                         .is_err()
                     {
                         tracing::warn!("Max subscribers for contract {} reached", key);

crates/core/src/operations/subscribe.rs

@@ -11,6 +11,7 @@ use crate::{
     message::{InnerMessage, NetMessage, Transaction},
     node::{NetworkBridge, OpManager, PeerId},
     ring::{Location, PeerKeyLocation, RingError},
+    transport::ObservedAddr,
 };
 use freenet_stdlib::{
     client_api::{ContractResponse, ErrorKind, HostResponse},
@@ -274,7 +275,11 @@ async fn complete_local_subscription(
     key: ContractKey,
 ) -> Result<(), OpError> {
     let subscriber = op_manager.ring.connection_manager.own_location();
-    if let Err(err) = op_manager.ring.add_subscriber(&key, subscriber.clone()) {
+    // Local subscription - no upstream_addr needed since it's our own peer
+    if let Err(err) = op_manager
+        .ring
+        .add_subscriber(&key, subscriber.clone(), None)
+    {
         tracing::warn!(
             %key,
             tx = %id,
@@ -305,7 +310,7 @@ pub(crate) struct SubscribeOp {
     state: Option<SubscribeState>,
     /// The address we received this operation's message from.
     /// Used for connection-based routing: responses are sent back to this address.
-    upstream_addr: Option<std::net::SocketAddr>,
+    upstream_addr: Option<ObservedAddr>,
 }
 
 impl SubscribeOp {
@@ -359,11 +364,16 @@ impl Operation for SubscribeOp {
             }
             Ok(None) => {
                 // new request to subscribe to a contract, initialize the machine
+                tracing::info!(
+                    tx = %id,
+                    ?source_addr,
+                    "subscribe: load_or_init creating new op with source_addr as upstream_addr"
+                );
                 Ok(OpInitialization {
                     op: Self {
                         state: Some(SubscribeState::ReceivedRequest),
                         id,
-                        upstream_addr: source_addr, // Connection-based routing: store who sent us this request
+                        upstream_addr: source_addr.map(ObservedAddr::new), // Connection-based routing: store who sent us this request
                     },
                     source_addr,
                 })
@@ -403,20 +413,19 @@ impl Operation for SubscribeOp {
                     target: _,
                     subscriber,
                 } => {
-                    // Fill in subscriber's external address from transport layer if unknown.
-                    // This is the key step where the first recipient (gateway) determines the
-                    // subscriber's external address from the actual packet source address.
+                    // ALWAYS use the transport-level source address when available.
+                    // This is critical for NAT peers: they may embed a "known" but wrong address
+                    // (e.g., 127.0.0.1:31337 for loopback). The transport address is the only
+                    // reliable way to route responses back through the NAT.
                     let mut subscriber = subscriber.clone();
-                    if subscriber.peer_addr.is_unknown() {
-                        if let Some(addr) = source_addr {
-                            subscriber.set_addr(addr);
-                            tracing::debug!(
-                                tx = %id,
-                                %key,
-                                subscriber_addr = %addr,
-                                "subscribe: filled subscriber address from source_addr"
-                            );
-                        }
+                    if let Some(addr) = source_addr {
+                        subscriber.set_addr(addr);
+                        tracing::debug!(
+                            tx = %id,
+                            %key,
+                            subscriber_addr = %addr,
+                            "subscribe: using transport source_addr for subscriber"
+                        );
                     }
 
                     tracing::debug!(
@@ -451,9 +460,10 @@ impl Operation for SubscribeOp {
                             "subscribe: handling RequestSub locally (contract available)"
                         );
 
+                        // Use upstream_addr for NAT routing - subscriber may embed wrong address
                         if op_manager
                             .ring
-                            .add_subscriber(key, subscriber.clone())
+                            .add_subscriber(key, subscriber.clone(), self.upstream_addr)
                             .is_err()
                         {
                             tracing::warn!(
@@ -523,6 +533,13 @@ impl Operation for SubscribeOp {
                             subscribed: true,
                         };
 
+                        tracing::info!(
+                            tx = %id,
+                            %key,
+                            upstream_addr = ?self.upstream_addr,
+                            "subscribe: creating ReturnSub with upstream_addr"
+                        );
+
                         return build_op_result(
                             self.id,
                             None,
@@ -725,9 +742,10 @@ impl Operation for SubscribeOp {
                         subscribers_before = ?before_direct,
                         "subscribe: attempting to register direct subscriber"
                     );
+                    // Use upstream_addr for NAT routing - subscriber may embed wrong address
                     if op_manager
                         .ring
-                        .add_subscriber(key, subscriber.clone())
+                        .add_subscriber(key, subscriber.clone(), self.upstream_addr)
                         .is_err()
                     {
                         tracing::warn!(
@@ -875,9 +893,10 @@ impl Operation for SubscribeOp {
                                 subscribers_before = ?before_upstream,
                                 "subscribe: attempting to register upstream link"
                             );
+                            // upstream_subscriber was stored in op state, no transport address available
                             if op_manager
                                 .ring
-                                .add_subscriber(key, upstream_subscriber.clone())
+                                .add_subscriber(key, upstream_subscriber.clone(), None)
                                 .is_err()
                             {
                                 tracing::warn!(
@@ -907,7 +926,12 @@ impl Operation for SubscribeOp {
                             subscribers_before = ?before_provider,
                             "subscribe: registering provider/subscription source"
                         );
-                        if op_manager.ring.add_subscriber(key, sender.clone()).is_err() {
+                        // Use upstream_addr for NAT routing - sender may embed wrong address
+                        if op_manager
+                            .ring
+                            .add_subscriber(key, sender.clone(), self.upstream_addr)
+                            .is_err()
+                        {
                             // concurrently it reached max number of subscribers for this contract
                             tracing::debug!(
                                 tx = %id,
@@ -964,17 +988,26 @@ fn build_op_result(
     id: Transaction,
     state: Option<SubscribeState>,
     msg: Option<SubscribeMsg>,
-    upstream_addr: Option<std::net::SocketAddr>,
+    upstream_addr: Option<ObservedAddr>,
 ) -> Result<OperationResult, OpError> {
     // For response messages (ReturnSub), use upstream_addr directly for routing.
     // This is more reliable than extracting from the message's target field, which
     // may have been looked up from connection_manager (subject to race conditions).
     // For forward messages (SeekNode, RequestSub, FetchRouting), use the message's target.
     let target_addr = match &msg {
-        Some(SubscribeMsg::ReturnSub { .. }) => upstream_addr,
+        // Convert ObservedAddr to SocketAddr at the transport boundary
+        Some(SubscribeMsg::ReturnSub { .. }) => upstream_addr.map(|a| a.socket_addr()),
         _ => msg.as_ref().and_then(|m| m.target_addr()),
     };
 
+    tracing::info!(
+        tx = %id,
+        msg_type = ?msg.as_ref().map(|m| std::any::type_name_of_val(m)),
+        ?upstream_addr,
+        ?target_addr,
+        "build_op_result: computed target_addr"
+    );
+
     let output_op = state.map(|state| SubscribeOp {
         id,
         state: Some(state),

crates/core/src/operations/update.rs

@@ -1005,10 +1005,10 @@ pub(crate) async fn request_update(
             .closest_potentially_caching(&key, [sender.peer().clone()].as_slice());
 
         if let Some(target) = remote_target {
-            // Subscribe to the contract
+            // Subscribe to the contract - sender is ourselves, no upstream_addr needed
             op_manager
                 .ring
-                .add_subscriber(&key, sender.clone())
+                .add_subscriber(&key, sender.clone(), None)
                 .map_err(|_| RingError::NoCachingPeers(key))?;
 
             target

crates/core/src/ring/mod.rs

@@ -20,7 +20,7 @@ use crate::topology::rate::Rate;
 use crate::topology::TopologyAdjustment;
 use crate::tracing::{NetEventLog, NetEventRegister};
 
-use crate::transport::TransportPublicKey;
+use crate::transport::{ObservedAddr, TransportPublicKey};
 use crate::util::Contains;
 use crate::{
     config::GlobalExecutor,
@@ -338,12 +338,18 @@ impl Ring {
     }
 
     /// Will return an error in case the max number of subscribers has been added.
+    ///
+    /// The `upstream_addr` parameter is the transport-level address from which the subscribe
+    /// message was received. This is used instead of the address embedded in `subscriber`
+    /// because NAT peers may embed incorrect addresses in their messages.
     pub fn add_subscriber(
         &self,
         contract: &ContractKey,
         subscriber: PeerKeyLocation,
+        upstream_addr: Option<ObservedAddr>,
     ) -> Result<(), ()> {
-        self.seeding_manager.add_subscriber(contract, subscriber)
+        self.seeding_manager
+            .add_subscriber(contract, subscriber, upstream_addr)
     }
 
     /// Remove a subscriber by peer ID from a specific contract

crates/core/src/ring/seeding.rs

@@ -1,4 +1,5 @@
 use super::{Location, PeerKeyLocation, Score};
+use crate::transport::ObservedAddr;
 use dashmap::{mapref::one::Ref as DmRef, DashMap};
 use freenet_stdlib::prelude::ContractKey;
 use tracing::{info, warn};
@@ -102,11 +103,23 @@ impl SeedingManager {
     }
 
     /// Will return an error in case the max number of subscribers has been added.
+    ///
+    /// The `upstream_addr` parameter is the transport-level address from which the subscribe
+    /// message was received. This is used instead of the address embedded in `subscriber`
+    /// because NAT peers may embed incorrect (e.g., loopback) addresses in their messages.
+    /// The transport address is the only reliable way to route back to them.
     pub fn add_subscriber(
         &self,
         contract: &ContractKey,
         subscriber: PeerKeyLocation,
+        upstream_addr: Option<ObservedAddr>,
     ) -> Result<(), ()> {
+        // Use the transport-level address if available, otherwise fall back to the embedded address
+        let subscriber = if let Some(addr) = upstream_addr {
+            PeerKeyLocation::new(subscriber.pub_key.clone(), addr.socket_addr())
+        } else {
+            subscriber
+        };
         let mut subs = self
             .subscribers
             .entry(*contract)
@@ -255,15 +268,15 @@ mod tests {
             Location::try_from(0.3).unwrap(),
         );
 
-        // Add subscribers
+        // Add subscribers (test setup - no upstream_addr)
         assert!(seeding_manager
-            .add_subscriber(&contract_key, peer_loc1.clone())
+            .add_subscriber(&contract_key, peer_loc1.clone(), None)
             .is_ok());
         assert!(seeding_manager
-            .add_subscriber(&contract_key, peer_loc2.clone())
+            .add_subscriber(&contract_key, peer_loc2.clone(), None)
             .is_ok());
         assert!(seeding_manager
-            .add_subscriber(&contract_key, peer_loc3.clone())
+            .add_subscriber(&contract_key, peer_loc3.clone(), None)
             .is_ok());
 
         // Verify all subscribers are present

crates/core/src/transport/mod.rs

@@ -33,6 +33,40 @@ pub(crate) use self::{
     peer_connection::PeerConnection,
 };
 
+/// Address observed at the transport layer (from UDP packet source).
+///
+/// This is the "ground truth" for NAT scenarios - it's the actual address we see
+/// at the network layer, not what the peer claims in protocol messages.
+///
+/// Using a newtype instead of raw `SocketAddr` makes the address semantics explicit
+/// and prevents accidental confusion with advertised/claimed addresses.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub struct ObservedAddr(SocketAddr);
+
+impl ObservedAddr {
+    /// Create a new observed address from a socket address.
+    pub fn new(addr: SocketAddr) -> Self {
+        Self(addr)
+    }
+
+    /// Get the underlying socket address.
+    pub fn socket_addr(&self) -> SocketAddr {
+        self.0
+    }
+}
+
+impl std::fmt::Display for ObservedAddr {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{}", self.0)
+    }
+}
+
+impl From<SocketAddr> for ObservedAddr {
+    fn from(addr: SocketAddr) -> Self {
+        Self(addr)
+    }
+}
+
 #[derive(Debug, thiserror::Error)]
 pub(crate) enum TransportError {
     #[error("transport handler channel closed, socket likely closed")]