Merge remote-tracking branch 'origin/main' into feat/dns

Author: Frando

iroh-bytes/src/protocol.rs

@@ -351,7 +351,7 @@ use crate::Hash;
 pub const MAX_MESSAGE_SIZE: usize = 1024 * 1024 * 100;
 
 /// The ALPN used with quic for the iroh bytes protocol.
-pub const ALPN: &[u8] = b"/iroh-bytes/3";
+pub const ALPN: &[u8] = b"/iroh-bytes/4";
 
 #[derive(Deserialize, Serialize, Debug, PartialEq, Eq, Clone, From)]
 /// A request to the provider

iroh-net/src/magicsock.rs

@@ -538,7 +538,8 @@ impl Inner {
                 } else {
                     // overwrite the first byte of the packets with zero.
                     // this makes quinn reliably and quickly ignore the packet as long as
-                    // [`quinn::EndpointConfig::grease_quic_bit`] is set to `true`.
+                    // [`quinn::EndpointConfig::grease_quic_bit`] is set to `false`
+                    // (which we always do in MagicEndpoint::bind).
                     buf[start] = 0u8;
                 }
                 start = end;

iroh-sync/Cargo.toml

@@ -39,8 +39,8 @@ tracing = "0.1"
 tokio = { version = "1", features = ["sync"] }
 
 # fs-store
-redb = { version = "1.5.1", optional = true }
-ouroboros = { version = "0.18", optional = true }
+redb = { version = "1.5.1" }
+ouroboros = { version = "0.18" }
 
 # net
 iroh-net = { version = "0.12.0", optional = true, path = "../iroh-net" }
@@ -59,7 +59,6 @@ proptest = "1.2.0"
 test-strategy = "0.3.1"
 
 [features]
-default = ["net", "fs-store", "metrics"]
+default = ["net", "metrics"]
 net = ["iroh-net", "tokio/io-util", "tokio-stream", "tokio-util", "quinn", "futures"]
-fs-store = ["redb", "ouroboros"]
 metrics = ["iroh-metrics"]

iroh-sync/src/actor.rs

@@ -16,7 +16,7 @@ use tracing::{debug, error, error_span, trace, warn};
 
 use crate::{
     ranger::Message,
-    store::{self, DownloadPolicy, ImportNamespaceOutcome, Query},
+    store::{fs::StoreInstance, DownloadPolicy, ImportNamespaceOutcome, Query, Store},
     Author, AuthorHeads, AuthorId, Capability, CapabilityKind, ContentStatus,
     ContentStatusCallback, Event, NamespaceId, NamespaceSecret, PeerIdBytes, Replica, SignedEntry,
     SyncOutcome,
@@ -170,8 +170,8 @@ pub struct OpenState {
 }
 
 #[derive(Debug)]
-struct OpenReplica<S: store::Store> {
-    replica: Replica<S::Instance>,
+struct OpenReplica {
+    replica: Replica<StoreInstance>,
     handles: usize,
     sync: bool,
 }
@@ -218,8 +218,8 @@ impl OpenOpts {
 #[allow(missing_docs)]
 impl SyncHandle {
     /// Spawn a sync actor and return a handle.
-    pub fn spawn<S: store::Store>(
-        store: S,
+    pub fn spawn(
+        store: Store,
         content_status_callback: Option<ContentStatusCallback>,
         me: String,
     ) -> SyncHandle {
@@ -524,14 +524,14 @@ impl Drop for SyncHandle {
     }
 }
 
-struct Actor<S: store::Store> {
-    store: S,
-    states: OpenReplicas<S>,
+struct Actor {
+    store: Store,
+    states: OpenReplicas,
     action_rx: flume::Receiver<Action>,
     content_status_callback: Option<ContentStatusCallback>,
 }
 
-impl<S: store::Store> Actor<S> {
+impl Actor {
     fn run(&mut self) -> Result<()> {
         while let Ok(action) = self.action_rx.recv() {
             trace!(%action, "tick");
@@ -741,24 +741,22 @@ impl<S: store::Store> Actor<S> {
     }
 }
 
-struct OpenReplicas<S: store::Store>(HashMap<NamespaceId, OpenReplica<S>>);
+#[derive(Default)]
+struct OpenReplicas(HashMap<NamespaceId, OpenReplica>);
 
-// We need a manual impl here because the derive won't work unless we'd restrict to S: Default.
-impl<S: store::Store> Default for OpenReplicas<S> {
-    fn default() -> Self {
-        Self(Default::default())
-    }
-}
-impl<S: store::Store> OpenReplicas<S> {
-    fn replica(&mut self, namespace: &NamespaceId) -> Result<&mut Replica<S::Instance>> {
+impl OpenReplicas {
+    fn replica(&mut self, namespace: &NamespaceId) -> Result<&mut Replica<StoreInstance>> {
         self.get_mut(namespace).map(|state| &mut state.replica)
     }
 
-    fn get_mut(&mut self, namespace: &NamespaceId) -> Result<&mut OpenReplica<S>> {
+    fn get_mut(&mut self, namespace: &NamespaceId) -> Result<&mut OpenReplica> {
         self.0.get_mut(namespace).context("replica not open")
     }
 
-    fn replica_if_syncing(&mut self, namespace: &NamespaceId) -> Result<&mut Replica<S::Instance>> {
+    fn replica_if_syncing(
+        &mut self,
+        namespace: &NamespaceId,
+    ) -> Result<&mut Replica<StoreInstance>> {
         let state = self.get_mut(namespace)?;
         if !state.sync {
             Err(anyhow!("sync is not enabled for replica"))
@@ -781,7 +779,7 @@ impl<S: store::Store> OpenReplicas<S> {
         &mut self,
         namespace: NamespaceId,
         opts: OpenOpts,
-        open_cb: impl Fn() -> Result<Replica<S::Instance>>,
+        open_cb: impl Fn() -> Result<Replica<StoreInstance>>,
     ) -> Result<()> {
         match self.0.entry(namespace) {
             hash_map::Entry::Vacant(e) => {
@@ -811,7 +809,7 @@ impl<S: store::Store> OpenReplicas<S> {
     fn close_with(
         &mut self,
         namespace: NamespaceId,
-        on_close: impl Fn(Replica<S::Instance>),
+        on_close: impl Fn(Replica<StoreInstance>),
     ) -> bool {
         match self.0.entry(namespace) {
             hash_map::Entry::Vacant(_e) => {
@@ -833,7 +831,7 @@ impl<S: store::Store> OpenReplicas<S> {
         }
     }
 
-    fn close_all_with(&mut self, on_close: impl Fn(Replica<S::Instance>)) {
+    fn close_all_with(&mut self, on_close: impl Fn(Replica<StoreInstance>)) {
         for (_namespace, state) in self.0.drain() {
             on_close(state.replica)
         }
@@ -855,7 +853,7 @@ fn iter_to_channel<T: Send + 'static>(
     Ok(())
 }
 
-fn get_author<S: store::Store>(store: &S, id: &AuthorId) -> Result<Author> {
+fn get_author(store: &Store, id: &AuthorId) -> Result<Author> {
     store.get_author(id)?.context("author not found")
 }
 
@@ -866,10 +864,10 @@ fn send_reply<T>(sender: oneshot::Sender<T>, value: T) -> Result<(), SendReplyEr
     sender.send(value).map_err(send_reply_error)
 }
 
-fn send_reply_with<T, S: store::Store>(
+fn send_reply_with<T>(
     sender: oneshot::Sender<Result<T>>,
-    this: &mut Actor<S>,
-    f: impl FnOnce(&mut Actor<S>) -> Result<T>,
+    this: &mut Actor,
+    f: impl FnOnce(&mut Actor) -> Result<T>,
 ) -> Result<(), SendReplyError> {
     sender.send(f(this)).map_err(send_reply_error)
 }
@@ -880,10 +878,12 @@ fn send_reply_error<T>(_err: T) -> SendReplyError {
 
 #[cfg(test)]
 mod tests {
+    use crate::store;
+
     use super::*;
     #[tokio::test]
     async fn open_close() -> anyhow::Result<()> {
-        let store = store::memory::Store::default();
+        let store = store::Store::memory();
         let sync = SyncHandle::spawn(store, None, "foo".into());
         let namespace = NamespaceSecret::new(&mut rand::rngs::OsRng {});
         let id = namespace.id();

iroh-sync/src/lib.rs

@@ -21,10 +21,14 @@
 //! sets over a network, based on recursively partitioning the sets and comparing fingerprints of
 //! the partitions to probabilistically detect whether a partition requires further work.
 //!
-//! The crate exposes a [generic storage interface](store::Store) with
-//! [in-memory](store::memory::Store) and [persistent, file-based](store::fs::Store)
-//! implementations. The latter makes use of [`redb`], an embedded key-value store, and persists
-//! the whole store with all replicas to a single file.
+//! The crate exposes a [generic storage interface](store::Store). There is an implementation
+//! of this interface, [store::fs::Store], that can be used either
+//! [in-memory](store::fs::Store::memory) or in
+//! [persistent, file-based](store::fs::Store::persistent) mode.
+//!
+//! Both modes make use of [`redb`], an embedded key-value store. When used
+//! in-memory, the store is backed by a `Vec<u8>`. When used in persistent mode,
+//! the store is backed by a single file on disk.
 //!
 //! [paper]: https://arxiv.org/abs/2212.13567
 #![deny(missing_docs, rustdoc::broken_intra_doc_links)]

iroh-sync/src/net/codec.rs

@@ -295,7 +295,7 @@ impl BobState {
 mod tests {
     use crate::{
         actor::OpenOpts,
-        store::{self, Query, Store},
+        store::{self, fs::StoreInstance, Query, Store},
         AuthorId, NamespaceSecret,
     };
     use anyhow::Result;
@@ -311,7 +311,7 @@ mod tests {
         let alice_peer_id = SecretKey::from_bytes(&[1u8; 32]).public();
         let bob_peer_id = SecretKey::from_bytes(&[2u8; 32]).public();
 
-        let alice_store = store::memory::Store::default();
+        let alice_store = store::Store::memory();
         // For now uses same author on both sides.
         let author = alice_store.new_author(&mut rng).unwrap();
 
@@ -322,7 +322,7 @@ mod tests {
             .hash_and_insert("hello bob", &author, "from alice")
             .unwrap();
 
-        let bob_store = store::memory::Store::default();
+        let bob_store = store::Store::memory();
         let mut bob_replica = bob_store.new_replica(namespace.clone()).unwrap();
         bob_replica
             .hash_and_insert("hello alice", &author, "from bob")
@@ -419,26 +419,26 @@ mod tests {
     #[tokio::test]
     async fn test_sync_many_authors_memory() -> Result<()> {
         let _guard = iroh_test::logging::setup();
-        let alice_store = store::memory::Store::default();
-        let bob_store = store::memory::Store::default();
+        let alice_store = store::Store::memory();
+        let bob_store = store::Store::memory();
         test_sync_many_authors(alice_store, bob_store).await
     }
 
     #[tokio::test]
     async fn test_sync_many_authors_fs() -> Result<()> {
         let _guard = iroh_test::logging::setup();
         let tmpdir = tempfile::tempdir()?;
-        let alice_store = store::fs::Store::new(tmpdir.path().join("a.db"))?;
-        let bob_store = store::fs::Store::new(tmpdir.path().join("b.db"))?;
+        let alice_store = store::fs::Store::persistent(tmpdir.path().join("a.db"))?;
+        let bob_store = store::fs::Store::persistent(tmpdir.path().join("b.db"))?;
         test_sync_many_authors(alice_store, bob_store).await
     }
 
     type Message = (AuthorId, Vec<u8>, Hash);
 
-    fn insert_messages<S: Store>(
+    fn insert_messages(
         mut rng: impl CryptoRngCore,
-        store: &S,
-        replica: &mut crate::sync::Replica<S::Instance>,
+        store: &Store,
+        replica: &mut crate::sync::Replica<StoreInstance>,
         num_authors: usize,
         msgs_per_author: usize,
         key_value_fn: impl Fn(&AuthorId, usize) -> (String, String),
@@ -459,7 +459,7 @@ mod tests {
         res
     }
 
-    fn get_messages<S: Store>(store: &S, namespace: NamespaceId) -> Vec<Message> {
+    fn get_messages(store: &Store, namespace: NamespaceId) -> Vec<Message> {
         let mut msgs = store
             .get_many(namespace, Query::all())
             .unwrap()
@@ -478,7 +478,7 @@ mod tests {
         msgs
     }
 
-    async fn test_sync_many_authors<S: Store>(alice_store: S, bob_store: S) -> Result<()> {
+    async fn test_sync_many_authors(alice_store: Store, bob_store: Store) -> Result<()> {
         let num_messages = &[1, 2, 5, 10];
         let num_authors = &[2, 3, 4, 5, 10];
         let mut rng = rand_chacha::ChaCha12Rng::seed_from_u64(99);
@@ -616,21 +616,21 @@ mod tests {
     #[tokio::test]
     async fn test_sync_timestamps_memory() -> Result<()> {
         let _guard = iroh_test::logging::setup();
-        let alice_store = store::memory::Store::default();
-        let bob_store = store::memory::Store::default();
+        let alice_store = store::Store::memory();
+        let bob_store = store::Store::memory();
         test_sync_timestamps(alice_store, bob_store).await
     }
 
     #[tokio::test]
     async fn test_sync_timestamps_fs() -> Result<()> {
         let _guard = iroh_test::logging::setup();
         let tmpdir = tempfile::tempdir()?;
-        let alice_store = store::fs::Store::new(tmpdir.path().join("a.db"))?;
-        let bob_store = store::fs::Store::new(tmpdir.path().join("b.db"))?;
+        let alice_store = store::fs::Store::persistent(tmpdir.path().join("a.db"))?;
+        let bob_store = store::fs::Store::persistent(tmpdir.path().join("b.db"))?;
         test_sync_timestamps(alice_store, bob_store).await
     }
 
-    async fn test_sync_timestamps<S: Store>(alice_store: S, bob_store: S) -> Result<()> {
+    async fn test_sync_timestamps(alice_store: Store, bob_store: Store) -> Result<()> {
         let mut rng = rand_chacha::ChaCha12Rng::seed_from_u64(99);
         let alice_node_pubkey = SecretKey::generate_with_rng(&mut rng).public();
         let bob_node_pubkey = SecretKey::generate_with_rng(&mut rng).public();