Merge branch 'main' into newtype-it

Author: rklaehn

Cargo.toml

@@ -21,7 +21,6 @@ quinn = { package = "iroh-quinn", version = "0.14.0" }
 n0-future = "0.2.0"
 n0-snafu = "0.2.0"
 range-collections = { version = "0.4.6", features = ["serde"] }
-redb = { version = "=2.4" }
 smallvec = { version = "1", features = ["serde", "const_new"] }
 snafu = "0.8.5"
 tokio = { version = "1.43.0", features = ["full"] }
@@ -41,10 +40,11 @@ iroh = "0.91.1"
 self_cell = "1.1.0"
 genawaiter = { version = "0.99.1", features = ["futures03"] }
 iroh-base = "0.91.1"
-reflink-copy = "0.1.24"
 irpc = { version = "0.7.0", features = ["rpc", "quinn_endpoint_setup", "spans", "stream", "derive"], default-features = false }
 iroh-metrics = { version = "0.35" }
 async-compression = { version = "0.4.30", features = ["lz4", "tokio"] }
+redb = { version = "=2.4", optional = true }
+reflink-copy = { version = "0.1.24", optional = true }
 
 [dev-dependencies]
 clap = { version = "4.5.31", features = ["derive"] }
@@ -67,7 +67,8 @@ concat_const = "0.2.0"
 [features]
 hide-proto-docs = []
 metrics = []
-default = ["hide-proto-docs"]
+default = ["hide-proto-docs", "fs-store"]
+fs-store = ["dep:redb", "dep:reflink-copy"]
 
 [patch.crates-io]
 iroh = { git = "https://github.com/n0-computer/iroh", branch = "main" }

examples/compression.rs

@@ -151,12 +151,11 @@ impl<C: Compression> ProtocolHandler for CompressedBlobsProtocol<C> {
         connection: iroh::endpoint::Connection,
     ) -> std::result::Result<(), iroh::protocol::AcceptError> {
         let connection_id = connection.stable_id() as u64;
-        let node_id = connection.remote_node_id()?;
         if let Err(cause) = self
             .events
             .client_connected(|| ClientConnected {
                 connection_id,
-                node_id,
+                node_id: connection.remote_node_id().ok(),
             })
             .await
         {

examples/limit.rs

@@ -81,20 +81,26 @@ fn limit_by_node_id(allowed_nodes: HashSet<NodeId>) -> EventSender {
     let mask = EventMask {
         // We want a request for each incoming connection so we can accept
         // or reject them. We don't need any other events.
-        connected: ConnectMode::Request,
+        connected: ConnectMode::Intercept,
         ..EventMask::DEFAULT
     };
     let (tx, mut rx) = EventSender::channel(32, mask);
     n0_future::task::spawn(async move {
         while let Some(msg) = rx.recv().await {
             if let ProviderMessage::ClientConnected(msg) = msg {
-                let node_id = msg.node_id;
-                let res = if allowed_nodes.contains(&node_id) {
-                    println!("Client connected: {node_id}");
-                    Ok(())
-                } else {
-                    println!("Client rejected: {node_id}");
-                    Err(AbortReason::Permission)
+                let res = match msg.node_id {
+                    Some(node_id) if allowed_nodes.contains(&node_id) => {
+                        println!("Client connected: {node_id}");
+                        Ok(())
+                    }
+                    Some(node_id) => {
+                        println!("Client rejected: {node_id}");
+                        Err(AbortReason::Permission)
+                    }
+                    None => {
+                        println!("Client rejected: no node id");
+                        Err(AbortReason::Permission)
+                    }
                 };
                 msg.tx.send(res).await.ok();
             }
@@ -108,7 +114,7 @@ fn limit_by_hash(allowed_hashes: HashSet<Hash>) -> EventSender {
         // We want to get a request for each get request that we can answer
         // with OK or not OK depending on the hash. We do not want detailed
         // events once it has been decided to handle a request.
-        get: RequestMode::Request,
+        get: RequestMode::Intercept,
         ..EventMask::DEFAULT
     };
     let (tx, mut rx) = EventSender::channel(32, mask);
@@ -136,7 +142,7 @@ fn throttle(delay_ms: u64) -> EventSender {
     let mask = EventMask {
         // We want to get requests for each sent user data blob, so we can add a delay.
         // Other than that, we don't need any events.
-        throttle: ThrottleMode::Throttle,
+        throttle: ThrottleMode::Intercept,
         ..EventMask::DEFAULT
     };
     let (tx, mut rx) = EventSender::channel(32, mask);
@@ -190,40 +196,30 @@ fn limit_max_connections(max_connections: usize) -> EventSender {
         // based on the current connection count if we want to accept or reject.
         // We also want detailed logging of events for the get request, so we can
         // detect when the request is finished one way or another.
-        get: RequestMode::RequestLog,
+        connected: ConnectMode::Intercept,
         ..EventMask::DEFAULT
     };
     let (tx, mut rx) = EventSender::channel(32, mask);
     n0_future::task::spawn(async move {
         let requests = ConnectionCounter::new(max_connections);
         while let Some(msg) = rx.recv().await {
-            if let ProviderMessage::GetRequestReceived(mut msg) = msg {
-                let connection_id = msg.connection_id;
-                let request_id = msg.request_id;
-                let res = requests.inc();
-                match res {
-                    Ok(n) => {
-                        println!("Accepting request {n}, id ({connection_id},{request_id})");
-                        msg.tx.send(Ok(())).await.ok();
-                    }
-                    Err(_) => {
-                        println!(
-                            "Connection limit of {max_connections} exceeded, rejecting request"
-                        );
-                        msg.tx.send(Err(AbortReason::RateLimited)).await.ok();
-                        continue;
-                    }
+            match msg {
+                ProviderMessage::ClientConnected(msg) => {
+                    let connection_id = msg.connection_id;
+                    let node_id = msg.node_id;
+                    let res = if let Ok(n) = requests.inc() {
+                        println!("Accepting connection {n}, node_id {node_id:?}, connection_id {connection_id}");
+                        Ok(())
+                    } else {
+                        Err(AbortReason::RateLimited)
+                    };
+                    msg.tx.send(res).await.ok();
                 }
-                let requests = requests.clone();
-                n0_future::task::spawn(async move {
-                    // just drain the per request events
-                    //
-                    // Note that we have requested updates for the request, now we also need to process them
-                    // otherwise the request will be aborted!
-                    while let Ok(Some(_)) = msg.rx.recv().await {}
-                    println!("Stopping request, id ({connection_id},{request_id})");
+                ProviderMessage::ConnectionClosed(msg) => {
                     requests.dec();
-                });
+                    println!("Connection closed, connection_id {}", msg.connection_id,);
+                }
+                _ => {}
             }
         }
     });

src/api.rs

@@ -98,7 +98,7 @@ pub enum ExportBaoError {
     #[snafu(display("encode error: {source}"))]
     ExportBaoInner { source: bao_tree::io::EncodeError },
     #[snafu(display("client error: {source}"))]
-    Progress { source: ProgressError },
+    ClientError { source: ProgressError },
 }
 
 impl From<ExportBaoError> for Error {
@@ -109,7 +109,7 @@ impl From<ExportBaoError> for Error {
             ExportBaoError::Request { source, .. } => Self::Io(source.into()),
             ExportBaoError::ExportBaoIo { source, .. } => Self::Io(source),
             ExportBaoError::ExportBaoInner { source, .. } => Self::Io(source.into()),
-            ExportBaoError::Progress { source, .. } => Self::Io(source.into()),
+            ExportBaoError::ClientError { source, .. } => Self::Io(source.into()),
         }
     }
 }
@@ -157,7 +157,7 @@ impl From<bao_tree::io::EncodeError> for ExportBaoError {
 
 impl From<ProgressError> for ExportBaoError {
     fn from(value: ProgressError) -> Self {
-        ProgressSnafu.into_error(value)
+        ClientSnafu.into_error(value)
     }
 }
 

src/api/blobs.rs

@@ -142,13 +142,15 @@ impl Blobs {
     /// clears the protections before.
     ///
     /// Users should rely only on garbage collection for blob deletion.
+    #[cfg(feature = "fs-store")]
     pub(crate) async fn delete_with_opts(&self, options: DeleteOptions) -> RequestResult<()> {
         trace!("{options:?}");
         self.client.rpc(options).await??;
         Ok(())
     }
 
     /// See [`Self::delete_with_opts`].
+    #[cfg(feature = "fs-store")]
     pub(crate) async fn delete(
         &self,
         hashes: impl IntoIterator<Item = impl Into<Hash>>,
@@ -501,6 +503,7 @@ impl Blobs {
         }
     }
 
+    #[allow(dead_code)]
     pub(crate) async fn clear_protected(&self) -> RequestResult<()> {
         let msg = ClearProtectedRequest;
         self.client.rpc(msg).await??;

src/api/blobs/reader.rs

@@ -214,6 +214,7 @@ impl tokio::io::AsyncSeek for BlobReader {
 }
 
 #[cfg(test)]
+#[cfg(feature = "fs-store")]
 mod tests {
     use bao_tree::ChunkRanges;
     use testresult::TestResult;

src/api/downloader.rs

@@ -34,7 +34,7 @@ pub struct Downloader {
 #[rpc_requests(message = SwarmMsg, alias = "Msg")]
 #[derive(Debug, Serialize, Deserialize)]
 enum SwarmProtocol {
-    #[rpc(tx = mpsc::Sender<DownloadProgessItem>)]
+    #[rpc(tx = mpsc::Sender<DownloadProgressItem>)]
     Download(DownloadRequest),
 }
 
@@ -46,7 +46,7 @@ struct DownloaderActor {
 }
 
 #[derive(Debug, Serialize, Deserialize)]
-pub enum DownloadProgessItem {
+pub enum DownloadProgressItem {
     #[serde(skip)]
     Error(anyhow::Error),
     TryProvider {
@@ -98,15 +98,15 @@ impl DownloaderActor {
 async fn handle_download(store: Store, pool: ConnectionPool, msg: DownloadMsg) {
     let DownloadMsg { inner, mut tx, .. } = msg;
     if let Err(cause) = handle_download_impl(store, pool, inner, &mut tx).await {
-        tx.send(DownloadProgessItem::Error(cause)).await.ok();
+        tx.send(DownloadProgressItem::Error(cause)).await.ok();
     }
 }
 
 async fn handle_download_impl(
     store: Store,
     pool: ConnectionPool,
     request: DownloadRequest,
-    tx: &mut mpsc::Sender<DownloadProgessItem>,
+    tx: &mut mpsc::Sender<DownloadProgressItem>,
 ) -> anyhow::Result<()> {
     match request.strategy {
         SplitStrategy::Split => handle_download_split_impl(store, pool, request, tx).await?,
@@ -127,7 +127,7 @@ async fn handle_download_split_impl(
     store: Store,
     pool: ConnectionPool,
     request: DownloadRequest,
-    tx: &mut mpsc::Sender<DownloadProgessItem>,
+    tx: &mut mpsc::Sender<DownloadProgressItem>,
 ) -> anyhow::Result<()> {
     let providers = request.providers;
     let requests = split_request(&request.request, &providers, &pool, &store, Drain).await?;
@@ -140,7 +140,7 @@ async fn handle_download_split_impl(
             let progress_tx = progress_tx.clone();
             async move {
                 let hash = request.hash;
-                let (tx, rx) = tokio::sync::mpsc::channel::<(usize, DownloadProgessItem)>(16);
+                let (tx, rx) = tokio::sync::mpsc::channel::<(usize, DownloadProgressItem)>(16);
                 progress_tx.send(rx).await.ok();
                 let sink = TokioMpscSenderSink(tx).with_map(move |x| (id, x));
                 let res = execute_get(&pool, Arc::new(request), &providers, &store, sink).await;
@@ -154,12 +154,12 @@ async fn handle_download_split_impl(
         into_stream(progress_rx)
             .flat_map(into_stream)
             .map(move |(id, item)| match item {
-                DownloadProgessItem::Progress(offset) => {
+                DownloadProgressItem::Progress(offset) => {
                     total += offset;
                     if let Some(prev) = offsets.insert(id, offset) {
                         total -= prev;
                     }
-                    DownloadProgessItem::Progress(total)
+                    DownloadProgressItem::Progress(total)
                 }
                 x => x,
             })
@@ -174,7 +174,7 @@ async fn handle_download_split_impl(
                     Some((_hash, Ok(()))) => {
                     }
                     Some((_hash, Err(_e))) => {
-                        tx.send(DownloadProgessItem::DownloadError).await?;
+                        tx.send(DownloadProgressItem::DownloadError).await?;
                     }
                     None => break,
                 }
@@ -298,19 +298,19 @@ impl<'de> Deserialize<'de> for DownloadRequest {
 pub type DownloadOptions = DownloadRequest;
 
 pub struct DownloadProgress {
-    fut: future::Boxed<irpc::Result<mpsc::Receiver<DownloadProgessItem>>>,
+    fut: future::Boxed<irpc::Result<mpsc::Receiver<DownloadProgressItem>>>,
 }
 
 impl DownloadProgress {
-    fn new(fut: future::Boxed<irpc::Result<mpsc::Receiver<DownloadProgessItem>>>) -> Self {
+    fn new(fut: future::Boxed<irpc::Result<mpsc::Receiver<DownloadProgressItem>>>) -> Self {
         Self { fut }
     }
 
-    pub async fn stream(self) -> irpc::Result<impl Stream<Item = DownloadProgessItem> + Unpin> {
+    pub async fn stream(self) -> irpc::Result<impl Stream<Item = DownloadProgressItem> + Unpin> {
         let rx = self.fut.await?;
         Ok(Box::pin(rx.into_stream().map(|item| match item {
             Ok(item) => item,
-            Err(e) => DownloadProgessItem::Error(e.into()),
+            Err(e) => DownloadProgressItem::Error(e.into()),
         })))
     }
 
@@ -320,8 +320,8 @@ impl DownloadProgress {
         tokio::pin!(stream);
         while let Some(item) = stream.next().await {
             match item? {
-                DownloadProgessItem::Error(e) => Err(e)?,
-                DownloadProgessItem::DownloadError => anyhow::bail!("Download error"),
+                DownloadProgressItem::Error(e) => Err(e)?,
+                DownloadProgressItem::DownloadError => anyhow::bail!("Download error"),
                 _ => {}
             }
         }
@@ -372,7 +372,7 @@ async fn split_request<'a>(
     providers: &Arc<dyn ContentDiscovery>,
     pool: &ConnectionPool,
     store: &Store,
-    progress: impl Sink<DownloadProgessItem, Error = irpc::channel::SendError>,
+    progress: impl Sink<DownloadProgressItem, Error = irpc::channel::SendError>,
 ) -> anyhow::Result<Box<dyn Iterator<Item = GetRequest> + Send + 'a>> {
     Ok(match request {
         FiniteRequest::Get(req) => {
@@ -428,13 +428,13 @@ async fn execute_get(
     request: Arc<GetRequest>,
     providers: &Arc<dyn ContentDiscovery>,
     store: &Store,
-    mut progress: impl Sink<DownloadProgessItem, Error = irpc::channel::SendError>,
+    mut progress: impl Sink<DownloadProgressItem, Error = irpc::channel::SendError>,
 ) -> anyhow::Result<()> {
     let remote = store.remote();
     let mut providers = providers.find_providers(request.content());
     while let Some(provider) = providers.next().await {
         progress
-            .send(DownloadProgessItem::TryProvider {
+            .send(DownloadProgressItem::TryProvider {
                 id: provider,
                 request: request.clone(),
             })
@@ -447,7 +447,7 @@ async fn execute_get(
         let local_bytes = local.local_bytes();
         let Ok(conn) = conn.await else {
             progress
-                .send(DownloadProgessItem::ProviderFailed {
+                .send(DownloadProgressItem::ProviderFailed {
                     id: provider,
                     request: request.clone(),
                 })
@@ -458,21 +458,21 @@ async fn execute_get(
             .execute_get_sink(
                 conn.clone(),
                 local.missing(),
-                (&mut progress).with_map(move |x| DownloadProgessItem::Progress(x + local_bytes)),
+                (&mut progress).with_map(move |x| DownloadProgressItem::Progress(x + local_bytes)),
             )
             .await
         {
             Ok(_stats) => {
                 progress
-                    .send(DownloadProgessItem::PartComplete {
+                    .send(DownloadProgressItem::PartComplete {
                         request: request.clone(),
                     })
                     .await?;
                 return Ok(());
             }
             Err(_cause) => {
                 progress
-                    .send(DownloadProgessItem::ProviderFailed {
+                    .send(DownloadProgressItem::ProviderFailed {
                         id: provider,
                         request: request.clone(),
                     })
@@ -521,6 +521,7 @@ impl ContentDiscovery for Shuffled {
 }
 
 #[cfg(test)]
+#[cfg(feature = "fs-store")]
 mod tests {
     use std::ops::Deref;