Git-Mirrors/veilid
1
0
Fork 0
mirror of https://gitlab.com/veilid/veilid.git synced 2024-12-24 06:49:38 -05:00
Code Issues Packages Projects Releases Wiki Activity

add namespacing to WASM TableStore

Browse Source
This commit is contained in:
Christien Rioux 2024-08-17 01:01:30 +00:00
parent 3716c5db78
commit 7eff6d12cc
22 changed files with 567 additions and 247 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
Cargo.lock generated
View File

@ -4739,6 +4739,16 @@ dependencies = [
"winapi-util",
]
[[package]]
name = "sanitize-filename"
version = "0.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2ed72fbaf78e6f2d41744923916966c4fbe3d7c74e3037a8ee482f1115572603"
dependencies = [
"lazy_static",
"regex",
]
[[package]]
name = "scc"
version = "2.1.14"
@ -6205,6 +6215,7 @@ dependencies = [
"range-set-blaze",
"rustls",
"rustls-pemfile",
"sanitize-filename",
"schemars",
"send_wrapper 0.6.0",
"serde",

4
Earthfile
View File

@ -18,6 +18,7 @@ FROM ubuntu:18.04
ENV ZIG_VERSION=0.13.0-dev.46+3648d7df1
ENV CMAKE_VERSION_MINOR=3.30
ENV CMAKE_VERSION_PATCH=3.30.1
ENV WASM_BINDGEN_CLI_VERSION=0.2.93
ENV RUSTUP_HOME=/usr/local/rustup
ENV RUSTUP_DIST_SERVER=https://static.rust-lang.org
ENV CARGO_HOME=/usr/local/cargo
@ -54,7 +55,8 @@ deps-rust:
RUN rustup target add x86_64-linux-android
# WASM
RUN rustup target add wasm32-unknown-unknown
RUN cargo install wasm-pack wasm-bindgen-cli
RUN cargo install wasm-pack
RUN cargo install -f wasm-bindgen-cli --version $WASM_BINDGEN_CLI_VERSION
# Caching tool
RUN cargo install cargo-chef
# Install Linux cross-platform tooling

2
veilid-cli/Cargo.toml
View File

@ -3,6 +3,8 @@
name = "veilid-cli"
version = "0.3.4"
# ---
description = "Client application for connecting to a Veilid headless node"
repository = "https://gitlab.com/veilid/veilid"
authors = ["Veilid Team <contact@veilid.com>"]
edition = "2021"
license = "MPL-2.0"

2
veilid-core/Cargo.toml
View File

@ -4,6 +4,7 @@ name = "veilid-core"
version = "0.3.4"
# ---
description = "Core library used to create a Veilid node and operate it as part of an application"
repository = "https://gitlab.com/veilid/veilid"
authors = ["Veilid Team <contact@veilid.com>"]
edition = "2021"
build = "build.rs"
@ -139,6 +140,7 @@ lz4_flex = { version = "0.11.3", default-features = false, features = [
"safe-decode",
] }
indent = "0.1.1"
sanitize-filename = "0.5.0"
# Dependencies for native builds only
# Linux, Windows, Mac, iOS, Android

95
veilid-core/src/core_context.rs
View File

@ -67,7 +67,13 @@ impl ServicesContext {
info!("Veilid API starting up");
info!("init api tracing");
ApiTracingLayer::init(self.update_callback.clone()).await;
let (program_name, namespace) = {
let config = self.config.get();
(config.program_name.clone(), config.namespace.clone())
};
ApiTracingLayer::add_callback(program_name, namespace, self.update_callback.clone())
.await?;
// Set up protected store
let protected_store = ProtectedStore::new(self.config.clone());
@ -178,7 +184,14 @@ impl ServicesContext {
info!("Veilid API shutdown complete");
// api logger terminate is idempotent
ApiTracingLayer::terminate().await;
let (program_name, namespace) = {
let config = self.config.get();
(config.program_name.clone(), config.namespace.clone())
};
if let Err(e) = ApiTracingLayer::remove_callback(program_name, namespace).await {
error!("Error removing callback from ApiTracingLayer: {}", e);
}
// send final shutdown update
(self.update_callback)(VeilidUpdate::Shutdown);
@ -212,17 +225,6 @@ impl VeilidCoreContext {
Self::new_common(update_callback, config).await
}
#[instrument(level = "trace", target = "core_context", err, skip_all)]
async fn new_with_config_json(
update_callback: UpdateCallback,
config_json: String,
) -> VeilidAPIResult<VeilidCoreContext> {
// Set up config from json
let mut config = VeilidConfig::new();
config.setup_from_json(config_json, update_callback.clone())?;
Self::new_common(update_callback, config).await
}
#[instrument(level = "trace", target = "core_context", err, skip_all)]
async fn new_with_config(
update_callback: UpdateCallback,
@ -283,12 +285,16 @@ impl VeilidCoreContext {
/////////////////////////////////////////////////////////////////////////////
lazy_static::lazy_static! {
static ref INITIALIZED: AsyncMutex<bool> = AsyncMutex::new(false);
static ref INITIALIZED: AsyncMutex<HashSet<(String,String)>> = AsyncMutex::new(HashSet::new());
}
/// Initialize a Veilid node.
///
/// Must be called only once at the start of an application.
/// Must be called only once per 'program_name + namespace' combination at the start of an application.
/// The 'config_callback' must return a unique 'program_name + namespace' combination per simulataneous call to api_startup.
/// You can use the same program_name multiple times to create separate storage locations.
/// Multiple namespaces for the same program_name will use the same databases and on-disk locations, but will partition keys internally
/// to keep the namespaces distict.
///
/// * `update_callback` - called when internal state of the Veilid node changes, for example, when app-level messages are received, when private routes die and need to be reallocated, or when routing table states change.
/// * `config_callback` - called at startup to supply a configuration object directly to Veilid.
@ -299,9 +305,22 @@ pub async fn api_startup(
update_callback: UpdateCallback,
config_callback: ConfigCallback,
) -> VeilidAPIResult<VeilidAPI> {
// Get the program_name and namespace we're starting up in
let program_name = *config_callback("program_name".to_owned())?
.downcast::<String>()
.map_err(|_| {
VeilidAPIError::invalid_argument("api_startup", "config_callback", "program_name")
})?;
let namespace = *config_callback("namespace".to_owned())?
.downcast::<String>()
.map_err(|_| {
VeilidAPIError::invalid_argument("api_startup", "config_callback", "namespace")
})?;
let init_key = (program_name, namespace);
// See if we have an API started up already
let mut initialized_lock = INITIALIZED.lock().await;
if *initialized_lock {
if initialized_lock.contains(&init_key) {
apibail_already_initialized!();
}
@ -312,14 +331,18 @@ pub async fn api_startup(
// Return an API object around our context
let veilid_api = VeilidAPI::new(context);
*initialized_lock = true;
initialized_lock.insert(init_key);
Ok(veilid_api)
}
/// Initialize a Veilid node, with the configuration in JSON format.
///
/// Must be called only once at the start of an application.
/// Must be called only once per 'program_name + namespace' combination at the start of an application.
/// The 'config_json' must specify a unique 'program_name + namespace' combination per simulataneous call to api_startup.
/// You can use the same program_name multiple times to create separate storage locations.
/// Multiple namespaces for the same program_name will use the same databases and on-disk locations, but will partition keys internally
/// to keep the namespaces distict.
///
/// * `update_callback` - called when internal state of the Veilid node changes, for example, when app-level messages are received, when private routes die and need to be reallocated, or when routing table states change.
/// * `config_json` - called at startup to supply a JSON configuration object.
@ -330,21 +353,11 @@ pub async fn api_startup_json(
update_callback: UpdateCallback,
config_json: String,
) -> VeilidAPIResult<VeilidAPI> {
// See if we have an API started up already
let mut initialized_lock = INITIALIZED.lock().await;
if *initialized_lock {
apibail_already_initialized!();
}
// Parse the JSON config
let config: VeilidConfigInner =
serde_json::from_str(&config_json).map_err(VeilidAPIError::generic)?;
// Create core context
let context = VeilidCoreContext::new_with_config_json(update_callback, config_json).await?;
// Return an API object around our context
let veilid_api = VeilidAPI::new(context);
*initialized_lock = true;
Ok(veilid_api)
api_startup_config(update_callback, config).await
}
/// Initialize a Veilid node, with the configuration object.
@ -360,9 +373,15 @@ pub async fn api_startup_config(
update_callback: UpdateCallback,
config: VeilidConfigInner,
) -> VeilidAPIResult<VeilidAPI> {
// Get the program_name and namespace we're starting up in
let program_name = config.program_name.clone();
let namespace = config.namespace.clone();
let init_key = (program_name, namespace);
// See if we have an API started up already
let mut initialized_lock = INITIALIZED.lock().await;
if *initialized_lock {
if initialized_lock.contains(&init_key) {
apibail_already_initialized!();
}
@ -372,7 +391,7 @@ pub async fn api_startup_config(
// Return an API object around our context
let veilid_api = VeilidAPI::new(context);
*initialized_lock = true;
initialized_lock.insert(init_key);
Ok(veilid_api)
}
@ -380,6 +399,12 @@ pub async fn api_startup_config(
#[instrument(level = "trace", target = "core_context", skip_all)]
pub(crate) async fn api_shutdown(context: VeilidCoreContext) {
let mut initialized_lock = INITIALIZED.lock().await;
let init_key = {
let config = context.config.get();
(config.program_name.clone(), config.namespace.clone())
};
context.shutdown().await;
*initialized_lock = false;
initialized_lock.remove(&init_key);
}

108
veilid-core/src/logging/api_tracing_layer.rs
View File

@ -6,9 +6,20 @@ use once_cell::sync::OnceCell;
use tracing_subscriber::*;
struct ApiTracingLayerInner {
update_callback: UpdateCallback,
update_callbacks: HashMap<(String, String), UpdateCallback>,
}
/// API Tracing layer for 'tracing' subscribers
///
/// For normal application use one should call ApiTracingLayer::init() and insert the
/// layer into your subscriber before calling api_startup() or api_startup_json().
///
/// For apps that call api_startup() many times concurrently with different 'namespace' or
/// 'program_name', you may want to disable api tracing as it can slow the system down
/// considerably. In those cases, deferring to buffered disk-based logging files is probably a better idea.
/// At the very least, no more verbose than info-level logging is recommended when using API tracing
/// with many copies of Veilid running.
#[derive(Clone)]
pub struct ApiTracingLayer {
inner: Arc<Mutex<Option<ApiTracingLayerInner>>>,
@ -17,28 +28,11 @@ pub struct ApiTracingLayer {
static API_LOGGER: OnceCell<ApiTracingLayer> = OnceCell::new();
impl ApiTracingLayer {
fn new_inner(update_callback: UpdateCallback) -> ApiTracingLayerInner {
ApiTracingLayerInner { update_callback }
}
#[instrument(level = "debug", skip(update_callback))]
pub async fn init(update_callback: UpdateCallback) {
let api_logger = API_LOGGER.get_or_init(|| ApiTracingLayer {
inner: Arc::new(Mutex::new(None)),
});
let apilogger_inner = Some(Self::new_inner(update_callback));
*api_logger.inner.lock() = apilogger_inner;
}
#[instrument(level = "debug")]
pub async fn terminate() {
if let Some(api_logger) = API_LOGGER.get() {
let mut inner = api_logger.inner.lock();
*inner = None;
}
}
pub fn get() -> ApiTracingLayer {
/// Initialize an ApiTracingLayer singleton
///
/// This must be inserted into your tracing subscriber before you
/// call api_startup() or api_startup_json() if you are going to use api tracing.
pub fn init() -> ApiTracingLayer {
API_LOGGER
.get_or_init(|| ApiTracingLayer {
inner: Arc::new(Mutex::new(None)),
@ -46,6 +40,66 @@ impl ApiTracingLayer {
.clone()
}
fn new_inner() -> ApiTracingLayerInner {
ApiTracingLayerInner {
update_callbacks: HashMap::new(),
}
}
#[instrument(level = "debug", skip(update_callback))]
pub(crate) async fn add_callback(
program_name: String,
namespace: String,
update_callback: UpdateCallback,
) -> VeilidAPIResult<()> {
let Some(api_logger) = API_LOGGER.get() else {
// Did not init, so skip this
return Ok(());
};
let mut inner = api_logger.inner.lock();
if inner.is_none() {
*inner = Some(Self::new_inner());
}
let key = (program_name, namespace);
if inner.as_ref().unwrap().update_callbacks.contains_key(&key) {
apibail_already_initialized!();
}
inner
.as_mut()
.unwrap()
.update_callbacks
.insert(key, update_callback);
return Ok(());
}
#[instrument(level = "debug")]
pub(crate) async fn remove_callback(
program_name: String,
namespace: String,
) -> VeilidAPIResult<()> {
let key = (program_name, namespace);
if let Some(api_logger) = API_LOGGER.get() {
let mut inner = api_logger.inner.lock();
if inner.is_none() {
apibail_not_initialized!();
}
if inner
.as_mut()
.unwrap()
.update_callbacks
.remove(&key)
.is_none()
{
apibail_not_initialized!();
}
if inner.as_mut().unwrap().update_callbacks.is_empty() {
*inner = None;
}
}
Ok(())
}
fn emit_log(&self, inner: &mut ApiTracingLayerInner, meta: &Metadata<'_>, message: String) {
let level = *meta.level();
let target = meta.target();
@ -88,11 +142,15 @@ impl ApiTracingLayer {
None
};
(inner.update_callback)(VeilidUpdate::Log(Box::new(VeilidLog {
let log_update = VeilidUpdate::Log(Box::new(VeilidLog {
log_level,
message,
backtrace,
})))
}));
for cb in inner.update_callbacks.values() {
(cb)(log_update.clone());
}
}
}

26
veilid-core/src/table_store/wasm.rs
View File

@ -3,21 +3,32 @@ pub use keyvaluedb_web::*;
#[derive(Clone)]
pub struct TableStoreDriver {
_config: VeilidConfig,
config: VeilidConfig,
}
impl TableStoreDriver {
pub(crate) fn new(config: VeilidConfig) -> Self {
Self { _config: config }
Self { config }
}
fn get_namespaced_table_name(&self, table: &str) -> String {
let c = self.config.get();
let namespace = c.namespace.clone();
if namespace.is_empty() {
table.to_owned()
} else {
format!("{}_{}", namespace, table)
}
}
pub async fn open(&self, table_name: &str, column_count: u32) -> VeilidAPIResult<Database> {
let db = Database::open(table_name, column_count, false)
let namespaced_table_name = self.get_namespaced_table_name(table_name);
let db = Database::open(&namespaced_table_name, column_count, false)
.await
.map_err(VeilidAPIError::generic)?;
log_tstore!(
"opened table store '{}' with {} columns",
table_name,
namespaced_table_name,
column_count
);
Ok(db)
@ -26,11 +37,12 @@ impl TableStoreDriver {
/// Delete a TableDB table by name
pub async fn delete(&self, table_name: &str) -> VeilidAPIResult<bool> {
if is_browser() {
let out = Database::delete(table_name).await.is_ok();
let namespaced_table_name = self.get_namespaced_table_name(table_name);
let out = Database::delete(&namespaced_table_name).await.is_ok();
if out {
log_tstore!("TableStore::delete {} deleted", table_name);
log_tstore!("TableStore::delete {} deleted", namespaced_table_name);
} else {
log_tstore!(debug "TableStore::delete {} not deleted", table_name);
log_tstore!(debug "TableStore::delete {} not deleted", namespaced_table_name);
}
Ok(out)
} else {

19
veilid-core/src/tests/common/test_veilid_config.rs
View File

@ -162,6 +162,21 @@ pub fn setup_veilid_core() -> (UpdateCallback, ConfigCallback) {
(Arc::new(update_callback), Arc::new(config_callback))
}
pub fn setup_veilid_core_with_namespace<S: AsRef<str>>(
namespace: S,
) -> (UpdateCallback, ConfigCallback) {
let namespace = namespace.as_ref().to_string();
(
Arc::new(update_callback),
Arc::new(move |key| {
if key.as_str() == "namespace" {
return Ok(Box::new(namespace.clone()));
}
config_callback(key)
}),
)
}
pub fn config_callback(key: String) -> ConfigCallbackReturn {
match key.as_str() {
"program_name" => Ok(Box::new(String::from("VeilidCoreTests"))),
@ -172,7 +187,7 @@ pub fn config_callback(key: String) -> ConfigCallbackReturn {
"block_store.directory" => Ok(Box::new(get_block_store_path())),
"block_store.delete" => Ok(Box::new(true)),
"protected_store.allow_insecure_fallback" => Ok(Box::new(true)),
"protected_store.always_use_insecure_storage" => Ok(Box::new(false)),
"protected_store.always_use_insecure_storage" => Ok(Box::new(true)),
"protected_store.directory" => Ok(Box::new(get_protected_store_path())),
"protected_store.delete" => Ok(Box::new(true)),
"protected_store.device_encryption_key_password" => Ok(Box::new("".to_owned())),
@ -306,7 +321,7 @@ pub async fn test_config() {
assert_eq!(inner.block_store.directory, get_block_store_path());
assert!(inner.block_store.delete);
assert!(inner.protected_store.allow_insecure_fallback);
assert!(!inner.protected_store.always_use_insecure_storage);
assert!(inner.protected_store.always_use_insecure_storage);
assert_eq!(inner.protected_store.directory, get_protected_store_path());
assert!(inner.protected_store.delete);
assert_eq!(

128
veilid-core/src/tests/common/test_veilid_core.rs
View File

@ -28,6 +28,7 @@ pub async fn test_startup_shutdown_from_config() {
},
protected_store: VeilidConfigProtectedStore {
allow_insecure_fallback: true,
always_use_insecure_storage: true,
directory: get_protected_store_path(),
device_encryption_key_password: "".to_owned(),
delete: true,
@ -44,7 +45,7 @@ pub async fn test_startup_shutdown_from_config() {
}
pub async fn test_attach_detach() {
info!("--- test normal order ---");
trace!("test_attach_detach: --- test normal order ---");
let (update_callback, config_callback) = setup_veilid_core();
let api = api_startup(update_callback, config_callback)
.await
@ -55,7 +56,7 @@ pub async fn test_attach_detach() {
sleep(2000).await;
api.shutdown().await;
info!("--- test auto detach ---");
trace!("test_attach_detach: --- test auto detach ---");
let (update_callback, config_callback) = setup_veilid_core();
let api = api_startup(update_callback, config_callback)
.await
@ -64,7 +65,7 @@ pub async fn test_attach_detach() {
sleep(5000).await;
api.shutdown().await;
info!("--- test detach without attach ---");
trace!("test_attach_detach: --- test detach without attach ---");
let (update_callback, config_callback) = setup_veilid_core();
let api = api_startup(update_callback, config_callback)
.await
@ -73,8 +74,129 @@ pub async fn test_attach_detach() {
api.shutdown().await;
}
pub async fn test_startup_shutdown_multiple() {
trace!("test_startup_shutdown_multiple: starting");
let namespaces = (0..10).map(|x| format!("ns_{}", x)).collect::<Vec<_>>();
let mut apis = vec![];
for ns in &namespaces {
let (update_callback, config_callback) = setup_veilid_core_with_namespace(ns);
let api = api_startup(update_callback, config_callback)
.await
.expect("startup failed");
apis.push(api);
}
trace!("test_startup_shutdown_multiple: shutting down");
for api in apis {
api.shutdown().await;
}
trace!("test_startup_shutdown_multiple: finished");
}
pub async fn test_startup_shutdown_from_config_multiple() {
trace!("test_startup_from_config_multiple: starting");
let namespaces = (0..10).map(|x| format!("ns_{}", x)).collect::<Vec<_>>();
let mut apis = vec![];
for ns in &namespaces {
let config = VeilidConfigInner {
program_name: "VeilidCoreTests".into(),
namespace: ns.to_owned(),
table_store: VeilidConfigTableStore {
directory: get_table_store_path(),
delete: true,
// ..Default::default()
},
block_store: VeilidConfigBlockStore {
directory: get_block_store_path(),
delete: true,
//..Default::default()
},
protected_store: VeilidConfigProtectedStore {
allow_insecure_fallback: true,
always_use_insecure_storage: true,
directory: get_protected_store_path(),
device_encryption_key_password: "".to_owned(),
delete: true,
..Default::default()
},
..Default::default()
};
let api = api_startup_config(Arc::new(|_: VeilidUpdate| {}), config)
.await
.expect("startup failed");
apis.push(api);
}
trace!("test_startup_from_config_multiple: shutting down");
for api in apis {
api.shutdown().await;
}
trace!("test_startup_from_config_multiple: finished");
}
pub async fn test_attach_detach_multiple() {
trace!("test_attach_detach_multiple: --- test normal order ---");
let namespaces = (0..10).map(|x| format!("ns_{}", x)).collect::<Vec<_>>();
let mut apis = vec![];
for ns in &namespaces {
let (update_callback, config_callback) = setup_veilid_core_with_namespace(ns);
let api = api_startup(update_callback, config_callback)
.await
.expect("startup failed");
apis.push(api);
}
for api in &apis {
api.attach().await.unwrap();
}
sleep(5000).await;
for api in &apis {
api.detach().await.unwrap();
}
sleep(2000).await;
for api in apis {
api.shutdown().await;
}
trace!("test_attach_detach_multiple: --- test auto detach ---");
let mut apis = vec![];
for ns in &namespaces {
let (update_callback, config_callback) = setup_veilid_core_with_namespace(ns);
let api = api_startup(update_callback, config_callback)
.await
.expect("startup failed");
apis.push(api);
}
for api in &apis {
api.attach().await.unwrap();
}
sleep(5000).await;
for api in apis {
api.shutdown().await;
}
trace!("test_attach_detach_multiple: --- test detach without attach ---");
let mut apis = vec![];
for ns in &namespaces {
let (update_callback, config_callback) = setup_veilid_core_with_namespace(ns);
let api = api_startup(update_callback, config_callback)
.await
.expect("startup failed");
apis.push(api);
}
for api in &apis {
assert!(api.detach().await.is_err());
}
for api in apis {
api.shutdown().await;
}
}
pub async fn test_all() {
test_startup_shutdown().await;
test_startup_shutdown_from_config().await;
test_attach_detach().await;
test_startup_shutdown_multiple().await;
test_startup_shutdown_from_config_multiple().await;
test_attach_detach_multiple().await;
}

11
veilid-core/src/veilid_api/api.rs
View File

@ -2,8 +2,9 @@ use super::*;
/////////////////////////////////////////////////////////////////////////////////////////////////////
struct VeilidAPIInner {
pub(super) struct VeilidAPIInner {
context: Option<VeilidCoreContext>,
pub(super) debug_cache: DebugCache,
}
impl fmt::Debug for VeilidAPIInner {
@ -35,7 +36,7 @@ impl Drop for VeilidAPIInner {
/// * Reply to `AppCall` RPCs.
#[derive(Clone, Debug)]
pub struct VeilidAPI {
inner: Arc<Mutex<VeilidAPIInner>>,
pub(super) inner: Arc<Mutex<VeilidAPIInner>>,
}
impl VeilidAPI {
@ -46,6 +47,12 @@ impl VeilidAPI {
Self {
inner: Arc::new(Mutex::new(VeilidAPIInner {
context: Some(context),
debug_cache: DebugCache {
imported_routes: Vec::new(),
opened_record_contexts: once_cell::sync::Lazy::new(
hashlink::LinkedHashMap::new,
),
},
})),
}
}

286
veilid-core/src/veilid_api/debug.rs
View File

@ -9,16 +9,11 @@ use once_cell::sync::Lazy;
use routing_table::*;
#[derive(Default)]
struct DebugCache {
imported_routes: Vec<RouteId>,
opened_record_contexts: Lazy<LinkedHashMap<TypedKey, RoutingContext>>,
pub(crate) struct DebugCache {
pub imported_routes: Vec<RouteId>,
pub opened_record_contexts: Lazy<LinkedHashMap<TypedKey, RoutingContext>>,
}
static DEBUG_CACHE: Mutex<DebugCache> = Mutex::new(DebugCache {
imported_routes: Vec::new(),
opened_record_contexts: Lazy::new(LinkedHashMap::new),
});
pub fn format_opt_ts(ts: Option<TimestampDuration>) -> String {
let Some(ts) = ts else {
return "---".to_owned();
@ -217,89 +212,6 @@ fn get_node_ref_modifiers(mut node_ref: NodeRef) -> impl FnOnce(&str) -> Option<
}
}
fn get_destination(
routing_table: RoutingTable,
) -> impl FnOnce(&str) -> SendPinBoxFuture<Option<Destination>> {
move |text| {
let text = text.to_owned();
Box::pin(async move {
// Safety selection
let (text, ss) = if let Some((first, second)) = text.split_once('+') {
let ss = get_safety_selection(routing_table.clone())(second)?;
(first, Some(ss))
} else {
(text.as_str(), None)
};
if text.is_empty() {
return None;
}
if &text[0..1] == "#" {
let rss = routing_table.route_spec_store();
// Private route
let text = &text[1..];
let private_route = if let Some(prid) = get_route_id(rss.clone(), false, true)(text)
{
rss.best_remote_private_route(&prid)?
} else {
let mut dc = DEBUG_CACHE.lock();
let n = get_number(text)?;
let prid = *dc.imported_routes.get(n)?;
let Some(private_route) = rss.best_remote_private_route(&prid) else {
// Remove imported route
dc.imported_routes.remove(n);
info!("removed dead imported route {}", n);
return None;
};
private_route
};
Some(Destination::private_route(
private_route,
ss.unwrap_or(SafetySelection::Unsafe(Sequencing::default())),
))
} else {
let (text, mods) = text
.split_once('/')
.map(|x| (x.0, Some(x.1)))
.unwrap_or((text, None));
if let Some((first, second)) = text.split_once('@') {
// Relay
let mut relay_nr = get_node_ref(routing_table.clone())(second)?;
let target_nr = get_node_ref(routing_table)(first)?;
if let Some(mods) = mods {
relay_nr = get_node_ref_modifiers(relay_nr)(mods)?;
}
let mut d = Destination::relay(relay_nr, target_nr);
if let Some(ss) = ss {
d = d.with_safety(ss)
}
Some(d)
} else {
// Direct
let mut target_nr =
resolve_node_ref(routing_table, ss.unwrap_or_default())(text).await?;
if let Some(mods) = mods {
target_nr = get_node_ref_modifiers(target_nr)(mods)?;
}
let mut d = Destination::direct(target_nr);
if let Some(ss) = ss {
d = d.with_safety(ss)
}
Some(d)
}
}
})
}
}
fn get_number<T: num_traits::Num + FromStr>(text: &str) -> Option<T> {
T::from_str(text).ok()
}
@ -548,34 +460,6 @@ async fn async_get_debug_argument_at<T, G: FnOnce(&str) -> SendPinBoxFuture<Opti
Ok(val)
}
fn get_opened_dht_record_context(
args: &[String],
context: &str,
key: &str,
arg: usize,
) -> VeilidAPIResult<(TypedKey, RoutingContext)> {
let dc = DEBUG_CACHE.lock();
let key = match get_debug_argument_at(args, arg, context, key, get_dht_key_no_safety)
.ok()
.or_else(|| {
// If unspecified, use the most recent key opened or created
dc.opened_record_contexts.back().map(|kv| kv.0).copied()
}) {
Some(k) => k,
None => {
apibail_missing_argument!("no keys are opened", "key");
}
};
// Get routing context for record
let Some(rc) = dc.opened_record_contexts.get(&key).cloned() else {
apibail_missing_argument!("key is not opened", "key");
};
Ok((key, rc))
}
pub fn print_data(data: &[u8], truncate_len: Option<usize>) -> String {
// check if message body is ascii printable
let mut printable = true;
@ -875,10 +759,7 @@ impl VeilidAPI {
Ok("Connections purged".to_owned())
} else if args[0] == "routes" {
// Purge route spec store
{
let mut dc = DEBUG_CACHE.lock();
dc.imported_routes.clear();
}
self.inner.lock().debug_cache.imported_routes.clear();
let rss = self.network_manager()?.routing_table().route_spec_store();
match rss.purge().await {
Ok(_) => Ok("Routes purged".to_owned()),
@ -960,7 +841,7 @@ impl VeilidAPI {
0,
"debug_resolve",
"destination",
get_destination(routing_table.clone()),
self.clone().get_destination(routing_table.clone()),
)
.await?;
@ -1004,7 +885,7 @@ impl VeilidAPI {
0,
"debug_ping",
"destination",
get_destination(routing_table),
self.clone().get_destination(routing_table),
)
.await?;
@ -1037,7 +918,7 @@ impl VeilidAPI {
arg,
"debug_app_message",
"destination",
get_destination(routing_table),
self.clone().get_destination(routing_table),
)
.await?;
@ -1073,7 +954,7 @@ impl VeilidAPI {
arg,
"debug_app_call",
"destination",
get_destination(routing_table),
self.clone().get_destination(routing_table),
)
.await?;
@ -1209,7 +1090,8 @@ impl VeilidAPI {
let out = match rss.release_route(route_id) {
true => {
// release imported
let mut dc = DEBUG_CACHE.lock();
let mut inner = self.inner.lock();
let dc = &mut inner.debug_cache;
for (n, ir) in dc.imported_routes.iter().enumerate() {
if *ir == route_id {
dc.imported_routes.remove(n);
@ -1351,7 +1233,8 @@ impl VeilidAPI {
.import_remote_private_route_blob(blob_dec)
.map_err(VeilidAPIError::generic)?;
let mut dc = DEBUG_CACHE.lock();
let mut inner = self.inner.lock();
let dc = &mut inner.debug_cache;
let n = dc.imported_routes.len();
let out = format!("Private route #{} imported: {}", n, route_id);
dc.imported_routes.push(route_id);
@ -1508,7 +1391,8 @@ impl VeilidAPI {
};
// Save routing context for record
let mut dc = DEBUG_CACHE.lock();
let mut inner = self.inner.lock();
let dc = &mut inner.debug_cache;
dc.opened_record_contexts.insert(*record.key(), rc);
Ok(format!(
@ -1552,14 +1436,17 @@ impl VeilidAPI {
};
// Save routing context for record
let mut dc = DEBUG_CACHE.lock();
let mut inner = self.inner.lock();
let dc = &mut inner.debug_cache;
dc.opened_record_contexts.insert(*record.key(), rc);
Ok(format!("Opened: {} : {:?}", key, record))
}
async fn debug_record_close(&self, args: Vec<String>) -> VeilidAPIResult<String> {
let (key, rc) = get_opened_dht_record_context(&args, "debug_record_close", "key", 1)?;
let (key, rc) =
self.clone()
.get_opened_dht_record_context(&args, "debug_record_close", "key", 1)?;
// Do a record close
if let Err(e) = rc.close_dht_record(key).await {
@ -1575,7 +1462,9 @@ impl VeilidAPI {
} else {
0
};
let (key, rc) = get_opened_dht_record_context(&args, "debug_record_set", "key", 1)?;
let (key, rc) =
self.clone()
.get_opened_dht_record_context(&args, "debug_record_set", "key", 1)?;
let subkey = get_debug_argument_at(
&args,
1 + opt_arg_add,
@ -1619,7 +1508,9 @@ impl VeilidAPI {
0
};
let (key, rc) = get_opened_dht_record_context(&args, "debug_record_get", "key", 1)?;
let (key, rc) =
self.clone()
.get_opened_dht_record_context(&args, "debug_record_get", "key", 1)?;
let subkey = get_debug_argument_at(
&args,
1 + opt_arg_add,
@ -1726,7 +1617,9 @@ impl VeilidAPI {
0
};
let (key, rc) = get_opened_dht_record_context(&args, "debug_record_watch", "key", 1)?;
let (key, rc) =
self.clone()
.get_opened_dht_record_context(&args, "debug_record_watch", "key", 1)?;
let mut rest_defaults = false;
let subkeys = get_debug_argument_at(
@ -1799,7 +1692,9 @@ impl VeilidAPI {
0
};
let (key, rc) = get_opened_dht_record_context(&args, "debug_record_watch", "key", 1)?;
let (key, rc) =
self.clone()
.get_opened_dht_record_context(&args, "debug_record_watch", "key", 1)?;
let subkeys = get_debug_argument_at(
&args,
1 + opt_arg_add,
@ -1832,7 +1727,9 @@ impl VeilidAPI {
0
};
let (key, rc) = get_opened_dht_record_context(&args, "debug_record_watch", "key", 1)?;
let (key, rc) =
self.clone()
.get_opened_dht_record_context(&args, "debug_record_watch", "key", 1)?;
let mut rest_defaults = false;
@ -2159,4 +2056,119 @@ table list
};
res
}
fn get_destination(
self,
routing_table: RoutingTable,
) -> impl FnOnce(&str) -> SendPinBoxFuture<Option<Destination>> {
move |text| {
let text = text.to_owned();
Box::pin(async move {
// Safety selection
let (text, ss) = if let Some((first, second)) = text.split_once('+') {
let ss = get_safety_selection(routing_table.clone())(second)?;
(first, Some(ss))
} else {
(text.as_str(), None)
};
if text.is_empty() {
return None;
}
if &text[0..1] == "#" {
let rss = routing_table.route_spec_store();
// Private route
let text = &text[1..];
let private_route =
if let Some(prid) = get_route_id(rss.clone(), false, true)(text) {
rss.best_remote_private_route(&prid)?
} else {
let mut inner = self.inner.lock();
let dc = &mut inner.debug_cache;
let n = get_number(text)?;
let prid = *dc.imported_routes.get(n)?;
let Some(private_route) = rss.best_remote_private_route(&prid) else {
// Remove imported route
dc.imported_routes.remove(n);
info!("removed dead imported route {}", n);
return None;
};
private_route
};
Some(Destination::private_route(
private_route,
ss.unwrap_or(SafetySelection::Unsafe(Sequencing::default())),
))
} else {
let (text, mods) = text
.split_once('/')
.map(|x| (x.0, Some(x.1)))
.unwrap_or((text, None));
if let Some((first, second)) = text.split_once('@') {
// Relay
let mut relay_nr = get_node_ref(routing_table.clone())(second)?;
let target_nr = get_node_ref(routing_table)(first)?;
if let Some(mods) = mods {
relay_nr = get_node_ref_modifiers(relay_nr)(mods)?;
}
let mut d = Destination::relay(relay_nr, target_nr);
if let Some(ss) = ss {
d = d.with_safety(ss)
}
Some(d)
} else {
// Direct
let mut target_nr =
resolve_node_ref(routing_table, ss.unwrap_or_default())(text).await?;
if let Some(mods) = mods {
target_nr = get_node_ref_modifiers(target_nr)(mods)?;
}
let mut d = Destination::direct(target_nr);
if let Some(ss) = ss {
d = d.with_safety(ss)
}
Some(d)
}
}
})
}
}
fn get_opened_dht_record_context(
self,
args: &[String],
context: &str,
key: &str,
arg: usize,
) -> VeilidAPIResult<(TypedKey, RoutingContext)> {
let mut inner = self.inner.lock();
let dc = &mut inner.debug_cache;
let key = match get_debug_argument_at(args, arg, context, key, get_dht_key_no_safety)
.ok()
.or_else(|| {
// If unspecified, use the most recent key opened or created
dc.opened_record_contexts.back().map(|kv| kv.0).copied()
}) {
Some(k) => k,
None => {
apibail_missing_argument!("no keys are opened", "key");
}
};
// Get routing context for record
let Some(rc) = dc.opened_record_contexts.get(&key).cloned() else {
apibail_missing_argument!("key is not opened", "key");
};
Ok((key, rc))
}
}

90
veilid-core/src/veilid_config.rs
View File

@ -327,7 +327,9 @@ pub fn get_default_ssl_directory(sub_path: &str) -> String {
}
/// Configure the Distributed Hash Table (DHT).
///
/// Defaults should be used here unless you are absolutely sure you know what you're doing.
/// If you change the count/fanout/timeout parameters, you may render your node inoperable
/// for correct DHT operations.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
#[cfg_attr(target_arch = "wasm32", derive(Tsify))]
pub struct VeilidConfigDHT {
@ -709,21 +711,40 @@ impl fmt::Display for VeilidConfigLogLevel {
}
}
/// Top level of the Veilid configuration tree
#[derive(Default, Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
#[cfg_attr(target_arch = "wasm32", derive(Tsify))]
pub struct VeilidConfigInner {
/// An identifier used to describe the program using veilid-core.
/// Used to partition storage locations in places like the ProtectedStore.
/// Must be non-empty and a valid filename for all Veilid-capable systems, which means
/// no backslashes or forward slashes in the name. Stick to a-z,0-9,_ and space and you should be fine.
///
/// Caution: If you change this string, there is no migration support. Your app's protected store and
/// table store will very likely experience data loss. Pick a program name and stick with it. This is
/// not a 'visible' identifier and it should uniquely identify your application.
pub program_name: String,
/// To run multiple Veilid nodes within the same application, either through a single process running
/// api_startup/api_startup_json multiple times, or your application running mulitple times side-by-side
/// there needs to be a key used to partition the application's storage (in the TableStore, ProtectedStore, etc).
/// An empty value here is the default, but if you run multiple veilid nodes concurrently, you should set this
/// to a string that uniquely identifies this -instance- within the same 'program_name'.
/// Must be a valid filename for all Veilid-capable systems, which means no backslashes or forward slashes
/// in the name. Stick to a-z,0-9,_ and space and you should be fine.
pub namespace: String,
/// Capabilities to enable for your application/node
pub capabilities: VeilidConfigCapabilities,
/// Configuring the protected store (keychain/keyring/etc)
pub protected_store: VeilidConfigProtectedStore,
/// Configuring the table store (persistent encrypted database)
pub table_store: VeilidConfigTableStore,
/// Configuring the block store (storage of large content-addressable content)
pub block_store: VeilidConfigBlockStore,
/// Configuring how Veilid interacts with the low level network
pub network: VeilidConfigNetwork,
}
/// The Veilid Configuration.
///
/// Veilid is configured.
/// The configuration built for each Veilid node during API startup
#[derive(Clone)]
pub struct VeilidConfig {
update_cb: Option<UpdateCallback>,
@ -749,27 +770,14 @@ impl VeilidConfig {
VeilidConfigInner::default()
}
pub fn new() -> Self {
pub(crate) fn new() -> Self {
Self {
update_cb: None,
inner: Arc::new(RwLock::new(Self::new_inner())),
}
}
pub fn setup_from_json(
&mut self,
config: String,
update_cb: UpdateCallback,
) -> VeilidAPIResult<()> {
self.update_cb = Some(update_cb);
self.with_mut(|inner| {
*inner = serde_json::from_str(&config).map_err(VeilidAPIError::generic)?;
Ok(())
})
}
pub fn setup_from_config(
pub(crate) fn setup_from_config(
&mut self,
config: VeilidConfigInner,
update_cb: UpdateCallback,
@ -782,7 +790,11 @@ impl VeilidConfig {
})
}
pub fn setup(&mut self, cb: ConfigCallback, update_cb: UpdateCallback) -> VeilidAPIResult<()> {
pub(crate) fn setup(
&mut self,
cb: ConfigCallback,
update_cb: UpdateCallback,
) -> VeilidAPIResult<()> {
self.update_cb = Some(update_cb);
self.with_mut(|inner| {
// Simple config transformation
@ -902,7 +914,7 @@ impl VeilidConfig {
})
}
pub fn get_veilid_state(&self) -> Box<VeilidStateConfig> {
pub(crate) fn get_veilid_state(&self) -> Box<VeilidStateConfig> {
let inner = self.inner.read();
Box::new(VeilidStateConfig {
config: inner.clone(),
@ -1039,10 +1051,42 @@ impl VeilidConfig {
})
}
fn validate(inner: &VeilidConfigInner) -> VeilidAPIResult<()> {
if inner.program_name.is_empty() {
fn validate_program_name(program_name: &str) -> VeilidAPIResult<()> {
if program_name.is_empty() {
apibail_generic!("Program name must not be empty in 'program_name'");
}
if !sanitize_filename::is_sanitized_with_options(
program_name,
sanitize_filename::OptionsForCheck {
windows: true,
truncate: true,
},
) {
apibail_generic!("'program_name' must not be an invalid filename");
}
Ok(())
}
fn validate_namespace(namespace: &str) -> VeilidAPIResult<()> {
if namespace.is_empty() {
return Ok(());
}
if !sanitize_filename::is_sanitized_with_options(
namespace,
sanitize_filename::OptionsForCheck {
windows: true,
truncate: true,
},
) {
apibail_generic!("'namespace' must not be an invalid filename");
}
Ok(())
}
fn validate(inner: &VeilidConfigInner) -> VeilidAPIResult<()> {
Self::validate_program_name(&inner.program_name)?;
Self::validate_namespace(&inner.namespace)?;
// if inner.network.protocol.udp.enabled {
// // Validate UDP settings

2
veilid-flutter/rust/Cargo.toml
View File

@ -3,6 +3,8 @@
name = "veilid-flutter"
version = "0.3.4"
# ---
description = "Flutter/Dart bindings for Veilid"
repository = "https://gitlab.com/veilid/veilid"
authors = ["Veilid Team <contact@veilid.com>"]
license = "MPL-2.0"
edition = "2021"

2
veilid-flutter/rust/src/dart_ffi.rs
View File

@ -345,7 +345,7 @@ pub extern "C" fn initialize_veilid_core(platform_config: FfiStr) {
platform_config.logging.api.level,
&platform_config.logging.api.ignore_log_targets,
);
let layer = veilid_core::ApiTracingLayer::get().with_filter(filter.clone());
let layer = veilid_core::ApiTracingLayer::init().with_filter(filter.clone());
filters.insert("api", filter);
layers.push(layer.boxed());
}

3
veilid-server/Cargo.toml
View File

@ -3,7 +3,8 @@
name = "veilid-server"
version = "0.3.4"
# ---
description = "Veilid Server"
description = "Veilid Headless Node"
repository = "https://gitlab.com/veilid/veilid"
authors = ["Veilid Team <contact@veilid.com>"]
license = "MPL-2.0"
edition = "2021"

2
veilid-server/src/veilid_logs.rs
View File

@ -208,7 +208,7 @@ impl VeilidLogs {
convert_loglevel(settingsr.logging.api.level),
&settingsr.logging.api.ignore_log_targets,
);
let layer = veilid_core::ApiTracingLayer::get().with_filter(filter.clone());
let layer = veilid_core::ApiTracingLayer::init().with_filter(filter.clone());
filters.insert("api", filter);
layers.push(layer.boxed());
}

6
veilid-tools/Cargo.toml
View File

@ -4,6 +4,7 @@ name = "veilid-tools"
version = "0.3.4"
# ---
description = "A collection of baseline tools for Rust development use by Veilid and Veilid-enabled Rust applications"
repository = "https://gitlab.com/veilid/veilid"
authors = ["Veilid Team <contact@veilid.com>"]
license = "MPL-2.0"
edition = "2021"
@ -84,7 +85,7 @@ nix = { version = "0.27.1", features = ["user"] }
# Dependencies for WASM builds only
[target.'cfg(target_arch = "wasm32")'.dependencies]
wasm-bindgen = "0.2.92"
js-sys = "0.3.69"
js-sys = "0.3.70"
wasm-bindgen-futures = "0.4.42"
async_executors = { version = "0.7.0", default-features = false }
getrandom = { version = "0.2", features = ["js"] }
@ -149,3 +150,6 @@ build_targets = [
]
deployment_target = "12.0"
build_id_prefix = "com.veilid.veilidtools"
[lints.rust]
unexpected_cfgs = { level = "warn", check-cfg = ['cfg(tokio_unstable)'] }

10
veilid-tools/src/spawn.rs
View File

@ -54,7 +54,7 @@ cfg_if! {
cfg_if! {
if #[cfg(feature="rt-async-std")] {
MustJoinHandle::new(async_std::task::Builder::new().name(name.to_string()).spawn(future).unwrap())
} else if #[cfg(all(feature="rt-tokio", feature="tracing"))] {
} else if #[cfg(all(tokio_unstable, feature="rt-tokio", feature="tracing"))] {
MustJoinHandle::new(tokio::task::Builder::new().name(name).spawn(future).unwrap())
} else if #[cfg(feature="rt-tokio")] {
let _name = name;
@ -70,7 +70,7 @@ cfg_if! {
cfg_if! {
if #[cfg(feature="rt-async-std")] {
MustJoinHandle::new(async_std::task::Builder::new().name(name.to_string()).local(future).unwrap())
} else if #[cfg(all(feature="rt-tokio", feature="tracing"))] {
} else if #[cfg(all(tokio_unstable, feature="rt-tokio", feature="tracing"))] {
MustJoinHandle::new(tokio::task::Builder::new().name(name).spawn_local(future).unwrap())
} else if #[cfg(feature="rt-tokio")] {
let _name = name;
@ -86,7 +86,7 @@ cfg_if! {
cfg_if! {
if #[cfg(feature="rt-async-std")] {
drop(async_std::task::Builder::new().name(name.to_string()).spawn(future).unwrap());
} else if #[cfg(all(feature="rt-tokio", feature="tracing"))] {
} else if #[cfg(all(tokio_unstable, feature="rt-tokio", feature="tracing"))] {
drop(tokio::task::Builder::new().name(name).spawn(future).unwrap());
} else if #[cfg(feature="rt-tokio")] {
let _name = name;
@ -102,7 +102,7 @@ cfg_if! {
cfg_if! {
if #[cfg(feature="rt-async-std")] {
drop(async_std::task::Builder::new().name(name.to_string()).local(future).unwrap());
} else if #[cfg(all(feature="rt-tokio", feature="tracing"))] {
} else if #[cfg(all(tokio_unstable, feature="rt-tokio", feature="tracing"))] {
drop(tokio::task::Builder::new().name(name).spawn_local(future).unwrap());
} else if #[cfg(feature="rt-tokio")] {
let _name = name;
@ -123,7 +123,7 @@ cfg_if! {
let _name = name;
// async_std::task::Builder blocking doesn't work like spawn_blocking()
async_std::task::spawn_blocking(blocking_task).await
} else if #[cfg(all(feature="rt-tokio", feature="tracing"))] {
} else if #[cfg(all(tokio_unstable, feature="rt-tokio", feature="tracing"))] {
tokio::task::Builder::new().name(name).spawn_blocking(blocking_task).unwrap().await.unwrap_or(err_result)
} else if #[cfg(feature="rt-tokio")] {
let _name = name;

1
veilid-wasm/Cargo.toml
View File

@ -3,6 +3,7 @@
name = "veilid-wasm"
version = "0.3.4"
# ---
description = "Veilid bindings for WebAssembly"
authors = ["Veilid Team <contact@veilid.com>"]
license = "MPL-2.0"
edition = "2021"

2
veilid-wasm/src/lib.rs
View File

@ -234,7 +234,7 @@ pub fn initialize_veilid_core(platform_config: String) {
platform_config.logging.api.level,
&platform_config.logging.api.ignore_log_targets,
);
let layer = veilid_core::ApiTracingLayer::get().with_filter(filter.clone());
let layer = veilid_core::ApiTracingLayer::init().with_filter(filter.clone());
filters.insert("api", filter);
layers.push(layer.boxed());
}

2
veilid-wasm/src/veilid_client_js.rs
View File

@ -67,7 +67,7 @@ impl VeilidClient {
platformConfig.logging.api.level,
&platformConfig.logging.api.ignore_log_targets,
);
let layer = veilid_core::ApiTracingLayer::get().with_filter(filter.clone());
let layer = veilid_core::ApiTracingLayer::init().with_filter(filter.clone());
filters.insert("api", filter);
layers.push(layer.boxed());
}

2
veilid-wasm/tests/package-lock.json generated
View File

@ -21,7 +21,7 @@
},
"../pkg": {
"name": "veilid-wasm",
"version": "0.3.2",
"version": "0.3.4",
"dev": true,
"license": "MPL-2.0"
},