Merge branch 'more-debug-commands' into 'main'

Debugging

See merge request veilid/veilid!188

veilid-core/src/network_manager/connection_manager.rs

@@ -404,4 +404,12 @@ impl ConnectionManager {
             let _ = sender.send_async(ConnectionManagerEvent::Dead(conn)).await;
         }
     }
+
+    pub async fn debug_print(&self) -> String {
+        //let inner = self.arc.inner.lock();
+        format!(
+            "Connection Table:\n\n{}",
+            self.arc.connection_table.debug_print_table()
+        )
+    }
 }

veilid-core/src/network_manager/connection_table.rs

@@ -72,6 +72,15 @@ impl ConnectionTable {
         }
     }
 
+    fn index_to_protocol(idx: usize) -> ProtocolType {
+        match idx {
+            0 => ProtocolType::TCP,
+            1 => ProtocolType::WS,
+            2 => ProtocolType::WSS,
+            _ => panic!("not a connection-oriented protocol"),
+        }
+    }
+
     #[instrument(level = "trace", skip(self))]
     pub async fn join(&self) {
         let mut unord = {
@@ -331,4 +340,23 @@ impl ConnectionTable {
         let conn = Self::remove_connection_records(&mut *inner, id);
         Some(conn)
     }
+
+    pub fn debug_print_table(&self) -> String {
+        let mut out = String::new();
+        let inner = self.inner.lock();
+        let cur_ts = get_aligned_timestamp();
+        for t in 0..inner.conn_by_id.len() {
+            out += &format!(
+                "  {} Connections: ({}/{})\n",
+                Self::index_to_protocol(t).to_string(),
+                inner.conn_by_id[t].len(),
+                inner.max_connections[t]
+            );
+
+            for (_, conn) in &inner.conn_by_id[t] {
+                out += &format!("    {}\n", conn.debug_print(cur_ts));
+            }
+        }
+        out
+    }
 }

veilid-core/src/network_manager/network_connection.rs

@@ -391,6 +391,17 @@ impl NetworkConnection {
                 .await;
         }.instrument(trace_span!("process_connection")))
     }
+
+    pub fn debug_print(&self, cur_ts: Timestamp) -> String {
+        format!("{} <- {} | {:x} | est {} sent {} rcvd {}",
+            self.descriptor.remote_address(), 
+            self.descriptor.local().map(|x| x.to_string()).unwrap_or("---".to_owned()),
+            self.connection_id.as_u64(),
+            debug_duration(cur_ts.as_u64().saturating_sub(self.established_time.as_u64())),
+            self.stats().last_message_sent_time.map(|ts| debug_duration(cur_ts.as_u64().saturating_sub(ts.as_u64())) ).unwrap_or("---".to_owned()),
+            self.stats().last_message_recv_time.map(|ts| debug_duration(cur_ts.as_u64().saturating_sub(ts.as_u64())) ).unwrap_or("---".to_owned()),
+        )
+    }
 }
 
 // Resolves ready when the connection loop has terminated

veilid-core/src/routing_table/debug.rs

@@ -73,6 +73,7 @@ impl RoutingTable {
             "   Self Transfer Stats: {:#?}\n\n",
             inner.self_transfer_stats
         );
+        out += &format!("   Version: {}\n\n", veilid_version_string());
 
         out
     }

veilid-core/src/routing_table/routing_domains.rs

@@ -470,7 +470,7 @@ impl RoutingDomainDetail for PublicInternetRoutingDomainDetail {
                 return ContactMethod::Unreachable;
             };
 
-            // Can we reach the full relay?
+            // Can we reach the inbound relay?
             if first_filtered_dial_info_detail_between_nodes(
                 node_a,
                 &node_b_relay,
@@ -480,11 +480,30 @@ impl RoutingDomainDetail for PublicInternetRoutingDomainDetail {
             )
             .is_some()
             {
+                ///////// Reverse connection
+
+                // Get the best match dial info for an reverse inbound connection from node B to node A
+                if let Some(reverse_did) = first_filtered_dial_info_detail_between_nodes(
+                    node_b,
+                    node_a,
+                    &dial_info_filter,
+                    sequencing,
+                    dif_sort.clone()
+                ) {
+                    // Can we receive a direct reverse connection?
+                    if !reverse_did.class.requires_signal() {
+                        return ContactMethod::SignalReverse(
+                            node_b_relay_id,
+                            node_b_id,
+                        );
+                    }
+                }
+
                 return ContactMethod::InboundRelay(node_b_relay_id);
             }
         }
 
-        // If node A can't reach the node by other means, it may need to use its own relay
+        // If node A can't reach the node by other means, it may need to use its outbound relay
         if peer_a.signed_node_info().node_info().network_class().outbound_wants_relay() {
             if let Some(node_a_relay_id) = peer_a.signed_node_info().relay_ids().get(best_ck) {
                 // Ensure it's not our relay we're trying to reach

veilid-core/src/routing_table/routing_table_inner.rs

@@ -1,7 +1,7 @@
 use super::*;
 use weak_table::PtrWeakHashSet;
 
-const RECENT_PEERS_TABLE_SIZE: usize = 64;
+pub const RECENT_PEERS_TABLE_SIZE: usize = 64;
 
 pub type EntryCounts = BTreeMap<(RoutingDomain, CryptoKind), usize>;
 //////////////////////////////////////////////////////////////////////////

veilid-core/src/rpc_processor/mod.rs

@@ -434,6 +434,11 @@ impl RPCProcessor {
 
     //////////////////////////////////////////////////////////////////////
 
+    /// Get waiting app call id for debugging purposes
+    pub fn get_app_call_ids(&self) -> Vec<OperationId> {
+        self.unlocked_inner.waiting_app_call_table.get_operation_ids()
+    }
+
     /// Determine if a SignedNodeInfo can be placed into the specified routing domain
     fn verify_node_info(
         &self,

veilid-core/src/rpc_processor/operation_waiter.rs

@@ -30,6 +30,7 @@ where
     C: Unpin + Clone,
 {
     context: C,
+    timestamp: Timestamp,
     eventual: EventualValue<(Option<Id>, T)>,
 }
 
@@ -82,6 +83,7 @@ where
         let e = EventualValue::new();
         let waiting_op = OperationWaitingOp {
             context,
+            timestamp: get_aligned_timestamp(),
             eventual: e.clone(),
         };
         if inner.waiting_op_table.insert(op_id, waiting_op).is_some() {
@@ -98,6 +100,18 @@ where
         }
     }
 
+    /// Get all waiting operation ids
+    pub fn get_operation_ids(&self) -> Vec<OperationId> {
+        let inner = self.inner.lock();
+        let mut opids: Vec<(OperationId, Timestamp)> = inner
+            .waiting_op_table
+            .iter()
+            .map(|x| (*x.0, x.1.timestamp))
+            .collect();
+        opids.sort_by(|a, b| a.1.cmp(&b.1));
+        opids.into_iter().map(|x| x.0).collect()
+    }
+
     /// Get operation context
     pub fn get_op_context(&self, op_id: OperationId) -> Result<C, RPCError> {
         let inner = self.inner.lock();

veilid-core/src/veilid_api/debug.rs

@@ -15,6 +15,35 @@ static DEBUG_CACHE: Mutex<DebugCache> = Mutex::new(DebugCache {
     imported_routes: Vec::new(),
 });
 
+fn format_opt_ts(ts: Option<TimestampDuration>) -> String {
+    let Some(ts) = ts else {
+        return "---".to_owned();
+    };
+    let ts = ts.as_u64();
+    let secs = timestamp_to_secs(ts);
+    if secs >= 1.0 {
+        format!("{:.2}s", timestamp_to_secs(ts))
+    } else {
+        format!("{:.2}ms", timestamp_to_secs(ts) * 1000.0)
+    }
+}
+
+fn format_opt_bps(bps: Option<ByteCount>) -> String {
+    let Some(bps) = bps else {
+        return "---".to_owned();
+    };
+    let bps = bps.as_u64();
+    if bps >= 1024u64 * 1024u64 * 1024u64 {
+        format!("{:.2}GB/s", (bps / (1024u64 * 1024u64)) as f64 / 1024.0)
+    } else if bps >= 1024u64 * 1024u64 {
+        format!("{:.2}MB/s", (bps / 1024u64) as f64 / 1024.0)
+    } else if bps >= 1024u64 {
+        format!("{:.2}KB/s", bps as f64 / 1024.0)
+    } else {
+        format!("{:.2}B/s", bps as f64)
+    }
+}
+
 fn get_bucket_entry_state(text: &str) -> Option<BucketEntryState> {
     if text == "dead" {
         Some(BucketEntryState::Dead)
@@ -165,82 +194,89 @@ fn get_node_ref_modifiers(mut node_ref: NodeRef) -> impl FnOnce(&str) -> Option<
     }
 }
 
-fn get_destination(routing_table: RoutingTable) -> impl FnOnce(&str) -> Option<Destination> {
+fn get_destination(
+    routing_table: RoutingTable,
+) -> impl FnOnce(&str) -> SendPinBoxFuture<Option<Destination>> {
     move |text| {
-        // Safety selection
+        let text = text.to_owned();
-        let (text, ss) = if let Some((first, second)) = text.split_once('+') {
+        Box::pin(async move {
-            let ss = get_safety_selection(routing_table.clone())(second)?;
+            // Safety selection
-            (first, Some(ss))
+            let (text, ss) = if let Some((first, second)) = text.split_once('+') {
-        } else {
+                let ss = get_safety_selection(routing_table.clone())(second)?;
-            (text, None)
+                (first, Some(ss))
-        };
+            } else {
-        if text.len() == 0 {
+                (text.as_str(), None)
-            return None;
+            };
-        }
+            if text.len() == 0 {
-        if &text[0..1] == "#" {
+                return None;
-            let rss = routing_table.route_spec_store();
+            }
+            if &text[0..1] == "#" {
+                let rss = routing_table.route_spec_store();
 
-            // Private route
+                // Private route
-            let text = &text[1..];
+                let text = &text[1..];
 
-            let private_route = if let Some(prid) = get_route_id(rss.clone(), false, true)(text) {
+                let private_route = if let Some(prid) = get_route_id(rss.clone(), false, true)(text)
-                let Some(private_route) = rss.best_remote_private_route(&prid) else {
+                {
+                    let Some(private_route) = rss.best_remote_private_route(&prid) else {
                     return None;
                 };
-                private_route
+                    private_route
-            } else {
+                } else {
-                let mut dc = DEBUG_CACHE.lock();
+                    let mut dc = DEBUG_CACHE.lock();
-                let n = get_number(text)?;
+                    let n = get_number(text)?;
-                let prid = dc.imported_routes.get(n)?.clone();
+                    let prid = dc.imported_routes.get(n)?.clone();
-                let Some(private_route) = rss.best_remote_private_route(&prid) else {
+                    let Some(private_route) = rss.best_remote_private_route(&prid) else {
                     // Remove imported route
                     dc.imported_routes.remove(n);
                     info!("removed dead imported route {}", n);
                     return None;
                 };
-                private_route
+                    private_route
-            };
+                };
 
-            Some(Destination::private_route(
+                Some(Destination::private_route(
-                private_route,
+                    private_route,
-                ss.unwrap_or(SafetySelection::Unsafe(Sequencing::default())),
+                    ss.unwrap_or(SafetySelection::Unsafe(Sequencing::default())),
-            ))
+                ))
-        } else {
-            let (text, mods) = text
-                .split_once('/')
-                .map(|x| (x.0, Some(x.1)))
-                .unwrap_or((text, None));
-            if let Some((first, second)) = text.split_once('@') {
-                // Relay
-                let mut relay_nr = get_node_ref(routing_table.clone())(second)?;
-                let target_nr = get_node_ref(routing_table)(first)?;
-
-                if let Some(mods) = mods {
-                    relay_nr = get_node_ref_modifiers(relay_nr)(mods)?;
-                }
-
-                let mut d = Destination::relay(relay_nr, target_nr);
-                if let Some(ss) = ss {
-                    d = d.with_safety(ss)
-                }
-
-                Some(d)
             } else {
-                // Direct
+                let (text, mods) = text
-                let mut target_nr = get_node_ref(routing_table)(text)?;
+                    .split_once('/')
+                    .map(|x| (x.0, Some(x.1)))
+                    .unwrap_or((text, None));
+                if let Some((first, second)) = text.split_once('@') {
+                    // Relay
+                    let mut relay_nr = get_node_ref(routing_table.clone())(second)?;
+                    let target_nr = get_node_ref(routing_table)(first)?;
 
-                if let Some(mods) = mods {
+                    if let Some(mods) = mods {
-                    target_nr = get_node_ref_modifiers(target_nr)(mods)?;
+                        relay_nr = get_node_ref_modifiers(relay_nr)(mods)?;
+                    }
+
+                    let mut d = Destination::relay(relay_nr, target_nr);
+                    if let Some(ss) = ss {
+                        d = d.with_safety(ss)
+                    }
+
+                    Some(d)
+                } else {
+                    // Direct
+                    let mut target_nr =
+                        resolve_node_ref(routing_table, ss.unwrap_or_default())(text).await?;
+
+                    if let Some(mods) = mods {
+                        target_nr = get_node_ref_modifiers(target_nr)(mods)?;
+                    }
+
+                    let mut d = Destination::direct(target_nr);
+                    if let Some(ss) = ss {
+                        d = d.with_safety(ss)
+                    }
+
+                    Some(d)
                 }
-
-                let mut d = Destination::direct(target_nr);
-                if let Some(ss) = ss {
-                    d = d.with_safety(ss)
-                }
-
-                Some(d)
             }
-        }
+        })
     }
 }
 
@@ -292,6 +328,44 @@ fn get_dht_key(
     }
 }
 
+fn resolve_node_ref(
+    routing_table: RoutingTable,
+    safety_selection: SafetySelection,
+) -> impl FnOnce(&str) -> SendPinBoxFuture<Option<NodeRef>> {
+    move |text| {
+        let text = text.to_owned();
+        Box::pin(async move {
+            let (text, mods) = text
+                .split_once('/')
+                .map(|x| (x.0, Some(x.1)))
+                .unwrap_or((&text, None));
+
+            let mut nr = if let Some(key) = get_public_key(text) {
+                let node_id = TypedKey::new(best_crypto_kind(), key);
+                routing_table
+                    .rpc_processor()
+                    .resolve_node(node_id, safety_selection)
+                    .await
+                    .ok()
+                    .flatten()?
+            } else if let Some(node_id) = get_typed_key(text) {
+                routing_table
+                    .rpc_processor()
+                    .resolve_node(node_id, safety_selection)
+                    .await
+                    .ok()
+                    .flatten()?
+            } else {
+                return None;
+            };
+            if let Some(mods) = mods {
+                nr = get_node_ref_modifiers(nr)(mods)?;
+            }
+            Some(nr)
+        })
+    }
+}
+
 fn get_node_ref(routing_table: RoutingTable) -> impl FnOnce(&str) -> Option<NodeRef> {
     move |text| {
         let (text, mods) = text
@@ -301,8 +375,8 @@ fn get_node_ref(routing_table: RoutingTable) -> impl FnOnce(&str) -> Option<Node
 
         let mut nr = if let Some(key) = get_public_key(text) {
             routing_table.lookup_any_node_ref(key).ok().flatten()?
-        } else if let Some(key) = get_typed_key(text) {
+        } else if let Some(node_id) = get_typed_key(text) {
-            routing_table.lookup_node_ref(key).ok().flatten()?
+            routing_table.lookup_node_ref(node_id).ok().flatten()?
         } else {
             return None;
         };
@@ -394,6 +468,19 @@ fn get_debug_argument<T, G: FnOnce(&str) -> Option<T>>(
     };
     Ok(val)
 }
+
+async fn async_get_debug_argument<T, G: FnOnce(&str) -> SendPinBoxFuture<Option<T>>>(
+    value: &str,
+    context: &str,
+    argument: &str,
+    getter: G,
+) -> VeilidAPIResult<T> {
+    let Some(val) = getter(value).await else {
+        apibail_invalid_argument!(context, argument, value);
+    };
+    Ok(val)
+}
+
 fn get_debug_argument_at<T, G: FnOnce(&str) -> Option<T>>(
     debug_args: &[String],
     pos: usize,
@@ -411,6 +498,23 @@ fn get_debug_argument_at<T, G: FnOnce(&str) -> Option<T>>(
     Ok(val)
 }
 
+async fn async_get_debug_argument_at<T, G: FnOnce(&str) -> SendPinBoxFuture<Option<T>>>(
+    debug_args: &[String],
+    pos: usize,
+    context: &str,
+    argument: &str,
+    getter: G,
+) -> VeilidAPIResult<T> {
+    if pos >= debug_args.len() {
+        apibail_missing_argument!(context, argument);
+    }
+    let value = &debug_args[pos];
+    let Some(val) = getter(value).await else {
+        apibail_invalid_argument!(context, argument, value);
+    };
+    Ok(val)
+}
+
 pub fn print_data(data: &[u8], truncate_len: Option<usize>) -> String {
     // check is message body is ascii printable
     let mut printable = true;
@@ -578,7 +682,32 @@ impl VeilidAPI {
     async fn debug_nodeinfo(&self, _args: String) -> VeilidAPIResult<String> {
         // Dump routing table entry
         let routing_table = self.network_manager()?.routing_table();
-        Ok(routing_table.debug_info_nodeinfo())
+        let connection_manager = self.network_manager()?.connection_manager();
+        let nodeinfo = routing_table.debug_info_nodeinfo();
+
+        // Dump core state
+        let state = self.get_state().await?;
+
+        let mut peertable = format!(
+            "Recent Peers: {} (max {})\n",
+            state.network.peers.len(),
+            RECENT_PEERS_TABLE_SIZE
+        );
+        for peer in state.network.peers {
+            peertable += &format!(
+                "   {} | {} | {} | {} down | {} up\n",
+                peer.node_ids.first().unwrap(),
+                peer.peer_address,
+                format_opt_ts(peer.peer_stats.latency.map(|l| l.average)),
+                format_opt_bps(Some(peer.peer_stats.transfer.down.average)),
+                format_opt_bps(Some(peer.peer_stats.transfer.up.average)),
+            );
+        }
+
+        // Dump connection table
+        let connman = connection_manager.debug_print().await;
+
+        Ok(format!("{}\n\n{}\n\n{}\n\n", nodeinfo, peertable, connman))
     }
 
     async fn debug_config(&self, args: String) -> VeilidAPIResult<String> {
@@ -749,15 +878,16 @@ impl VeilidAPI {
 
         let args: Vec<String> = args.split_whitespace().map(|s| s.to_owned()).collect();
 
-        let dest = get_debug_argument_at(
+        let dest = async_get_debug_argument_at(
             &args,
             0,
             "debug_ping",
             "destination",
             get_destination(routing_table),
-        )?;
+        )
+        .await?;
 
-        // Dump routing table entry
+        // Send a StatusQ
         let out = match rpc
             .rpc_call_status(dest)
             .await
@@ -772,6 +902,109 @@ impl VeilidAPI {
         Ok(format!("{:#?}", out))
     }
 
+    async fn debug_app_message(&self, args: String) -> VeilidAPIResult<String> {
+        let netman = self.network_manager()?;
+        let routing_table = netman.routing_table();
+        let rpc = netman.rpc_processor();
+
+        let (arg, rest) = args.split_once(' ').unwrap_or((&args, ""));
+        let rest = rest.trim_start().to_owned();
+
+        let dest = async_get_debug_argument(
+            arg,
+            "debug_app_message",
+            "destination",
+            get_destination(routing_table),
+        )
+        .await?;
+
+        let data = get_debug_argument(&rest, "debug_app_message", "data", get_data)?;
+        let data_len = data.len();
+
+        // Send a AppMessage
+        let out = match rpc
+            .rpc_call_app_message(dest, data)
+            .await
+            .map_err(VeilidAPIError::internal)?
+        {
+            NetworkResult::Value(_) => format!("Sent {} bytes", data_len),
+            r => {
+                return Ok(r.to_string());
+            }
+        };
+
+        Ok(out)
+    }
+
+    async fn debug_app_call(&self, args: String) -> VeilidAPIResult<String> {
+        let netman = self.network_manager()?;
+        let routing_table = netman.routing_table();
+        let rpc = netman.rpc_processor();
+
+        let (arg, rest) = args.split_once(' ').unwrap_or((&args, ""));
+        let rest = rest.trim_start().to_owned();
+
+        let dest = async_get_debug_argument(
+            arg,
+            "debug_app_call",
+            "destination",
+            get_destination(routing_table),
+        )
+        .await?;
+
+        let data = get_debug_argument(&rest, "debug_app_call", "data", get_data)?;
+        let data_len = data.len();
+
+        // Send a AppMessage
+        let out = match rpc
+            .rpc_call_app_call(dest, data)
+            .await
+            .map_err(VeilidAPIError::internal)?
+        {
+            NetworkResult::Value(v) => format!(
+                "Sent {} bytes, received: {}",
+                data_len,
+                print_data(&v.answer, Some(512))
+            ),
+            r => {
+                return Ok(r.to_string());
+            }
+        };
+
+        Ok(out)
+    }
+
+    async fn debug_app_reply(&self, args: String) -> VeilidAPIResult<String> {
+        let netman = self.network_manager()?;
+        let rpc = netman.rpc_processor();
+
+        let (call_id, data) = if args.starts_with("#") {
+            let (arg, rest) = args[1..].split_once(' ').unwrap_or((&args, ""));
+            let call_id =
+                OperationId::new(u64::from_str_radix(arg, 16).map_err(VeilidAPIError::generic)?);
+            let rest = rest.trim_start().to_owned();
+            let data = get_debug_argument(&rest, "debug_app_reply", "data", get_data)?;
+            (call_id, data)
+        } else {
+            let call_id = rpc
+                .get_app_call_ids()
+                .first()
+                .cloned()
+                .ok_or_else(|| VeilidAPIError::generic("no app calls waiting"))?;
+            let data = get_debug_argument(&args, "debug_app_reply", "data", get_data)?;
+            (call_id, data)
+        };
+
+        let data_len = data.len();
+
+        // Send a AppCall Reply
+        self.app_call_reply(call_id, data)
+            .await
+            .map_err(VeilidAPIError::internal)?;
+
+        Ok(format!("Replied with {} bytes", data_len))
+    }
+
     async fn debug_route_allocate(&self, args: Vec<String>) -> VeilidAPIResult<String> {
         // [ord|*ord] [rel] [<count>] [in|out] [avoid_node_id]
 
@@ -1388,6 +1621,9 @@ detach
 restart network
 contact <node>[<modifiers>]
 ping <destination>
+appmessage <destination> <data>
+appcall <destination> <data>
+appreply [#id] <data>
 relay <relay> [public|local]
 punish list
 route allocate [ord|*ord] [rel] [<count>] [in|out]
@@ -1465,6 +1701,12 @@ record list <local|remote>
                 self.debug_relay(rest).await
             } else if arg == "ping" {
                 self.debug_ping(rest).await
+            } else if arg == "appmessage" {
+                self.debug_app_message(rest).await
+            } else if arg == "appcall" {
+                self.debug_app_call(rest).await
+            } else if arg == "appreply" {
+                self.debug_app_reply(rest).await
             } else if arg == "contact" {
                 self.debug_contact(rest).await
             } else if arg == "nodeinfo" {

veilid-flutter/lib/default_config.dart

@@ -173,14 +173,14 @@ Future<VeilidConfig> getDefaultVeilidConfig(String programName) async {
         ws: VeilidConfigWS(
           connect: true,
           listen: !kIsWeb,
-          maxConnections: 32,
+          maxConnections: 1024,
           listenAddress: '',
           path: 'ws',
         ),
         wss: VeilidConfigWSS(
           connect: true,
           listen: false,
-          maxConnections: 32,
+          maxConnections: 1024,
           listenAddress: '',
           path: 'ws',
         ),