feat: add iroh-sync and integrate into iroh node

Cargo.toml

@@ -5,6 +5,7 @@ members = [
   "iroh-bytes",
   "iroh-gossip",
   "iroh-metrics",
+  "iroh-sync",
 ]
 
 [profile.release]

deny.toml

@@ -4,8 +4,10 @@ multiple-versions = "allow"
 [licenses]
 allow = [
       "Apache-2.0",
+      "Apache-2.0 WITH LLVM-exception",
       "BSD-2-Clause",
       "BSD-3-Clause",
+      "BSL-1.0", # BOSL license
       "ISC",
       "MIT",
       "OpenSSL",

iroh-bytes/src/protocol.rs

@@ -172,7 +172,7 @@ impl GetRequest {
 }
 
 /// Write the given data to the provider sink, with a unsigned varint length prefix.
-pub(crate) async fn write_lp<W: AsyncWrite + Unpin>(writer: &mut W, data: &[u8]) -> Result<()> {
+pub async fn write_lp<W: AsyncWrite + Unpin>(writer: &mut W, data: &[u8]) -> Result<()> {
     ensure!(
         data.len() < MAX_MESSAGE_SIZE,
         "sending message is too large"
@@ -193,7 +193,7 @@ pub(crate) async fn write_lp<W: AsyncWrite + Unpin>(writer: &mut W, data: &[u8])
 ///
 /// The message as raw bytes.  If the end of the stream is reached and there is no partial
 /// message, returns `None`.
-pub(crate) async fn read_lp(
+pub async fn read_lp(
     mut reader: impl AsyncRead + Unpin,
     buffer: &mut BytesMut,
 ) -> Result<Option<Bytes>> {

iroh-bytes/src/util.rs

@@ -83,6 +83,18 @@ impl From<[u8; 32]> for Hash {
     }
 }
 
+impl From<Hash> for [u8; 32] {
+    fn from(value: Hash) -> Self {
+        *value.as_bytes()
+    }
+}
+
+impl From<&[u8; 32]> for Hash {
+    fn from(value: &[u8; 32]) -> Self {
+        Hash(blake3::Hash::from(*value))
+    }
+}
+
 impl PartialOrd for Hash {
     fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
         Some(self.0.as_bytes().cmp(other.0.as_bytes()))
@@ -202,6 +214,12 @@ impl From<anyhow::Error> for RpcError {
     }
 }
 
+impl From<std::io::Error> for RpcError {
+    fn from(e: std::io::Error) -> Self {
+        RpcError(serde_error::Error::new(&e))
+    }
+}
+
 /// A serializable result type for use in RPC responses.
 #[allow(dead_code)]
 pub type RpcResult<T> = result::Result<T, RpcError>;

iroh-gossip/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "iroh-gossip"
-version = "0.4.1"
+version = "0.5.1"
 edition = "2021"
 readme = "README.md"
 description = "gossip messages over broadcast trees"
@@ -14,7 +14,7 @@ rust-version = "1.67"
 [dependencies]
 # proto dependencies (required)
 anyhow = { version = "1", features = ["backtrace"] }
-blake3 = "1.3.3"
+blake3 = { package = "iroh-blake3", version = "1.4.3"}
 bytes = { version = "1.4.0", features = ["serde"] }
 data-encoding = "2.4.0"
 derive_more = { version = "1.0.0-beta.1", features = ["add", "debug", "display", "from", "try_into"] }

iroh-gossip/src/net.rs

@@ -17,15 +17,15 @@ use tokio::{
     sync::{broadcast, mpsc, oneshot, watch},
     task::JoinHandle,
 };
-use tracing::{debug, warn};
+use tracing::{debug, trace, warn};
 
 use self::util::{read_message, write_message, Dialer, Timers};
 use crate::proto::{self, TopicId};
 
 pub mod util;
 
 /// ALPN protocol name
-pub const GOSSIP_ALPN: &[u8] = b"n0/iroh-gossip/0";
+pub const GOSSIP_ALPN: &[u8] = b"/iroh-gossip/0";
 /// Maximum message size is limited to 1024 bytes.
 pub const MAX_MESSAGE_SIZE: usize = 1024;
 
@@ -265,6 +265,20 @@ struct IrohInfo {
     derp_region: Option<u16>,
 }
 
+impl IrohInfo {
+    async fn from_endpoint(endpoint: &MagicEndpoint) -> anyhow::Result<Self> {
+        Ok(Self {
+            addrs: endpoint
+                .local_endpoints()
+                .await?
+                .iter()
+                .map(|ep| ep.addr)
+                .collect(),
+            derp_region: endpoint.my_derp().await,
+        })
+    }
+}
+
 /// Whether a connection is initiated by us (Dial) or by the remote peer (Accept)
 #[derive(Debug)]
 enum ConnOrigin {
@@ -356,11 +370,7 @@ impl Actor {
                     }
                 },
                 _ = self.on_endpoints_rx.changed() => {
-                    let endpoints = self.on_endpoints_rx.borrow().clone();
-                    let info = IrohInfo {
-                        addrs: endpoints.iter().map(|ep| ep.addr).collect(),
-                        derp_region: self.endpoint.my_derp().await
-                    };
+                    let info = IrohInfo::from_endpoint(&self.endpoint).await?;
                     let peer_data = postcard::to_stdvec(&info)?;
                     self.handle_in_event(InEvent::UpdatePeerData(peer_data.into()), Instant::now()).await?;
                 }
@@ -469,13 +479,21 @@ impl Actor {
 
     async fn handle_in_event(&mut self, event: InEvent, now: Instant) -> anyhow::Result<()> {
         let me = *self.state.me();
-        debug!(me = ?me, "handle in_event  {event:?}");
+        if matches!(event, InEvent::TimerExpired(_)) {
+            trace!(me = ?me, "handle in_event  {event:?}");
+        } else {
+            debug!(me = ?me, "handle in_event  {event:?}");
+        };
         if let InEvent::PeerDisconnected(peer) = &event {
             self.conn_send_tx.remove(peer);
         }
         let out = self.state.handle(event, now);
         for event in out {
-            debug!(me = ?me, "handle out_event {event:?}");
+            if matches!(event, OutEvent::ScheduleTimer(_, _)) {
+                trace!(me = ?me, "handle out_event {event:?}");
+            } else {
+                debug!(me = ?me, "handle out_event {event:?}");
+            };
             match event {
                 OutEvent::SendMessage(peer_id, message) => {
                     if let Some(send) = self.conn_send_tx.get(&peer_id) {
@@ -518,10 +536,14 @@ impl Actor {
                 OutEvent::PeerData(peer, data) => match postcard::from_bytes::<IrohInfo>(&data) {
                     Err(err) => warn!("Failed to decode PeerData from {peer}: {err}"),
                     Ok(info) => {
-                        debug!("add known addrs for {peer}: {info:?}...");
-                        self.endpoint
+                        debug!(me = ?self.endpoint.peer_id(), peer = ?peer, "add known addrs: {info:?}");
+                        if let Err(err) = self
+                            .endpoint
                             .add_known_addrs(peer, info.derp_region, &info.addrs)
-                            .await?;
+                            .await
+                        {
+                            debug!(me = ?self.endpoint.peer_id(), peer = ?peer, "add known failed: {err:?}");
+                        }
                     }
                 },
             }

iroh-gossip/src/proto/util.rs

@@ -56,7 +56,7 @@ macro_rules! idbytes_impls {
             }
         }
 
-        impl<T: Into<[u8; 32]>> From<T> for $ty {
+        impl<T: Into<[u8; 32]>> std::convert::From<T> for $ty {
             fn from(value: T) -> Self {
                 Self::from_bytes(value.into())
             }

iroh-net/src/key.rs

@@ -50,6 +50,18 @@ impl PublicKey {
         self.public.as_bytes()
     }
 
+    /// Construct a `PublicKey` from a slice of bytes.
+    ///
+    /// # Warning
+    ///
+    /// This will return a [`SignatureError`] if the bytes passed into this method do not represent
+    /// a valid `ed25519_dalek` curve point. Will never fail for bytes return from [`Self::as_bytes`].
+    /// See [`VerifyingKey::from_bytes`] for details.
+    pub fn from_bytes(bytes: &[u8; 32]) -> Result<Self, SignatureError> {
+        let public = VerifyingKey::from_bytes(bytes)?;
+        Ok(public.into())
+    }
+
     fn public_crypto_box(&self) -> crypto_box::PublicKey {
         crypto_box::PublicKey::from_bytes(self.public_crypto_box)
     }
@@ -74,6 +86,15 @@ impl TryFrom<&[u8]> for PublicKey {
     }
 }
 
+impl TryFrom<&[u8; 32]> for PublicKey {
+    type Error = SignatureError;
+
+    #[inline]
+    fn try_from(bytes: &[u8; 32]) -> Result<Self, Self::Error> {
+        Self::from_bytes(bytes)
+    }
+}
+
 impl AsRef<[u8]> for PublicKey {
     fn as_ref(&self) -> &[u8] {
         self.as_bytes()
@@ -233,6 +254,12 @@ impl SecretKey {
         self.secret.to_bytes()
     }
 
+    /// Create a secret key from its byte representation.
+    pub fn from_bytes(bytes: &[u8; 32]) -> Self {
+        let secret = SigningKey::from_bytes(bytes);
+        secret.into()
+    }
+
     fn secret_crypto_box(&self) -> &crypto_box::SecretKey {
         self.secret_crypto_box
             .get_or_init(|| secret_ed_box(&self.secret))
@@ -250,8 +277,7 @@ impl From<SigningKey> for SecretKey {
 
 impl From<[u8; 32]> for SecretKey {
     fn from(value: [u8; 32]) -> Self {
-        let secret = SigningKey::from(value);
-        secret.into()
+        Self::from_bytes(&value)
     }
 }
 

iroh-sync/Cargo.toml

@@ -0,0 +1,41 @@
+[package]
+name = "iroh-sync"
+version = "0.5.1"
+edition = "2021"
+readme = "README.md"
+description = "Iroh sync"
+license = "MIT/Apache-2.0"
+authors = ["n0 team"]
+repository = "https://github.com/n0-computer/iroh"
+
+[dependencies]
+anyhow = "1"
+blake3 = { package = "iroh-blake3", version = "1.4.3"}
+crossbeam = "0.8.2"
+derive_more = { version = "1.0.0-beta.1", features = ["debug", "display", "from", "try_into"] }
+ed25519-dalek = { version = "2.0.0", features = ["serde", "rand_core"] }
+flume = "0.10"
+iroh-bytes = { version = "0.5.0", path = "../iroh-bytes" }
+iroh-metrics = { version = "0.5.0", path = "../iroh-metrics", optional = true }
+once_cell = "1.18.0"
+postcard = { version = "1", default-features = false, features = ["alloc", "use-std", "experimental-derive"] }
+rand = "0.8.5"
+rand_core = "0.6.4"
+serde = { version = "1.0.164", features = ["derive"] }
+url = "2.4"
+bytes = "1"
+parking_lot = "0.12.1"
+hex = "0.4"
+
+# fs-store
+redb = { version = "1.0.5", optional = true }
+ouroboros = { version = "0.17", optional = true }
+
+[dev-dependencies]
+tokio = { version = "1", features = ["sync", "macros"] }
+tempfile = "3.4"
+
+[features]
+default = ["fs-store", "metrics"]
+fs-store = ["redb", "ouroboros"]
+metrics = ["iroh-metrics"]