Git-Mirrors/veilid
1
0
Fork 0
mirror of https://gitlab.com/veilid/veilid.git synced 2025-02-22 07:29:57 -05:00
Code Issues Packages Projects Releases Wiki Activity

Relay debugging

Browse Source
This commit is contained in:
Christien Rioux 2025-02-16 03:03:05 +00:00
parent df573368c3
commit 7ba1c0e3a7
33 changed files with 513 additions and 342 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
veilid-core/src/network_manager/address_check.rs
View File

@ -299,10 +299,14 @@ impl AddressCheck {
flow: Flow, // the flow used
reporting_ipblock: IpAddr, // the IP block this report came from
) -> bool {
let acckey =
AddressCheckCacheKey(routing_domain, flow.protocol_type(), flow.address_type());
// Don't do this if we aren't ever going to use it
if !self.config.detect_address_changes {
return false;
}
// Add the currently seen socket address into the consistency table
let acckey =
AddressCheckCacheKey(routing_domain, flow.protocol_type(), flow.address_type());
let cache = self
.address_consistency_table
.entry(acckey)

4
veilid-core/src/network_manager/direct_boot.rs
View File

@ -10,9 +10,9 @@ impl NetworkManager {
let bootstrap_nodes = routing_table.find_bootstrap_nodes_filtered(2);
// Serialize out peer info
let bootstrap_peerinfo: Vec<PeerInfo> = bootstrap_nodes
let bootstrap_peerinfo: Vec<Arc<PeerInfo>> = bootstrap_nodes
.iter()
.filter_map(|nr| nr.make_peer_info(RoutingDomain::PublicInternet))
.filter_map(|nr| nr.get_peer_info(RoutingDomain::PublicInternet))
.collect();
let json_bytes = serialize_json(bootstrap_peerinfo).as_bytes().to_vec();

14
veilid-core/src/network_manager/mod.rs
View File

@ -107,6 +107,7 @@ struct NodeContactMethodCacheKey {
target_node_info_ts: Timestamp,
target_node_ref_filter: NodeRefFilter,
target_node_ref_sequencing: Sequencing,
dial_info_failures_map: BTreeMap<DialInfo, Timestamp>,
}
enum SendDataToExistingFlowResult {
@ -1179,19 +1180,6 @@ impl NetworkManager {
self.net().restart_network();
}
/// If some other subsystem believes our dial info is no longer valid, this will trigger
/// a re-check of the dial info and network class
pub fn set_needs_dial_info_check(&self, routing_domain: RoutingDomain) {
match routing_domain {
RoutingDomain::LocalNetwork => {
// nothing here yet
}
RoutingDomain::PublicInternet => {
self.net().set_needs_public_dial_info_check(None);
}
}
}
// Report peer info changes
fn peer_info_change_event_handler(&self, evt: Arc<PeerInfoChangeEvent>) {
let mut inner = self.inner.lock();

73
veilid-core/src/network_manager/native/discovery_context.rs
View File

@ -414,37 +414,48 @@ impl DiscoveryContext {
&self,
unord: &mut FuturesUnordered<SendPinBoxFuture<Option<DetectionResult>>>,
) {
let external_1 = self.inner.lock().external_info.first().cloned().unwrap();
let external_infos = self.inner.lock().external_info.clone();
let this = self.clone();
let do_no_nat_fut: SendPinBoxFuture<Option<DetectionResult>> = Box::pin(async move {
// Do a validate_dial_info on the external address from a redirected node
if this
.validate_dial_info(external_1.node.clone(), external_1.dial_info.clone(), true)
.await
{
// Add public dial info with Direct dialinfo class
Some(DetectionResult {
config: this.config,
ddi: DetectedDialInfo::Detected(DialInfoDetail {
dial_info: external_1.dial_info.clone(),
class: DialInfoClass::Direct,
}),
external_address_types: AddressTypeSet::only(external_1.address.address_type()),
})
} else {
// Add public dial info with Blocked dialinfo class
Some(DetectionResult {
config: this.config,
ddi: DetectedDialInfo::Detected(DialInfoDetail {
dial_info: external_1.dial_info.clone(),
class: DialInfoClass::Blocked,
}),
external_address_types: AddressTypeSet::only(external_1.address.address_type()),
})
}
});
unord.push(do_no_nat_fut);
// Have all the external validator nodes check us
for external_info in external_infos {
let this = self.clone();
let do_no_nat_fut: SendPinBoxFuture<Option<DetectionResult>> = Box::pin(async move {
// Do a validate_dial_info on the external address from a redirected node
if this
.validate_dial_info(
external_info.node.clone(),
external_info.dial_info.clone(),
true,
)
.await
{
// Add public dial info with Direct dialinfo class
Some(DetectionResult {
config: this.config,
ddi: DetectedDialInfo::Detected(DialInfoDetail {
dial_info: external_info.dial_info.clone(),
class: DialInfoClass::Direct,
}),
external_address_types: AddressTypeSet::only(
external_info.address.address_type(),
),
})
} else {
// Add public dial info with Blocked dialinfo class
Some(DetectionResult {
config: this.config,
ddi: DetectedDialInfo::Detected(DialInfoDetail {
dial_info: external_info.dial_info.clone(),
class: DialInfoClass::Blocked,
}),
external_address_types: AddressTypeSet::only(
external_info.address.address_type(),
),
})
}
});
unord.push(do_no_nat_fut);
}
}
// If we know we are behind NAT check what kind
@ -725,7 +736,7 @@ impl DiscoveryContext {
// NAT Detection
///////////
// If our local interface list contains external_1 then there is no NAT in place
// If our local interface list contains any of the external addresses then there is no NAT in place
let local_address_in_external_info = self
.inner
.lock()

69
veilid-core/src/network_manager/native/mod.rs
View File

@ -31,6 +31,8 @@ use std::path::{Path, PathBuf};
/////////////////////////////////////////////////////////////////
pub const MAX_DIAL_INFO_FAILURE_COUNT: usize = 100;
pub const UPDATE_OUTBOUND_ONLY_NETWORK_CLASS_PERIOD_SECS: u32 = 10;
pub const UPDATE_NETWORK_CLASS_TASK_TICK_PERIOD_SECS: u32 = 1;
pub const NETWORK_INTERFACES_TASK_TICK_PERIOD_SECS: u32 = 1;
pub const UPNP_TASK_TICK_PERIOD_SECS: u32 = 1;
@ -82,15 +84,17 @@ struct NetworkInner {
/// set if the network needs to be restarted due to a low level configuration change
/// such as dhcp release or change of address or interfaces being added or removed
network_needs_restart: bool,
/// the number of consecutive dial info failures per routing domain,
/// which may indicate the network is down for that domain
dial_info_failure_count: BTreeMap<RoutingDomain, usize>,
/// the next time we are allowed to check for better dialinfo when we are OutboundOnly
next_outbound_only_dial_info_check: Timestamp,
/// join handles for all the low level network background tasks
join_handles: Vec<MustJoinHandle<()>>,
/// stop source for shutting down the low level network background tasks
stop_source: Option<StopSource>,
/// set if we need to calculate our public dial info again
needs_public_dial_info_check: bool,
/// set if we have yet to clear the network during public dial info checking
network_already_cleared: bool,
/// the punishment closure to enax
public_dial_info_check_punishment: Option<Box<dyn FnOnce() + Send + 'static>>,
/// Actual bound addresses per protocol
@ -151,8 +155,9 @@ impl Network {
fn new_inner() -> NetworkInner {
NetworkInner {
network_needs_restart: false,
dial_info_failure_count: BTreeMap::new(),
next_outbound_only_dial_info_check: Timestamp::default(),
needs_public_dial_info_check: false,
network_already_cleared: false,
public_dial_info_check_punishment: None,
join_handles: Vec::new(),
stop_source: None,
@ -312,11 +317,42 @@ impl Network {
dial_info: DialInfo,
fut: F,
) -> EyreResult<NetworkResult<T>> {
let opt_routing_domain = self
.routing_table()
.routing_domain_for_address(dial_info.address());
let network_result = fut.await?;
if matches!(network_result, NetworkResult::NoConnection(_)) {
// Dial info failure
self.network_manager()
.address_filter()
.set_dial_info_failed(dial_info);
// Increment consecutive failure count for this routing domain
if let Some(rd) = opt_routing_domain {
let dial_info_failure_count = {
let mut inner = self.inner.lock();
*inner
.dial_info_failure_count
.entry(rd)
.and_modify(|x| *x += 1)
.or_insert(1)
};
if dial_info_failure_count == MAX_DIAL_INFO_FAILURE_COUNT {
log_net!(debug "Node may be offline. Exceeded maximum dial info failure count for {:?}", rd);
// todo: what operations should we perform here?
// self.set_needs_dial_info_check(rd);
}
}
} else {
// Dial info success
// Clear failure count for this routing domain
if let Some(rd) = opt_routing_domain {
let mut inner = self.inner.lock();
inner.dial_info_failure_count.remove(&rd);
}
}
Ok(network_result)
}
@ -793,6 +829,20 @@ impl Network {
editor_local_network.publish();
}
if self.config().with(|c| c.network.detect_address_changes) {
// Say we need to detect the public dialinfo
self.set_needs_public_dial_info_check(None);
} else {
// Warn if we have no public dialinfo, because we're not going to magically find some
// with detect address changes turned off
let pi = routing_table.get_current_peer_info(RoutingDomain::PublicInternet);
if !pi.signed_node_info().has_any_dial_info() {
warn!(
"This node has no valid public dial info.\nConfigure this node with a static public IP address and correct firewall rules."
);
}
}
Ok(StartupDisposition::Success)
}
@ -918,6 +968,17 @@ impl Network {
}
//////////////////////////////////////////
// pub fn set_needs_dial_info_check(&self, routing_domain: RoutingDomain) {
// match routing_domain {
// RoutingDomain::LocalNetwork => {
// // nothing here yet
// }
// RoutingDomain::PublicInternet => {
// self.set_needs_public_dial_info_check(None);
// }
// }
// }
pub fn set_needs_public_dial_info_check(
&self,
punishment: Option<Box<dyn FnOnce() + Send + 'static>>,

27
veilid-core/src/network_manager/native/network_state.rs
View File

@ -14,12 +14,10 @@ pub struct ProtocolConfig {
pub(super) struct NetworkState {
/// the calculated protocol configuration for inbound/outbound protocols
pub protocol_config: ProtocolConfig,
/// does our network have ipv4 on any network?
/// does our network have ipv4 on any interface?
pub enable_ipv4: bool,
/// does our network have ipv6 on the global internet?
pub enable_ipv6_global: bool,
/// does our network have ipv6 on the local network?
pub enable_ipv6_local: bool,
/// does our network have ipv6 on any interface?
pub enable_ipv6: bool,
/// The list of stable interface addresses we have last seen
pub stable_interface_addresses: Vec<IpAddr>,
/// The local networks (network+mask) most recently seen
@ -84,21 +82,15 @@ impl Network {
// determine if we have ipv4/ipv6 addresses
let mut enable_ipv4 = false;
let mut enable_ipv6_global = false;
let mut enable_ipv6_local = false;
let mut enable_ipv6 = false;
let stable_interface_addresses = self.make_stable_interface_addresses();
for addr in stable_interface_addresses.iter().copied() {
for addr in stable_interface_addresses.iter() {
if addr.is_ipv4() {
enable_ipv4 = true;
} else if addr.is_ipv6() {
let address = Address::from_ip_addr(addr);
if address.is_global() {
enable_ipv6_global = true;
} else if address.is_local() {
enable_ipv6_local = true;
}
enable_ipv6 = true;
}
}
@ -141,10 +133,8 @@ impl Network {
family_global.insert(AddressType::IPV4);
family_local.insert(AddressType::IPV4);
}
if enable_ipv6_global {
if enable_ipv6 {
family_global.insert(AddressType::IPV6);
}
if enable_ipv6_local {
family_local.insert(AddressType::IPV6);
}
@ -178,8 +168,7 @@ impl Network {
Ok(NetworkState {
protocol_config,
enable_ipv4,
enable_ipv6_global,
enable_ipv6_local,
enable_ipv6,
stable_interface_addresses,
local_networks,
})

8
veilid-core/src/network_manager/native/network_tcp.rs
View File

@ -350,6 +350,14 @@ impl Network {
.or_default();
bapp.push(addr);
log_net!(
debug
"set_preferred_local_address: {:?} {:?} -> {:?}",
protocol_type,
addr,
PeerAddress::new(SocketAddress::from_socket_addr(addr), protocol_type)
);
Self::set_preferred_local_address(
&mut inner,
PeerAddress::new(SocketAddress::from_socket_addr(addr), protocol_type),

8
veilid-core/src/network_manager/native/network_udp.rs
View File

@ -158,6 +158,14 @@ impl Network {
.or_default();
bapp.push(addr);
log_net!(
debug
"set_preferred_local_address: {:?} {:?} -> {:?}",
ProtocolType::UDP,
addr,
PeerAddress::new(SocketAddress::from_socket_addr(addr), ProtocolType::UDP)
);
Self::set_preferred_local_address(
&mut inner,
PeerAddress::new(

68
veilid-core/src/network_manager/native/tasks/mod.rs
View File

@ -39,6 +39,42 @@ impl Network {
}
}
// Determine if we need to check for public dialinfo
fn needs_update_network_class_tick(&self) -> bool {
let public_internet_network_class = self
.routing_table()
.get_network_class(RoutingDomain::PublicInternet);
let needs_public_dial_info_check = self.needs_public_dial_info_check();
if needs_public_dial_info_check
|| public_internet_network_class == NetworkClass::Invalid
|| (public_internet_network_class == NetworkClass::OutboundOnly
&& self.inner.lock().next_outbound_only_dial_info_check <= Timestamp::now())
{
let routing_table = self.routing_table();
let rth = routing_table.get_routing_table_health();
// We want at least two live entries per crypto kind before we start doing this (bootstrap)
let mut has_at_least_two = true;
for ck in VALID_CRYPTO_KINDS {
if rth
.live_entry_counts
.get(&(RoutingDomain::PublicInternet, ck))
.copied()
.unwrap_or_default()
< 2
{
has_at_least_two = false;
break;
}
}
has_at_least_two
} else {
false
}
}
#[instrument(level = "trace", target = "net", name = "Network::tick", skip_all, err)]
pub async fn tick(&self) -> EyreResult<()> {
let Ok(_guard) = self.startup_lock.enter() else {
@ -62,36 +98,8 @@ impl Network {
// Check our network interfaces to see if they have changed
self.network_interfaces_task.tick().await?;
// Check our public dial info to see if it has changed
let public_internet_network_class = self
.routing_table()
.get_network_class(RoutingDomain::PublicInternet)
.unwrap_or(NetworkClass::Invalid);
let needs_public_dial_info_check = self.needs_public_dial_info_check();
if public_internet_network_class == NetworkClass::Invalid
|| needs_public_dial_info_check
{
let routing_table = self.routing_table();
let rth = routing_table.get_routing_table_health();
// We want at least two live entries per crypto kind before we start doing this (bootstrap)
let mut has_at_least_two = true;
for ck in VALID_CRYPTO_KINDS {
if rth
.live_entry_counts
.get(&(RoutingDomain::PublicInternet, ck))
.copied()
.unwrap_or_default()
< 2
{
has_at_least_two = false;
break;
}
}
if has_at_least_two {
self.update_network_class_task.tick().await?;
}
if self.needs_update_network_class_tick() {
self.update_network_class_task.tick().await?;
}
}

4
veilid-core/src/network_manager/native/tasks/network_interfaces_task.rs
View File

@ -9,7 +9,8 @@ impl Network {
_t: Timestamp,
) -> EyreResult<()> {
// Network lock ensures only one task operating on the low level network state
// can happen at the same time.
// can happen at the same time. Try lock is here to give preference to other longer
// running processes like update_network_class_task.
let _guard = match self.network_task_lock.try_lock() {
Ok(v) => v,
Err(_) => {
@ -68,7 +69,6 @@ impl Network {
// If any of the new addresses were PublicInternet addresses, re-run public dial info check
if public_internet_changed {
// inner.network_needs_restart = true;
self.set_needs_public_dial_info_check(None);
}

59
veilid-core/src/network_manager/native/tasks/update_network_class_task.rs
View File

@ -14,14 +14,9 @@ impl Network {
t: Timestamp,
) -> EyreResult<()> {
// Network lock ensures only one task operating on the low level network state
// can happen at the same time.
let _guard = match self.network_task_lock.try_lock() {
Ok(v) => v,
Err(_) => {
// If we can't get the lock right now, then
return Ok(());
}
};
// can happen at the same time. This a blocking lock so we can ensure this runs
// as soon as network_interfaces_task is finished
let _guard = self.network_task_lock.lock().await;
// Do the public dial info check
let finished = self.do_public_dial_info_check(stop_token, l, t).await?;
@ -29,7 +24,13 @@ impl Network {
// Done with public dial info check
if finished {
let mut inner = self.inner.lock();
// Note that we did the check successfully
inner.needs_public_dial_info_check = false;
// Don't try to re-do OutboundOnly dialinfo for another 10 seconds
inner.next_outbound_only_dial_info_check = Timestamp::now()
+ TimestampDuration::new_secs(UPDATE_OUTBOUND_ONLY_NETWORK_CLASS_PERIOD_SECS)
}
Ok(())
@ -94,7 +95,7 @@ impl Network {
) -> EyreResult<bool> {
// Figure out if we can optimize TCP/WS checking since they are often on the same port
let (protocol_config, inbound_protocol_map) = {
let mut inner = self.inner.lock();
let inner = self.inner.lock();
let Some(protocol_config) = inner
.network_state
.as_ref()
@ -103,9 +104,6 @@ impl Network {
bail!("should not be doing public dial info check before we have an initial network state");
};
// Allow network to be cleared if external addresses change
inner.network_already_cleared = false;
let mut inbound_protocol_map =
HashMap::<(AddressType, LowLevelProtocolType, u16), Vec<ProtocolType>>::new();
for at in protocol_config.family_global {
@ -171,6 +169,7 @@ impl Network {
// Wait for all discovery futures to complete and apply discoverycontexts
let mut external_address_types = AddressTypeSet::new();
let mut detection_results = HashMap::<DiscoveryContextConfig, DetectionResult>::new();
loop {
match unord
.next()
@ -185,8 +184,34 @@ impl Network {
// Add the external address kinds to the set we've seen
external_address_types |= dr.external_address_types;
// Import the dialinfo
self.update_with_detection_result(&mut editor, &inbound_protocol_map, dr);
// Get best detection result for each discovery context config
if let Some(cur_dr) = detection_results.get_mut(&dr.config) {
let ddi = &mut cur_dr.ddi;
// Upgrade existing dialinfo
match ddi {
DetectedDialInfo::SymmetricNAT => {
// Whatever we got is better than or equal to symmetric
*ddi = dr.ddi;
}
DetectedDialInfo::Detected(cur_did) => match dr.ddi {
DetectedDialInfo::SymmetricNAT => {
// Nothing is worse than this
}
DetectedDialInfo::Detected(did) => {
// Pick the best dial info class we detected
// because some nodes could be degenerate and if any node can validate a
// better dial info class we should go with it and leave the
// degenerate nodes in the dust to fade into obscurity
if did.class < cur_did.class {
cur_did.class = did.class;
}
}
},
}
cur_dr.external_address_types |= dr.external_address_types;
} else {
detection_results.insert(dr.config, dr);
}
}
Ok(Some(None)) => {
// Found no dial info for this protocol/address combination
@ -202,6 +227,12 @@ impl Network {
}
}
// Apply best effort coalesced detection results
for (_, dr) in detection_results {
// Import the dialinfo
self.update_with_detection_result(&mut editor, &inbound_protocol_map, dr);
}
// See if we have any discovery contexts that did not complete for a
// particular protocol type if its external address type was supported.
let mut success = true;

125
veilid-core/src/network_manager/send_data.rs
View File

@ -397,9 +397,6 @@ impl NetworkManager {
}))
}
/// Figure out how to reach a node from our own node over the best routing domain and reference the nodes we want to access
/// Uses NodeRefs to ensure nodes are referenced, this is not a part of 'RoutingTable' because RoutingTable is not
/// allowed to use NodeRefs due to recursive locking
#[instrument(level = "trace", target = "net", skip_all, err)]
pub fn get_node_contact_method(
&self,
@ -430,48 +427,76 @@ impl NetworkManager {
let peer_a = routing_table.get_current_peer_info(routing_domain);
let own_node_info_ts = peer_a.signed_node_info().timestamp();
// Peer B is the target node, get just the timestamp for the cache check
let target_node_info_ts = match target_node_ref.operate(|_rti, e| {
e.signed_node_info(routing_domain)
.map(|sni| sni.timestamp())
}) {
Some(ts) => ts,
None => {
log_net!(
"no node info for node {:?} in {:?}",
target_node_ref,
routing_domain
);
return Ok(NodeContactMethod::Unreachable);
// Peer B is the target node, get the whole peer info now
let Some(peer_b) = target_node_ref.get_peer_info(routing_domain) else {
log_net!("no node info for node {:?}", target_node_ref);
return Ok(NodeContactMethod::Unreachable);
};
// Calculate the dial info failures map
let address_filter = self.address_filter();
let dial_info_failures_map = {
let mut dial_info_failures_map = BTreeMap::<DialInfo, Timestamp>::new();
for did in peer_b
.signed_node_info()
.node_info()
.filtered_dial_info_details(DialInfoDetail::NO_SORT, &|_| true)
{
if let Some(ts) = address_filter.get_dial_info_failed_ts(&did.dial_info) {
dial_info_failures_map.insert(did.dial_info, ts);
}
}
dial_info_failures_map
};
// Get cache key
let mut ncm_key = NodeContactMethodCacheKey {
let ncm_key = NodeContactMethodCacheKey {
node_ids: target_node_ref.node_ids(),
own_node_info_ts,
target_node_info_ts,
target_node_info_ts: peer_b.signed_node_info().timestamp(),
target_node_ref_filter: target_node_ref.filter(),
target_node_ref_sequencing: target_node_ref.sequencing(),
dial_info_failures_map,
};
if let Some(ncm) = self.inner.lock().node_contact_method_cache.get(&ncm_key) {
return Ok(ncm.clone());
}
// Peer B is the target node, get the whole peer info now
let peer_b = match target_node_ref.make_peer_info(routing_domain) {
Some(pi) => Arc::new(pi),
None => {
log_net!("no node info for node {:?}", target_node_ref);
return Ok(NodeContactMethod::Unreachable);
}
};
// Update the key's timestamp to ensure we avoid any race conditions
ncm_key.target_node_info_ts = peer_b.signed_node_info().timestamp();
// Calculate the node contact method
let routing_table = self.routing_table();
let ncm = Self::get_node_contact_method_uncached(
&routing_table,
routing_domain,
target_node_ref,
peer_a,
peer_b,
&ncm_key,
)?;
// Cache this
self.inner
.lock()
.node_contact_method_cache
.insert(ncm_key, ncm.clone());
Ok(ncm)
}
/// Figure out how to reach a node from our own node over the best routing domain and reference the nodes we want to access
/// Uses NodeRefs to ensure nodes are referenced, this is not a part of 'RoutingTable' because RoutingTable is not
/// allowed to use NodeRefs due to recursive locking
#[instrument(level = "trace", target = "net", skip_all, err)]
fn get_node_contact_method_uncached(
routing_table: &RoutingTable,
routing_domain: RoutingDomain,
target_node_ref: FilteredNodeRef,
peer_a: Arc<PeerInfo>,
peer_b: Arc<PeerInfo>,
ncm_key: &NodeContactMethodCacheKey,
) -> EyreResult<NodeContactMethod> {
// Dial info filter comes from the target node ref but must be filtered by this node's outbound capabilities
let dial_info_filter = target_node_ref.dial_info_filter().filtered(
&DialInfoFilter::all()
DialInfoFilter::all()
.with_address_type_set(peer_a.signed_node_info().node_info().address_types())
.with_protocol_type_set(peer_a.signed_node_info().node_info().outbound_protocols()),
);
@ -487,26 +512,17 @@ impl NetworkManager {
// }
// Deprioritize dial info that have recently failed
let address_filter = self.address_filter();
let mut dial_info_failures_map = BTreeMap::<DialInfo, Timestamp>::new();
for did in peer_b
.signed_node_info()
.node_info()
.filtered_dial_info_details(DialInfoDetail::NO_SORT, |_| true)
{
if let Some(ts) = address_filter.get_dial_info_failed_ts(&did.dial_info) {
dial_info_failures_map.insert(did.dial_info, ts);
}
}
let dif_sort: Option<Arc<DialInfoDetailSort>> = if dial_info_failures_map.is_empty() {
let dif_sort: Option<DialInfoDetailSort> = if ncm_key.dial_info_failures_map.is_empty() {
None
} else {
Some(Arc::new(move |a: &DialInfoDetail, b: &DialInfoDetail| {
let ats = dial_info_failures_map
Some(Box::new(|a: &DialInfoDetail, b: &DialInfoDetail| {
let ats = ncm_key
.dial_info_failures_map
.get(&a.dial_info)
.copied()
.unwrap_or_default();
let bts = dial_info_failures_map
let bts = ncm_key
.dial_info_failures_map
.get(&b.dial_info)
.copied()
.unwrap_or_default();
@ -521,7 +537,7 @@ impl NetworkManager {
peer_b.clone(),
dial_info_filter,
sequencing,
dif_sort,
dif_sort.as_ref(),
);
// Translate the raw contact method to a referenced contact method
@ -549,7 +565,7 @@ impl NetworkManager {
let tighten = peer_a
.signed_node_info()
.node_info()
.filtered_dial_info_details(DialInfoDetail::NO_SORT, |did| {
.filtered_dial_info_details(DialInfoDetail::NO_SORT, &|did| {
did.matches_filter(&dial_info_filter)
})
.iter()
@ -634,11 +650,6 @@ impl NetworkManager {
}
};
// Cache this
self.inner
.lock()
.node_contact_method_cache
.insert(ncm_key, ncm.clone());
Ok(ncm)
}
@ -806,10 +817,13 @@ impl NetworkManager {
// punch should come through and create a real 'last connection' for us if this succeeds
network_result_try!(
self.net()
.send_hole_punch(hole_punch_did.dial_info.clone())
.send_data_to_dial_info(hole_punch_did.dial_info.clone(), Vec::new())
.await?
);
// Add small delay to encourage packets to be delivered in order
sleep(HOLE_PUNCH_DELAY_MS).await;
// Issue the signal
let rpc = self.rpc_processor();
network_result_try!(rpc
@ -823,6 +837,13 @@ impl NetworkManager {
.await
.wrap_err("failed to send signal")?);
// Another hole punch after the signal for UDP redundancy
network_result_try!(
self.net()
.send_data_to_dial_info(hole_punch_did.dial_info, Vec::new())
.await?
);
// Wait for the return receipt
let inbound_nr = match eventual_value
.timeout_at(stop_token)

2
veilid-core/src/network_manager/types/address.rs
View File

@ -72,7 +72,7 @@ impl Address {
pub fn canonical(&self) -> Address {
match self {
Address::IPV4(v4) => Address::IPV4(*v4),
Address::IPV6(v6) => match v6.to_ipv4() {
Address::IPV6(v6) => match v6.to_ipv4_mapped() {
Some(v4) => Address::IPV4(v4),
None => Address::IPV6(*v6),
},

4
veilid-core/src/network_manager/types/dial_info_filter.rs
View File

@ -38,7 +38,7 @@ impl DialInfoFilter {
self.address_type_set = address_set;
self
}
pub fn filtered(mut self, other_dif: &DialInfoFilter) -> Self {
pub fn filtered(mut self, other_dif: DialInfoFilter) -> Self {
self.protocol_type_set &= other_dif.protocol_type_set;
self.address_type_set &= other_dif.address_type_set;
self
@ -54,7 +54,7 @@ impl DialInfoFilter {
Sequencing::EnsureOrdered => (
true,
self.filtered(
&DialInfoFilter::all().with_protocol_type_set(ProtocolType::all_ordered_set()),
DialInfoFilter::all().with_protocol_type_set(ProtocolType::all_ordered_set()),
),
),
}

12
veilid-core/src/network_manager/wasm/mod.rs
View File

@ -336,6 +336,18 @@ impl Network {
.await
}
// Send hole punch attempt to a specific dialinfo. May not be appropriate for all protocols.
// Returns a flow for the connection used to send the data
#[instrument(level = "trace", target = "net", err, skip(self))]
pub async fn send_hole_punch(
&self,
dial_info: DialInfo,
) -> EyreResult<NetworkResult<UniqueFlow>> {
return Ok(NetworkResult::ServiceUnavailable(
"unimplemented for this platform".to_owned(),
));
}
/////////////////////////////////////////////////////////////////
pub async fn startup_internal(&self) -> EyreResult<StartupDisposition> {

58
veilid-core/src/routing_table/bucket_entry.rs
View File

@ -182,6 +182,9 @@ pub(crate) struct BucketEntryInner {
local_network: BucketEntryLocalNetwork,
/// Statistics gathered for the peer
peer_stats: PeerStats,
/// The peer info cache for speedy access to fully encapsulated per-routing-domain peer info
#[serde(skip)]
peer_info_cache: Mutex<BTreeMap<RoutingDomain, Option<Arc<PeerInfo>>>>,
/// The accounting for the latency statistics
#[serde(skip)]
latency_stats_accounting: LatencyStatsAccounting,
@ -277,7 +280,7 @@ impl BucketEntryInner {
/// Add a node id for a particular crypto kind.
/// Returns Ok(Some(node)) any previous existing node id associated with that crypto kind
/// Returns Ok(None) if no previous existing node id was associated with that crypto kind
/// Returns Ok(None) if no previous existing node id was associated with that crypto kind, or one existed but nothing changed.
/// Results Err() if this operation would add more crypto kinds than we support
pub fn add_node_id(&mut self, node_id: TypedKey) -> EyreResult<Option<TypedKey>> {
let total_node_id_count = self.validated_node_ids.len() + self.unsupported_node_ids.len();
@ -292,8 +295,13 @@ impl BucketEntryInner {
if old_node_id == node_id {
return Ok(None);
}
// Won't change number of crypto kinds
// Won't change number of crypto kinds, but the node id changed
node_ids.add(node_id);
// Also clear the peerinfo cache since the node ids changed
let mut pi_cache = self.peer_info_cache.lock();
pi_cache.clear();
return Ok(Some(old_node_id));
}
// Check to ensure we aren't adding more crypto kinds than we support
@ -301,6 +309,11 @@ impl BucketEntryInner {
bail!("too many crypto kinds for this node");
}
node_ids.add(node_id);
// Also clear the peerinfo cache since the node ids changed
let mut pi_cache = self.peer_info_cache.lock();
pi_cache.clear();
Ok(None)
}
@ -314,7 +327,14 @@ impl BucketEntryInner {
&mut self.unsupported_node_ids
};
node_ids.remove(crypto_kind)
let opt_dead_id = node_ids.remove(crypto_kind);
if opt_dead_id.is_some() {
// Also clear the peerinfo cache since the node ids changed
let mut pi_cache = self.peer_info_cache.lock();
pi_cache.clear();
}
opt_dead_id
}
pub fn best_node_id(&self) -> TypedKey {
@ -413,9 +433,6 @@ impl BucketEntryInner {
move |e1, e2| Self::cmp_fastest_reliable(cur_ts, e1, e2)
}
// xxx: if we ever implement a 'remove_signed_node_info' to take nodes out of a routing domain
// then we need to call 'on_entry_node_info_updated' with that removal. as of right now
// this never happens, because we only have one routing domain implemented.
pub fn update_signed_node_info(
&mut self,
routing_domain: RoutingDomain,
@ -429,7 +446,10 @@ impl BucketEntryInner {
// See if we have an existing signed_node_info to update or not
let mut node_info_changed = false;
let mut had_previous_node_info = false;
if let Some(current_sni) = opt_current_sni {
had_previous_node_info = true;
// Always allow overwriting unsigned node (bootstrap)
if current_sni.has_any_signature() {
// If the timestamp hasn't changed or is less, ignore this update
@ -480,6 +500,12 @@ impl BucketEntryInner {
self.clear_last_flows_except_latest();
}
// Clear the peerinfo cache since the node info changed or was added
if node_info_changed || !had_previous_node_info {
let mut pi_cache = self.peer_info_cache.lock();
pi_cache.remove(&routing_domain);
}
node_info_changed
}
@ -535,16 +561,29 @@ impl BucketEntryInner {
opt_current_sni.as_ref().map(|s| s.as_ref())
}
pub fn make_peer_info(&self, routing_domain: RoutingDomain) -> Option<PeerInfo> {
pub fn get_peer_info(&self, routing_domain: RoutingDomain) -> Option<Arc<PeerInfo>> {
// Return cached peer info if we have it
let mut pi_cache = self.peer_info_cache.lock();
if let Some(opt_pi) = pi_cache.get(&routing_domain).cloned() {
return opt_pi;
}
// Create a new peerinfo
let opt_current_sni = match routing_domain {
RoutingDomain::LocalNetwork => &self.local_network.signed_node_info,
RoutingDomain::PublicInternet => &self.public_internet.signed_node_info,
};
// Peer info includes all node ids, even unvalidated ones
let node_ids = self.node_ids();
opt_current_sni
let opt_pi = opt_current_sni
.as_ref()
.map(|s| PeerInfo::new(routing_domain, node_ids, *s.clone()))
.map(|s| Arc::new(PeerInfo::new(routing_domain, node_ids, *s.clone())));
// Cache the peerinfo
pi_cache.insert(routing_domain, opt_pi.clone());
// Return the peerinfo
opt_pi
}
pub fn best_routing_domain(
@ -1138,6 +1177,7 @@ impl BucketEntry {
transfer: TransferStatsDownUp::default(),
state: StateStats::default(),
},
peer_info_cache: Mutex::new(BTreeMap::new()),
latency_stats_accounting: LatencyStatsAccounting::new(),
transfer_stats_accounting: TransferStatsAccounting::new(),
state_stats_accounting: Mutex::new(StateStatsAccounting::new()),

10
veilid-core/src/routing_table/find_peers.rs
View File

@ -133,9 +133,9 @@ impl RoutingTable {
filters,
// transform
|rti, entry| {
entry.unwrap().with(rti, |_rti, e| {
Arc::new(e.make_peer_info(routing_domain).unwrap())
})
entry
.unwrap()
.with(rti, |_rti, e| e.get_peer_info(routing_domain).unwrap())
},
) {
Ok(v) => v,
@ -187,8 +187,8 @@ impl RoutingTable {
if d_far < d_near {
let warning = format!(
r#"peer: {}
near (key): {}
far (self): {}
near (key): {}
far (self): {}
d_near: {}
d_far: {}
cmp: {:?}"#,

6
veilid-core/src/routing_table/mod.rs
View File

@ -552,7 +552,7 @@ impl RoutingTable {
peer_b: Arc<PeerInfo>,
dial_info_filter: DialInfoFilter,
sequencing: Sequencing,
dif_sort: Option<Arc<DialInfoDetailSort>>,
dif_sort: Option<&DialInfoDetailSort>,
) -> ContactMethod {
self.inner.read().get_contact_method(
routing_domain,
@ -586,7 +586,7 @@ impl RoutingTable {
/// Return the domain's currently registered network class
#[cfg_attr(all(target_arch = "wasm32", target_os = "unknown"), expect(dead_code))]
pub fn get_network_class(&self, routing_domain: RoutingDomain) -> Option<NetworkClass> {
pub fn get_network_class(&self, routing_domain: RoutingDomain) -> NetworkClass {
self.inner.read().get_network_class(routing_domain)
}
@ -796,7 +796,7 @@ impl RoutingTable {
e.with(rti, |_rti, e| {
if let Some(ni) = e.node_info(routing_domain) {
if ni
.first_filtered_dial_info_detail(DialInfoDetail::NO_SORT, |did| {
.first_filtered_dial_info_detail(DialInfoDetail::NO_SORT, &|did| {
did.matches_filter(&dial_info_filter)
})
.is_some()

2
veilid-core/src/routing_table/node_ref/filtered_node_ref.rs
View File

@ -63,7 +63,7 @@ impl FilteredNodeRef {
}
pub fn merge_filter(&mut self, filter: NodeRefFilter) {
self.filter = self.filter.filtered(&filter);
self.filter = self.filter.filtered(filter);
}
pub fn set_sequencing(&mut self, sequencing: Sequencing) {

4
veilid-core/src/routing_table/node_ref/node_ref_filter.rs
View File

@ -49,11 +49,11 @@ impl NodeRefFilter {
self.dial_info_filter = self.dial_info_filter.with_address_type_set(address_set);
self
}
pub fn filtered(mut self, other_filter: &NodeRefFilter) -> Self {
pub fn filtered(mut self, other_filter: NodeRefFilter) -> Self {
self.routing_domain_set &= other_filter.routing_domain_set;
self.dial_info_filter = self
.dial_info_filter
.filtered(&other_filter.dial_info_filter);
.filtered(other_filter.dial_info_filter);
self
}
#[expect(dead_code)]

8
veilid-core/src/routing_table/node_ref/traits.rs
View File

@ -70,8 +70,8 @@ pub(crate) trait NodeRefCommonTrait: NodeRefAccessorsTrait + NodeRefOperateTrait
self.operate(|_rti, e| e.peer_stats().clone())
}
fn make_peer_info(&self, routing_domain: RoutingDomain) -> Option<PeerInfo> {
self.operate(|_rti, e| e.make_peer_info(routing_domain))
fn get_peer_info(&self, routing_domain: RoutingDomain) -> Option<Arc<PeerInfo>> {
self.operate(|_rti, e| e.get_peer_info(routing_domain))
}
fn node_info(&self, routing_domain: RoutingDomain) -> Option<NodeInfo> {
self.operate(|_rti, e| e.node_info(routing_domain).cloned())
@ -150,7 +150,7 @@ pub(crate) trait NodeRefCommonTrait: NodeRefAccessorsTrait + NodeRefOperateTrait
self.operate(|_rt, e| {
for routing_domain in routing_domain_set {
if let Some(ni) = e.node_info(routing_domain) {
if let Some(did) = ni.first_filtered_dial_info_detail(sort, filter) {
if let Some(did) = ni.first_filtered_dial_info_detail(sort.as_ref(), &filter) {
return Some(did);
}
}
@ -177,7 +177,7 @@ pub(crate) trait NodeRefCommonTrait: NodeRefAccessorsTrait + NodeRefOperateTrait
self.operate(|_rt, e| {
for routing_domain in routing_domain_set {
if let Some(ni) = e.node_info(routing_domain) {
let mut dids = ni.filtered_dial_info_details(sort, filter);
let mut dids = ni.filtered_dial_info_details(sort.as_ref(), &filter);
out.append(&mut dids);
}
}

16
veilid-core/src/routing_table/route_spec_store/mod.rs
View File

@ -493,11 +493,9 @@ impl RouteSpecStore {
let nodes_pi: Vec<Arc<PeerInfo>> = nodes
.iter()
.map(|nr| {
Arc::new(
nr.locked(rti)
.make_peer_info(RoutingDomain::PublicInternet)
.unwrap(),
)
nr.locked(rti)
.get_peer_info(RoutingDomain::PublicInternet)
.unwrap()
})
.collect();
@ -1260,14 +1258,14 @@ impl RouteSpecStore {
let pi = rti
.with_node_entry(node_id, |entry| {
entry.with(rti, |_rti, e| {
e.make_peer_info(RoutingDomain::PublicInternet)
e.get_peer_info(RoutingDomain::PublicInternet)
})
})
.flatten();
if pi.is_none() {
apibail_internal!("peer info should exist for route but doesn't");
}
RouteNode::PeerInfo(Arc::new(pi.unwrap()))
RouteNode::PeerInfo(pi.unwrap())
},
next_hop: Some(route_hop_data),
};
@ -1504,14 +1502,14 @@ impl RouteSpecStore {
let pi = rti
.with_node_entry(node_id, |entry| {
entry.with(rti, |_rti, e| {
e.make_peer_info(RoutingDomain::PublicInternet)
e.get_peer_info(RoutingDomain::PublicInternet)
})
})
.flatten();
if pi.is_none() {
apibail_internal!("peer info should exist for route but doesn't");
}
RouteNode::PeerInfo(Arc::new(pi.unwrap()))
RouteNode::PeerInfo(pi.unwrap())
},
next_hop: Some(route_hop_data),
}

20
veilid-core/src/routing_table/routing_table_inner/mod.rs
View File

@ -219,7 +219,7 @@ impl RoutingTableInner {
peer_b: Arc<PeerInfo>,
dial_info_filter: DialInfoFilter,
sequencing: Sequencing,
dif_sort: Option<Arc<DialInfoDetailSort>>,
dif_sort: Option<&DialInfoDetailSort>,
) -> ContactMethod {
self.with_routing_domain(routing_domain, |rdd| {
rdd.get_contact_method(self, peer_a, peer_b, dial_info_filter, sequencing, dif_sort)
@ -256,7 +256,7 @@ impl RoutingTableInner {
}
/// Return the domain's currently registered network class
pub fn get_network_class(&self, routing_domain: RoutingDomain) -> Option<NetworkClass> {
pub fn get_network_class(&self, routing_domain: RoutingDomain) -> NetworkClass {
self.with_routing_domain(routing_domain, |rdd| rdd.network_class())
}
@ -624,7 +624,7 @@ impl RoutingTableInner {
// New node id, get the old peer info if we don't have it yet
if old_peer_infos.is_empty() {
for rd in RoutingDomainSet::all() {
if let Some(old_peer_info) = e.make_peer_info(rd) {
if let Some(old_peer_info) = e.get_peer_info(rd) {
old_peer_infos.push(old_peer_info);
}
}
@ -667,7 +667,7 @@ impl RoutingTableInner {
if !old_peer_infos.is_empty() {
let mut new_peer_infos = vec![];
for rd in RoutingDomainSet::all() {
if let Some(new_peer_info) = e.make_peer_info(rd) {
if let Some(new_peer_info) = e.get_peer_info(rd) {
new_peer_infos.push(new_peer_info);
}
}
@ -895,13 +895,13 @@ impl RoutingTableInner {
let mut updated = false;
let mut old_peer_info = None;
let nr = self.create_node_ref(&node_ids, |_rti, e| {
old_peer_info = e.make_peer_info(routing_domain);
old_peer_info = e.get_peer_info(routing_domain);
updated = e.update_signed_node_info(routing_domain, &signed_node_info);
})?;
// Process any new or updated PeerInfo
if old_peer_info.is_none() || updated {
let new_peer_info = nr.locked(self).make_peer_info(routing_domain);
let new_peer_info = nr.locked(self).get_peer_info(routing_domain);
self.on_entry_peer_info_updated(old_peer_info, new_peer_info);
}
@ -940,8 +940,8 @@ impl RoutingTableInner {
/// 2. nodes are removed that don't have any peer info
fn on_entry_peer_info_updated(
&mut self,
old_peer_info: Option<PeerInfo>,
new_peer_info: Option<PeerInfo>,
old_peer_info: Option<Arc<PeerInfo>>,
new_peer_info: Option<Arc<PeerInfo>>,
) {
let (routing_domain, node_ids) = match (old_peer_info.as_ref(), new_peer_info.as_ref()) {
(None, None) => {
@ -1114,9 +1114,7 @@ impl RoutingTableInner {
) -> Arc<PeerInfo> {
match entry {
None => own_peer_info.clone(),
Some(entry) => {
entry.with_inner(|e| Arc::new(e.make_peer_info(routing_domain).unwrap()))
}
Some(entry) => entry.with_inner(|e| e.get_peer_info(routing_domain).unwrap()),
}
}

11
veilid-core/src/routing_table/routing_table_inner/routing_domains/local_network/mod.rs
View File

@ -55,7 +55,7 @@ impl RoutingDomainDetail for LocalNetworkRoutingDomainDetail {
RoutingDomain::LocalNetwork
}
fn network_class(&self) -> Option<NetworkClass> {
fn network_class(&self) -> NetworkClass {
self.common.network_class()
}
fn outbound_protocols(&self) -> ProtocolTypeSet {
@ -67,11 +67,14 @@ impl RoutingDomainDetail for LocalNetworkRoutingDomainDetail {
fn address_types(&self) -> AddressTypeSet {
self.common.address_types()
}
fn compatible_address_types(&self) -> AddressTypeSet {
AddressType::IPV4 | AddressType::IPV6
}
fn capabilities(&self) -> Vec<Capability> {
self.common.capabilities()
}
fn requires_relay(&self) -> Option<RelayKind> {
self.common.requires_relay()
self.common.requires_relay(self.compatible_address_types())
}
fn relay_node(&self) -> Option<FilteredNodeRef> {
self.common.relay_node()
@ -182,7 +185,7 @@ impl RoutingDomainDetail for LocalNetworkRoutingDomainDetail {
peer_b: Arc<PeerInfo>,
dial_info_filter: DialInfoFilter,
sequencing: Sequencing,
dif_sort: Option<Arc<DialInfoDetailSort>>,
dif_sort: Option<&DialInfoDetailSort>,
) -> ContactMethod {
// Get the nodeinfos for convenience
let node_a = peer_a.signed_node_info().node_info();
@ -198,7 +201,7 @@ impl RoutingDomainDetail for LocalNetworkRoutingDomainDetail {
if let Some(target_did) = first_filtered_dial_info_detail_between_nodes(
node_a,
node_b,
&dial_info_filter,
dial_info_filter,
sequencing,
dif_sort,
) {

57
veilid-core/src/routing_table/routing_table_inner/routing_domains/mod.rs
View File

@ -13,10 +13,11 @@ pub trait RoutingDomainDetail {
// Common accessors
#[expect(dead_code)]
fn routing_domain(&self) -> RoutingDomain;
fn network_class(&self) -> Option<NetworkClass>;
fn network_class(&self) -> NetworkClass;
fn outbound_protocols(&self) -> ProtocolTypeSet;
fn inbound_protocols(&self) -> ProtocolTypeSet;
fn address_types(&self) -> AddressTypeSet;
fn compatible_address_types(&self) -> AddressTypeSet;
fn capabilities(&self) -> Vec<Capability>;
fn requires_relay(&self) -> Option<RelayKind>;
fn relay_node(&self) -> Option<FilteredNodeRef>;
@ -47,7 +48,7 @@ pub trait RoutingDomainDetail {
peer_b: Arc<PeerInfo>,
dial_info_filter: DialInfoFilter,
sequencing: Sequencing,
dif_sort: Option<Arc<DialInfoDetailSort>>,
dif_sort: Option<&DialInfoDetailSort>,
) -> ContactMethod;
// Set last relay keepalive time
@ -63,13 +64,13 @@ trait RoutingDomainDetailCommonAccessors: RoutingDomainDetail {
fn first_filtered_dial_info_detail_between_nodes(
from_node: &NodeInfo,
to_node: &NodeInfo,
dial_info_filter: &DialInfoFilter,
dial_info_filter: DialInfoFilter,
sequencing: Sequencing,
dif_sort: Option<Arc<DialInfoDetailSort>>,
dif_sort: Option<&DialInfoDetailSort>,
) -> Option<DialInfoDetail> {
// Consider outbound capabilities
let dial_info_filter = (*dial_info_filter).filtered(
&DialInfoFilter::all()
let dial_info_filter = dial_info_filter.filtered(
DialInfoFilter::all()
.with_address_type_set(from_node.address_types())
.with_protocol_type_set(from_node.outbound_protocols()),
);
@ -79,9 +80,9 @@ fn first_filtered_dial_info_detail_between_nodes(
// based on an external preference table, for example the one kept by
// AddressFilter to deprioritize dialinfo that have recently failed to connect
let (ordered, dial_info_filter) = dial_info_filter.apply_sequencing(sequencing);
let sort: Option<Box<DialInfoDetailSort>> = if ordered {
let sort: Option<DialInfoDetailSort> = if ordered {
if let Some(dif_sort) = dif_sort {
Some(Box::new(move |a, b| {
Some(Box::new(|a, b| {
let mut ord = dif_sort(a, b);
if ord == core::cmp::Ordering::Equal {
ord = DialInfoDetail::ordered_sequencing_sort(a, b);
@ -89,12 +90,12 @@ fn first_filtered_dial_info_detail_between_nodes(
ord
}))
} else {
Some(Box::new(move |a, b| {
Some(Box::new(|a, b| {
DialInfoDetail::ordered_sequencing_sort(a, b)
}))
}
} else if let Some(dif_sort) = dif_sort {
Some(Box::new(move |a, b| dif_sort(a, b)))
Some(Box::new(|a, b| dif_sort(a, b)))
} else {
None
};
@ -106,7 +107,7 @@ fn first_filtered_dial_info_detail_between_nodes(
// Get the best match dial info for node B if we have it
let direct_filter = |did: &DialInfoDetail| did.matches_filter(&dial_info_filter);
to_node.first_filtered_dial_info_detail(sort, direct_filter)
to_node.first_filtered_dial_info_detail(sort.as_ref(), &direct_filter)
}
#[derive(Debug)]
@ -141,22 +142,18 @@ impl RoutingDomainDetailCommon {
///////////////////////////////////////////////////////////////////////
// Accessors
pub fn network_class(&self) -> Option<NetworkClass> {
pub fn network_class(&self) -> NetworkClass {
cfg_if! {
if #[cfg(all(target_arch = "wasm32", target_os = "unknown"))] {
Some(NetworkClass::WebApp)
NetworkClass::WebApp
} else {
if self.address_types.is_empty() {
None
NetworkClass::Invalid
}
else if self.dial_info_details.is_empty() {
if self.relay_node.is_none() {
None
} else {
Some(NetworkClass::OutboundOnly)
}
NetworkClass::OutboundOnly
} else {
Some(NetworkClass::InboundCapable)
NetworkClass::InboundCapable
}
}
}
@ -178,23 +175,30 @@ impl RoutingDomainDetailCommon {
self.capabilities.clone()
}
pub fn requires_relay(&self) -> Option<RelayKind> {
match self.network_class()? {
pub fn requires_relay(&self, compatible_address_types: AddressTypeSet) -> Option<RelayKind> {
match self.network_class() {
NetworkClass::InboundCapable => {
let mut all_inbound_set: HashSet<(ProtocolType, AddressType)> = HashSet::new();
let mut address_types = AddressTypeSet::empty();
for p in self.inbound_protocols {
for a in self.address_types {
all_inbound_set.insert((p, a));
}
}
for did in &self.dial_info_details {
// Request an inbound relay if any of our dialinfo require one
if did.class.requires_relay() {
return Some(RelayKind::Inbound);
}
let ib = (did.dial_info.protocol_type(), did.dial_info.address_type());
all_inbound_set.remove(&ib);
address_types |= did.dial_info.address_type();
}
if !all_inbound_set.is_empty() {
// Request an inbound relay if any of our inbound protocols do not have dialinfo for all address types
// we want to support, or if this routing domain doesn't support the full range of compatible address types
// for the routing domain
if !all_inbound_set.is_empty() || address_types != compatible_address_types {
return Some(RelayKind::Inbound);
}
}
@ -316,7 +320,7 @@ impl RoutingDomainDetailCommon {
let routing_table = rti.routing_table();
let node_info = NodeInfo::new(
self.network_class().unwrap_or(NetworkClass::Invalid),
self.network_class(),
self.outbound_protocols,
self.address_types,
VALID_ENVELOPE_VERSIONS.to_vec(),
@ -326,9 +330,10 @@ impl RoutingDomainDetailCommon {
);
let relay_info = if let Some(rn) = &self.relay_node {
let opt_relay_pi = rn.locked(rti).make_peer_info(self.routing_domain);
let opt_relay_pi = rn.locked(rti).get_peer_info(self.routing_domain);
if let Some(relay_pi) = opt_relay_pi {
let (_routing_domain, relay_ids, relay_sni) = relay_pi.destructure();
let (_routing_domain, relay_ids, relay_sni) =
relay_pi.as_ref().clone().destructure();
match relay_sni {
SignedNodeInfo::Direct(d) => Some((relay_ids, d)),
SignedNodeInfo::Relayed(_) => {

39
veilid-core/src/routing_table/routing_table_inner/routing_domains/public_internet/mod.rs
View File

@ -36,7 +36,7 @@ impl RoutingDomainDetail for PublicInternetRoutingDomainDetail {
RoutingDomain::PublicInternet
}
fn network_class(&self) -> Option<NetworkClass> {
fn network_class(&self) -> NetworkClass {
self.common.network_class()
}
fn outbound_protocols(&self) -> ProtocolTypeSet {
@ -48,11 +48,14 @@ impl RoutingDomainDetail for PublicInternetRoutingDomainDetail {
fn address_types(&self) -> AddressTypeSet {
self.common.address_types()
}
fn compatible_address_types(&self) -> AddressTypeSet {
AddressType::IPV4 | AddressType::IPV6
}
fn capabilities(&self) -> Vec<Capability> {
self.common.capabilities()
}
fn requires_relay(&self) -> Option<RelayKind> {
self.common.requires_relay()
self.common.requires_relay(self.compatible_address_types())
}
fn relay_node(&self) -> Option<FilteredNodeRef> {
self.common.relay_node()
@ -160,7 +163,7 @@ impl RoutingDomainDetail for PublicInternetRoutingDomainDetail {
peer_b: Arc<PeerInfo>,
dial_info_filter: DialInfoFilter,
sequencing: Sequencing,
dif_sort: Option<Arc<DialInfoDetailSort>>,
dif_sort: Option<&DialInfoDetailSort>,
) -> ContactMethod {
let ip6_prefix_size = rti
.config()
@ -191,9 +194,9 @@ impl RoutingDomainDetail for PublicInternetRoutingDomainDetail {
first_filtered_dial_info_detail_between_nodes(
node_a,
node_b,
&dial_info_filter,
dial_info_filter,
sequencing,
dif_sort.clone(),
dif_sort,
)
})
.flatten()
@ -228,9 +231,9 @@ impl RoutingDomainDetail for PublicInternetRoutingDomainDetail {
if first_filtered_dial_info_detail_between_nodes(
node_a,
node_b_relay,
&dial_info_filter,
dial_info_filter,
sequencing,
dif_sort.clone(),
dif_sort,
)
.is_some()
{
@ -242,9 +245,9 @@ impl RoutingDomainDetail for PublicInternetRoutingDomainDetail {
if let Some(reverse_did) = first_filtered_dial_info_detail_between_nodes(
node_b,
node_a,
&dial_info_filter,
dial_info_filter,
sequencing,
dif_sort.clone(),
dif_sort,
) {
// Ensure we aren't on the same public IP address (no hairpin nat)
if reverse_did.dial_info.ip_addr() != target_did.dial_info.ip_addr() {
@ -262,22 +265,22 @@ impl RoutingDomainDetail for PublicInternetRoutingDomainDetail {
// Does node B have a direct udp dialinfo node A can reach?
let udp_dial_info_filter = dial_info_filter
.filtered(&DialInfoFilter::all().with_protocol_type(ProtocolType::UDP));
.filtered(DialInfoFilter::all().with_protocol_type(ProtocolType::UDP));
if let Some(target_udp_did) = first_filtered_dial_info_detail_between_nodes(
node_a,
node_b,
&udp_dial_info_filter,
udp_dial_info_filter,
sequencing,
dif_sort.clone(),
dif_sort,
) {
// Does node A have a direct udp dialinfo that node B can reach?
if let Some(reverse_udp_did) =
first_filtered_dial_info_detail_between_nodes(
node_b,
node_a,
&udp_dial_info_filter,
udp_dial_info_filter,
sequencing,
dif_sort.clone(),
dif_sort,
)
{
// Ensure we aren't on the same public IP address (no hairpin nat)
@ -322,9 +325,9 @@ impl RoutingDomainDetail for PublicInternetRoutingDomainDetail {
if first_filtered_dial_info_detail_between_nodes(
node_a,
node_b_relay,
&dial_info_filter,
dial_info_filter,
sequencing,
dif_sort.clone(),
dif_sort,
)
.is_some()
{
@ -337,9 +340,9 @@ impl RoutingDomainDetail for PublicInternetRoutingDomainDetail {
first_filtered_dial_info_detail_between_nodes(
node_b,
node_a,
&dial_info_filter,
dial_info_filter,
sequencing,
dif_sort.clone(),
dif_sort,
)
})
.flatten()

4
veilid-core/src/routing_table/tasks/bootstrap.rs
View File

@ -330,12 +330,12 @@ impl RoutingTable {
peers: Vec<Arc<PeerInfo>>,
stop_token: StopToken,
) -> EyreResult<()> {
log_rtab!(debug " bootstrapped peers: {:?}", &peers);
log_rtab!(debug " bootstrap peers: {:?}", &peers);
// Get crypto kinds to bootstrap
let crypto_kinds = self.get_bootstrap_crypto_kinds();
log_rtab!(debug " bootstrapped crypto kinds: {:?}", &crypto_kinds);
log_rtab!(debug " bootstrap crypto kinds: {:?}", &crypto_kinds);
// Run all bootstrap operations concurrently
let mut unord = FuturesUnordered::<SendPinBoxFuture<()>>::new();

20
veilid-core/src/routing_table/tasks/relay_management.rs
View File

@ -6,14 +6,8 @@ impl RoutingTable {
fn public_internet_wants_relay(&self) -> Option<RelayKind> {
let own_peer_info = self.get_current_peer_info(RoutingDomain::PublicInternet);
let own_node_info = own_peer_info.signed_node_info().node_info();
let network_class = own_node_info.network_class();
// Never give a relay to something with an invalid network class
if matches!(network_class, NetworkClass::Invalid) {
return None;
}
// If we -need- a relay always request one
// If we need a relay, always request one
let requires_relay = self
.inner
.read()
@ -22,16 +16,6 @@ impl RoutingTable {
return Some(rk);
}
// If we don't always need a relay, but we don't have support for
// all the address types then we should request one anyway
let mut address_types = AddressTypeSet::empty();
for did in own_node_info.dial_info_detail_list() {
address_types |= did.dial_info.address_type();
}
if address_types != AddressTypeSet::all() {
return Some(RelayKind::Inbound);
}
// If we are behind some NAT, then we should get ourselves a relay just
// in case we need to navigate hairpin NAT to our own network
let mut inbound_addresses = HashSet::<SocketAddr>::new();
@ -200,7 +184,7 @@ impl RoutingTable {
// Disqualify nodes that don't cover all our inbound ports for tcp and udp
// as we need to be able to use the relay for keepalives for all nat mappings
let mut low_level_protocol_ports = mapped_port_info.low_level_protocol_ports.clone();
let dids = node_info.filtered_dial_info_details(DialInfoDetail::NO_SORT, |did| {
let dids = node_info.filtered_dial_info_details(DialInfoDetail::NO_SORT, &|did| {
did.matches_filter(&outbound_dif)
});
for did in &dids {

7
veilid-core/src/routing_table/types/dial_info_detail.rs
View File

@ -19,7 +19,8 @@ impl MatchesDialInfoFilter for DialInfoDetail {
}
}
pub type DialInfoDetailSort = dyn Fn(&DialInfoDetail, &DialInfoDetail) -> core::cmp::Ordering;
pub type DialInfoDetailSort<'a> =
Box<dyn Fn(&DialInfoDetail, &DialInfoDetail) -> core::cmp::Ordering + 'a>;
impl DialInfoDetail {
pub fn ordered_sequencing_sort(a: &DialInfoDetail, b: &DialInfoDetail) -> core::cmp::Ordering {
@ -29,7 +30,5 @@ impl DialInfoDetail {
}
a.class.cmp(&b.class)
}
pub const NO_SORT: std::option::Option<
for<'r, 's> fn(&'r DialInfoDetail, &'s DialInfoDetail) -> std::cmp::Ordering,
> = None::<fn(&DialInfoDetail, &DialInfoDetail) -> core::cmp::Ordering>;
pub const NO_SORT: std::option::Option<&DialInfoDetailSort<'static>> = None::<_>;
}

10
veilid-core/src/routing_table/types/node_info.rs
View File

@ -85,10 +85,10 @@ impl NodeInfo {
&self.dial_info_detail_list
}
pub fn first_filtered_dial_info_detail<S, F>(
pub fn first_filtered_dial_info_detail<'a, S, F>(
&self,
sort: Option<S>,
filter: F,
sort: Option<&'a S>,
filter: &'a F,
) -> Option<DialInfoDetail>
where
S: Fn(&DialInfoDetail, &DialInfoDetail) -> std::cmp::Ordering,
@ -114,8 +114,8 @@ impl NodeInfo {
pub fn filtered_dial_info_details<S, F>(
&self,
sort: Option<S>,
filter: F,
sort: Option<&S>,
filter: &F,
) -> Vec<DialInfoDetail>
where
S: Fn(&DialInfoDetail, &DialInfoDetail) -> std::cmp::Ordering,

4
veilid-core/src/rpc_processor/mod.rs
View File

@ -738,7 +738,7 @@ impl RPCProcessor {
// No private route was specified for the request
// but we are using a safety route, so we must create an empty private route
// Destination relay is ignored for safety routed operations
let peer_info = match destination_node_ref.make_peer_info(routing_domain) {
let peer_info = match destination_node_ref.get_peer_info(routing_domain) {
None => {
return Ok(NetworkResult::no_connection_other(
"No peer info for stub private route",
@ -748,7 +748,7 @@ impl RPCProcessor {
};
let private_route = PrivateRoute::new_stub(
destination_node_ref.best_node_id(),
RouteNode::PeerInfo(Arc::new(peer_info)),
RouteNode::PeerInfo(peer_info),
);
// Wrap with safety route

86
veilid-flutter/example/pubspec.lock
View File

@ -13,10 +13,10 @@ packages:
dependency: transitive
description:
name: async
sha256: "947bfcf187f74dbc5e146c9eb9c0f10c9f8b30743e341481c1e2ed3ecc18c20c"
sha256: d2872f9c19731c2e5f10444b14686eb7cc85c76274bd6c16e1816bff9a3bab63
url: "https://pub.dev"
source: hosted
version: "2.11.0"
version: "2.12.0"
async_tools:
dependency: transitive
description:
@ -29,10 +29,10 @@ packages:
dependency: transitive
description:
name: boolean_selector
sha256: "6cfb5af12253eaf2b368f07bacc5a80d1301a071c73360d746b7f2e32d762c66"
sha256: "8aab1771e1243a5063b8b0ff68042d67334e3feab9e95b9490f9a6ebf73b42ea"
url: "https://pub.dev"
source: hosted
version: "2.1.1"
version: "2.1.2"
change_case:
dependency: transitive
description:
@ -45,10 +45,10 @@ packages:
dependency: transitive
description:
name: characters
sha256: "04a925763edad70e8443c99234dc3328f442e811f1d8fd1a72f1c8ad0f69a605"
sha256: f71061c654a3380576a52b451dd5532377954cf9dbd272a78fc8479606670803
url: "https://pub.dev"
source: hosted
version: "1.3.0"
version: "1.4.0"
charcode:
dependency: transitive
description:
@ -61,18 +61,18 @@ packages:
dependency: transitive
description:
name: clock
sha256: cb6d7f03e1de671e34607e909a7213e31d7752be4fb66a86d29fe1eb14bfb5cf
sha256: fddb70d9b5277016c77a80201021d40a2247104d9f4aa7bab7157b7e3f05b84b
url: "https://pub.dev"
source: hosted
version: "1.1.1"
version: "1.1.2"
collection:
dependency: transitive
description:
name: collection
sha256: a1ace0a119f20aabc852d165077c036cd864315bd99b7eaa10a60100341941bf
sha256: "2f5709ae4d3d59dd8f7cd309b4e023046b57d8a6c82130785d2b0e5868084e76"
url: "https://pub.dev"
source: hosted
version: "1.19.0"
version: "1.19.1"
convert:
dependency: transitive
description:
@ -101,10 +101,10 @@ packages:
dependency: transitive
description:
name: fake_async
sha256: "511392330127add0b769b75a987850d136345d9227c6b94c96a04cf4a391bf78"
sha256: "6a95e56b2449df2273fd8c45a662d6947ce1ebb7aafe80e550a3f68297f3cacc"
url: "https://pub.dev"
source: hosted
version: "1.3.1"
version: "1.3.2"
ffi:
dependency: transitive
description:
@ -117,10 +117,10 @@ packages:
dependency: transitive
description:
name: file
sha256: "5fc22d7c25582e38ad9a8515372cd9a93834027aacf1801cf01164dac0ffa08c"
sha256: a3b4f84adafef897088c160faf7dfffb7696046cb13ae90b508c2cbc95d3b8d4
url: "https://pub.dev"
source: hosted
version: "7.0.0"
version: "7.0.1"
fixnum:
dependency: transitive
description:
@ -195,18 +195,18 @@ packages:
dependency: transitive
description:
name: leak_tracker
sha256: "7bb2830ebd849694d1ec25bf1f44582d6ac531a57a365a803a6034ff751d2d06"
sha256: c35baad643ba394b40aac41080300150a4f08fd0fd6a10378f8f7c6bc161acec
url: "https://pub.dev"
source: hosted
version: "10.0.7"
version: "10.0.8"
leak_tracker_flutter_testing:
dependency: transitive
description:
name: leak_tracker_flutter_testing
sha256: "9491a714cca3667b60b5c420da8217e6de0d1ba7a5ec322fab01758f6998f379"
sha256: f8b613e7e6a13ec79cfdc0e97638fddb3ab848452eff057653abd3edba760573
url: "https://pub.dev"
source: hosted
version: "3.0.8"
version: "3.0.9"
leak_tracker_testing:
dependency: transitive
description:
@ -243,10 +243,10 @@ packages:
dependency: transitive
description:
name: matcher
sha256: d2323aa2060500f906aa31a895b4030b6da3ebdcc5619d14ce1aada65cd161cb
sha256: dc58c723c3c24bf8d3e2d3ad3f2f9d7bd9cf43ec6feaa64181775e60190153f2
url: "https://pub.dev"
source: hosted
version: "0.12.16+1"
version: "0.12.17"
material_color_utilities:
dependency: transitive
description:
@ -259,18 +259,18 @@ packages:
dependency: transitive
description:
name: meta
sha256: bdb68674043280c3428e9ec998512fb681678676b3c54e773629ffe74419f8c7
sha256: e3641ec5d63ebf0d9b41bd43201a66e3fc79a65db5f61fc181f04cd27aab950c
url: "https://pub.dev"
source: hosted
version: "1.15.0"
version: "1.16.0"
path:
dependency: "direct main"
description:
name: path
sha256: "087ce49c3f0dc39180befefc60fdb4acd8f8620e5682fe2476afd0b3688bb4af"
sha256: "75cca69d1490965be98c73ceaea117e8a04dd21217b37b292c9ddbec0d955bc5"
url: "https://pub.dev"
source: hosted
version: "1.9.0"
version: "1.9.1"
path_provider:
dependency: "direct main"
description:
@ -323,10 +323,10 @@ packages:
dependency: transitive
description:
name: platform
sha256: "9b71283fc13df574056616011fb138fd3b793ea47cc509c189a6c3fa5f8a1a65"
sha256: "5d6b1b0036a5f331ebc77c850ebc8506cbc1e9416c27e59b439f917a902a4984"
url: "https://pub.dev"
source: hosted
version: "3.1.5"
version: "3.1.6"
plugin_platform_interface:
dependency: transitive
description:
@ -339,10 +339,10 @@ packages:
dependency: transitive
description:
name: process
sha256: "21e54fd2faf1b5bdd5102afd25012184a6793927648ea81eea80552ac9405b32"
sha256: "107d8be718f120bbba9dcd1e95e3bd325b1b4a4f07db64154635ba03f2567a0d"
url: "https://pub.dev"
source: hosted
version: "5.0.2"
version: "5.0.3"
quiver:
dependency: transitive
description:
@ -360,34 +360,34 @@ packages:
dependency: transitive
description:
name: source_span
sha256: "53e943d4206a5e30df338fd4c6e7a077e02254531b138a15aec3bd143c1a8b3c"
sha256: "254ee5351d6cb365c859e20ee823c3bb479bf4a293c22d17a9f1bf144ce86f7c"
url: "https://pub.dev"
source: hosted
version: "1.10.0"
version: "1.10.1"
stack_trace:
dependency: transitive
description:
name: stack_trace
sha256: "9f47fd3630d76be3ab26f0ee06d213679aa425996925ff3feffdec504931c377"
sha256: "8b27215b45d22309b5cddda1aa2b19bdfec9df0e765f2de506401c071d38d1b1"
url: "https://pub.dev"
source: hosted
version: "1.12.0"
version: "1.12.1"
stream_channel:
dependency: transitive
description:
name: stream_channel
sha256: ba2aa5d8cc609d96bbb2899c28934f9e1af5cddbd60a827822ea467161eb54e7
sha256: "969e04c80b8bcdf826f8f16579c7b14d780458bd97f56d107d3950fdbeef059d"
url: "https://pub.dev"
source: hosted
version: "2.1.2"
version: "2.1.4"
string_scanner:
dependency: transitive
description:
name: string_scanner
sha256: "688af5ed3402a4bde5b3a6c15fd768dbf2621a614950b17f04626c431ab3c4c3"
sha256: "921cd31725b72fe181906c6a94d987c78e3b98c2e205b397ea399d4054872b43"
url: "https://pub.dev"
source: hosted
version: "1.3.0"
version: "1.4.1"
sync_http:
dependency: transitive
description:
@ -416,18 +416,18 @@ packages:
dependency: transitive
description:
name: term_glyph
sha256: a29248a84fbb7c79282b40b8c72a1209db169a2e0542bce341da992fe1bc7e84
sha256: "7f554798625ea768a7518313e58f83891c7f5024f88e46e7182a4558850a4b8e"
url: "https://pub.dev"
source: hosted
version: "1.2.1"
version: "1.2.2"
test_api:
dependency: transitive
description:
name: test_api
sha256: "664d3a9a64782fcdeb83ce9c6b39e78fd2971d4e37827b9b06c3aa1edc5e760c"
sha256: fb31f383e2ee25fbbfe06b40fe21e1e458d14080e3c67e7ba0acfde4df4e0bbd
url: "https://pub.dev"
source: hosted
version: "0.7.3"
version: "0.7.4"
typed_data:
dependency: transitive
description:
@ -462,10 +462,10 @@ packages:
dependency: transitive
description:
name: vm_service
sha256: f6be3ed8bd01289b34d679c2b62226f63c0e69f9fd2e50a6b3c1c729a961041b
sha256: "0968250880a6c5fe7edc067ed0a13d4bae1577fe2771dcf3010d52c4a9d3ca14"
url: "https://pub.dev"
source: hosted
version: "14.3.0"
version: "14.3.1"
webdriver:
dependency: transitive
description:
@ -499,5 +499,5 @@ packages:
source: hosted
version: "0.0.6"
sdks:
dart: ">=3.5.0 <4.0.0"
dart: ">=3.7.0-0 <4.0.0"
flutter: ">=3.24.0"

8
veilid-server/src/settings.rs
View File

@ -177,8 +177,8 @@ core:
public_watch_limit: 32
member_watch_limit: 8
max_watch_expiration_ms: 600000
upnp: true
detect_address_changes: true
upnp: false
detect_address_changes: false
restricted_nat_retries: 0
tls:
certificate_path: '%CERTIFICATE_PATH%'
@ -1851,8 +1851,8 @@ mod tests {
assert_eq!(s.core.network.dht.member_watch_limit, 8u32);
assert_eq!(s.core.network.dht.max_watch_expiration_ms, 600_000u32);
//
assert!(s.core.network.upnp);
assert!(s.core.network.detect_address_changes);
assert!(!s.core.network.upnp);
assert!(!s.core.network.detect_address_changes);
assert_eq!(s.core.network.restricted_nat_retries, 0u32);
//
assert_eq!(