liz/veilid
1
0
Fork 0
Code Pull Requests Packages Releases Activity

Merge branch veilid:main into fix-lockapi-dependency

Browse Source
This commit is contained in:
Pedro Nunes
2023-08-23 15:52:56 +00:00
parent 59f4899bca 1fff6cfdcd
commit aa824c176f
36 changed files with 1058 additions and 929 deletions
Download Patch File Download Diff File Expand all files Collapse all files
veilid-core/Cargo.toml
+68 -22
View File
@@ -13,8 +13,24 @@ crate-type = ["cdylib", "staticlib", "rlib"]
# Common features
default = ["enable-crypto-vld0"]
rt-async-std = ["async-std", "async-std-resolver", "async_executors/async_std", "rtnetlink/smol_socket", "veilid-tools/rt-async-std"]
rt-tokio = ["tokio", "tokio-util", "tokio-stream", "trust-dns-resolver/tokio-runtime", "async_executors/tokio_tp", "async_executors/tokio_io", "async_executors/tokio_timer", "rtnetlink/tokio_socket", "veilid-tools/rt-tokio"]
rt-async-std = [
"async-std",
"async-std-resolver",
"async_executors/async_std",
"rtnetlink/smol_socket",
"veilid-tools/rt-async-std",
]
rt-tokio = [
"tokio",
"tokio-util",
"tokio-stream",
"trust-dns-resolver/tokio-runtime",
"async_executors/tokio_tp",
"async_executors/tokio_io",
"async_executors/tokio_timer",
"rtnetlink/tokio_socket",
"veilid-tools/rt-tokio",
]
rt-wasm-bindgen = ["veilid-tools/rt-wasm-bindgen", "async_executors/bindgen"]
# Crypto support features
@@ -36,7 +52,7 @@ network-result-extra = ["veilid-tools/network-result-extra"]
[dependencies]
# Tools
veilid-tools = { path = "../veilid-tools", features = [ "tracing" ] }
veilid-tools = { path = "../veilid-tools", features = ["tracing"] }
paste = "1.0.14"
once_cell = "1.18.0"
owning_ref = "0.4.1"
@@ -57,7 +73,7 @@ eyre = "0.6.8"
thiserror = "1.0.47"
# Data structures
enumset = { version= "1.1.2", features = ["serde"] }
enumset = { version = "1.1.2", features = ["serde"] }
keyvaluedb = { path = "../external/keyvaluedb/keyvaluedb" }
range-set-blaze = "0.1.9"
weak-table = "0.3.2"
@@ -65,16 +81,32 @@ generic-array = "0.14.7"
hashlink = { path = "../external/hashlink", features = ["serde_impl"] }
# System
futures-util = { version = "0.3.28", default_features = false, features = ["alloc"] }
futures-util = { version = "0.3.28", default_features = false, features = [
"alloc",
] }
flume = { version = "0.11.0", features = ["async"] }
parking_lot = "0.12.1"
lock_api = "0.4.10"
stop-token = { version = "0.7.0", default-features = false }
# Crypto
ed25519-dalek = { version = "2.0.0", default_features = false, features = ["alloc", "rand_core", "digest"] }
x25519-dalek = { version = "2.0.0", default_features = false, features = ["alloc", "static_secrets"] }
curve25519-dalek = { version = "4.0.0", default_features = false, features = ["alloc"] }
ed25519-dalek = { version = "2.0.0", default_features = false, features = [
"alloc",
"rand_core",
"digest",
"zeroize",
] }
x25519-dalek = { version = "2.0.0", default_features = false, features = [
"alloc",
"static_secrets",
"zeroize",
"precomputed-tables",
] }
curve25519-dalek = { version = "4.0.0", default_features = false, features = [
"alloc",
"zeroize",
"precomputed-tables",
] }
blake3 = { version = "1.4.1" }
chacha20poly1305 = "0.10.1"
chacha20 = "0.9.1"
@@ -83,17 +115,20 @@ argon2 = "0.5.1"
# Network
async-std-resolver = { version = "0.22.0", optional = true }
trust-dns-resolver = { version = "0.22.0", optional = true }
enum-as-inner = "=0.5.1" # temporary fix for trust-dns-resolver v0.22.0
enum-as-inner = "=0.5.1" # temporary fix for trust-dns-resolver v0.22.0
# Serialization
capnp = { version = "0.17.2", default_features = false }
serde = { version = "1.0.183", features = ["derive" ] }
serde = { version = "1.0.183", features = ["derive"] }
serde_json = { version = "1.0.105" }
serde-big-array = "0.5.1"
json = "0.12.4"
data-encoding = { version = "2.4.0" }
schemars = "0.8.12"
lz4_flex = { version = "0.11.1", default-features = false, features = ["safe-encode", "safe-decode"] }
lz4_flex = { version = "0.11.1", default-features = false, features = [
"safe-encode",
"safe-decode",
] }
# Dependencies for native builds only
# Linux, Windows, Mac, iOS, Android
@@ -107,12 +142,17 @@ libc = "0.2.147"
nix = "0.26.2"
# System
async-std = { version = "1.12.0", features = ["unstable"], optional = true}
tokio = { version = "1.32.0", features = ["full"], optional = true}
tokio-util = { version = "0.7.8", features = ["compat"], optional = true}
tokio-stream = { version = "0.1.14", features = ["net"], optional = true}
async-std = { version = "1.12.0", features = ["unstable"], optional = true }
tokio = { version = "1.32.0", features = ["full"], optional = true }
tokio-util = { version = "0.7.8", features = ["compat"], optional = true }
tokio-stream = { version = "0.1.14", features = ["net"], optional = true }
async-io = { version = "1.13.0" }
futures-util = { version = "0.3.28", default-features = false, features = ["async-await", "sink", "std", "io"] }
futures-util = { version = "0.3.28", default-features = false, features = [
"async-await",
"sink",
"std",
"io",
] }
# Data structures
keyring-manager = { path = "../external/keyring-manager" }
@@ -120,7 +160,7 @@ keyvaluedb-sqlite = { path = "../external/keyvaluedb/keyvaluedb-sqlite" }
# Network
async-tungstenite = { version = "0.23.0", features = ["async-tls"] }
igd = { path = "../external/rust-igd" }
igd = { path = "../external/rust-igd" }
async-tls = "0.12.0"
webpki = "0.22.0"
webpki-roots = "0.25.2"
@@ -135,7 +175,10 @@ socket2 = { version = "0.5.3", features = ["all"] }
getrandom = { version = "0.2.4", features = ["js"] }
# System
async_executors = { version = "0.7.0", default-features = false, features = [ "bindgen", "timer" ]}
async_executors = { version = "0.7.0", default-features = false, features = [
"bindgen",
"timer",
] }
async-lock = "2.8.0"
wasm-bindgen = "0.2.87"
js-sys = "0.3.64"
@@ -182,14 +225,17 @@ ifstructs = "0.1.1"
# Dependencies for Linux or Android
[target.'cfg(any(target_os = "android", target_os = "linux"))'.dependencies]
rtnetlink = { version = "=0.13.0", default-features = false}
rtnetlink = { version = "=0.13.0", default-features = false }
netlink-sys = { version = "=0.8.5" }
netlink-packet-route = { version = "=0.17.0" }
# Dependencies for Windows
[target.'cfg(target_os = "windows")'.dependencies]
winapi = { version = "0.3.9", features = [ "iptypes", "iphlpapi" ] }
windows = { version = "0.51.1", features = [ "Win32_NetworkManagement_Dns", "Win32_Foundation" ]}
winapi = { version = "0.3.9", features = ["iptypes", "iphlpapi"] }
windows = { version = "0.51.1", features = [
"Win32_NetworkManagement_Dns",
"Win32_Foundation",
] }
windows-permissions = "0.2.4"
# Dependencies for iOS
@@ -208,7 +254,7 @@ features = ["bundled"]
serial_test = "2.0.0"
[target.'cfg(not(target_arch = "wasm32"))'.dev-dependencies]
simplelog = { version = "0.12.1", features=["test"] }
simplelog = { version = "0.12.1", features = ["test"] }
[target.'cfg(target_arch = "wasm32")'.dev-dependencies]
wasm-bindgen-test = "0.3.37"
veilid-core/src/lib.rs
+3 -1
View File
@@ -61,7 +61,7 @@ pub fn veilid_version() -> (u32, u32, u32) {
#[cfg(target_os = "android")]
pub use intf::android::veilid_core_setup_android;
pub static DEFAULT_LOG_IGNORE_LIST: [&str; 21] = [
pub static DEFAULT_LOG_IGNORE_LIST: [&str; 23] = [
"mio",
"h2",
"hyper",
@@ -83,6 +83,8 @@ pub static DEFAULT_LOG_IGNORE_LIST: [&str; 21] = [
"trust_dns_resolver",
"trust_dns_proto",
"attohttpc",
"ws_stream_wasm",
"keyvaluedb_web",
];
use cfg_if::*;
veilid-core/src/network_manager/address_filter.rs
+61
View File
@@ -4,6 +4,8 @@ use alloc::collections::btree_map::Entry;
// XXX: Move to config eventually?
const PUNISHMENT_DURATION_MIN: usize = 60;
const MAX_PUNISHMENTS_BY_NODE_ID: usize = 65536;
const DIAL_INFO_FAILURE_DURATION_MIN: usize = 10;
const MAX_DIAL_INFO_FAILURES: usize = 65536;
#[derive(ThisError, Debug, Clone, Copy, PartialEq, Eq)]
pub enum AddressFilterError {
@@ -28,6 +30,7 @@ struct AddressFilterInner {
punishments_by_ip4: BTreeMap<Ipv4Addr, Timestamp>,
punishments_by_ip6_prefix: BTreeMap<Ipv6Addr, Timestamp>,
punishments_by_node_id: BTreeMap<TypedKey, Timestamp>,
dial_info_failures: BTreeMap<DialInfo, Timestamp>,
}
struct AddressFilterUnlockedInner {
@@ -36,6 +39,7 @@ struct AddressFilterUnlockedInner {
max_connections_per_ip6_prefix_size: usize,
max_connection_frequency_per_min: usize,
punishment_duration_min: usize,
dial_info_failure_duration_min: usize,
routing_table: RoutingTable,
}
@@ -56,6 +60,10 @@ impl fmt::Debug for AddressFilterUnlockedInner {
&self.max_connection_frequency_per_min,
)
.field("punishment_duration_min", &self.punishment_duration_min)
.field(
"dial_info_failure_duration_min",
&self.dial_info_failure_duration_min,
)
.finish()
}
}
@@ -78,6 +86,7 @@ impl AddressFilter {
max_connection_frequency_per_min: c.network.max_connection_frequency_per_min
as usize,
punishment_duration_min: PUNISHMENT_DURATION_MIN,
dial_info_failure_duration_min: DIAL_INFO_FAILURE_DURATION_MIN,
routing_table,
}),
inner: Arc::new(Mutex::new(AddressFilterInner {
@@ -88,10 +97,17 @@ impl AddressFilter {
punishments_by_ip4: BTreeMap::new(),
punishments_by_ip6_prefix: BTreeMap::new(),
punishments_by_node_id: BTreeMap::new(),
dial_info_failures: BTreeMap::new(),
})),
}
}
// When the network restarts, some of the address filter can be cleared
pub fn restart(&self) {
let mut inner = self.inner.lock();
inner.dial_info_failures.clear();
}
fn purge_old_timestamps(&self, inner: &mut AddressFilterInner, cur_ts: Timestamp) {
// v4
{
@@ -180,6 +196,22 @@ impl AddressFilter {
}
}
}
// dial info
{
let mut dead_keys = Vec::<DialInfo>::new();
for (key, value) in &mut inner.dial_info_failures {
// Drop failures older than the failure duration
if cur_ts.as_u64().saturating_sub(value.as_u64())
> self.unlocked_inner.dial_info_failure_duration_min as u64 * 60_000_000u64
{
dead_keys.push(key.clone());
}
}
for key in dead_keys {
log_net!(debug ">>> DIALINFO PERMIT: {}", key);
inner.dial_info_failures.remove(&key);
}
}
}
fn is_ip_addr_punished_inner(&self, inner: &AddressFilterInner, ipblock: IpAddr) -> bool {
@@ -198,6 +230,14 @@ impl AddressFilter {
false
}
fn get_dial_info_failed_ts_inner(
&self,
inner: &AddressFilterInner,
dial_info: &DialInfo,
) -> Option<Timestamp> {
inner.dial_info_failures.get(dial_info).copied()
}
pub fn is_ip_addr_punished(&self, addr: IpAddr) -> bool {
let inner = self.inner.lock();
let ipblock = ip_to_ipblock(
@@ -207,6 +247,27 @@ impl AddressFilter {
self.is_ip_addr_punished_inner(&*inner, ipblock)
}
pub fn get_dial_info_failed_ts(&self, dial_info: &DialInfo) -> Option<Timestamp> {
let inner = self.inner.lock();
self.get_dial_info_failed_ts_inner(&*inner, dial_info)
}
pub fn set_dial_info_failed(&self, dial_info: DialInfo) {
let ts = get_aligned_timestamp();
let mut inner = self.inner.lock();
if inner.dial_info_failures.len() >= MAX_DIAL_INFO_FAILURES {
log_net!(debug ">>> DIALINFO FAILURE TABLE FULL: {}", dial_info);
return;
}
log_net!(debug ">>> DIALINFO FAILURE: {:?}", dial_info);
inner
.dial_info_failures
.entry(dial_info)
.and_modify(|v| *v = ts)
.or_insert(ts);
}
pub fn punish_ip_addr(&self, addr: IpAddr) {
log_net!(debug ">>> PUNISHED: {}", addr);
let ts = get_aligned_timestamp();
veilid-core/src/network_manager/mod.rs
+4 -14
View File
@@ -211,20 +211,7 @@ impl NetworkManager {
// Make the network key
let network_key = {
let c = config.get();
let network_key_password = if let Some(nkp) = c.network.network_key_password.clone() {
Some(nkp)
} else {
if c.network
.routing_table
.bootstrap
.contains(&"bootstrap.veilid.net".to_owned())
{
None
} else {
Some(c.network.routing_table.bootstrap.join(","))
}
};
let network_key_password = c.network.network_key_password.clone();
let network_key = if let Some(network_key_password) = network_key_password {
if !network_key_password.is_empty() {
info!("Using network key");
@@ -380,6 +367,9 @@ impl NetworkManager {
return Ok(());
}
// Clean address filter for things that should not be persistent
self.address_filter().restart();
// Create network components
let connection_manager = ConnectionManager::new(self.clone());
let net = Network::new(
veilid-core/src/network_manager/native/mod.rs
+180 -149
View File
@@ -384,6 +384,21 @@ impl Network {
////////////////////////////////////////////////////////////
// Record DialInfo failures
pub async fn record_dial_info_failure<T, F: Future<Output = EyreResult<NetworkResult<T>>>>(
&self,
dial_info: DialInfo,
fut: F,
) -> EyreResult<NetworkResult<T>> {
let network_result = fut.await?;
if matches!(network_result, NetworkResult::NoConnection(_)) {
self.network_manager()
.address_filter()
.set_dial_info_failed(dial_info);
}
Ok(network_result)
}
// Send data to a dial info, unbound, using a new connection from a random port
// This creates a short-lived connection in the case of connection-oriented protocols
// for the purpose of sending this one message.
@@ -394,59 +409,62 @@ impl Network {
dial_info: DialInfo,
data: Vec<u8>,
) -> EyreResult<NetworkResult<()>> {
let data_len = data.len();
let connect_timeout_ms = {
let c = self.config.get();
c.network.connection_initial_timeout_ms
};
self.record_dial_info_failure(dial_info.clone(), async move {
let data_len = data.len();
let connect_timeout_ms = {
let c = self.config.get();
c.network.connection_initial_timeout_ms
};
if self
.network_manager()
.address_filter()
.is_ip_addr_punished(dial_info.address().to_ip_addr())
{
return Ok(NetworkResult::no_connection_other("punished"));
}
if self
.network_manager()
.address_filter()
.is_ip_addr_punished(dial_info.address().to_ip_addr())
{
return Ok(NetworkResult::no_connection_other("punished"));
}
match dial_info.protocol_type() {
ProtocolType::UDP => {
let peer_socket_addr = dial_info.to_socket_addr();
let h = RawUdpProtocolHandler::new_unspecified_bound_handler(&peer_socket_addr)
match dial_info.protocol_type() {
ProtocolType::UDP => {
let peer_socket_addr = dial_info.to_socket_addr();
let h = RawUdpProtocolHandler::new_unspecified_bound_handler(&peer_socket_addr)
.await
.wrap_err("create socket failure")?;
let _ = network_result_try!(h
.send_message(data, peer_socket_addr)
.await
.map(NetworkResult::Value)
.wrap_err("send message failure")?);
}
ProtocolType::TCP => {
let peer_socket_addr = dial_info.to_socket_addr();
let pnc = network_result_try!(RawTcpProtocolHandler::connect(
None,
peer_socket_addr,
connect_timeout_ms
)
.await
.wrap_err("create socket failure")?;
let _ = network_result_try!(h
.send_message(data, peer_socket_addr)
.wrap_err("connect failure")?);
network_result_try!(pnc.send(data).await.wrap_err("send failure")?);
}
ProtocolType::WS | ProtocolType::WSS => {
let pnc = network_result_try!(WebsocketProtocolHandler::connect(
None,
&dial_info,
connect_timeout_ms
)
.await
.map(NetworkResult::Value)
.wrap_err("send message failure")?);
.wrap_err("connect failure")?);
network_result_try!(pnc.send(data).await.wrap_err("send failure")?);
}
}
ProtocolType::TCP => {
let peer_socket_addr = dial_info.to_socket_addr();
let pnc = network_result_try!(RawTcpProtocolHandler::connect(
None,
peer_socket_addr,
connect_timeout_ms
)
.await
.wrap_err("connect failure")?);
network_result_try!(pnc.send(data).await.wrap_err("send failure")?);
}
ProtocolType::WS | ProtocolType::WSS => {
let pnc = network_result_try!(WebsocketProtocolHandler::connect(
None,
&dial_info,
connect_timeout_ms
)
.await
.wrap_err("connect failure")?);
network_result_try!(pnc.send(data).await.wrap_err("send failure")?);
}
}
// Network accounting
self.network_manager()
.stats_packet_sent(dial_info.to_ip_addr(), ByteCount::new(data_len as u64));
// Network accounting
self.network_manager()
.stats_packet_sent(dial_info.to_ip_addr(), ByteCount::new(data_len as u64));
Ok(NetworkResult::Value(()))
Ok(NetworkResult::Value(()))
})
.await
}
// Send data to a dial info, unbound, using a new connection from a random port
@@ -461,85 +479,95 @@ impl Network {
data: Vec<u8>,
timeout_ms: u32,
) -> EyreResult<NetworkResult<Vec<u8>>> {
let data_len = data.len();
let connect_timeout_ms = {
let c = self.config.get();
c.network.connection_initial_timeout_ms
};
self.record_dial_info_failure(dial_info.clone(), async move {
let data_len = data.len();
let connect_timeout_ms = {
let c = self.config.get();
c.network.connection_initial_timeout_ms
};
if self
.network_manager()
.address_filter()
.is_ip_addr_punished(dial_info.address().to_ip_addr())
{
return Ok(NetworkResult::no_connection_other("punished"));
}
match dial_info.protocol_type() {
ProtocolType::UDP => {
let peer_socket_addr = dial_info.to_socket_addr();
let h = RawUdpProtocolHandler::new_unspecified_bound_handler(&peer_socket_addr)
.await
.wrap_err("create socket failure")?;
network_result_try!(h
.send_message(data, peer_socket_addr)
.await
.wrap_err("send message failure")?);
self.network_manager()
.stats_packet_sent(dial_info.to_ip_addr(), ByteCount::new(data_len as u64));
// receive single response
let mut out = vec![0u8; MAX_MESSAGE_SIZE];
let (recv_len, recv_addr) = network_result_try!(timeout(
timeout_ms,
h.recv_message(&mut out).instrument(Span::current())
)
.await
.into_network_result())
.wrap_err("recv_message failure")?;
let recv_socket_addr = recv_addr.remote_address().to_socket_addr();
self.network_manager()
.stats_packet_rcvd(recv_socket_addr.ip(), ByteCount::new(recv_len as u64));
// if the from address is not the same as the one we sent to, then drop this
if recv_socket_addr != peer_socket_addr {
bail!("wrong address");
}
out.resize(recv_len, 0u8);
Ok(NetworkResult::Value(out))
if self
.network_manager()
.address_filter()
.is_ip_addr_punished(dial_info.address().to_ip_addr())
{
return Ok(NetworkResult::no_connection_other("punished"));
}
ProtocolType::TCP | ProtocolType::WS | ProtocolType::WSS => {
let pnc = network_result_try!(match dial_info.protocol_type() {
ProtocolType::UDP => unreachable!(),
ProtocolType::TCP => {
let peer_socket_addr = dial_info.to_socket_addr();
RawTcpProtocolHandler::connect(None, peer_socket_addr, connect_timeout_ms)
.await
.wrap_err("connect failure")?
}
ProtocolType::WS | ProtocolType::WSS => {
WebsocketProtocolHandler::connect(None, &dial_info, connect_timeout_ms)
.await
.wrap_err("connect failure")?
}
});
network_result_try!(pnc.send(data).await.wrap_err("send failure")?);
self.network_manager()
.stats_packet_sent(dial_info.to_ip_addr(), ByteCount::new(data_len as u64));
match dial_info.protocol_type() {
ProtocolType::UDP => {
let peer_socket_addr = dial_info.to_socket_addr();
let h = RawUdpProtocolHandler::new_unspecified_bound_handler(&peer_socket_addr)
.await
.wrap_err("create socket failure")?;
network_result_try!(h
.send_message(data, peer_socket_addr)
.await
.wrap_err("send message failure")?);
self.network_manager()
.stats_packet_sent(dial_info.to_ip_addr(), ByteCount::new(data_len as u64));
let out = network_result_try!(network_result_try!(timeout(timeout_ms, pnc.recv())
// receive single response
let mut out = vec![0u8; MAX_MESSAGE_SIZE];
let (recv_len, recv_addr) = network_result_try!(timeout(
timeout_ms,
h.recv_message(&mut out).instrument(Span::current())
)
.await
.into_network_result())
.wrap_err("recv failure")?);
.wrap_err("recv_message failure")?;
self.network_manager()
.stats_packet_rcvd(dial_info.to_ip_addr(), ByteCount::new(out.len() as u64));
let recv_socket_addr = recv_addr.remote_address().to_socket_addr();
self.network_manager()
.stats_packet_rcvd(recv_socket_addr.ip(), ByteCount::new(recv_len as u64));
Ok(NetworkResult::Value(out))
// if the from address is not the same as the one we sent to, then drop this
if recv_socket_addr != peer_socket_addr {
bail!("wrong address");
}
out.resize(recv_len, 0u8);
Ok(NetworkResult::Value(out))
}
ProtocolType::TCP | ProtocolType::WS | ProtocolType::WSS => {
let pnc = network_result_try!(match dial_info.protocol_type() {
ProtocolType::UDP => unreachable!(),
ProtocolType::TCP => {
let peer_socket_addr = dial_info.to_socket_addr();
RawTcpProtocolHandler::connect(
None,
peer_socket_addr,
connect_timeout_ms,
)
.await
.wrap_err("connect failure")?
}
ProtocolType::WS | ProtocolType::WSS => {
WebsocketProtocolHandler::connect(None, &dial_info, connect_timeout_ms)
.await
.wrap_err("connect failure")?
}
});
network_result_try!(pnc.send(data).await.wrap_err("send failure")?);
self.network_manager()
.stats_packet_sent(dial_info.to_ip_addr(), ByteCount::new(data_len as u64));
let out =
network_result_try!(network_result_try!(timeout(timeout_ms, pnc.recv())
.await
.into_network_result())
.wrap_err("recv failure")?);
self.network_manager().stats_packet_rcvd(
dial_info.to_ip_addr(),
ByteCount::new(out.len() as u64),
);
Ok(NetworkResult::Value(out))
}
}
}
})
.await
}
#[cfg_attr(feature="verbose-tracing", instrument(level="trace", err, skip(self, data), fields(data.len = data.len())))]
@@ -561,7 +589,7 @@ impl Network {
network_result_value_or_log!(ph.clone()
.send_message(data.clone(), peer_socket_addr)
.await
.wrap_err("sending data to existing conection")? => [ format!(": data.len={}, descriptor={:?}", data.len(), descriptor) ]
.wrap_err("sending data to existing connection")? => [ format!(": data.len={}, descriptor={:?}", data.len(), descriptor) ]
{ return Ok(Some(data)); } );
// Network accounting
@@ -609,41 +637,44 @@ impl Network {
dial_info: DialInfo,
data: Vec<u8>,
) -> EyreResult<NetworkResult<ConnectionDescriptor>> {
let data_len = data.len();
let connection_descriptor;
if dial_info.protocol_type() == ProtocolType::UDP {
// Handle connectionless protocol
let peer_socket_addr = dial_info.to_socket_addr();
let ph = match self.find_best_udp_protocol_handler(&peer_socket_addr, &None) {
Some(ph) => ph,
None => bail!("no appropriate UDP protocol handler for dial_info"),
};
connection_descriptor = network_result_try!(ph
.send_message(data, peer_socket_addr)
.await
.wrap_err("failed to send data to dial info")?);
} else {
// Handle connection-oriented protocols
let conn = network_result_try!(
self.connection_manager()
.get_or_create_connection(dial_info.clone())
.await?
);
self.record_dial_info_failure(dial_info.clone(), async move {
let data_len = data.len();
let connection_descriptor;
if dial_info.protocol_type() == ProtocolType::UDP {
// Handle connectionless protocol
let peer_socket_addr = dial_info.to_socket_addr();
let ph = match self.find_best_udp_protocol_handler(&peer_socket_addr, &None) {
Some(ph) => ph,
None => bail!("no appropriate UDP protocol handler for dial_info"),
};
connection_descriptor = network_result_try!(ph
.send_message(data, peer_socket_addr)
.await
.wrap_err("failed to send data to dial info")?);
} else {
// Handle connection-oriented protocols
let conn = network_result_try!(
self.connection_manager()
.get_or_create_connection(dial_info.clone())
.await?
);
if let ConnectionHandleSendResult::NotSent(_) = conn.send_async(data).await {
return Ok(NetworkResult::NoConnection(io::Error::new(
io::ErrorKind::ConnectionReset,
"failed to send",
)));
if let ConnectionHandleSendResult::NotSent(_) = conn.send_async(data).await {
return Ok(NetworkResult::NoConnection(io::Error::new(
io::ErrorKind::ConnectionReset,
"failed to send",
)));
}
connection_descriptor = conn.connection_descriptor();
}
connection_descriptor = conn.connection_descriptor();
}
// Network accounting
self.network_manager()
.stats_packet_sent(dial_info.to_ip_addr(), ByteCount::new(data_len as u64));
// Network accounting
self.network_manager()
.stats_packet_sent(dial_info.to_ip_addr(), ByteCount::new(data_len as u64));
Ok(NetworkResult::value(connection_descriptor))
Ok(NetworkResult::value(connection_descriptor))
})
.await
}
/////////////////////////////////////////////////////////////////
veilid-core/src/network_manager/send_data.rs
+19
View File
@@ -357,6 +357,24 @@ impl NetworkManager {
// log_net!(debug "Node contact failing over to Ordered for {}", target_node_ref.to_string().cyan());
// sequencing = Sequencing::PreferOrdered;
// }
// 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().all_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<dyn Fn(&DialInfoDetail, &DialInfoDetail) -> core::cmp::Ordering>> = if dial_info_failures_map.is_empty() {
None
} else {
Some(Arc::new(move |a: &DialInfoDetail, b: &DialInfoDetail| {
let ats = dial_info_failures_map.get(&a.dial_info).copied().unwrap_or_default();
let bts = dial_info_failures_map.get(&b.dial_info).copied().unwrap_or_default();
ats.cmp(&bts)
}))
};
// Get the best contact method with these parameters from the routing domain
let cm = routing_table.get_contact_method(
@@ -365,6 +383,7 @@ impl NetworkManager {
&peer_b,
dial_info_filter,
sequencing,
dif_sort,
);
// Translate the raw contact method to a referenced contact method
veilid-core/src/network_manager/wasm/mod.rs
+126 -98
View File
@@ -118,47 +118,66 @@ impl Network {
/////////////////////////////////////////////////////////////////
// Record DialInfo failures
pub async fn record_dial_info_failure<T, F: Future<Output = EyreResult<NetworkResult<T>>>>(
&self,
dial_info: DialInfo,
fut: F,
) -> EyreResult<NetworkResult<T>> {
let network_result = fut.await?;
if matches!(network_result, NetworkResult::NoConnection(_)) {
self.network_manager()
.address_filter()
.set_dial_info_failed(dial_info);
}
Ok(network_result)
}
#[cfg_attr(feature="verbose-tracing", instrument(level="trace", err, skip(self, data), fields(data.len = data.len())))]
pub async fn send_data_unbound_to_dial_info(
&self,
dial_info: DialInfo,
data: Vec<u8>,
) -> EyreResult<NetworkResult<()>> {
let data_len = data.len();
let timeout_ms = {
let c = self.config.get();
c.network.connection_initial_timeout_ms
};
self.record_dial_info_failure(dial_info.clone(), async move {
let data_len = data.len();
let timeout_ms = {
let c = self.config.get();
c.network.connection_initial_timeout_ms
};
if self
.network_manager()
.address_filter()
.is_ip_addr_punished(dial_info.address().to_ip_addr())
{
return Ok(NetworkResult::no_connection_other("punished"));
}
match dial_info.protocol_type() {
ProtocolType::UDP => {
bail!("no support for UDP protocol")
if self
.network_manager()
.address_filter()
.is_ip_addr_punished(dial_info.address().to_ip_addr())
{
return Ok(NetworkResult::no_connection_other("punished"));
}
ProtocolType::TCP => {
bail!("no support for TCP protocol")
}
ProtocolType::WS | ProtocolType::WSS => {
let pnc =
network_result_try!(WebsocketProtocolHandler::connect(&dial_info, timeout_ms)
.await
.wrap_err("connect failure")?);
network_result_try!(pnc.send(data).await.wrap_err("send failure")?);
}
};
// Network accounting
self.network_manager()
.stats_packet_sent(dial_info.to_ip_addr(), ByteCount::new(data_len as u64));
match dial_info.protocol_type() {
ProtocolType::UDP => {
bail!("no support for UDP protocol")
}
ProtocolType::TCP => {
bail!("no support for TCP protocol")
}
ProtocolType::WS | ProtocolType::WSS => {
let pnc = network_result_try!(WebsocketProtocolHandler::connect(
&dial_info, timeout_ms
)
.await
.wrap_err("connect failure")?);
network_result_try!(pnc.send(data).await.wrap_err("send failure")?);
}
};
Ok(NetworkResult::Value(()))
// Network accounting
self.network_manager()
.stats_packet_sent(dial_info.to_ip_addr(), ByteCount::new(data_len as u64));
Ok(NetworkResult::Value(()))
})
.await
}
// Send data to a dial info, unbound, using a new connection from a random port
@@ -173,53 +192,59 @@ impl Network {
data: Vec<u8>,
timeout_ms: u32,
) -> EyreResult<NetworkResult<Vec<u8>>> {
let data_len = data.len();
let connect_timeout_ms = {
let c = self.config.get();
c.network.connection_initial_timeout_ms
};
self.record_dial_info_failure(dial_info.clone(), async move {
let data_len = data.len();
let connect_timeout_ms = {
let c = self.config.get();
c.network.connection_initial_timeout_ms
};
if self
.network_manager()
.address_filter()
.is_ip_addr_punished(dial_info.address().to_ip_addr())
{
return Ok(NetworkResult::no_connection_other("punished"));
}
if self
.network_manager()
.address_filter()
.is_ip_addr_punished(dial_info.address().to_ip_addr())
{
return Ok(NetworkResult::no_connection_other("punished"));
}
match dial_info.protocol_type() {
ProtocolType::UDP => {
bail!("no support for UDP protocol")
}
ProtocolType::TCP => {
bail!("no support for TCP protocol")
}
ProtocolType::WS | ProtocolType::WSS => {
let pnc = network_result_try!(match dial_info.protocol_type() {
ProtocolType::UDP => unreachable!(),
ProtocolType::TCP => unreachable!(),
ProtocolType::WS | ProtocolType::WSS => {
WebsocketProtocolHandler::connect(&dial_info, connect_timeout_ms)
match dial_info.protocol_type() {
ProtocolType::UDP => {
bail!("no support for UDP protocol")
}
ProtocolType::TCP => {
bail!("no support for TCP protocol")
}
ProtocolType::WS | ProtocolType::WSS => {
let pnc = network_result_try!(match dial_info.protocol_type() {
ProtocolType::UDP => unreachable!(),
ProtocolType::TCP => unreachable!(),
ProtocolType::WS | ProtocolType::WSS => {
WebsocketProtocolHandler::connect(&dial_info, connect_timeout_ms)
.await
.wrap_err("connect failure")?
}
});
network_result_try!(pnc.send(data).await.wrap_err("send failure")?);
self.network_manager()
.stats_packet_sent(dial_info.to_ip_addr(), ByteCount::new(data_len as u64));
let out =
network_result_try!(network_result_try!(timeout(timeout_ms, pnc.recv())
.await
.wrap_err("connect failure")?
}
});
.into_network_result())
.wrap_err("recv failure")?);
network_result_try!(pnc.send(data).await.wrap_err("send failure")?);
self.network_manager()
.stats_packet_sent(dial_info.to_ip_addr(), ByteCount::new(data_len as u64));
self.network_manager().stats_packet_rcvd(
dial_info.to_ip_addr(),
ByteCount::new(out.len() as u64),
);
let out = network_result_try!(network_result_try!(timeout(timeout_ms, pnc.recv())
.await
.into_network_result())
.wrap_err("recv failure")?);
self.network_manager()
.stats_packet_rcvd(dial_info.to_ip_addr(), ByteCount::new(out.len() as u64));
Ok(NetworkResult::Value(out))
Ok(NetworkResult::Value(out))
}
}
}
})
.await
}
#[cfg_attr(feature="verbose-tracing", instrument(level="trace", err, skip(self, data), fields(data.len = data.len())))]
@@ -273,34 +298,37 @@ impl Network {
dial_info: DialInfo,
data: Vec<u8>,
) -> EyreResult<NetworkResult<ConnectionDescriptor>> {
let data_len = data.len();
if dial_info.protocol_type() == ProtocolType::UDP {
bail!("no support for UDP protocol");
}
if dial_info.protocol_type() == ProtocolType::TCP {
bail!("no support for TCP protocol");
}
self.record_dial_info_failure(dial_info.clone(), async move {
let data_len = data.len();
if dial_info.protocol_type() == ProtocolType::UDP {
bail!("no support for UDP protocol");
}
if dial_info.protocol_type() == ProtocolType::TCP {
bail!("no support for TCP protocol");
}
// Handle connection-oriented protocols
let conn = network_result_try!(
self.connection_manager()
.get_or_create_connection(dial_info.clone())
.await?
);
// Handle connection-oriented protocols
let conn = network_result_try!(
self.connection_manager()
.get_or_create_connection(dial_info.clone())
.await?
);
if let ConnectionHandleSendResult::NotSent(_) = conn.send_async(data).await {
return Ok(NetworkResult::NoConnection(io::Error::new(
io::ErrorKind::ConnectionReset,
"failed to send",
)));
}
let connection_descriptor = conn.connection_descriptor();
if let ConnectionHandleSendResult::NotSent(_) = conn.send_async(data).await {
return Ok(NetworkResult::NoConnection(io::Error::new(
io::ErrorKind::ConnectionReset,
"failed to send",
)));
}
let connection_descriptor = conn.connection_descriptor();
// Network accounting
self.network_manager()
.stats_packet_sent(dial_info.to_ip_addr(), ByteCount::new(data_len as u64));
// Network accounting
self.network_manager()
.stats_packet_sent(dial_info.to_ip_addr(), ByteCount::new(data_len as u64));
Ok(NetworkResult::value(connection_descriptor))
Ok(NetworkResult::value(connection_descriptor))
})
.await
}
/////////////////////////////////////////////////////////////////
@@ -320,8 +348,8 @@ impl Network {
}
// XXX: See issue #92
let family_global = AddressTypeSet::all();
let family_local = AddressTypeSet::all();
let family_global = AddressTypeSet::from(AddressType::IPV4);
let family_local = AddressTypeSet::from(AddressType::IPV4);
ProtocolConfig {
outbound,
veilid-core/src/routing_table/bucket_entry.rs
+1 -1
View File
@@ -297,7 +297,7 @@ impl BucketEntryInner {
// If we're updating an entry's node info, purge all
// but the last connection in our last connections list
// because the dial info could have changed and its safer to just reconnect.
// because the dial info could have changed and it's safer to just reconnect.
// The latest connection would have been the once we got the new node info
// over so that connection is still valid.
if node_info_changed {
veilid-core/src/routing_table/mod.rs
+2
View File
@@ -538,6 +538,7 @@ impl RoutingTable {
peer_b: &PeerInfo,
dial_info_filter: DialInfoFilter,
sequencing: Sequencing,
dif_sort: Option<Arc<dyn Fn(&DialInfoDetail, &DialInfoDetail) -> core::cmp::Ordering>>,
) -> ContactMethod {
self.inner.read().get_contact_method(
routing_domain,
@@ -545,6 +546,7 @@ impl RoutingTable {
peer_b,
dial_info_filter,
sequencing,
dif_sort,
)
}
veilid-core/src/routing_table/route_spec_store/route_spec_store.rs
+4
View File
@@ -401,6 +401,7 @@ impl RouteSpecStore {
current_node,
DialInfoFilter::all(),
sequencing,
None,
);
if matches!(cm, ContactMethod::Unreachable) {
reachable = false;
@@ -415,6 +416,7 @@ impl RouteSpecStore {
current_node,
DialInfoFilter::all(),
Sequencing::EnsureOrdered,
None,
);
if matches!(cm, ContactMethod::Unreachable) {
can_do_sequenced = false;
@@ -438,6 +440,7 @@ impl RouteSpecStore {
current_node,
DialInfoFilter::all(),
sequencing,
None,
);
if matches!(cm, ContactMethod::Unreachable) {
reachable = false;
@@ -452,6 +455,7 @@ impl RouteSpecStore {
current_node,
DialInfoFilter::all(),
Sequencing::EnsureOrdered,
None,
);
if matches!(cm, ContactMethod::Unreachable) {
can_do_sequenced = false;
veilid-core/src/routing_table/routing_domains.rs
+54 -17
View File
@@ -220,6 +220,7 @@ pub trait RoutingDomainDetail {
peer_b: &PeerInfo,
dial_info_filter: DialInfoFilter,
sequencing: Sequencing,
dif_sort: Option<Arc<dyn Fn(&DialInfoDetail, &DialInfoDetail) -> core::cmp::Ordering>>,
) -> ContactMethod;
}
@@ -245,6 +246,7 @@ fn first_filtered_dial_info_detail_between_nodes(
to_node: &NodeInfo,
dial_info_filter: &DialInfoFilter,
sequencing: Sequencing,
dif_sort: Option<Arc<dyn Fn(&DialInfoDetail, &DialInfoDetail) -> core::cmp::Ordering>>
) -> Option<DialInfoDetail> {
let dial_info_filter = dial_info_filter.clone().filtered(
&DialInfoFilter::all()
@@ -253,11 +255,28 @@ fn first_filtered_dial_info_detail_between_nodes(
);
// Apply sequencing and get sort
// Include sorting by external dial info sort for rotating through dialinfo
// 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.with_sequencing(sequencing);
let sort = if ordered {
Some(DialInfoDetail::ordered_sequencing_sort)
let sort: Option<Box<dyn Fn(&DialInfoDetail, &DialInfoDetail) -> core::cmp::Ordering>> = if ordered {
if let Some(dif_sort) = dif_sort {
Some(Box::new(move |a, b| {
let mut ord = dif_sort(a,b);
if ord == core::cmp::Ordering::Equal {
ord = DialInfoDetail::ordered_sequencing_sort(a,b);
}
ord
}))
} else {
Some(Box::new(move |a,b| { DialInfoDetail::ordered_sequencing_sort(a,b) }))
}
} else {
None
if let Some(dif_sort) = dif_sort {
Some(Box::new(move |a,b| { dif_sort(a,b) }))
} else {
None
}
};
// If the filter is dead then we won't be able to connect
@@ -287,6 +306,7 @@ impl RoutingDomainDetail for PublicInternetRoutingDomainDetail {
peer_b: &PeerInfo,
dial_info_filter: DialInfoFilter,
sequencing: Sequencing,
dif_sort: Option<Arc<dyn Fn(&DialInfoDetail, &DialInfoDetail) -> core::cmp::Ordering>>,
) -> ContactMethod {
// Get the nodeinfos for convenience
let node_a = peer_a.signed_node_info().node_info();
@@ -304,7 +324,7 @@ impl RoutingDomainDetail for PublicInternetRoutingDomainDetail {
// Get the best match dial info for node B if we have it
if let Some(target_did) =
first_filtered_dial_info_detail_between_nodes(node_a, node_b, &dial_info_filter, sequencing)
first_filtered_dial_info_detail_between_nodes(node_a, node_b, &dial_info_filter, sequencing, dif_sort.clone())
{
// Do we need to signal before going inbound?
if !target_did.class.requires_signal() {
@@ -334,6 +354,7 @@ impl RoutingDomainDetail for PublicInternetRoutingDomainDetail {
node_b_relay,
&dial_info_filter,
sequencing,
dif_sort.clone(),
)
.is_some()
{
@@ -347,6 +368,7 @@ impl RoutingDomainDetail for PublicInternetRoutingDomainDetail {
node_a,
&dial_info_filter,
sequencing,
dif_sort.clone()
) {
// Ensure we aren't on the same public IP address (no hairpin nat)
if reverse_did.dial_info.to_ip_addr()
@@ -373,6 +395,7 @@ impl RoutingDomainDetail for PublicInternetRoutingDomainDetail {
node_b,
&udp_dial_info_filter,
sequencing,
dif_sort.clone()
) {
// 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(
@@ -380,6 +403,7 @@ impl RoutingDomainDetail for PublicInternetRoutingDomainDetail {
node_a,
&udp_dial_info_filter,
sequencing,
dif_sort.clone(),
) {
// Ensure we aren't on the same public IP address (no hairpin nat)
if reverse_udp_did.dial_info.to_ip_addr()
@@ -422,6 +446,7 @@ impl RoutingDomainDetail for PublicInternetRoutingDomainDetail {
&node_b_relay,
&dial_info_filter,
sequencing,
dif_sort.clone()
)
.is_some()
{
@@ -496,6 +521,7 @@ impl RoutingDomainDetail for LocalNetworkRoutingDomainDetail {
peer_b: &PeerInfo,
dial_info_filter: DialInfoFilter,
sequencing: Sequencing,
dif_sort: Option<Arc<dyn Fn(&DialInfoDetail, &DialInfoDetail) -> core::cmp::Ordering>>,
) -> ContactMethod {
// Scope the filter down to protocols node A can do outbound
let dial_info_filter = dial_info_filter.filtered(
@@ -504,20 +530,31 @@ impl RoutingDomainDetail for LocalNetworkRoutingDomainDetail {
.with_protocol_type_set(peer_a.signed_node_info().node_info().outbound_protocols()),
);
// Get first filtered dialinfo
let (sort, dial_info_filter) = match sequencing {
Sequencing::NoPreference => (None, dial_info_filter),
Sequencing::PreferOrdered => (
Some(DialInfoDetail::ordered_sequencing_sort),
dial_info_filter,
),
Sequencing::EnsureOrdered => (
Some(DialInfoDetail::ordered_sequencing_sort),
dial_info_filter.filtered(
&DialInfoFilter::all().with_protocol_type_set(ProtocolType::all_ordered_set()),
),
),
// Apply sequencing and get sort
// Include sorting by external dial info sort for rotating through dialinfo
// 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.with_sequencing(sequencing);
let sort: Option<Box<dyn Fn(&DialInfoDetail, &DialInfoDetail) -> core::cmp::Ordering>> = if ordered {
if let Some(dif_sort) = dif_sort {
Some(Box::new(move |a, b| {
let mut ord = dif_sort(a,b);
if ord == core::cmp::Ordering::Equal {
ord = DialInfoDetail::ordered_sequencing_sort(a,b);
}
ord
}))
} else {
Some(Box::new(move |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) }))
} else {
None
}
};
// If the filter is dead then we won't be able to connect
if dial_info_filter.is_dead() {
return ContactMethod::Unreachable;
veilid-core/src/routing_table/routing_table_inner.rs
+2 -1
View File
@@ -226,9 +226,10 @@ impl RoutingTableInner {
peer_b: &PeerInfo,
dial_info_filter: DialInfoFilter,
sequencing: Sequencing,
dif_sort: Option<Arc<dyn Fn(&DialInfoDetail, &DialInfoDetail) -> core::cmp::Ordering>>,
) -> ContactMethod {
self.with_routing_domain(routing_domain, |rdd| {
rdd.get_contact_method(self, peer_a, peer_b, dial_info_filter, sequencing)
rdd.get_contact_method(self, peer_a, peer_b, dial_info_filter, sequencing, dif_sort)
})
}
veilid-core/src/rpc_processor/coders/signed_value_data.rs
+1 -1
View File
@@ -25,7 +25,7 @@ pub fn decode_signed_value_data(
let signature = decode_signature512(&sr);
Ok(SignedValueData::new(
ValueData::new_with_seq(seq, data, writer),
ValueData::new_with_seq(seq, data, writer).map_err(RPCError::protocol)?,
signature,
))
}
veilid-core/src/storage_manager/get_value.rs
+1 -1
View File
@@ -203,7 +203,7 @@ impl StorageManager {
}
}
/// Handle a recieved 'Get Value' query
/// Handle a received 'Get Value' query
pub async fn inbound_get_value(
&self,
key: TypedKey,
veilid-core/src/storage_manager/mod.rs
+2 -2
View File
@@ -402,9 +402,9 @@ impl StorageManager {
return Ok(None);
}
let seq = last_signed_value_data.value_data().seq();
ValueData::new_with_seq(seq + 1, data, writer.key)
ValueData::new_with_seq(seq + 1, data, writer.key)?
} else {
ValueData::new(data, writer.key)
ValueData::new(data, writer.key)?
};
// Validate with schema
veilid-core/src/storage_manager/set_value.rs
+1 -1
View File
@@ -173,7 +173,7 @@ impl StorageManager {
}
}
/// Handle a recieved 'Set Value' query
/// Handle a received 'Set Value' query
/// Returns a None if the value passed in was set
/// Returns a Some(current value) if the value was older and the current value was kept
pub async fn inbound_set_value(
veilid-core/src/tests/common/test_veilid_config.rs
+3 -3
View File
@@ -167,8 +167,8 @@ pub fn setup_veilid_core() -> (UpdateCallback, ConfigCallback) {
fn config_callback(key: String) -> ConfigCallbackReturn {
match key.as_str() {
"program_name" => Ok(Box::new(String::from("VeilidCoreTests"))),
"namespace" => Ok(Box::new(String::from(""))),
"capabilities.disable" => Ok(Box::new(Vec::<FourCC>::new())),
"namespace" => Ok(Box::<String>::default()),
"capabilities.disable" => Ok(Box::<Vec::<FourCC>>::default()),
"table_store.directory" => Ok(Box::new(get_table_store_path())),
"table_store.delete" => Ok(Box::new(true)),
"block_store.directory" => Ok(Box::new(get_block_store_path())),
@@ -193,7 +193,7 @@ fn config_callback(key: String) -> ConfigCallbackReturn {
"network.network_key_password" => Ok(Box::new(Option::<String>::None)),
"network.routing_table.node_id" => Ok(Box::new(TypedKeyGroup::new())),
"network.routing_table.node_id_secret" => Ok(Box::new(TypedSecretGroup::new())),
"network.routing_table.bootstrap" => Ok(Box::new(Vec::<String>::new())),
"network.routing_table.bootstrap" => Ok(Box::<Vec::<String>>::default()),
"network.routing_table.limit_over_attached" => Ok(Box::new(64u32)),
"network.routing_table.limit_fully_attached" => Ok(Box::new(32u32)),
"network.routing_table.limit_attached_strong" => Ok(Box::new(16u32)),
veilid-core/src/veilid_api/tests/fixtures.rs
+1 -1
View File
@@ -207,6 +207,6 @@ pub fn fix_veilidvaluechange() -> VeilidValueChange {
key: fix_typedkey(),
subkeys: vec![1, 2, 3, 4],
count: 5,
value: ValueData::new_with_seq(23, b"ValueData".to_vec(), fix_cryptokey()),
value: ValueData::new_with_seq(23, b"ValueData".to_vec(), fix_cryptokey()).unwrap(),
}
}
veilid-core/src/veilid_api/types/dht/value_data.rs
+47 -7
View File
@@ -1,4 +1,5 @@
use super::*;
use veilid_api::VeilidAPIResult;
#[derive(Clone, Default, PartialOrd, PartialEq, Eq, Ord, Serialize, Deserialize, JsonSchema)]
pub struct ValueData {
@@ -17,17 +18,25 @@ pub struct ValueData {
impl ValueData {
pub const MAX_LEN: usize = 32768;
pub fn new(data: Vec<u8>, writer: PublicKey) -> Self {
assert!(data.len() <= Self::MAX_LEN);
Self {
pub fn new(data: Vec<u8>, writer: PublicKey) -> VeilidAPIResult<Self> {
if data.len() > Self::MAX_LEN {
apibail_generic!("invalid size");
}
Ok(Self {
seq: 0,
data,
writer,
}
})
}
pub fn new_with_seq(seq: ValueSeqNum, data: Vec<u8>, writer: PublicKey) -> Self {
assert!(data.len() <= Self::MAX_LEN);
Self { seq, data, writer }
pub fn new_with_seq(
seq: ValueSeqNum,
data: Vec<u8>,
writer: PublicKey,
) -> VeilidAPIResult<Self> {
if data.len() > Self::MAX_LEN {
apibail_generic!("invalid size");
}
Ok(Self { seq, data, writer })
}
pub fn seq(&self) -> ValueSeqNum {
@@ -56,3 +65,34 @@ impl fmt::Debug for ValueData {
.finish()
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn value_data_ok() {
assert!(ValueData::new(vec![0; ValueData::MAX_LEN], CryptoKey { bytes: [0; 32] }).is_ok());
assert!(ValueData::new_with_seq(
0,
vec![0; ValueData::MAX_LEN],
CryptoKey { bytes: [0; 32] }
)
.is_ok());
}
#[test]
fn value_data_too_long() {
assert!(ValueData::new(
vec![0; ValueData::MAX_LEN + 1],
CryptoKey { bytes: [0; 32] }
)
.is_err());
assert!(ValueData::new_with_seq(
0,
vec![0; ValueData::MAX_LEN + 1],
CryptoKey { bytes: [0; 32] }
)
.is_err());
}
}