fix(put): use hop-by-hop routing instead of embedded origin field (#2243)

Co-authored-by: Claude Opus 4.5 <noreply@anthropic.com>

Author: sanity

.github/workflows/ci.yml

@@ -126,7 +126,8 @@ jobs:
       - name: Test
         # Limit test threads to reduce resource contention on high-core-count runners
         # Without this, 64+ parallel tests cause timing-sensitive network tests to fail
-        run: cargo test --workspace --no-default-features --features trace,websocket,redb -- --test-threads=8
+        # --nocapture shows logs in real-time instead of buffering until test completion
+        run: cargo test --workspace --no-default-features --features trace,websocket,redb -- --test-threads=8 --nocapture
 
   six_peer_regression:
     name: six-peer-regression

crates/core/src/contract/executor.rs

@@ -462,35 +462,6 @@ impl ComposeNetworkMessage<operations::subscribe::SubscribeOp> for SubscribeCont
     }
 }
 
-#[allow(unused)]
-struct PutContract {
-    contract: ContractContainer,
-    state: WrappedState,
-    related_contracts: RelatedContracts<'static>,
-}
-
-impl ComposeNetworkMessage<operations::put::PutOp> for PutContract {
-    fn initiate_op(self, op_manager: &OpManager) -> operations::put::PutOp {
-        let PutContract {
-            contract,
-            state,
-            related_contracts,
-        } = self;
-        operations::put::start_op(
-            contract,
-            related_contracts,
-            state,
-            op_manager.ring.max_hops_to_live,
-            false,
-        )
-    }
-
-    async fn resume_op(op: operations::put::PutOp, op_manager: &OpManager) -> Result<(), OpError> {
-        operations::put::request_put(op_manager, op).await
-    }
-}
-
-#[allow(unused)]
 struct UpdateContract {
     key: ContractKey,
     new_state: WrappedState,

crates/core/src/message.rs

@@ -3,7 +3,7 @@
 //! See `architecture.md`.
 
 use std::{
-    borrow::{Borrow, Cow},
+    borrow::Cow,
     fmt::Display,
     net::SocketAddr,
     time::{Duration, SystemTime},
@@ -98,7 +98,8 @@ impl Transaction {
         self.id.0.to_le_bytes()
     }
 
-    fn elapsed(&self) -> Duration {
+    /// Returns the elapsed time since this transaction was created.
+    pub fn elapsed(&self) -> Duration {
         let current_unix_epoch_ts = SystemTime::now()
             .duration_since(SystemTime::UNIX_EPOCH)
             .expect("now should be always be later than unix epoch")
@@ -264,8 +265,6 @@ mod sealed_msg_type {
 pub(crate) trait MessageStats {
     fn id(&self) -> &Transaction;
 
-    fn target(&self) -> Option<PeerKeyLocation>;
-
     fn requested_location(&self) -> Option<Location>;
 }
 
@@ -397,8 +396,6 @@ impl From<NetMessage> for semver::Version {
 pub trait InnerMessage: Into<NetMessage> {
     fn id(&self) -> &Transaction;
 
-    fn target(&self) -> Option<impl Borrow<PeerKeyLocation>>;
-
     fn requested_location(&self) -> Option<Location>;
 }
 
@@ -551,12 +548,6 @@ impl MessageStats for NetMessage {
         }
     }
 
-    fn target(&self) -> Option<PeerKeyLocation> {
-        match self {
-            NetMessage::V1(msg) => msg.target(),
-        }
-    }
-
     fn requested_location(&self) -> Option<Location> {
         match self {
             NetMessage::V1(msg) => msg.requested_location(),
@@ -578,19 +569,6 @@ impl MessageStats for NetMessageV1 {
         }
     }
 
-    fn target(&self) -> Option<PeerKeyLocation> {
-        match self {
-            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()),
-            NetMessageV1::Update(op) => op.target().as_ref().map(|b| b.borrow().clone()),
-            NetMessageV1::Aborted(_) => None,
-            NetMessageV1::Unsubscribed { .. } => None,
-            NetMessageV1::ProximityCache { .. } => None,
-        }
-    }
-
     fn requested_location(&self) -> Option<Location> {
         match self {
             NetMessageV1::Connect(op) => op.requested_location(),

crates/core/src/node/network_bridge/p2p_protoc.rs

@@ -477,6 +477,49 @@ impl OpManager {
         Ok(())
     }
 
+    /// Peek at the next hop address for an operation without removing it.
+    /// Used by hop-by-hop routing to determine where to send initial outbound messages.
+    /// Returns None if the operation doesn't exist or doesn't have a next hop address.
+    pub fn peek_next_hop_addr(&self, id: &Transaction) -> Option<std::net::SocketAddr> {
+        if self.ops.completed.contains(id) || self.ops.under_progress.contains(id) {
+            return None;
+        }
+        match id.transaction_type() {
+            TransactionType::Connect => self
+                .ops
+                .connect
+                .get(id)
+                .and_then(|op| op.get_next_hop_addr()),
+            TransactionType::Put => self.ops.put.get(id).and_then(|op| op.get_next_hop_addr()),
+            TransactionType::Get => self.ops.get.get(id).and_then(|op| op.get_next_hop_addr()),
+            TransactionType::Subscribe => self
+                .ops
+                .subscribe
+                .get(id)
+                .and_then(|op| op.get_next_hop_addr()),
+            TransactionType::Update => self
+                .ops
+                .update
+                .get(id)
+                .and_then(|op| op.get_next_hop_addr()),
+        }
+    }
+
+    /// Peek at the full target peer (including public key) without removing the operation.
+    /// Used when establishing new connections where we need the public key for handshake.
+    pub fn peek_target_peer(&self, id: &Transaction) -> Option<PeerKeyLocation> {
+        if self.ops.completed.contains(id) || self.ops.under_progress.contains(id) {
+            return None;
+        }
+        match id.transaction_type() {
+            TransactionType::Connect => {
+                self.ops.connect.get(id).and_then(|op| op.get_target_peer())
+            }
+            // Other operations only store addresses, not full peer info
+            _ => None,
+        }
+    }
+
     pub fn pop(&self, id: &Transaction) -> Result<Option<OpEnum>, OpNotAvailable> {
         if self.ops.completed.contains(id) {
             return Err(OpNotAvailable::Completed);
@@ -653,9 +696,11 @@ impl OpManager {
     /// Notify the operation manager that a transaction is being transacted over the network.
     pub fn sending_transaction(&self, peer: &PeerKeyLocation, msg: &NetMessage) {
         let transaction = msg.id();
-        if let (Some(recipient), Some(target)) = (msg.target(), msg.requested_location()) {
+        // With hop-by-hop routing, record the request using the peer we're sending to
+        // and the message's requested location (contract location)
+        if let Some(target_loc) = msg.requested_location() {
             self.ring
-                .record_request(recipient.clone(), target, transaction.transaction_type());
+                .record_request(peer.clone(), target_loc, transaction.transaction_type());
         }
         if let Some(peer_addr) = peer.socket_addr() {
             self.ring
@@ -767,7 +812,13 @@ async fn garbage_cleanup_task<ER: NetEventRegister>(
                     } else {
                         ops.under_progress.remove(&tx);
                         ops.completed.remove(&tx);
-                        tracing::debug!("Transaction timed out: {tx}");
+                        tracing::info!(
+                            tx = %tx,
+                            tx_type = ?tx.transaction_type(),
+                            elapsed_ms = tx.elapsed().as_millis(),
+                            ttl_ms = crate::config::OPERATION_TTL.as_millis(),
+                            "Transaction timed out"
+                        );
 
                         // Check if this is a child operation and propagate timeout to parent
                         if let Some(parent_tx) = sub_op_tracker.get_parent(tx) {
@@ -812,7 +863,13 @@ async fn garbage_cleanup_task<ER: NetEventRegister>(
                         TransactionType::Update => ops.update.remove(&tx).is_some(),
                     };
                     if removed {
-                        tracing::debug!("Transaction timed out: {tx}");
+                        tracing::info!(
+                            tx = %tx,
+                            tx_type = ?tx.transaction_type(),
+                            elapsed_ms = tx.elapsed().as_millis(),
+                            ttl_ms = crate::config::OPERATION_TTL.as_millis(),
+                            "Transaction timed out"
+                        );
 
                         // Check if this is a child operation and propagate timeout to parent
                         if let Some(parent_tx) = sub_op_tracker.get_parent(tx) {