Bootstrap Client Module

Cargo.toml

@@ -51,6 +51,8 @@ futures = "0.3"
 tempfile = "3.14.0"
 # kitsune2 internally uses a mix of std::io::Error and thiserror derivation.
 thiserror = "2.0.3"
+# ureq is used in the bootstrap client module.
+ureq = "2.10.1"
 # url validation used by the api crate.
 url = "2.5.4"
 # kitsune2 uses tracing to log events. A consumer can choose any subscriber
@@ -68,8 +70,11 @@ prost-build = "0.13.3"
 # Please be careful to only include them in dev dependencies or move them
 # above this section.
 # --- dev-dependencies ---
+kitsune2_bootstrap_srv = { path = "crates/bootstrap_srv" }
 kitsune2_memory = { path = "crates/memory" }
 kitsune2_test_utils = { path = "crates/test_utils" }
 
-ureq = "2.10.1"
+rand = "0.8.5"
+# this is also used by the binary bootstrap_srv. But, since this monorepo
+# is largely libraries, leaving this in this section.
 tracing-subscriber = "0.3"

Makefile

@@ -2,9 +2,9 @@
 # tasks both in automation and locally until we figure out better
 # release automation tools.
 
-.PHONY: all fmt clippy doc build test
+.PHONY: all static-toml fmt clippy doc build test
 
-all: fmt clippy doc test
+all: static-toml fmt clippy doc build test
 
 static-toml:
 	taplo format --check

crates/api/src/agent.rs

@@ -105,6 +105,17 @@ pub trait Verifier: std::fmt::Debug {
 /// Trait-object [Verifier].
 pub type DynVerifier = Arc<dyn Verifier + 'static + Send + Sync>;
 
+impl Verifier for DynVerifier {
+    fn verify(
+        &self,
+        agent_info: &AgentInfo,
+        message: &[u8],
+        signature: &[u8],
+    ) -> bool {
+        (**self).verify(agent_info, message, signature)
+    }
+}
+
 /// A "Local" agent is an agent that is connected to the local Kitsune2 node,
 /// and is able to sign messages and agent infos.
 pub trait LocalAgent: Signer + 'static + Send + Sync + std::fmt::Debug {
@@ -170,7 +181,6 @@ pub struct AgentInfo {
 }
 
 /// Signed agent information.
-#[derive(Debug)]
 pub struct AgentInfoSigned {
     /// The decoded information associated with this agent.
     agent_info: AgentInfo,
@@ -182,6 +192,14 @@ pub struct AgentInfoSigned {
     signature: bytes::Bytes,
 }
 
+impl std::fmt::Debug for AgentInfoSigned {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_str("AgentInfoSigned(")?;
+        f.write_str(&self.encoded)?;
+        f.write_str(")")
+    }
+}
+
 impl AgentInfoSigned {
     /// Generate a signed agent info by signing an agent info.
     pub async fn sign<S: Signer>(
@@ -215,16 +233,44 @@ impl AgentInfoSigned {
         }
         let v: Ref = serde_json::from_slice(encoded)
             .map_err(|e| K2Error::other_src("decoding agent_info", e))?;
-        let agent_info: AgentInfo = serde_json::from_str(&v.agent_info)
+        Self::inner_decode_one(verifier, v.agent_info, v.signature)
+    }
+
+    /// Decode a canonical json encoding of a list of signed agent infos.
+    pub fn decode_list<V: Verifier>(
+        verifier: &V,
+        encoded: &[u8],
+    ) -> K2Result<Vec<K2Result<std::sync::Arc<Self>>>> {
+        #[derive(serde::Deserialize)]
+        #[serde(rename_all = "camelCase")]
+        struct Ref {
+            agent_info: String,
+            #[serde(with = "crate::serde_bytes_base64")]
+            signature: bytes::Bytes,
+        }
+        let v: Vec<Ref> = serde_json::from_slice(encoded)
+            .map_err(|e| K2Error::other_src("decoding agent_info", e))?;
+        Ok(v.into_iter()
+            .map(|v| {
+                Self::inner_decode_one(verifier, v.agent_info, v.signature)
+            })
+            .collect())
+    }
+
+    fn inner_decode_one<V: Verifier>(
+        verifier: &V,
+        agent_info: String,
+        signature: bytes::Bytes,
+    ) -> K2Result<std::sync::Arc<Self>> {
+        let info: AgentInfo = serde_json::from_str(&agent_info)
             .map_err(|e| K2Error::other_src("decoding inner agent_info", e))?;
-        if !verifier.verify(&agent_info, v.agent_info.as_bytes(), &v.signature)
-        {
+        if !verifier.verify(&info, agent_info.as_bytes(), &signature) {
             return Err(K2Error::other("InvalidSignature"));
         }
         Ok(std::sync::Arc::new(Self {
-            agent_info,
-            encoded: v.agent_info,
-            signature: v.signature,
+            agent_info: info,
+            encoded: agent_info,
+            signature,
         }))
     }
 

crates/api/src/bootstrap.rs

@@ -0,0 +1,41 @@
+//! Kitsune2 bootstrap related types.
+
+use crate::*;
+use std::sync::Arc;
+
+/// Method for bootstrapping WAN discovery of peers.
+///
+/// The internal implementation will take care of whatever polling
+/// or managing of message queues is required to be notified of
+/// remote peers both on initialization and over runtime.
+pub trait Bootstrap: 'static + Send + Sync + std::fmt::Debug {
+    /// Put an agent info onto a bootstrap server.
+    ///
+    /// This method takes responsibility for retrying the send in the case
+    /// of server error until such time as:
+    /// - the Put succeeds
+    /// - we receive a new info that supercedes the previous
+    /// - or the info expires
+    fn put(&self, info: Arc<agent::AgentInfoSigned>);
+}
+
+/// Trait-object [Bootstrap].
+pub type DynBootstrap = Arc<dyn Bootstrap>;
+
+/// A factory for constructing Bootstrap instances.
+pub trait BootstrapFactory: 'static + Send + Sync + std::fmt::Debug {
+    /// Help the builder construct a default config from the chosen
+    /// module factories.
+    fn default_config(&self, config: &mut config::Config) -> K2Result<()>;
+
+    /// Construct a bootstrap instance.
+    fn create(
+        &self,
+        builder: Arc<builder::Builder>,
+        peer_store: peer_store::DynPeerStore,
+        space: SpaceId,
+    ) -> BoxFut<'static, K2Result<DynBootstrap>>;
+}
+
+/// Trait-object [BootstrapFactory].
+pub type DynBootstrapFactory = Arc<dyn BootstrapFactory>;

crates/api/src/builder.rs

@@ -27,6 +27,10 @@ pub struct Builder {
     /// [peer_store::PeerStore] instances.
     pub peer_store: peer_store::DynPeerStoreFactory,
 
+    /// The [bootstrap::BootstrapFactory] to be used for creating
+    /// [bootstrap::Bootstrap] instances for initial WAN discovery.
+    pub bootstrap: bootstrap::DynBootstrapFactory,
+
     /// The [transport::TransportFactory] to be used for creating
     /// [transport::Transport] instances.
     pub transport: transport::DynTransportFactory,
@@ -43,12 +47,14 @@ impl Builder {
             kitsune,
             space,
             peer_store,
+            bootstrap,
             transport,
         } = self;
 
         kitsune.default_config(config)?;
         space.default_config(config)?;
         peer_store.default_config(config)?;
+        bootstrap.default_config(config)?;
         transport.default_config(config)?;
 
         Ok(())

crates/api/src/config.rs

@@ -61,6 +61,19 @@ impl Config {
         Ok(())
     }
 
+    /// Update a module config with new values. Note, this will overwrite
+    /// ALL config values. If you would like to only update some, please
+    /// first call [Self::get_module_config], update the appropriate values,
+    /// then call this function again to set them.
+    pub fn set_module_config<M: ModConfig>(
+        &mut self,
+        module_name: String,
+        config: &M,
+    ) -> K2Result<()> {
+        self.0.insert(module_name, tc(config)?);
+        Ok(())
+    }
+
     /// When kitsune2 is initializing, it will call the factory function
     /// for all of its modules with an immutable reference to this config
     /// struct. Each of those modules may choose to call this function

crates/api/src/lib.rs

@@ -40,6 +40,7 @@ pub mod agent;
 pub mod arc;
 pub use arc::*;
 
+pub mod bootstrap;
 pub mod builder;
 pub mod config;
 pub mod kitsune;

crates/bootstrap_srv/Cargo.toml

@@ -27,6 +27,8 @@ serde = { workspace = true }
 serde_json = { workspace = true }
 tempfile = { workspace = true }
 tokio = { workspace = true, features = ["time", "rt", "rt-multi-thread"] }
+tracing = { workspace = true }
+tracing-subscriber = { workspace = true, features = ["env-filter", "json"] }
 
 [dev-dependencies]
 ureq = { workspace = true }

crates/bootstrap_srv/src/bin/kitsune2-bootstrap-srv.rs

@@ -14,6 +14,9 @@ pub struct Args {
     /// on a single given machine, you can set this "production" mode.
     #[arg(long)]
     pub production: bool,
+    /// Output tracing in json format.
+    #[arg(long)]
+    pub json: bool,
     // TODO - Implement the ability to specify TLS certificates
     // TODO - Implement the ability to specify the listening address
     // TODO - Implement the ability to override any other relevant
@@ -23,13 +26,28 @@ pub struct Args {
 fn main() {
     let args = <Args as clap::Parser>::parse();
 
+    let t = tracing_subscriber::fmt()
+        .with_env_filter(
+            tracing_subscriber::EnvFilter::builder()
+                .with_default_directive(tracing::Level::DEBUG.into())
+                .from_env_lossy(),
+        )
+        .with_file(true)
+        .with_line_number(true);
+
+    if args.json {
+        let _ = t.json().try_init();
+    } else {
+        let _ = t.try_init();
+    }
+
     let config = if args.production {
         Config::production()
     } else {
         Config::testing()
     };
 
-    println!("{args:?}--{config:?}");
+    tracing::info!(?args, ?config);
 
     let (send, recv) = std::sync::mpsc::channel();
 
@@ -42,8 +60,8 @@ fn main() {
 
     let _ = recv.recv();
 
-    println!("Terminating...");
+    tracing::info!("Terminating...");
     drop(srv);
-    println!("Done.");
+    tracing::info!("Exit Process.");
     std::process::exit(0);
 }

crates/bootstrap_srv/src/server.rs

@@ -19,7 +19,7 @@ struct ThreadGuard(&'static str);
 
 impl Drop for ThreadGuard {
     fn drop(&mut self) {
-        eprintln!("{}", self.0);
+        tracing::debug!("{}", self.0);
     }
 }
 
@@ -37,6 +37,9 @@ pub struct BootstrapSrv {
 
 impl Drop for BootstrapSrv {
     fn drop(&mut self) {
+        let _g = ThreadGuard("Server Shutdown Complete!");
+
+        tracing::debug!("begin server shutdown...");
         let _ = self.shutdown();
     }
 }
@@ -68,7 +71,10 @@ impl BootstrapSrv {
 
         // get the address that was assigned
         let addrs = server.server_addrs().to_vec();
-        println!("Listening at {:?}", addrs);
+        for addr in addrs.iter() {
+            // print these separately incase someone wants to parse them
+            tracing::info!(addr = format!("{:?}", addr), "Listening");
+        }
 
         // spawn our worker threads
         let mut workers = Vec::with_capacity(config.worker_thread_count + 1);
@@ -104,11 +110,16 @@ impl BootstrapSrv {
         let mut is_err = false;
         self.cont.store(false, std::sync::atomic::Ordering::SeqCst);
         drop(self.server.take());
-        for worker in self.workers.drain(..) {
-            if worker.join().is_err() {
+        while !self.workers.is_empty() {
+            tracing::debug!(
+                "waiting on {} threads to close...",
+                self.workers.len()
+            );
+            if self.workers.pop().unwrap().join().is_err() {
                 is_err = true;
             }
         }
+        tracing::debug!("all threads closed.");
         if is_err {
             Err(std::io::Error::other("Failure shutting down worker thread"))
         } else {
@@ -127,7 +138,7 @@ fn prune_worker(
     cont: Arc<std::sync::atomic::AtomicBool>,
     space_map: crate::SpaceMap,
 ) -> std::io::Result<()> {
-    let _g = ThreadGuard("WARN: prune_worker thread has ended");
+    let _g = ThreadGuard("prune_worker thread has ended");
 
     let mut last_check = std::time::Instant::now();
 
@@ -151,7 +162,7 @@ fn worker(
     recv: HttpReceiver,
     space_map: crate::SpaceMap,
 ) -> std::io::Result<()> {
-    let _g = ThreadGuard("WARN: worker thread has ended");
+    let _g = ThreadGuard("worker thread has ended");
 
     while cont.load(std::sync::atomic::Ordering::SeqCst) {
         let (req, res) = match recv.recv() {
@@ -168,6 +179,7 @@ fn worker(
 
         handler.handle(req)?;
     }
+
     Ok(())
 }
 

crates/core/Cargo.toml

@@ -15,9 +15,24 @@ bytes = { workspace = true }
 ed25519-dalek = { workspace = true }
 futures = { workspace = true }
 kitsune2_api = { workspace = true }
+kitsune2_test_utils = { workspace = true, optional = true }
 serde = { workspace = true }
 serde_json = { workspace = true }
-tokio = { workspace = true, features = ["sync", "rt"] }
+tokio = { workspace = true, features = ["rt", "sync", "time"] }
+ureq = { workspace = true }
 
 [dev-dependencies]
+axum = { workspace = true, default-features = false, features = [
+  "http1",
+  "tokio",
+] }
+ed25519-dalek = { workspace = true, features = ["rand_core"] }
+kitsune2_core = { path = ".", features = ["test_utils"] }
+rand = { workspace = true }
 tokio = { workspace = true, features = ["full"] }
+
+[features]
+default = []
+
+# additional utilities for testing
+test_utils = ["dep:kitsune2_test_utils"]

crates/core/src/factories.rs

@@ -9,5 +9,8 @@ pub use core_space::*;
 mod mem_peer_store;
 pub use mem_peer_store::*;
 
+mod core_bootstrap;
+pub use core_bootstrap::*;
+
 mod mem_transport;
 pub use mem_transport::*;