feat: add proximity-based update forwarding

Implements proximity cache to track which neighbors have cached contracts.
UPDATE operations now forward to neighbors who have the contract cached,
not just explicit subscribers. This reduces update propagation failures
in the network.

Key changes:
- New ProximityCacheManager tracks neighbor cache states
- Immediate cache addition announcements
- Batched cache removal announcements to reduce network traffic
- UPDATE operation combines subscribers with proximity-based neighbors
- PUT/GET operations announce cache additions after seeding

Addresses #1848

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

Co-Authored-By: Claude <noreply@anthropic.com>

Author: sanity

crates/core/src/message.rs

@@ -10,7 +10,7 @@ use std::{
 
 use crate::{
     client_events::{ClientId, HostResult},
-    node::PeerId,
+    node::{proximity_cache::ProximityCacheMessage, PeerId},
     operations::{
         connect::ConnectMsg, get::GetMsg, put::PutMsg, subscribe::SubscribeMsg, update::UpdateMsg,
     },
@@ -255,6 +255,10 @@ pub(crate) enum NetMessageV1 {
     },
     Update(UpdateMsg),
     Aborted(Transaction),
+    ProximityCache {
+        from: PeerId,
+        message: ProximityCacheMessage,
+    },
 }
 
 trait Versioned {
@@ -279,6 +283,7 @@ impl Versioned for NetMessageV1 {
             NetMessageV1::Unsubscribed { .. } => semver::Version::new(1, 0, 0),
             NetMessageV1::Update(_) => semver::Version::new(1, 0, 0),
             NetMessageV1::Aborted(_) => semver::Version::new(1, 0, 0),
+            NetMessageV1::ProximityCache { .. } => semver::Version::new(1, 0, 0),
         }
     }
 }
@@ -339,6 +344,11 @@ pub(crate) enum NodeEvent {
         target: PeerId,
         msg: Box<NetMessage>,
     },
+    #[allow(dead_code)] // Reserved for future proximity cache broadcasting
+    BroadcastProximityCache {
+        from: PeerId,
+        message: crate::node::proximity_cache::ProximityCacheMessage,
+    },
 }
 
 #[derive(Debug, Clone)]
@@ -418,6 +428,12 @@ impl Display for NodeEvent {
             NodeEvent::SendMessage { target, msg } => {
                 write!(f, "SendMessage (to {target}, tx: {})", msg.id())
             }
+            NodeEvent::BroadcastProximityCache { from, message } => {
+                write!(
+                    f,
+                    "BroadcastProximityCache (from {from}, message: {message:?})"
+                )
+            }
         }
     }
 }
@@ -452,6 +468,7 @@ impl MessageStats for NetMessageV1 {
             NetMessageV1::Update(op) => op.id(),
             NetMessageV1::Aborted(tx) => tx,
             NetMessageV1::Unsubscribed { transaction, .. } => transaction,
+            NetMessageV1::ProximityCache { .. } => Transaction::NULL,
         }
     }
 
@@ -464,6 +481,7 @@ impl MessageStats for NetMessageV1 {
             NetMessageV1::Update(op) => op.target().as_ref().map(|b| b.borrow().clone()),
             NetMessageV1::Aborted(_) => None,
             NetMessageV1::Unsubscribed { .. } => None,
+            NetMessageV1::ProximityCache { .. } => None,
         }
     }
 
@@ -476,6 +494,7 @@ impl MessageStats for NetMessageV1 {
             NetMessageV1::Update(op) => op.requested_location(),
             NetMessageV1::Aborted(_) => None,
             NetMessageV1::Unsubscribed { .. } => None,
+            NetMessageV1::ProximityCache { .. } => None,
         }
     }
 }
@@ -495,6 +514,9 @@ impl Display for NetMessage {
                 Unsubscribed { key, from, .. } => {
                     write!(f, "Unsubscribed {{  key: {key}, from: {from} }}")?;
                 }
+                ProximityCache { from, message } => {
+                    write!(f, "ProximityCache {{ from: {from}, message: {message:?} }}")?;
+                }
             },
         };
         write!(f, "}}")

crates/core/src/node/mod.rs

@@ -71,6 +71,7 @@ mod message_processor;
 mod network_bridge;
 mod op_state_manager;
 mod p2p_impl;
+pub(crate) mod proximity_cache;
 mod request_router;
 pub(crate) mod testing_impl;
 

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

@@ -815,6 +815,30 @@ impl P2pConnManager {
                                     Err(e) => tracing::error!("Failed to send local subscribe response to result router: {}", e),
                                 }
                             }
+                            NodeEvent::BroadcastProximityCache { from, message } => {
+                                // Broadcast ProximityCache message to all connected peers
+                                tracing::debug!(
+                                    %from,
+                                    ?message,
+                                    peer_count = ctx.connections.len(),
+                                    "Broadcasting ProximityCache message to connected peers"
+                                );
+
+                                use crate::message::{NetMessage, NetMessageV1};
+                                let msg = NetMessage::V1(NetMessageV1::ProximityCache {
+                                    from: from.clone(),
+                                    message: message.clone(),
+                                });
+
+                                for peer in ctx.connections.keys() {
+                                    if peer != &from {
+                                        tracing::debug!(%peer, "Sending ProximityCache to peer");
+                                        if let Err(e) = ctx.bridge.send(peer, msg.clone()).await {
+                                            tracing::warn!(%peer, "Failed to send ProximityCache: {}", e);
+                                        }
+                                    }
+                                }
+                            }
                             NodeEvent::Disconnect { cause } => {
                                 tracing::info!(
                                     "Disconnecting from network{}",

crates/core/src/node/op_state_manager.rs

@@ -32,7 +32,10 @@ use crate::{
     ring::{ConnectionManager, LiveTransactionTracker, Ring},
 };
 
-use super::{network_bridge::EventLoopNotificationsSender, NetEventRegister, NodeConfig};
+use super::{
+    network_bridge::EventLoopNotificationsSender, proximity_cache::ProximityCacheManager,
+    NetEventRegister, NodeConfig,
+};
 
 #[cfg(debug_assertions)]
 macro_rules! check_id_op {
@@ -77,6 +80,8 @@ pub(crate) struct OpManager {
     pub peer_ready: Arc<AtomicBool>,
     /// Whether this node is a gateway
     pub is_gateway: bool,
+    /// Proximity cache manager for tracking which neighbors have which contracts
+    pub proximity_cache: Arc<ProximityCacheManager>,
 }
 
 impl OpManager {
@@ -126,6 +131,8 @@ impl OpManager {
             tracing::debug!("Regular peer node: peer_ready will be set after first handshake");
         }
 
+        let proximity_cache = Arc::new(ProximityCacheManager::new());
+
         Ok(Self {
             ring,
             ops,
@@ -135,6 +142,7 @@ impl OpManager {
             result_router_tx,
             peer_ready,
             is_gateway,
+            proximity_cache,
         })
     }