Time slice

Cargo.toml

@@ -2,10 +2,16 @@
 members = [
   "crates/api",
   "crates/bootstrap_srv",
+  "crates/dht",
+  "crates/memory",
 ]
 resolver = "2"
 
 [workspace.dependencies]
+# Self dependencies for workspace crates to depend on each other.
+# For example, most crates will depend on the api crate.
+kitsune2_api = { version = "0.0.1-alpha", path = "crates/api" }
+
 # debugging is far easier when you can see short byte arrays
 # as base64 instead of decimal u8s.
 base64 = "0.22.1"
@@ -26,6 +32,8 @@ serde = { version = "1.0.215", features = ["derive"] }
 # kitsune2 agent info is serialized as json to improve debugability of
 # bootstrapping. So, we need a json library.
 serde_json = "1.0.132"
+# Simplify writing async code
+futures = "0.3"
 # bootstrap_srv uses tempfiles as virtual memory for storage instead of RAM.
 tempfile = "3.14.0"
 # kitsune2 internally uses a mix of std::io::Error and thiserror derivation.
@@ -39,3 +47,4 @@ tiny_http = "0.12.0"
 # --- dev-dependencies ---
 tokio = "1.41.1"
 ureq = "2.10.1"
+kitsune2_memory = { path = "crates/memory" }

crates/api/Cargo.toml

@@ -16,6 +16,7 @@ bytes = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }
 thiserror = { workspace = true }
+futures = { workspace = true }
 
 [dev-dependencies]
 tokio = { workspace = true, features = ["full"] }

crates/api/src/lib.rs

@@ -47,3 +47,5 @@ mod timestamp;
 pub use timestamp::*;
 
 pub mod agent;
+pub mod op_store;
+pub use op_store::*;

crates/api/src/op_store.rs

@@ -0,0 +1,83 @@
+//! Kitsune2 op store types.
+
+use crate::{K2Result, OpId, Timestamp};
+use futures::future::BoxFuture;
+use std::cmp::Ordering;
+use std::sync::Arc;
+
+/// An op with metadata.
+///
+/// This is the basic unit of data in the kitsune2 system.
+#[derive(Debug, Clone, Eq, PartialEq)]
+pub struct MetaOp {
+    /// The id of the op.
+    pub op_id: OpId,
+
+    /// The creation timestamp of the op.
+    ///
+    /// This must be the same for everyone who sees this op.
+    ///
+    /// The host must reject the op if the timestamp does not agree with any timestamps inside the
+    /// op data.
+    pub timestamp: Timestamp,
+
+    /// The actual op data.
+    pub op_data: Vec<u8>,
+}
+
+impl Ord for MetaOp {
+    fn cmp(&self, other: &Self) -> Ordering {
+        (&self.timestamp, &self.op_id).cmp(&(&other.timestamp, &other.op_id))
+    }
+}
+
+impl PartialOrd for MetaOp {
+    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
+        Some(self.cmp(other))
+    }
+}
+
+/// The API that a kitsune2 host must implement to provide data persistence for kitsune2.
+pub trait OpStore: 'static + Send + Sync + std::fmt::Debug {
+    /// Process incoming ops.
+    ///
+    /// Pass the incoming ops to the host for processing. The host is expected to store the ops
+    /// if it is able to process them.
+    fn process_incoming_ops(
+        &self,
+        op_list: Vec<MetaOp>,
+    ) -> BoxFuture<'_, K2Result<()>>;
+
+    /// Retrieve a batch of ops from the host by time range.
+    ///
+    /// This must be the timestamp of the op, not the time that we saw the op or chose to store it.
+    /// The returned ops must be ordered by timestamp, ascending.
+    fn retrieve_op_hashes_in_time_slice(
+        &self,
+        start: Timestamp,
+        end: Timestamp,
+    ) -> BoxFuture<'_, K2Result<Vec<OpId>>>;
+
+    /// Store the combined hash of a time slice.
+    fn store_slice_hash(
+        &self,
+        slice_id: u64,
+        slice_hash: Vec<u8>,
+    ) -> BoxFuture<'_, K2Result<()>>;
+
+    /// Count the number of stored time slices.
+    fn slice_hash_count(&self) -> BoxFuture<'_, K2Result<u64>>;
+
+    /// Retrieve the combined hash of a time slice.
+    ///
+    /// If the slice is not found, return None.
+    /// If the time slice is present, then it must be identical to what was stored by Kitsune2
+    /// using [Self::store_slice_hash].
+    fn retrieve_slice_hash(
+        &self,
+        slice_id: u64,
+    ) -> BoxFuture<'_, K2Result<Option<Vec<u8>>>>;
+}
+
+/// Trait-object version of kitsune2 op store.
+pub type DynOpStore = Arc<dyn OpStore>;

crates/api/src/timestamp.rs

@@ -46,6 +46,36 @@ impl std::ops::Add<std::time::Duration> for Timestamp {
     }
 }
 
+impl std::ops::AddAssign<std::time::Duration> for Timestamp {
+    fn add_assign(&mut self, rhs: std::time::Duration) {
+        self.0 += rhs.as_micros() as i64;
+    }
+}
+
+impl std::ops::Sub<std::time::Duration> for Timestamp {
+    type Output = Result<Timestamp, ()>;
+
+    fn sub(self, rhs: std::time::Duration) -> Self::Output {
+        if self.0 < rhs.as_micros() as i64 {
+            Err(())
+        } else {
+            Ok(Timestamp(self.0 - rhs.as_micros() as i64))
+        }
+    }
+}
+
+impl std::ops::Sub for Timestamp {
+    type Output = Result<std::time::Duration, ()>;
+
+    fn sub(self, rhs: Self) -> Self::Output {
+        if self.0 < rhs.0 {
+            Err(())
+        } else {
+            Ok(std::time::Duration::from_micros((self.0 - rhs.0) as u64))
+        }
+    }
+}
+
 impl From<std::time::SystemTime> for Timestamp {
     fn from(t: std::time::SystemTime) -> Self {
         Self(

crates/dht/Cargo.toml

@@ -0,0 +1,20 @@
+[package]
+name = "kitsune2_dht"
+version = "0.0.1-alpha"
+description = "The DHT model for kitsune2"
+license = "Apache-2.0"
+homepage = "https://github.com/holochain/kitsune2"
+documentation = "https://docs.rs/kitsune2_api"
+authors = ["Holochain Core Dev Team <[email protected]>"]
+keywords = ["holochain", "holo", "p2p", "dht", "networking"]
+categories = ["network-programming"]
+edition = "2021"
+
+[dependencies]
+kitsune2_api = { workspace = true }
+bytes = { workspace = true }
+
+[dev-dependencies]
+kitsune2_memory = { workspace = true }
+
+tokio = { workspace = true, features = ["macros", "rt"] }

crates/dht/src/constant.rs

@@ -0,0 +1,3 @@
+use std::time::Duration;
+
+pub const UNIT_TIME: Duration = Duration::from_secs(15 * 60); // 15 minutes

crates/dht/src/lib.rs

@@ -0,0 +1,5 @@
+pub mod constant;
+pub use constant::*;
+
+pub mod time;
+pub use time::*;