checkpoint

veilid-core/src/rpc_processor/coders/operations/answer.rs

@@ -14,6 +14,9 @@ impl RPCAnswer {
     pub fn detail(&self) -> &RPCAnswerDetail {
         &self.detail
     }
+    pub fn into_detail(self) -> RPCAnswerDetail {
+        self.detail
+    }
     pub fn desc(&self) -> &'static str {
         self.detail.desc()
     }

veilid-core/src/rpc_processor/coders/operations/operation.rs

@@ -93,6 +93,10 @@ impl RPCOperation {
         &self.kind
     }
 
+    pub fn into_kind(&self) -> RPCOperationKind {
+        self.kind
+    }
+
     pub fn decode(
         operation_reader: &veilid_capnp::operation::Reader,
         sender_node_id: &DHTKey,

veilid-core/src/rpc_processor/coders/operations/operation_cancel_tunnel.rs

@@ -3,7 +3,7 @@ use rpc_processor::*;
 
 #[derive(Debug, Clone)]
 pub struct RPCOperationCancelTunnelQ {
-    id: TunnelId,
+    pub id: TunnelId,
 }
 
 impl RPCOperationCancelTunnelQ {

veilid-core/src/rpc_processor/coders/operations/operation_complete_tunnel.rs

@@ -3,10 +3,10 @@ use rpc_processor::*;
 
 #[derive(Debug, Clone)]
 pub struct RPCOperationCompleteTunnelQ {
-    id: TunnelId,
+    pub id: TunnelId,
-    local_mode: TunnelMode,
+    pub local_mode: TunnelMode,
-    depth: u8,
+    pub depth: u8,
-    endpoint: TunnelEndpoint,
+    pub endpoint: TunnelEndpoint,
 }
 
 impl RPCOperationCompleteTunnelQ {

veilid-core/src/rpc_processor/coders/operations/operation_find_block.rs

@@ -3,7 +3,7 @@ use rpc_processor::*;
 
 #[derive(Debug, Clone)]
 pub struct RPCOperationFindBlockQ {
-    block_id: DHTKey,
+    pub block_id: DHTKey,
 }
 
 impl RPCOperationFindBlockQ {
@@ -28,9 +28,9 @@ impl RPCOperationFindBlockQ {
 
 #[derive(Debug, Clone)]
 pub struct RPCOperationFindBlockA {
-    data: Vec<u8>,
+    pub data: Vec<u8>,
-    suppliers: Vec<PeerInfo>,
+    pub suppliers: Vec<PeerInfo>,
-    peers: Vec<PeerInfo>,
+    pub peers: Vec<PeerInfo>,
 }
 
 impl RPCOperationFindBlockA {

veilid-core/src/rpc_processor/coders/operations/operation_find_node.rs

@@ -3,7 +3,7 @@ use rpc_processor::*;
 
 #[derive(Debug, Clone)]
 pub struct RPCOperationFindNodeQ {
-    node_id: DHTKey,
+    pub node_id: DHTKey,
 }
 
 impl RPCOperationFindNodeQ {
@@ -26,7 +26,7 @@ impl RPCOperationFindNodeQ {
 
 #[derive(Debug, Clone)]
 pub struct RPCOperationFindNodeA {
-    peers: Vec<PeerInfo>,
+    pub peers: Vec<PeerInfo>,
 }
 
 impl RPCOperationFindNodeA {

veilid-core/src/rpc_processor/coders/operations/operation_get_value.rs

@@ -3,7 +3,7 @@ use rpc_processor::*;
 
 #[derive(Debug, Clone)]
 pub struct RPCOperationGetValueQ {
-    key: ValueKey,
+    pub key: ValueKey,
 }
 
 impl RPCOperationGetValueQ {

veilid-core/src/rpc_processor/coders/operations/operation_node_info_update.rs

@@ -3,7 +3,7 @@ use rpc_processor::*;
 
 #[derive(Debug, Clone)]
 pub struct RPCOperationNodeInfoUpdate {
-    signed_node_info: SignedNodeInfo,
+    pub signed_node_info: SignedNodeInfo,
 }
 
 impl RPCOperationNodeInfoUpdate {

veilid-core/src/rpc_processor/coders/operations/operation_return_receipt.rs

@@ -3,7 +3,7 @@ use rpc_processor::*;
 
 #[derive(Debug, Clone)]
 pub struct RPCOperationReturnReceipt {
-    receipt: Vec<u8>,
+    pub receipt: Vec<u8>,
 }
 
 impl RPCOperationReturnReceipt {

veilid-core/src/rpc_processor/coders/operations/operation_route.rs

@@ -3,9 +3,9 @@ use rpc_processor::*;
 
 #[derive(Debug, Clone)]
 struct RoutedOperation {
-    signatures: Vec<DHTSignature>,
+    pub signatures: Vec<DHTSignature>,
-    nonce: Nonce,
+    pub nonce: Nonce,
-    data: Vec<u8>,
+    pub data: Vec<u8>,
 }
 
 impl RoutedOperation {
@@ -62,8 +62,8 @@ impl RoutedOperation {
 
 #[derive(Debug, Clone)]
 pub struct RPCOperationRoute {
-    safety_route: SafetyRoute,
+    pub safety_route: SafetyRoute,
-    operation: RoutedOperation,
+    pub operation: RoutedOperation,
 }
 
 impl RPCOperationRoute {

veilid-core/src/rpc_processor/coders/operations/operation_set_value.rs

@@ -3,8 +3,8 @@ use rpc_processor::*;
 
 #[derive(Debug, Clone)]
 pub struct RPCOperationSetValueQ {
-    key: ValueKey,
+    pub key: ValueKey,
-    value: ValueData,
+    pub value: ValueData,
 }
 
 impl RPCOperationSetValueQ {

veilid-core/src/rpc_processor/coders/operations/operation_signal.rs

@@ -3,7 +3,7 @@ use rpc_processor::*;
 
 #[derive(Debug, Clone)]
 pub struct RPCOperationSignal {
-    signal_info: SignalInfo,
+    pub signal_info: SignalInfo,
 }
 
 impl RPCOperationSignal {

veilid-core/src/rpc_processor/coders/operations/operation_start_tunnel.rs

@@ -3,9 +3,9 @@ use rpc_processor::*;
 
 #[derive(Debug, Clone)]
 pub struct RPCOperationStartTunnelQ {
-    id: TunnelId,
+    pub id: TunnelId,
-    local_mode: TunnelMode,
+    pub local_mode: TunnelMode,
-    depth: u8,
+    pub depth: u8,
 }
 
 impl RPCOperationStartTunnelQ {

veilid-core/src/rpc_processor/coders/operations/operation_status.rs

@@ -3,7 +3,7 @@ use rpc_processor::*;
 
 #[derive(Debug, Clone)]
 pub struct RPCOperationStatusQ {
-    node_status: NodeStatus,
+    pub node_status: NodeStatus,
 }
 
 impl RPCOperationStatusQ {
@@ -26,8 +26,8 @@ impl RPCOperationStatusQ {
 
 #[derive(Debug, Clone)]
 pub struct RPCOperationStatusA {
-    node_status: NodeStatus,
+    pub node_status: NodeStatus,
-    sender_info: SenderInfo,
+    pub sender_info: SenderInfo,
 }
 
 impl RPCOperationStatusA {

veilid-core/src/rpc_processor/coders/operations/operation_supply_block.rs

@@ -3,7 +3,7 @@ use rpc_processor::*;
 
 #[derive(Debug, Clone)]
 pub struct RPCOperationSupplyBlockQ {
-    block_id: DHTKey,
+    pub block_id: DHTKey,
 }
 
 impl RPCOperationSupplyBlockQ {

veilid-core/src/rpc_processor/coders/operations/operation_validate_dial_info.rs

@@ -3,9 +3,9 @@ use rpc_processor::*;
 
 #[derive(Debug, Clone)]
 pub struct RPCOperationValidateDialInfo {
-    dial_info: DialInfo,
+    pub dial_info: DialInfo,
-    receipt: Vec<u8>,
+    pub receipt: Vec<u8>,
-    redirect: bool,
+    pub redirect: bool,
 }
 
 impl RPCOperationValidateDialInfo {

veilid-core/src/rpc_processor/coders/operations/operation_value_changed.rs

@@ -3,8 +3,8 @@ use rpc_processor::*;
 
 #[derive(Debug, Clone)]
 pub struct RPCOperationValueChanged {
-    key: ValueKey,
+    pub key: ValueKey,
-    value: ValueData,
+    pub value: ValueData,
 }
 
 impl RPCOperationValueChanged {

veilid-core/src/rpc_processor/coders/operations/operation_watch_value.rs

@@ -3,7 +3,7 @@ use rpc_processor::*;
 
 #[derive(Debug, Clone)]
 pub struct RPCOperationWatchValueQ {
-    key: ValueKey,
+    pub key: ValueKey,
 }
 
 impl RPCOperationWatchValueQ {
@@ -26,8 +26,8 @@ impl RPCOperationWatchValueQ {
 
 #[derive(Debug, Clone)]
 pub struct RPCOperationWatchValueA {
-    expiration: u64,
+    pub expiration: u64,
-    peers: Vec<PeerInfo>,
+    pub peers: Vec<PeerInfo>,
 }
 
 impl RPCOperationWatchValueA {

veilid-core/src/rpc_processor/coders/operations/question.rs

@@ -18,6 +18,12 @@ impl RPCQuestion {
     pub fn detail(&self) -> &RPCQuestionDetail {
         &self.detail
     }
+    pub fn into_detail(self) -> RPCQuestionDetail {
+        self.detail
+    }
+    pub fn into_respond_to_detail(self) -> (RespondTo, RPCQuestionDetail) {
+        (self.respond_to, self.detail)
+    }
     pub fn desc(&self) -> &'static str {
         self.detail.desc()
     }

veilid-core/src/rpc_processor/coders/operations/statement.rs

@@ -14,6 +14,9 @@ impl RPCStatement {
     pub fn detail(&self) -> &RPCStatementDetail {
         &self.detail
     }
+    pub fn into_detail(self) -> RPCQuestionDetail {
+        self.detail
+    }
     pub fn desc(&self) -> &'static str {
         self.detail.desc()
     }

veilid-core/src/rpc_processor/coders/tunnel.rs

@@ -47,7 +47,10 @@ pub fn decode_tunnel_endpoint(
     reader: &veilid_capnp::tunnel_endpoint::Reader,
 ) -> Result<TunnelEndpoint, RPCError> {
     let mode = decode_tunnel_mode(reader.get_mode().map_err(map_error_capnp_notinschema!())?);
-    let description = reader.get_description();
+    let description = reader
+        .get_description()
+        .map_err(map_error_capnp_error!())?
+        .to_owned();
 
     Ok(TunnelEndpoint { mode, description })
 }

veilid-core/src/rpc_processor/rpc_cancel_tunnel.rs

@@ -0,0 +1,7 @@
+use super::*;
+
+impl RPCProcessor {
+    pub(crate) async fn process_cancel_tunnel_q(&self, msg: RPCMessage) -> Result<(), RPCError> {
+        Err(rpc_error_unimplemented("process_cancel_tunnel_q"))
+    }
+}

veilid-core/src/rpc_processor/rpc_complete_tunnel.rs

@@ -0,0 +1,7 @@
+use super::*;
+
+impl RPCProcessor {
+    pub(crate) async fn process_complete_tunnel_q(&self, msg: RPCMessage) -> Result<(), RPCError> {
+        Err(rpc_error_unimplemented("process_complete_tunnel_q"))
+    }
+}

veilid-core/src/rpc_processor/rpc_find_block.rs

@@ -0,0 +1,7 @@
+use super::*;
+
+impl RPCProcessor {
+    pub(crate) async fn process_find_block_q(&self, msg: RPCMessage) -> Result<(), RPCError> {
+        Err(rpc_error_unimplemented("process_find_block_q"))
+    }
+}

veilid-core/src/rpc_processor/rpc_find_node.rs

@@ -0,0 +1,148 @@
+use super::*;
+
+impl RPCProcessor {
+    // Send FindNodeQ RPC request, receive FindNodeA answer
+    // Can be sent via all methods including relays and routes
+    pub async fn rpc_call_find_node(
+        self,
+        dest: Destination,
+        key: DHTKey,
+        safety_route: Option<&SafetyRouteSpec>,
+        respond_to: RespondTo,
+    ) -> Result<FindNodeAnswer, RPCError> {
+        let find_node_q_msg = {
+            let mut find_node_q_msg = ::capnp::message::Builder::new_default();
+            let mut question = find_node_q_msg.init_root::<veilid_capnp::operation::Builder>();
+            question.set_op_id(self.get_next_op_id());
+            let mut respond_to_builder = question.reborrow().init_respond_to();
+            respond_to.encode(&mut respond_to_builder)?;
+            let detail = question.reborrow().init_detail();
+            let mut fnq = detail.init_find_node_q();
+            let mut node_id_builder = fnq.reborrow().init_node_id();
+            encode_public_key(&key, &mut node_id_builder)?;
+
+            find_node_q_msg.into_reader()
+        };
+
+        // Send the find_node request
+        let waitable_reply = self
+            .request(dest, find_node_q_msg, safety_route)
+            .await?
+            .unwrap();
+
+        // Wait for reply
+        let (rpcreader, latency) = self.wait_for_reply(waitable_reply).await?;
+
+        let response_operation = rpcreader
+            .reader
+            .get_root::<veilid_capnp::operation::Reader>()
+            .map_err(map_error_capnp_error!())
+            .map_err(logthru_rpc!())?;
+        let find_node_a = match response_operation
+            .get_detail()
+            .which()
+            .map_err(map_error_capnp_notinschema!())
+            .map_err(logthru_rpc!())?
+        {
+            veilid_capnp::operation::detail::FindNodeA(a) => {
+                a.map_err(map_error_internal!("Invalid FindNodeA"))?
+            }
+            _ => return Err(rpc_error_internal("Incorrect RPC answer for question")),
+        };
+
+        let peers_reader = find_node_a
+            .get_peers()
+            .map_err(map_error_internal!("Missing peers"))?;
+        let mut peers = Vec::<PeerInfo>::with_capacity(
+            peers_reader
+                .len()
+                .try_into()
+                .map_err(map_error_internal!("too many peers"))?,
+        );
+        for p in peers_reader.iter() {
+            let peer_info = decode_peer_info(&p, true)?;
+
+            if !self.filter_peer_scope(&peer_info.signed_node_info.node_info) {
+                return Err(rpc_error_invalid_format(
+                    "find_node response has invalid peer scope",
+                ));
+            }
+
+            peers.push(peer_info);
+        }
+
+        let out = FindNodeAnswer { latency, peers };
+
+        Ok(out)
+    }
+
+    pub(crate) async fn process_find_node_q(&self, rpcreader: RPCMessage) -> Result<(), RPCError> {
+        //
+        let reply_msg = {
+            let operation = rpcreader
+                .reader
+                .get_root::<veilid_capnp::operation::Reader>()
+                .map_err(map_error_capnp_error!())
+                .map_err(logthru_rpc!())?;
+
+            // find_node must always want an answer
+            if !self.wants_answer(&operation)? {
+                return Err(rpc_error_invalid_format("find_node_q should want answer"));
+            }
+
+            // get findNodeQ reader
+            let fnq_reader = match operation.get_detail().which() {
+                Ok(veilid_capnp::operation::detail::Which::FindNodeQ(Ok(x))) => x,
+                _ => panic!("invalid operation type in process_find_node_q"),
+            };
+
+            // get the node id we want to look up
+            let target_node_id = decode_public_key(
+                &fnq_reader
+                    .get_node_id()
+                    .map_err(map_error_capnp_error!())
+                    .map_err(logthru_rpc!())?,
+            );
+
+            // add node information for the requesting node to our routing table
+            let routing_table = self.routing_table();
+
+            // find N nodes closest to the target node in our routing table
+            let own_peer_info = routing_table.get_own_peer_info();
+            let own_peer_info_is_valid = own_peer_info.signed_node_info.is_valid();
+
+            let closest_nodes = routing_table.find_closest_nodes(
+                target_node_id,
+                // filter
+                Some(move |_k, v| {
+                    RoutingTable::filter_has_valid_signed_node_info(v, own_peer_info_is_valid)
+                }),
+                // transform
+                move |k, v| RoutingTable::transform_to_peer_info(k, v, &own_peer_info),
+            );
+            log_rpc!(">>>> Returning {} closest peers", closest_nodes.len());
+
+            // Send find_node answer
+            let mut reply_msg = ::capnp::message::Builder::new_default();
+            let mut answer = reply_msg.init_root::<veilid_capnp::operation::Builder>();
+            answer.set_op_id(operation.get_op_id());
+            let mut respond_to = answer.reborrow().init_respond_to();
+            respond_to.set_none(());
+            let detail = answer.reborrow().init_detail();
+            let fna = detail.init_find_node_a();
+            let mut peers_builder = fna.init_peers(
+                closest_nodes
+                    .len()
+                    .try_into()
+                    .map_err(map_error_internal!("invalid closest nodes list length"))?,
+            );
+            for (i, closest_node) in closest_nodes.iter().enumerate() {
+                let mut pi_builder = peers_builder.reborrow().get(i as u32);
+                encode_peer_info(closest_node, &mut pi_builder)?;
+            }
+            reply_msg.into_reader()
+        };
+
+        self.reply(rpcreader, reply_msg, None).await
+    }
+}

veilid-core/src/rpc_processor/rpc_get_value.rs

@@ -0,0 +1,7 @@
+use super::*;
+
+impl RPCProcessor {
+    pub(crate) async fn process_get_value_q(&self, msg: RPCMessage) -> Result<(), RPCError> {
+        Err(rpc_error_unimplemented("process_get_value_q"))
+    }
+}

veilid-core/src/rpc_processor/rpc_node_info_update.rs

@@ -0,0 +1,78 @@
+use super::*;
+
+impl RPCProcessor {
+    // Sends a our node info to another node
+    // Can be sent via all methods including relays and routes
+    pub async fn rpc_call_node_info_update(
+        &self,
+        dest: Destination,
+        safety_route: Option<&SafetyRouteSpec>,
+    ) -> Result<(), RPCError> {
+        let sni_msg = {
+            let mut sni_msg = ::capnp::message::Builder::new_default();
+            let mut question = sni_msg.init_root::<veilid_capnp::operation::Builder>();
+            question.set_op_id(self.get_next_op_id());
+            let mut respond_to = question.reborrow().init_respond_to();
+            respond_to.set_none(());
+            let detail = question.reborrow().init_detail();
+            let niu_builder = detail.init_node_info_update();
+            let mut sni_builder = niu_builder.init_signed_node_info();
+            let sni = self.routing_table().get_own_signed_node_info();
+            encode_signed_node_info(&sni, &mut sni_builder)?;
+
+            sni_msg.into_reader()
+        };
+
+        // Send the node_info_update request
+        self.request(dest, sni_msg, safety_route).await?;
+
+        Ok(())
+    }
+
+    pub(crate) async fn process_node_info_update(
+        &self,
+        rpcreader: RPCMessage,
+    ) -> Result<(), RPCError> {
+        //
+        let sender_node_id = rpcreader.header.envelope.get_sender_id();
+        let signed_node_info = {
+            let operation = rpcreader
+                .reader
+                .get_root::<veilid_capnp::operation::Reader>()
+                .map_err(map_error_capnp_error!())
+                .map_err(logthru_rpc!())?;
+
+            // This should never want an answer
+            if self.wants_answer(&operation)? {
+                return Err(rpc_error_invalid_format(
+                    "node_info_update should not want answer",
+                ));
+            }
+
+            // get nodeInfoUpdate reader
+            let niumsg_reader = match operation.get_detail().which() {
+                Ok(veilid_capnp::operation::detail::Which::NodeInfoUpdate(Ok(x))) => x,
+                _ => panic!("invalid operation type in process_node_info_update"),
+            };
+
+            // Parse out fields
+            let sni_reader = niumsg_reader
+                .get_signed_node_info()
+                .map_err(map_error_internal!("no valid signed node info"))?;
+            decode_signed_node_info(&sni_reader, &sender_node_id, true)?
+        };
+
+        // Update our routing table with signed node info
+        if !self.filter_peer_scope(&signed_node_info.node_info) {
+            return Err(rpc_error_invalid_format(
+                "node_info_update has invalid peer scope",
+            ));
+        }
+        let _ = self
+            .routing_table()
+            .register_node_with_signed_node_info(sender_node_id, signed_node_info)
+            .map_err(RPCError::Internal)?;
+
+        Ok(())
+    }
+}

veilid-core/src/rpc_processor/rpc_return_receipt.rs

@@ -0,0 +1,76 @@
+use super::*;
+
+impl RPCProcessor {
+    // Sends a unidirectional in-band return receipt
+    // Can be sent via all methods including relays and routes
+    pub async fn rpc_call_return_receipt<D: AsRef<[u8]>>(
+        &self,
+        dest: Destination,
+        safety_route: Option<&SafetyRouteSpec>,
+        receipt: D,
+    ) -> Result<(), RPCError> {
+        let receipt = receipt.as_ref();
+
+        let rr_msg = {
+            let mut rr_msg = ::capnp::message::Builder::new_default();
+            let mut question = rr_msg.init_root::<veilid_capnp::operation::Builder>();
+            question.set_op_id(self.get_next_op_id());
+            let mut respond_to = question.reborrow().init_respond_to();
+            respond_to.set_none(());
+            let detail = question.reborrow().init_detail();
+            let rr_builder = detail.init_return_receipt();
+            let r_builder = rr_builder.init_receipt(receipt.len().try_into().map_err(
+                map_error_protocol!("invalid receipt length in return receipt"),
+            )?);
+            r_builder.copy_from_slice(receipt);
+
+            rr_msg.into_reader()
+        };
+
+        // Send the return receipt request
+        self.request(dest, rr_msg, safety_route).await?;
+
+        Ok(())
+    }
+
+    pub(crate) async fn process_return_receipt(
+        &self,
+        rpcreader: RPCMessage,
+    ) -> Result<(), RPCError> {
+        let receipt = {
+            let operation = rpcreader
+                .reader
+                .get_root::<veilid_capnp::operation::Reader>()
+                .map_err(map_error_capnp_error!())
+                .map_err(logthru_rpc!())?;
+
+            // This should never want an answer
+            if self.wants_answer(&operation)? {
+                return Err(rpc_error_invalid_format(
+                    "return receipt should not want answer",
+                ));
+            }
+
+            // get returnReceipt reader
+            let rr_reader = match operation.get_detail().which() {
+                Ok(veilid_capnp::operation::detail::Which::ReturnReceipt(Ok(x))) => x,
+                _ => panic!("invalid operation type in process_return_receipt"),
+            };
+
+            // Get receipt
+            rr_reader
+                .get_receipt()
+                .map_err(map_error_internal!(
+                    "no valid receipt in process_return_receipt"
+                ))?
+                .to_vec()
+        };
+
+        // Handle it
+        let network_manager = self.network_manager();
+        network_manager
+            .handle_in_band_receipt(receipt, rpcreader.header.peer_noderef)
+            .await
+            .map_err(map_error_string!())
+    }
+}

veilid-core/src/rpc_processor/rpc_route.rs

@@ -0,0 +1,10 @@
+use super::*;
+
+impl RPCProcessor {
+    // xxx do not process latency for routed messages
+
+    pub(crate) async fn process_route(&self, _rpcreader: RPCMessage) -> Result<(), RPCError> {
+        // xxx do not process latency for routed messages
+        Err(rpc_error_unimplemented("process_route"))
+    }
+}

veilid-core/src/rpc_processor/rpc_set_value.rs

@@ -0,0 +1,7 @@
+use super::*;
+
+impl RPCProcessor {
+    pub(crate) async fn process_set_value_q(&self, msg: RPCMessage) -> Result<(), RPCError> {
+        Err(rpc_error_unimplemented("process_set_value_q"))
+    }
+}

veilid-core/src/rpc_processor/rpc_signal.rs

@@ -0,0 +1,61 @@
+use super::*;
+
+impl RPCProcessor {
+    // Sends a unidirectional signal to a node
+    // Can be sent via all methods including relays and routes
+    pub async fn rpc_call_signal(
+        &self,
+        dest: Destination,
+        safety_route: Option<&SafetyRouteSpec>,
+        signal_info: SignalInfo,
+    ) -> Result<(), RPCError> {
+        let sig_msg = {
+            let mut sig_msg = ::capnp::message::Builder::new_default();
+            let mut question = sig_msg.init_root::<veilid_capnp::operation::Builder>();
+            question.set_op_id(self.get_next_op_id());
+            let mut respond_to = question.reborrow().init_respond_to();
+            respond_to.set_none(());
+            let detail = question.reborrow().init_detail();
+            let mut sig_builder = detail.init_signal();
+            encode_signal_info(&signal_info, &mut sig_builder)?;
+
+            sig_msg.into_reader()
+        };
+
+        // Send the signal request
+        self.request(dest, sig_msg, safety_route).await?;
+
+        Ok(())
+    }
+
+    pub(crate) async fn process_signal(&self, rpcreader: RPCMessage) -> Result<(), RPCError> {
+        let signal_info = {
+            let operation = rpcreader
+                .reader
+                .get_root::<veilid_capnp::operation::Reader>()
+                .map_err(map_error_capnp_error!())
+                .map_err(logthru_rpc!())?;
+
+            // This should never want an answer
+            if self.wants_answer(&operation)? {
+                return Err(rpc_error_invalid_format("signal should not want answer"));
+            }
+
+            // get signal reader
+            let sig_reader = match operation.get_detail().which() {
+                Ok(veilid_capnp::operation::detail::Which::Signal(Ok(x))) => x,
+                _ => panic!("invalid operation type in process_signal"),
+            };
+
+            // Get signal info
+            decode_signal_info(&sig_reader)?
+        };
+
+        // Handle it
+        let network_manager = self.network_manager();
+        network_manager
+            .handle_signal(signal_info)
+            .await
+            .map_err(map_error_string!())
+    }
+}

veilid-core/src/rpc_processor/rpc_start_tunnel.rs

@@ -0,0 +1,7 @@
+use super::*;
+
+impl RPCProcessor {
+    pub(crate) async fn process_start_tunnel_q(&self, msg: RPCMessage) -> Result<(), RPCError> {
+        Err(rpc_error_unimplemented("process_start_tunnel_q"))
+    }
+}

veilid-core/src/rpc_processor/rpc_status.rs

@@ -0,0 +1,98 @@
+use super::*;
+
+impl RPCProcessor {
+    // Send StatusQ RPC request, receive StatusA answer
+    // Can be sent via relays, but not via routes
+    pub async fn rpc_call_status(
+        self,
+        peer: NodeRef,
+    ) -> Result<Answer<RPCOperationStatusA>, RPCError> {
+        let node_status = self.network_manager().generate_node_status();
+        let status_q = RPCOperationStatusQ { node_status };
+        let respond_to = self.make_respond_to_sender(peer.clone());
+        let question = RPCQuestion::new(respond_to, RPCQuestionDetail::StatusQ(status_q));
+
+        // Send the info request
+        let waitable_reply = self
+            .question(Destination::Direct(peer.clone()), question, None)
+            .await?;
+
+        // Note what kind of ping this was and to what peer scope
+        let send_data_kind = waitable_reply.send_data_kind;
+
+        // Wait for reply
+        let (msg, latency) = self.wait_for_reply(waitable_reply).await?;
+
+        // Get the right answer type
+        let status_a = match msg.operation.into_kind() {
+            RPCOperationKind::Answer(a) => match a.into_detail() {
+                RPCAnswerDetail::StatusA(a) => a,
+                _ => return Err(rpc_error_invalid_format("not a status answer")),
+            },
+            _ => return Err(rpc_error_invalid_format("not an answer")),
+        };
+
+        // Update latest node status in routing table
+        peer.operate_mut(|e| {
+            e.update_node_status(status_a.node_status.clone());
+        });
+
+        // Report sender_info IP addresses to network manager
+        if let Some(socket_address) = status_a.sender_info.socket_address {
+            match send_data_kind {
+                SendDataKind::LocalDirect => {
+                    self.network_manager()
+                        .report_local_socket_address(socket_address, peer)
+                        .await;
+                }
+                SendDataKind::GlobalDirect => {
+                    self.network_manager()
+                        .report_global_socket_address(socket_address, peer)
+                        .await;
+                }
+                SendDataKind::GlobalIndirect => {
+                    // Do nothing in this case, as the socket address returned here would be for any node other than ours
+                }
+            }
+        }
+
+        Ok(Answer::new(latency, status_a))
+    }
+
+    pub(crate) async fn process_status_q(&self, msg: RPCMessage) -> Result<(), RPCError> {
+        let peer_noderef = msg.header.peer_noderef.clone();
+
+        // Get the question
+        let status_q = match msg.operation.kind() {
+            RPCOperationKind::Question(q) => match q.detail() {
+                RPCQuestionDetail::StatusQ(q) => q,
+                _ => panic!("not a status question"),
+            },
+            _ => panic!("not a question"),
+        };
+
+        // update node status for the requesting node to our routing table
+        if let Some(sender_nr) = msg.opt_sender_nr.clone() {
+            // Update latest node status in routing table for the statusq sender
+            sender_nr.operate_mut(|e| {
+                e.update_node_status(status_q.node_status.clone());
+            });
+        }
+
+        // Make status answer
+        let node_status = self.network_manager().generate_node_status();
+        let sender_info = Self::generate_sender_info(peer_noderef).await;
+        let status_a = RPCOperationStatusA {
+            node_status,
+            sender_info,
+        };
+
+        // Send status answer
+        self.answer(
+            msg,
+            RPCAnswer::new(RPCAnswerDetail::StatusA(status_a)),
+            None,
+        )
+        .await
+    }
+}

veilid-core/src/rpc_processor/rpc_supply_block.rs

@@ -0,0 +1,7 @@
+use super::*;
+
+impl RPCProcessor {
+    pub(crate) async fn process_supply_block_q(&self, msg: RPCMessage) -> Result<(), RPCError> {
+        Err(rpc_error_unimplemented("process_supply_block_q"))
+    }
+}

veilid-core/src/rpc_processor/rpc_validate_dial_info.rs

@@ -0,0 +1,207 @@
+use super::*;
+
+impl RPCProcessor {
+    // Can only be sent directly, not via relays or routes
+    pub async fn rpc_call_validate_dial_info(
+        &self,
+        peer: NodeRef,
+        dial_info: DialInfo,
+        redirect: bool,
+    ) -> Result<bool, RPCError> {
+        let network_manager = self.network_manager();
+        let receipt_time = ms_to_us(
+            self.config
+                .get()
+                .network
+                .dht
+                .validate_dial_info_receipt_time_ms,
+        );
+        //
+        let (vdi_msg, eventual_value) = {
+            let mut vdi_msg = ::capnp::message::Builder::new_default();
+            let mut question = vdi_msg.init_root::<veilid_capnp::operation::Builder>();
+            question.set_op_id(self.get_next_op_id());
+            let mut respond_to = question.reborrow().init_respond_to();
+            respond_to.set_none(());
+            let detail = question.reborrow().init_detail();
+            let mut vdi_builder = detail.init_validate_dial_info();
+
+            // Generate receipt and waitable eventual so we can see if we get the receipt back
+            let (receipt, eventual_value) = network_manager
+                .generate_single_shot_receipt(receipt_time, [])
+                .map_err(map_error_string!())?;
+
+            vdi_builder.set_redirect(redirect);
+            let mut di_builder = vdi_builder.reborrow().init_dial_info();
+            encode_dial_info(&dial_info, &mut di_builder)?;
+            let r_builder = vdi_builder.init_receipt(receipt.len().try_into().map_err(
+                map_error_protocol!("invalid receipt length in validate dial info"),
+            )?);
+            r_builder.copy_from_slice(&receipt);
+
+            (vdi_msg.into_reader(), eventual_value)
+        };
+
+        // Send the validate_dial_info request
+        // This can only be sent directly, as relays can not validate dial info
+        self.request(Destination::Direct(peer), vdi_msg, None)
+            .await?;
+
+        log_net!(debug "waiting for validate_dial_info receipt");
+        // Wait for receipt
+        match eventual_value.await.take_value().unwrap() {
+            ReceiptEvent::ReturnedInBand { inbound_noderef: _ } => Err(rpc_error_internal(
+                "validate_dial_info receipt should be returned out-of-band",
+            )),
+            ReceiptEvent::ReturnedOutOfBand => {
+                log_net!(debug "validate_dial_info receipt returned");
+                Ok(true)
+            }
+            ReceiptEvent::Expired => {
+                log_net!(debug "validate_dial_info receipt expired");
+                Ok(false)
+            }
+            ReceiptEvent::Cancelled => {
+                Err(rpc_error_internal("receipt was dropped before expiration"))
+            }
+        }
+    }
+
+    pub(crate) async fn process_validate_dial_info(
+        &self,
+        rpcreader: RPCMessage,
+    ) -> Result<(), RPCError> {
+        //
+        let (redirect, dial_info, receipt) = {
+            let operation = rpcreader
+                .reader
+                .get_root::<veilid_capnp::operation::Reader>()
+                .map_err(map_error_capnp_error!())
+                .map_err(logthru_rpc!())?;
+
+            // This should never want an answer
+            if self.wants_answer(&operation)? {
+                return Err(rpc_error_invalid_format(
+                    "validate dial info should not want answer",
+                ));
+            }
+
+            // get validateDialInfo reader
+            let vdi_reader = match operation.get_detail().which() {
+                Ok(veilid_capnp::operation::detail::Which::ValidateDialInfo(Ok(x))) => x,
+                _ => panic!("invalid operation type in process_validate_dial_info"),
+            };
+
+            // Parse out fields
+            let redirect = vdi_reader.get_redirect();
+            let dial_info = decode_dial_info(&vdi_reader.get_dial_info().map_err(
+                map_error_internal!("no valid dial info in process_validate_dial_info"),
+            )?)?;
+            let receipt = vdi_reader
+                .get_receipt()
+                .map_err(map_error_internal!(
+                    "no valid receipt in process_validate_dial_info"
+                ))?
+                .to_vec();
+
+            (redirect, dial_info, receipt)
+        };
+
+        // Redirect this request if we are asked to
+        if redirect {
+            // Find peers capable of validating this dial info
+            // We filter on the -outgoing- protocol capability status not the node's dial info
+            // Use the address type though, to ensure we reach an ipv6 capable node if this is
+            // an ipv6 address
+            let routing_table = self.routing_table();
+            let filter = DialInfoFilter::global().with_address_type(dial_info.address_type());
+            let sender_id = rpcreader.header.envelope.get_sender_id();
+            let node_count = {
+                let c = self.config.get();
+                c.network.dht.max_find_node_count as usize
+            };
+            let mut peers = routing_table.find_fast_public_nodes_filtered(node_count, &filter);
+            if peers.is_empty() {
+                return Err(rpc_error_internal(format!(
+                    "no peers matching filter '{:?}'",
+                    filter
+                )));
+            }
+            for peer in &mut peers {
+                // Ensure the peer is not the one asking for the validation
+                if peer.node_id() == sender_id {
+                    continue;
+                }
+
+                // Release the filter on the peer because we don't need to send the redirect with the filter
+                // we just wanted to make sure we only selected nodes that were capable of
+                // using the correct protocol for the dial info being validated
+                peer.set_filter(None);
+
+                // Ensure the peer's status is known and that it is capable of
+                // making outbound connections for the dial info we want to verify
+                // and if this peer can validate dial info
+                let can_contact_dial_info = peer.operate(|e: &BucketEntryInner| {
+                    if let Some(ni) = e.node_info() {
+                        ni.outbound_protocols.contains(dial_info.protocol_type())
+                            && ni.can_validate_dial_info()
+                    } else {
+                        false
+                    }
+                });
+                if !can_contact_dial_info {
+                    continue;
+                }
+
+                // See if this peer will validate dial info
+                let will_validate_dial_info = peer.operate(|e: &BucketEntryInner| {
+                    if let Some(status) = &e.peer_stats().status {
+                        status.will_validate_dial_info
+                    } else {
+                        true
+                    }
+                });
+                if !will_validate_dial_info {
+                    continue;
+                }
+
+                // Make a copy of the request, without the redirect flag
+                let vdi_msg_reader = {
+                    let mut vdi_msg = ::capnp::message::Builder::new_default();
+                    let mut question = vdi_msg.init_root::<veilid_capnp::operation::Builder>();
+                    question.set_op_id(self.get_next_op_id());
+                    let mut respond_to = question.reborrow().init_respond_to();
+                    respond_to.set_none(());
+                    let detail = question.reborrow().init_detail();
+                    let mut vdi_builder = detail.init_validate_dial_info();
+                    vdi_builder.set_redirect(false);
+                    let mut di_builder = vdi_builder.reborrow().init_dial_info();
+                    encode_dial_info(&dial_info, &mut di_builder)?;
+                    let r_builder = vdi_builder.init_receipt(receipt.len().try_into().map_err(
+                        map_error_protocol!("invalid receipt length in process_validate_dial_info"),
+                    )?);
+                    r_builder.copy_from_slice(&receipt);
+                    vdi_msg.into_reader()
+                };
+
+                // Send the validate_dial_info request until we succeed
+                self.request(Destination::Direct(peer.clone()), vdi_msg_reader, None)
+                    .await?;
+            }
+            return Ok(());
+        };
+
+        // Otherwise send a return receipt directly
+        // Possibly from an alternate port
+        let network_manager = self.network_manager();
+        network_manager
+            .send_out_of_band_receipt(dial_info.clone(), receipt)
+            .await
+            .map_err(map_error_string!())
+            .map_err(
+                logthru_net!(error "failed to send direct receipt to dial info: {}", dial_info),
+            )?;
+
+        Ok(())
+    }
+}

veilid-core/src/rpc_processor/rpc_value_changed.rs

@@ -0,0 +1,7 @@
+use super::*;
+
+impl RPCProcessor {
+    pub(crate) async fn process_value_changed(&self, msg: RPCMessage) -> Result<(), RPCError> {
+        Err(rpc_error_unimplemented("process_value_changed"))
+    }
+}

veilid-core/src/rpc_processor/rpc_watch_value.rs

@@ -0,0 +1,7 @@
+use super::*;
+
+impl RPCProcessor {
+    pub(crate) async fn process_watch_value_q(&self, msg: RPCMessage) -> Result<(), RPCError> {
+        Err(rpc_error_unimplemented("process_watch_value_q"))
+    }
+}

veilid-core/src/veilid_api/mod.rs

@@ -1887,89 +1887,56 @@ impl VeilidAPI {
     ////////////////////////////////////////////////////////////////
     // Direct Node Access (pretty much for testing only)
 
-    #[instrument(level = "debug", err, skip(self))]
+    // #[instrument(level = "debug", err, skip(self))]
-    pub async fn status(&self, node_id: NodeId) -> Result<StatusAnswer, VeilidAPIError> {
+    // pub async fn search_dht(&self, node_id: NodeId) -> Result<PeerInfo, VeilidAPIError> {
-        let rpc = self.rpc_processor()?;
+    //     let rpc_processor = self.rpc_processor()?;
-        let routing_table = rpc.routing_table();
+    //     let config = self.config()?;
-        let node_ref = match routing_table.lookup_node_ref(node_id.key) {
+    //     let (count, fanout, timeout) = {
-            None => return Err(VeilidAPIError::NodeNotFound { node_id }),
+    //         let c = config.get();
-            Some(nr) => nr,
+    //         (
-        };
+    //             c.network.dht.resolve_node_count,
-        let status_answer = rpc
+    //             c.network.dht.resolve_node_fanout,
-            .rpc_call_status(node_ref)
+    //             c.network.dht.resolve_node_timeout_ms.map(ms_to_us),
-            .await
+    //         )
-            .map_err(map_rpc_error!())?;
+    //     };
-        Ok(status_answer)
-    }
 
-    #[instrument(level = "debug", err, skip(self))]
+    //     let node_ref = rpc_processor
-    pub async fn validate_dial_info(
+    //         .search_dht_single_key(node_id.key, count, fanout, timeout)
-        &self,
+    //         .await
-        node_id: NodeId,
+    //         .map_err(map_rpc_error!())?;
-        dial_info: DialInfo,
-        redirect: bool,
-    ) -> Result<bool, VeilidAPIError> {
-        let rpc = self.rpc_processor()?;
-        let routing_table = rpc.routing_table();
-        let node_ref = match routing_table.lookup_node_ref(node_id.key) {
-            None => return Err(VeilidAPIError::NodeNotFound { node_id }),
-            Some(nr) => nr,
-        };
-        rpc.rpc_call_validate_dial_info(node_ref.clone(), dial_info, redirect)
-            .await
-            .map_err(map_rpc_error!())
-    }
 
-    #[instrument(level = "debug", err, skip(self))]
+    //     let answer = node_ref.peer_info();
-    pub async fn search_dht(&self, node_id: NodeId) -> Result<PeerInfo, VeilidAPIError> {
+    //     if let Some(answer) = answer {
-        let rpc_processor = self.rpc_processor()?;
+    //         Ok(answer)
-        let config = self.config()?;
+    //     } else {
-        let (count, fanout, timeout) = {
+    //         Err(VeilidAPIError::NoPeerInfo {
-            let c = config.get();
+    //             node_id: NodeId::new(node_ref.node_id()),
-            (
+    //         })
-                c.network.dht.resolve_node_count,
+    //     }
-                c.network.dht.resolve_node_fanout,
+    // }
-                c.network.dht.resolve_node_timeout_ms.map(ms_to_us),
-            )
-        };
 
-        let node_ref = rpc_processor
+    // #[instrument(level = "debug", err, skip(self))]
-            .search_dht_single_key(node_id.key, count, fanout, timeout)
+    // pub async fn search_dht_multi(&self, node_id: NodeId) -> Result<Vec<PeerInfo>, VeilidAPIError> {
-            .await
+    //     let rpc_processor = self.rpc_processor()?;
-            .map_err(map_rpc_error!())?;
+    //     let config = self.config()?;
+    //     let (count, fanout, timeout) = {
+    //         let c = config.get();
+    //         (
+    //             c.network.dht.resolve_node_count,
+    //             c.network.dht.resolve_node_fanout,
+    //             c.network.dht.resolve_node_timeout_ms.map(ms_to_us),
+    //         )
+    //     };
 
-        let answer = node_ref.peer_info();
+    //     let node_refs = rpc_processor
-        if let Some(answer) = answer {
+    //         .search_dht_multi_key(node_id.key, count, fanout, timeout)
-            Ok(answer)
+    //         .await
-        } else {
+    //         .map_err(map_rpc_error!())?;
-            Err(VeilidAPIError::NoPeerInfo {
-                node_id: NodeId::new(node_ref.node_id()),
-            })
-        }
-    }
 
-    #[instrument(level = "debug", err, skip(self))]
+    //     let answer = node_refs.iter().filter_map(|x| x.peer_info()).collect();
-    pub async fn search_dht_multi(&self, node_id: NodeId) -> Result<Vec<PeerInfo>, VeilidAPIError> {
-        let rpc_processor = self.rpc_processor()?;
-        let config = self.config()?;
-        let (count, fanout, timeout) = {
-            let c = config.get();
-            (
-                c.network.dht.resolve_node_count,
-                c.network.dht.resolve_node_fanout,
-                c.network.dht.resolve_node_timeout_ms.map(ms_to_us),
-            )
-        };
 
-        let node_refs = rpc_processor
+    //     Ok(answer)
-            .search_dht_multi_key(node_id.key, count, fanout, timeout)
+    // }
-            .await
-            .map_err(map_rpc_error!())?;
-
-        let answer = node_refs.iter().filter_map(|x| x.peer_info()).collect();
-
-        Ok(answer)
-    }
 
     ////////////////////////////////////////////////////////////////
     // Safety / Private Route Handling