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

Merge branch 'multiple-instances' into 'main'

Browse Source 
add namespacing to WASM TableStore

See merge request veilid/veilid!315
This commit is contained in:
Christien Rioux 2024-08-17 01:01:30 +00:00
commit 99e824829b
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"
},