refactor: refine fire-and-forget UPDATE semantics (#2038)

Author: sanity

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

@@ -1662,7 +1662,6 @@ fn extract_sender_from_message(msg: &NetMessage) -> Option<PeerKeyLocation> {
             // Update messages have sender in some variants
             NetMessageV1::Update(update_msg) => match update_msg {
                 UpdateMsg::SeekNode { sender, .. } => Some(sender.clone()),
-                UpdateMsg::SuccessfulUpdate { sender, .. } => Some(sender.clone()),
                 UpdateMsg::Broadcasting { sender, .. } => Some(sender.clone()),
                 UpdateMsg::BroadcastTo { sender, .. } => Some(sender.clone()),
                 _ => None,

crates/core/src/operations/mod.rs

@@ -133,8 +133,21 @@ where
                 if let Some(target) = msg.target() {
                     tracing::debug!(%id, %target, "sending updated op state");
                     network_bridge.send(&target.peer, msg).await?;
+                    op_manager.push(id, updated_state).await?;
+                } else {
+                    tracing::debug!(%id, "queueing op state for local processing");
+                    debug_assert!(
+                        matches!(
+                            msg,
+                            NetMessage::V1(NetMessageV1::Update(
+                                crate::operations::update::UpdateMsg::Broadcasting { .. }
+                            ))
+                        ),
+                        "Only Update::Broadcasting messages should be re-queued locally"
+                    );
+                    op_manager.notify_op_change(msg, updated_state).await?;
+                    return Err(OpError::StatePushed);
                 }
-                op_manager.push(id, updated_state).await?;
             }
         }
 

crates/core/src/operations/put.rs

@@ -477,6 +477,40 @@ impl Operation for PutOp {
                     let sender = op_manager.ring.connection_manager.own_location();
                     let mut broadcasted_to = *broadcasted_to;
 
+                    if upstream.peer == sender.peer {
+                        // Originator reached the subscription tree. This path should be filtered
+                        // out by the deduplication layer, so treat it as a warning if it happens
+                        // to help surface potential bugs.
+                        tracing::warn!(
+                            tx = %id,
+                            %key,
+                            "PUT originator re-entered broadcast loop; dedup should have completed"
+                        );
+                        new_state = Some(PutState::Finished { key: *key });
+                    } else {
+                        // Notify the upstream hop right away so the request
+                        // path does not wait for the broadcast to finish.
+                        let ack = PutMsg::SuccessfulPut {
+                            id: *id,
+                            target: upstream.clone(),
+                            key: *key,
+                            sender: sender.clone(),
+                        };
+
+                        tracing::trace!(
+                            tx = %id,
+                            %key,
+                            upstream = %upstream.peer,
+                            "Forwarding SuccessfulPut upstream before broadcast"
+                        );
+
+                        conn_manager
+                            .send(&upstream.peer, NetMessage::from(ack))
+                            .await?;
+
+                        new_state = None;
+                    }
+
                     // Broadcast to all peers in parallel
                     let mut broadcasting = Vec::with_capacity(broadcast_to.len());
                     for peer in broadcast_to.iter() {
@@ -526,14 +560,7 @@ impl Operation for PutOp {
                         "Successfully broadcasted put into contract {key} to {broadcasted_to} peers"
                     );
 
-                    // Subscriber nodes have been notified of the change, the operation is completed
-                    return_msg = Some(PutMsg::SuccessfulPut {
-                        id: *id,
-                        target: upstream.clone(),
-                        key: *key,
-                        sender,
-                    });
-                    new_state = None;
+                    return_msg = None;
                 }
                 PutMsg::SuccessfulPut { id, .. } => {
                     match self.state {