diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 6976e98..d0fac45 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -32,16 +32,18 @@ jobs:
components: rustfmt
- name: Run tests
run: |
- cargo check --no-default-features --features tokio
- cargo check --no-default-features --features tokio,sparse
- cargo check --no-default-features --features tokio,sparse,cache
- cargo check --no-default-features --features async-std
- cargo check --no-default-features --features async-std,sparse
- cargo check --no-default-features --features async-std,sparse,cache
+ cargo check --all-targets --no-default-features --features tokio
+ cargo check --all-targets --no-default-features --features tokio,sparse
+ cargo check --all-targets --no-default-features --features tokio,sparse,cache
+ cargo check --all-targets --no-default-features --features async-std
+ cargo check --all-targets --no-default-features --features async-std,sparse
+ cargo check --all-targets --no-default-features --features async-std,sparse,cache
cargo test --no-default-features --features js_interop_tests,tokio
+ cargo test --no-default-features --features js_interop_tests,tokio,shared-core
cargo test --no-default-features --features js_interop_tests,tokio,sparse
cargo test --no-default-features --features js_interop_tests,tokio,sparse,cache
cargo test --no-default-features --features js_interop_tests,async-std
+ cargo test --no-default-features --features js_interop_tests,async-std,shared-core
cargo test --no-default-features --features js_interop_tests,async-std,sparse
cargo test --no-default-features --features js_interop_tests,async-std,sparse,cache
cargo test --benches --no-default-features --features tokio
@@ -57,16 +59,18 @@ jobs:
components: rustfmt
- name: Run tests
run: |
- cargo check --no-default-features --features tokio
- cargo check --no-default-features --features tokio,sparse
- cargo check --no-default-features --features tokio,sparse,cache
- cargo check --no-default-features --features async-std
- cargo check --no-default-features --features async-std,sparse
- cargo check --no-default-features --features async-std,sparse,cache
+ cargo check --all-targets --no-default-features --features tokio
+ cargo check --all-targets --no-default-features --features tokio,sparse
+ cargo check --all-targets --no-default-features --features tokio,sparse,cache
+ cargo check --all-targets --no-default-features --features async-std
+ cargo check --all-targets --no-default-features --features async-std,sparse
+ cargo check --all-targets --no-default-features --features async-std,sparse,cache
cargo test --no-default-features --features tokio
+ cargo test --no-default-features --features tokio,shared-core
cargo test --no-default-features --features tokio,sparse
cargo test --no-default-features --features tokio,sparse,cache
cargo test --no-default-features --features async-std
+ cargo test --no-default-features --features async-std,shared-core
cargo test --no-default-features --features async-std,sparse
cargo test --no-default-features --features async-std,sparse,cache
cargo test --benches --no-default-features --features tokio
@@ -82,16 +86,18 @@ jobs:
components: rustfmt
- name: Run tests
run: |
- cargo check --no-default-features --features tokio
- cargo check --no-default-features --features tokio,sparse
- cargo check --no-default-features --features tokio,sparse,cache
- cargo check --no-default-features --features async-std
- cargo check --no-default-features --features async-std,sparse
- cargo check --no-default-features --features async-std,sparse,cache
+ cargo check --all-targets --no-default-features --features tokio
+ cargo check --all-targets --no-default-features --features tokio,sparse
+ cargo check --all-targets --no-default-features --features tokio,sparse,cache
+ cargo check --all-targets --no-default-features --features async-std
+ cargo check --all-targets --no-default-features --features async-std,sparse
+ cargo check --all-targets --no-default-features --features async-std,sparse,cache
cargo test --no-default-features --features js_interop_tests,tokio
+ cargo test --no-default-features --features js_interop_tests,tokio,shared-core
cargo test --no-default-features --features js_interop_tests,tokio,sparse
cargo test --no-default-features --features js_interop_tests,tokio,sparse,cache
cargo test --no-default-features --features js_interop_tests,async-std
+ cargo test --no-default-features --features js_interop_tests,async-std,shared-core
cargo test --no-default-features --features js_interop_tests,async-std,sparse
cargo test --no-default-features --features js_interop_tests,async-std,sparse,cache
cargo test --benches --no-default-features --features tokio
diff --git a/Cargo.toml b/Cargo.toml
index 4c4cdad..1ad0d43 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -39,6 +39,8 @@ futures = "0.3"
crc32fast = "1"
intmap = "2"
moka = { version = "0.12", optional = true, features = ["sync"] }
+async-broadcast = { version = "0.7.1", optional = true }
+async-lock = {version = "3.4.0", optional = true }
[target.'cfg(not(target_arch = "wasm32"))'.dependencies]
random-access-disk = { version = "3", default-features = false }
@@ -59,7 +61,9 @@ test-log = { version = "0.2.11", default-features = false, features = ["trace"]
tracing-subscriber = { version = "0.3.16", features = ["env-filter", "fmt"] }
[features]
-default = ["async-std", "sparse"]
+default = ["tokio", "sparse", "replication"]
+replication = ["dep:async-broadcast"]
+shared-core = ["replication", "dep:async-lock"]
sparse = ["random-access-disk/sparse"]
tokio = ["random-access-disk/tokio"]
async-std = ["random-access-disk/async-std"]
diff --git a/benches/memory.rs b/benches/memory.rs
index bb97685..7c916fb 100644
--- a/benches/memory.rs
+++ b/benches/memory.rs
@@ -25,7 +25,7 @@ async fn create_hypercore(page_size: usize) -> Result<Hypercore, HypercoreError>
let storage = Storage::open(
|_| {
Box::pin(async move {
- Ok(Box::new(RandomAccessMemory::new(page_size)) as Box<dyn StorageTraits>)
+ Ok(Box::new(RandomAccessMemory::new(page_size)) as Box<dyn StorageTraits + Send>)
})
},
false,
@@ -44,7 +44,7 @@ async fn create_hypercore(page_size: usize) -> Result<Hypercore, HypercoreError>
let storage = Storage::open(
|_| {
Box::pin(async move {
- Ok(Box::new(RandomAccessMemory::new(page_size)) as Box<dyn StorageTraits>)
+ Ok(Box::new(RandomAccessMemory::new(page_size)) as Box<dyn StorageTraits + Send>)
})
},
false,
diff --git a/src/common/node.rs b/src/common/node.rs
index 7e339d3..1c78144 100644
--- a/src/common/node.rs
+++ b/src/common/node.rs
@@ -14,16 +14,21 @@ pub(crate) struct NodeByteRange {
pub(crate) length: u64,
}
-/// Nodes that are persisted to disk.
+/// Nodes of the Merkle Tree that are persisted to disk.
// TODO: replace `hash: Vec<u8>` with `hash: Hash`. This requires patching /
// rewriting the Blake2b crate to support `.from_bytes()` to serialize from
// disk.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Node {
+ /// This node's index in the Merkle tree
pub(crate) index: u64,
+ /// Hash of the data in this node
pub(crate) hash: Vec<u8>,
+ /// Number of bytes in this [`Node::data`]
pub(crate) length: u64,
+ /// Index of this nodes parent
pub(crate) parent: u64,
+ /// Hypercore's data. Can be receieved after the rest of the node, so it's optional.
pub(crate) data: Option<Vec<u8>>,
pub(crate) blank: bool,
}
diff --git a/src/common/peer.rs b/src/common/peer.rs
index c71b981..b420317 100644
--- a/src/common/peer.rs
+++ b/src/common/peer.rs
@@ -1,6 +1,7 @@
//! Types needed for passing information with with peers.
//! hypercore-protocol-rs uses these types and wraps them
//! into wire messages.
+
use crate::Node;
#[derive(Debug, Clone, PartialEq)]
@@ -20,7 +21,7 @@ pub struct RequestSeek {
}
#[derive(Debug, Clone, PartialEq)]
-/// Request of a DataUpgrade from peer
+/// Request for a DataUpgrade from peer
pub struct RequestUpgrade {
/// Hypercore start index
pub start: u64,
@@ -79,7 +80,7 @@ pub struct DataBlock {
pub index: u64,
/// Data block value in bytes
pub value: Vec<u8>,
- /// TODO: document
+ /// Nodes of the merkle tree
pub nodes: Vec<Node>,
}
@@ -104,11 +105,11 @@ pub struct DataSeek {
#[derive(Debug, Clone, PartialEq)]
/// TODO: Document
pub struct DataUpgrade {
- /// TODO: Document
+ /// Starting block of this upgrade response
pub start: u64,
- /// TODO: Document
+ /// Number of blocks in this upgrade response
pub length: u64,
- /// TODO: Document
+ /// The nodes of the merkle tree
pub nodes: Vec<Node>,
/// TODO: Document
pub additional_nodes: Vec<Node>,
diff --git a/src/core.rs b/src/core.rs
index 886ff98..cf82049 100644
--- a/src/core.rs
+++ b/src/core.rs
@@ -48,10 +48,12 @@ pub struct Hypercore {
pub(crate) bitfield: Bitfield,
skip_flush_count: u8, // autoFlush in Javascript
header: Header,
+ #[cfg(feature = "replication")]
+ events: crate::replication::events::Events,
}
/// Response from append, matches that of the Javascript result
-#[derive(Debug)]
+#[derive(Debug, PartialEq)]
pub struct AppendOutcome {
/// Length of the hypercore after append
pub length: u64,
@@ -60,7 +62,7 @@ pub struct AppendOutcome {
}
/// Info about the hypercore
-#[derive(Debug)]
+#[derive(Debug, PartialEq)]
pub struct Info {
/// Length of the hypercore
pub length: u64,
@@ -247,6 +249,8 @@ impl Hypercore {
bitfield,
header,
skip_flush_count: 0,
+ #[cfg(feature = "replication")]
+ events: crate::replication::events::Events::new(),
})
}
@@ -321,6 +325,14 @@ impl Hypercore {
if self.should_flush_bitfield_and_tree_and_oplog() {
self.flush_bitfield_and_tree_and_oplog(false).await?;
}
+
+ #[cfg(feature = "replication")]
+ {
+ let _ = self.events.send(crate::replication::events::DataUpgrade {});
+ let _ = self
+ .events
+ .send(crate::replication::events::Have::from(&bitfield_update));
+ }
}
// Return the new value
@@ -330,10 +342,27 @@ impl Hypercore {
})
}
+ #[cfg(feature = "replication")]
+ /// Subscribe to core events relevant to replication
+ pub fn event_subscribe(&self) -> async_broadcast::Receiver<crate::replication::events::Event> {
+ self.events.channel.new_receiver()
+ }
+
+ /// Check if core has the block at the given `index` locally
+ #[instrument(ret, skip(self))]
+ pub fn has(&self, index: u64) -> bool {
+ self.bitfield.get(index)
+ }
+
/// Read value at given index, if any.
#[instrument(err, skip(self))]
pub async fn get(&mut self, index: u64) -> Result<Option<Vec<u8>>, HypercoreError> {
if !self.bitfield.get(index) {
+ #[cfg(feature = "replication")]
+ // if not in this core, emit Event::Get(index)
+ {
+ self.events.send_on_get(index);
+ }
return Ok(None);
}
@@ -522,12 +551,12 @@ impl Hypercore {
self.storage.flush_infos(&outcome.infos_to_flush).await?;
self.header = outcome.header;
- if let Some(bitfield_update) = bitfield_update {
+ if let Some(bitfield_update) = &bitfield_update {
// Write to bitfield
- self.bitfield.update(&bitfield_update);
+ self.bitfield.update(bitfield_update);
// Contiguous length is known only now
- update_contiguous_length(&mut self.header, &self.bitfield, &bitfield_update);
+ update_contiguous_length(&mut self.header, &self.bitfield, bitfield_update);
}
// Commit changeset to in-memory tree
@@ -537,6 +566,21 @@ impl Hypercore {
if self.should_flush_bitfield_and_tree_and_oplog() {
self.flush_bitfield_and_tree_and_oplog(false).await?;
}
+
+ #[cfg(feature = "replication")]
+ {
+ if proof.upgrade.is_some() {
+ // Notify replicator if we receieved an upgrade
+ let _ = self.events.send(crate::replication::events::DataUpgrade {});
+ }
+
+ // Notify replicator if we receieved a bitfield update
+ if let Some(ref bitfield) = bitfield_update {
+ let _ = self
+ .events
+ .send(crate::replication::events::Have::from(bitfield));
+ }
+ }
Ok(true)
}
@@ -725,7 +769,7 @@ fn update_contiguous_length(
}
#[cfg(test)]
-mod tests {
+pub(crate) mod tests {
use super::*;
#[async_std::test]
@@ -1091,7 +1135,9 @@ mod tests {
Ok(())
}
- async fn create_hypercore_with_data(length: u64) -> Result<Hypercore, HypercoreError> {
+ pub(crate) async fn create_hypercore_with_data(
+ length: u64,
+ ) -> Result<Hypercore, HypercoreError> {
let signing_key = generate_signing_key();
create_hypercore_with_data_and_key_pair(
length,
@@ -1103,7 +1149,7 @@ mod tests {
.await
}
- async fn create_hypercore_with_data_and_key_pair(
+ pub(crate) async fn create_hypercore_with_data_and_key_pair(
length: u64,
key_pair: PartialKeypair,
) -> Result<Hypercore, HypercoreError> {
diff --git a/src/lib.rs b/src/lib.rs
index 24d8627..eae3b21 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -1,4 +1,4 @@
-#![forbid(unsafe_code, bad_style, future_incompatible)]
+#![forbid(unsafe_code, future_incompatible)]
#![forbid(rust_2018_idioms, rust_2018_compatibility)]
#![forbid(missing_debug_implementations)]
#![forbid(missing_docs)]
@@ -74,6 +74,8 @@
pub mod encoding;
pub mod prelude;
+#[cfg(feature = "replication")]
+pub mod replication;
mod bitfield;
mod builder;
diff --git a/src/replication/events.rs b/src/replication/events.rs
new file mode 100644
index 0000000..b9c07df
--- /dev/null
+++ b/src/replication/events.rs
@@ -0,0 +1,177 @@
+//! events related to replication
+use crate::{common::BitfieldUpdate, HypercoreError};
+use async_broadcast::{broadcast, InactiveReceiver, Receiver, Sender};
+
+static MAX_EVENT_QUEUE_CAPACITY: usize = 32;
+
+/// Event emitted by [`crate::Hypercore::event_subscribe`]
+#[derive(Debug, Clone)]
+/// Emitted when [`crate::Hypercore::get`] is called when the block is missing.
+pub struct Get {
+ /// Index of the requested block
+ pub index: u64,
+ /// When the block is gotten this emits an event
+ pub get_result: Sender<()>,
+}
+
+/// Emitted when
+#[derive(Debug, Clone)]
+pub struct DataUpgrade {}
+
+/// Emitted when core gets new blocks
+#[derive(Debug, Clone)]
+pub struct Have {
+ /// Starting index of the blocks we have
+ pub start: u64,
+ /// The number of blocks
+ pub length: u64,
+ /// TODO
+ pub drop: bool,
+}
+
+impl From<&BitfieldUpdate> for Have {
+ fn from(
+ BitfieldUpdate {
+ start,
+ length,
+ drop,
+ }: &BitfieldUpdate,
+ ) -> Self {
+ Have {
+ start: *start,
+ length: *length,
+ drop: *drop,
+ }
+ }
+}
+
+#[derive(Debug, Clone)]
+/// Core events relevant to replication
+pub enum Event {
+ /// Emmited when core.get(i) happens for a missing block
+ Get(Get),
+ /// Emmitted when data.upgrade applied
+ DataUpgrade(DataUpgrade),
+ /// Emmitted when core gets new blocks
+ Have(Have),
+}
+
+/// Derive From<msg> for Enum where enum variant and msg have the same name
+macro_rules! impl_from_for_enum_variant {
+ ($enum_name:ident, $variant_and_msg_name:ident) => {
+ impl From<$variant_and_msg_name> for $enum_name {
+ fn from(value: $variant_and_msg_name) -> Self {
+ $enum_name::$variant_and_msg_name(value)
+ }
+ }
+ };
+}
+
+impl_from_for_enum_variant!(Event, Get);
+impl_from_for_enum_variant!(Event, DataUpgrade);
+impl_from_for_enum_variant!(Event, Have);
+
+#[derive(Debug)]
+pub(crate) struct Events {
+ /// Channel for core events
+ pub(crate) channel: Sender<Event>,
+ /// Kept around so `Events::channel` stays open.
+ _receiver: InactiveReceiver<Event>,
+}
+
+impl Events {
+ pub(crate) fn new() -> Self {
+ let (mut channel, receiver) = broadcast(MAX_EVENT_QUEUE_CAPACITY);
+ channel.set_await_active(false);
+ let mut _receiver = receiver.deactivate();
+ // Message sending is best effort. Is msg queue fills up, remove old messages to make place
+ // for new ones.
+ _receiver.set_overflow(true);
+ Self { channel, _receiver }
+ }
+
+ /// The internal channel errors on send when no replicators are subscribed,
+ /// For now we don't consider that an error, but just in case, we return a Result in case
+ /// we want to change this or add another fail path later.
+ pub(crate) fn send<T: Into<Event>>(&self, evt: T) -> Result<(), HypercoreError> {
+ let _errs_when_no_replicators_subscribed = self.channel.try_broadcast(evt.into());
+ Ok(())
+ }
+
+ /// Send a [`Get`] messages and return [`Receiver`] that will receive a message when block is
+ /// gotten.
+ pub(crate) fn send_on_get(&self, index: u64) -> Receiver<()> {
+ let (mut tx, rx) = broadcast(1);
+ tx.set_await_active(false);
+ let _ = self.send(Get {
+ index,
+ get_result: tx,
+ });
+ rx
+ }
+}
+
+#[cfg(test)]
+mod test {
+ use super::*;
+ use crate::replication::CoreMethodsError;
+
+ #[async_std::test]
+ async fn test_events() -> Result<(), CoreMethodsError> {
+ let mut core = crate::core::tests::create_hypercore_with_data(0).await?;
+
+ // Check that appending data emits a DataUpgrade and Have event
+
+ let mut rx = core.event_subscribe();
+ let handle = async_std::task::spawn(async move {
+ let mut out = vec![];
+ loop {
+ if out.len() == 2 {
+ return (out, rx);
+ }
+ if let Ok(evt) = rx.recv().await {
+ out.push(evt);
+ }
+ }
+ });
+ core.append(b"foo").await?;
+ let (res, mut rx) = handle.await;
+ assert!(matches!(res[0], Event::DataUpgrade(_)));
+ assert!(matches!(
+ res[1],
+ Event::Have(Have {
+ start: 0,
+ length: 1,
+ drop: false
+ })
+ ));
+ // no messages in queue
+ assert!(rx.is_empty());
+
+ // Check that Hypercore::get for missing data emits a Get event
+
+ let handle = async_std::task::spawn(async move {
+ let mut out = vec![];
+ loop {
+ if out.len() == 1 {
+ return (out, rx);
+ }
+ if let Ok(evt) = rx.recv().await {
+ out.push(evt);
+ }
+ }
+ });
+ assert_eq!(core.get(1).await?, None);
+ let (res, rx) = handle.await;
+ assert!(matches!(
+ res[0],
+ Event::Get(Get {
+ index: 1,
+ get_result: _
+ })
+ ));
+ // no messages in queue
+ assert!(rx.is_empty());
+ Ok(())
+ }
+}
diff --git a/src/replication/mod.rs b/src/replication/mod.rs
new file mode 100644
index 0000000..166cb30
--- /dev/null
+++ b/src/replication/mod.rs
@@ -0,0 +1,93 @@
+//! External interface for replication
+pub mod events;
+#[cfg(feature = "shared-core")]
+pub mod shared_core;
+
+#[cfg(feature = "shared-core")]
+pub use shared_core::SharedCore;
+
+use crate::{
+ AppendOutcome, HypercoreError, Info, PartialKeypair, Proof, RequestBlock, RequestSeek,
+ RequestUpgrade,
+};
+
+pub use events::Event;
+
+use async_broadcast::Receiver;
+use std::future::Future;
+
+/// Methods related to just this core's information
+pub trait CoreInfo {
+ /// Get core info (see: [`crate::Hypercore::info`]
+ fn info(&self) -> impl Future<Output = Info> + Send;
+ /// Get the key_pair (see: [`crate::Hypercore::key_pair`]
+ fn key_pair(&self) -> impl Future<Output = PartialKeypair> + Send;
+}
+
+/// Error for ReplicationMethods trait
+#[derive(thiserror::Error, Debug)]
+pub enum ReplicationMethodsError {
+ /// Error from hypercore
+ #[error("Got a hypercore error: [{0}]")]
+ HypercoreError(#[from] HypercoreError),
+ /// Error from CoreMethods
+ #[error("Got a CoreMethods error: [{0}]")]
+ CoreMethodsError(#[from] CoreMethodsError),
+}
+
+/// Methods needed for replication
+pub trait ReplicationMethods: CoreInfo + Send {
+ /// ref Core::verify_and_apply_proof
+ fn verify_and_apply_proof(
+ &self,
+ proof: &Proof,
+ ) -> impl Future<Output = Result<bool, ReplicationMethodsError>> + Send;
+ /// ref Core::missing_nodes
+ fn missing_nodes(
+ &self,
+ index: u64,
+ ) -> impl Future<Output = Result<u64, ReplicationMethodsError>> + Send;
+ /// ref Core::create_proof
+ fn create_proof(
+ &self,
+ block: Option<RequestBlock>,
+ hash: Option<RequestBlock>,
+ seek: Option<RequestSeek>,
+ upgrade: Option<RequestUpgrade>,
+ ) -> impl Future<Output = Result<Option<Proof>, ReplicationMethodsError>> + Send;
+ /// subscribe to core events
+ fn event_subscribe(&self) -> impl Future<Output = Receiver<Event>>;
+}
+
+/// Error for CoreMethods trait
+#[derive(thiserror::Error, Debug)]
+pub enum CoreMethodsError {
+ /// Error from hypercore
+ #[error("Got a hypercore error [{0}]")]
+ HypercoreError(#[from] HypercoreError),
+}
+
+/// Trait for things that consume [`crate::Hypercore`] can instead use this trait
+/// so they can use all Hypercore-like things such as `SharedCore`.
+pub trait CoreMethods: CoreInfo {
+ /// Check if the core has the block at the given index locally
+ fn has(&self, index: u64) -> impl Future<Output = bool> + Send;
+
+ /// get a block
+ fn get(
+ &self,
+ index: u64,
+ ) -> impl Future<Output = Result<Option<Vec<u8>>, CoreMethodsError>> + Send;
+
+ /// Append data to the core
+ fn append(
+ &self,
+ data: &[u8],
+ ) -> impl Future<Output = Result<AppendOutcome, CoreMethodsError>> + Send;
+
+ /// Append a batch of data to the core
+ fn append_batch<A: AsRef<[u8]>, B: AsRef<[A]> + Send>(
+ &self,
+ batch: B,
+ ) -> impl Future<Output = Result<AppendOutcome, CoreMethodsError>> + Send;
+}
diff --git a/src/replication/shared_core.rs b/src/replication/shared_core.rs
new file mode 100644
index 0000000..f30de47
--- /dev/null
+++ b/src/replication/shared_core.rs
@@ -0,0 +1,224 @@
+//! Implementation of a Hypercore that can have multiple owners. Along with implementations of all
+//! the hypercore traits.
+use crate::{
+ AppendOutcome, Hypercore, Info, PartialKeypair, Proof, RequestBlock, RequestSeek,
+ RequestUpgrade,
+};
+use async_broadcast::Receiver;
+use async_lock::Mutex;
+use std::{future::Future, sync::Arc};
+
+use super::{
+ CoreInfo, CoreMethods, CoreMethodsError, Event, ReplicationMethods, ReplicationMethodsError,
+};
+
+/// Hypercore that can have multiple owners
+#[derive(Debug, Clone)]
+pub struct SharedCore(pub Arc<Mutex<Hypercore>>);
+
+impl From<Hypercore> for SharedCore {
+ fn from(core: Hypercore) -> Self {
+ SharedCore(Arc::new(Mutex::new(core)))
+ }
+}
+impl SharedCore {
+ /// Create a shared core from a [`Hypercore`]
+ pub fn from_hypercore(core: Hypercore) -> Self {
+ SharedCore(Arc::new(Mutex::new(core)))
+ }
+}
+
+impl CoreInfo for SharedCore {
+ fn info(&self) -> impl Future<Output = Info> + Send {
+ async move {
+ let core = &self.0.lock().await;
+ core.info()
+ }
+ }
+
+ fn key_pair(&self) -> impl Future<Output = PartialKeypair> + Send {
+ async move {
+ let core = &self.0.lock().await;
+ core.key_pair().clone()
+ }
+ }
+}
+
+impl ReplicationMethods for SharedCore {
+ fn verify_and_apply_proof(
+ &self,
+ proof: &Proof,
+ ) -> impl Future<Output = Result<bool, ReplicationMethodsError>> {
+ async move {
+ let mut core = self.0.lock().await;
+ Ok(core.verify_and_apply_proof(proof).await?)
+ }
+ }
+
+ fn missing_nodes(
+ &self,
+ index: u64,
+ ) -> impl Future<Output = Result<u64, ReplicationMethodsError>> {
+ async move {
+ let mut core = self.0.lock().await;
+ Ok(core.missing_nodes(index).await?)
+ }
+ }
+
+ fn create_proof(
+ &self,
+ block: Option<RequestBlock>,
+ hash: Option<RequestBlock>,
+ seek: Option<RequestSeek>,
+ upgrade: Option<RequestUpgrade>,
+ ) -> impl Future<Output = Result<Option<Proof>, ReplicationMethodsError>> {
+ async move {
+ let mut core = self.0.lock().await;
+ Ok(core.create_proof(block, hash, seek, upgrade).await?)
+ }
+ }
+
+ fn event_subscribe(&self) -> impl Future<Output = Receiver<Event>> {
+ async move { self.0.lock().await.event_subscribe() }
+ }
+}
+
+impl CoreMethods for SharedCore {
+ fn has(&self, index: u64) -> impl Future<Output = bool> + Send {
+ async move {
+ let core = self.0.lock().await;
+ core.has(index)
+ }
+ }
+ fn get(
+ &self,
+ index: u64,
+ ) -> impl Future<Output = Result<Option<Vec<u8>>, CoreMethodsError>> + Send {
+ async move {
+ let mut core = self.0.lock().await;
+ Ok(core.get(index).await?)
+ }
+ }
+
+ fn append(
+ &self,
+ data: &[u8],
+ ) -> impl Future<Output = Result<AppendOutcome, CoreMethodsError>> + Send {
+ async move {
+ let mut core = self.0.lock().await;
+ Ok(core.append(data).await?)
+ }
+ }
+
+ fn append_batch<A: AsRef<[u8]>, B: AsRef<[A]> + Send>(
+ &self,
+ batch: B,
+ ) -> impl Future<Output = Result<AppendOutcome, CoreMethodsError>> + Send {
+ async move {
+ let mut core = self.0.lock().await;
+ Ok(core.append_batch(batch).await?)
+ }
+ }
+}
+
+#[cfg(test)]
+mod tests {
+
+ use super::*;
+
+ use crate::core::tests::{create_hypercore_with_data, create_hypercore_with_data_and_key_pair};
+ #[async_std::test]
+ async fn shared_core_methods() -> Result<(), CoreMethodsError> {
+ let core = crate::core::tests::create_hypercore_with_data(0).await?;
+ let core = SharedCore::from(core);
+
+ // check CoreInfo
+ let info = core.info().await;
+ assert_eq!(
+ info,
+ crate::core::Info {
+ length: 0,
+ byte_length: 0,
+ contiguous_length: 0,
+ fork: 0,
+ writeable: true,
+ }
+ );
+
+ // key_pair is random, nothing to test here
+ let _kp = core.key_pair().await;
+
+ // check CoreMethods
+ assert_eq!(core.has(0).await, false);
+ assert_eq!(core.get(0).await?, None);
+ let res = core.append(b"foo").await?;
+ assert_eq!(
+ res,
+ AppendOutcome {
+ length: 1,
+ byte_length: 3
+ }
+ );
+ assert_eq!(core.has(0).await, true);
+ assert_eq!(core.get(0).await?, Some(b"foo".into()));
+ let res = core.append_batch([b"hello", b"world"]).await?;
+ assert_eq!(
+ res,
+ AppendOutcome {
+ length: 3,
+ byte_length: 13
+ }
+ );
+ assert_eq!(core.has(2).await, true);
+ assert_eq!(core.get(2).await?, Some(b"world".into()));
+ Ok(())
+ }
+
+ #[async_std::test]
+ async fn shared_core_replication_methods() -> Result<(), ReplicationMethodsError> {
+ let main = create_hypercore_with_data(10).await?;
+ let clone = create_hypercore_with_data_and_key_pair(
+ 0,
+ PartialKeypair {
+ public: main.key_pair.public,
+ secret: None,
+ },
+ )
+ .await?;
+
+ let main = SharedCore::from(main);
+ let clone = SharedCore::from(clone);
+
+ let index = 6;
+ let nodes = clone.missing_nodes(index).await?;
+ let proof = main
+ .create_proof(
+ None,
+ Some(RequestBlock { index, nodes }),
+ None,
+ Some(RequestUpgrade {
+ start: 0,
+ length: 10,
+ }),
+ )
+ .await?
+ .unwrap();
+ assert!(clone.verify_and_apply_proof(&proof).await?);
+ let main_info = main.info().await;
+ let clone_info = clone.info().await;
+ assert_eq!(main_info.byte_length, clone_info.byte_length);
+ assert_eq!(main_info.length, clone_info.length);
+ assert!(main.get(6).await?.is_some());
+ assert!(clone.get(6).await?.is_none());
+
+ // Fetch data for index 6 and verify it is found
+ let index = 6;
+ let nodes = clone.missing_nodes(index).await?;
+ let proof = main
+ .create_proof(Some(RequestBlock { index, nodes }), None, None, None)
+ .await?
+ .unwrap();
+ assert!(clone.verify_and_apply_proof(&proof).await?);
+ Ok(())
+ }
+}
diff --git a/src/storage/mod.rs b/src/storage/mod.rs
index 333da2b..ad4b68a 100644
--- a/src/storage/mod.rs
+++ b/src/storage/mod.rs
@@ -147,11 +147,7 @@ impl Storage {
instruction.index,
&buf,
)),
- Err(RandomAccessError::OutOfBounds {
- offset: _,
- end: _,
- length,
- }) => {
+ Err(RandomAccessError::OutOfBounds { length, .. }) => {
if instruction.allow_miss {
Ok(StoreInfo::new_content_miss(
instruction.store.clone(),
diff --git a/src/tree/merkle_tree_changeset.rs b/src/tree/merkle_tree_changeset.rs
index be28873..9305302 100644
--- a/src/tree/merkle_tree_changeset.rs
+++ b/src/tree/merkle_tree_changeset.rs
@@ -10,8 +10,8 @@ use crate::{
/// first create the changes to this changeset, get out information from this to put to the oplog,
/// and the commit the changeset to the tree.
///
-/// This is called "MerkleTreeBatch" in Javascript, see:
-/// https://github.com/hypercore-protocol/hypercore/blob/master/lib/merkle-tree.js
+/// This is called "MerkleTreeBatch" in Javascript, source
+/// [here](https://github.com/holepunchto/hypercore/blob/88a1a2f1ebe6e33102688225516c4e882873f710/lib/merkle-tree.js#L44).
#[derive(Debug)]
pub(crate) struct MerkleTreeChangeset {
pub(crate) length: u64,
diff --git a/tests/js/package.json b/tests/js/package.json
index 2c5db7d..e5132c9 100644
--- a/tests/js/package.json
+++ b/tests/js/package.json
@@ -5,6 +5,6 @@
"step": "node interop.js"
},
"dependencies": {
- "hypercore": "^10"
+ "hypercore": "10.31.12"
}
}