Merge pull request #8 from n0-computer/try-out-new-rpc

refactor: use quic-rpc v0.15.0 and adapt code

Author: rklaehn

Cargo.lock

@@ -37,8 +37,8 @@ parking_lot = { version = "0.12.1", optional = true }
 pin-project = "1.1.5"
 portable-atomic = { version = "1", optional = true }
 postcard = { version = "1", default-features = false, features = ["alloc", "use-std", "experimental-derive"] }
-quic-rpc = { version = "0.13.0", optional = true }
-quic-rpc-derive = { version = "0.13.0", optional = true }
+quic-rpc = { version = "0.15.0", optional = true }
+quic-rpc-derive = { version = "0.15.0", optional = true }
 quinn = { package = "iroh-quinn", version = "0.12", features = ["ring"] }
 rand = "0.8"
 range-collections = "0.4.0"
@@ -82,9 +82,6 @@ fs-store = ["dep:reflink-copy", "redb", "dep:redb_v1", "dep:tempfile"]
 metrics = ["iroh-metrics/metrics"]
 redb = ["dep:redb"]
 rpc = ["dep:quic-rpc", "dep:quic-rpc-derive", "dep:nested_enum_utils", "dep:strum", "dep:futures-util", "dep:ref-cast", "dep:portable-atomic", "dep:walkdir", "downloader"]
-ref-cast = ["dep:ref-cast"]
-portable-atomic = ["dep:portable-atomic"]
-walkdir = ["dep:walkdir"]
 
 [package.metadata.docs.rs]
 all-features = true

Cargo.toml

@@ -21,6 +21,7 @@ allow = [
       "Unicode-DFS-2016",
       "Zlib",
       "MPL-2.0", # https://fossa.com/blog/open-source-software-licenses-101-mozilla-public-license-2-0/
+      "Unicode-3.0"
 ]
 
 [[licenses.clarify]]

deny.toml

@@ -30,9 +30,9 @@ use proto::{
         CreateRequest as TagsCreateRequest, DeleteRequest as TagDeleteRequest,
         ListRequest as TagListRequest, SetRequest as TagsSetRequest, SyncMode,
     },
-    RpcError, RpcResult,
+    Request, RpcError, RpcResult, RpcService,
 };
-use quic_rpc::server::{RpcChannel, RpcServerError};
+use quic_rpc::server::{ChannelTypes, RpcChannel, RpcServerError};
 
 use crate::{
     export::ExportProgress,
@@ -57,31 +57,29 @@ const RPC_BLOB_GET_CHANNEL_CAP: usize = 2;
 
 impl<D: crate::store::Store> Blobs<D> {
     /// Handle an RPC request
-    pub async fn handle_rpc_request<S, C>(
+    pub async fn handle_rpc_request<C>(
         self: Arc<Self>,
-        msg: crate::rpc::proto::Request,
-        chan: RpcChannel<crate::rpc::proto::RpcService, C, S>,
+        msg: Request,
+        chan: RpcChannel<RpcService, C>,
     ) -> std::result::Result<(), RpcServerError<C>>
     where
-        S: quic_rpc::Service,
-        C: quic_rpc::ServiceEndpoint<S>,
+        C: ChannelTypes<RpcService>,
     {
-        use crate::rpc::proto::Request::*;
+        use Request::*;
         match msg {
             Blobs(msg) => self.handle_blobs_request(msg, chan).await,
             Tags(msg) => self.handle_tags_request(msg, chan).await,
         }
     }
 
     /// Handle a tags request
-    pub async fn handle_tags_request<S, C>(
+    pub async fn handle_tags_request<C>(
         self: Arc<Self>,
         msg: proto::tags::Request,
-        chan: RpcChannel<proto::RpcService, C, S>,
+        chan: RpcChannel<proto::RpcService, C>,
     ) -> std::result::Result<(), RpcServerError<C>>
     where
-        S: quic_rpc::Service,
-        C: quic_rpc::ServiceEndpoint<S>,
+        C: ChannelTypes<proto::RpcService>,
     {
         use proto::tags::Request::*;
         match msg {
@@ -93,14 +91,13 @@ impl<D: crate::store::Store> Blobs<D> {
     }
 
     /// Handle a blobs request
-    pub async fn handle_blobs_request<Sv, C>(
+    pub async fn handle_blobs_request<C>(
         self: Arc<Self>,
         msg: proto::blobs::Request,
-        chan: RpcChannel<proto::RpcService, C, Sv>,
+        chan: RpcChannel<proto::RpcService, C>,
     ) -> std::result::Result<(), RpcServerError<C>>
     where
-        Sv: quic_rpc::Service,
-        C: quic_rpc::ServiceEndpoint<Sv>,
+        C: ChannelTypes<proto::RpcService>,
     {
         use proto::blobs::Request::*;
         match msg {

src/rpc.rs

@@ -73,8 +73,8 @@ use genawaiter::sync::{Co, Gen};
 use iroh_net::NodeAddr;
 use portable_atomic::{AtomicU64, Ordering};
 use quic_rpc::{
-    client::{BoxStreamSync, BoxedServiceConnection},
-    RpcClient,
+    client::{BoxStreamSync, BoxedConnector},
+    Connector, RpcClient,
 };
 use serde::{Deserialize, Serialize};
 use tokio::io::{AsyncRead, AsyncReadExt, ReadBuf};
@@ -87,6 +87,7 @@ use crate::{
     format::collection::{Collection, SimpleStore},
     get::db::DownloadProgress as BytesDownloadProgress,
     net_protocol::BlobDownloadRequest,
+    rpc::proto::RpcService,
     store::{BaoBlobSize, ConsistencyCheckProgress, ExportFormat, ExportMode, ValidateProgress},
     util::SetTagOption,
     BlobFormat, Hash, Tag,
@@ -105,20 +106,16 @@ use crate::rpc::proto::blobs::{
 
 /// Iroh blobs client.
 #[derive(Debug, Clone)]
-pub struct Client<
-    C = BoxedServiceConnection<crate::rpc::proto::RpcService>,
-    S = crate::rpc::proto::RpcService,
-> {
-    pub(super) rpc: RpcClient<crate::rpc::proto::RpcService, C, S>,
+pub struct Client<C = BoxedConnector<RpcService>> {
+    pub(super) rpc: RpcClient<RpcService, C>,
 }
 
-impl<C, S> Client<C, S>
+impl<C> Client<C>
 where
-    S: quic_rpc::Service,
-    C: quic_rpc::ServiceConnection<S>,
+    C: Connector<RpcService>,
 {
     /// Create a new client
-    pub fn new(rpc: RpcClient<crate::rpc::proto::RpcService, C, S>) -> Self {
+    pub fn new(rpc: RpcClient<RpcService, C>) -> Self {
         Self { rpc }
     }
 
@@ -147,7 +144,7 @@ where
     /// A batch is a context in which temp tags are created and data is added to the node. Temp tags
     /// are automatically deleted when the batch is dropped, leading to the data being garbage collected
     /// unless a permanent tag is created for it.
-    pub async fn batch(&self) -> Result<Batch<C, S>> {
+    pub async fn batch(&self) -> Result<Batch<C>> {
         let (updates, mut stream) = self.rpc.bidi(BatchCreateRequest).await?;
         let BatchCreateResponse::Id(batch) = stream.next().await.context("expected scope id")??;
         let rpc = self.rpc.clone();
@@ -457,15 +454,14 @@ where
         Ok(())
     }
 
-    fn tags_client(&self) -> tags::Client<C, S> {
+    fn tags_client(&self) -> tags::Client<C> {
         tags::Client::new(self.rpc.clone())
     }
 }
 
-impl<C, S> SimpleStore for Client<C, S>
+impl<C> SimpleStore for Client<C>
 where
-    S: quic_rpc::Service,
-    C: quic_rpc::ServiceConnection<S>,
+    C: Connector<RpcService>,
 {
     async fn load(&self, hash: Hash) -> anyhow::Result<Bytes> {
         self.read_to_bytes(hash).await
@@ -882,26 +878,24 @@ impl Reader {
     }
 
     /// todo make private again
-    pub async fn from_rpc_read<C, S>(
-        rpc: &RpcClient<crate::rpc::proto::RpcService, C, S>,
+    pub async fn from_rpc_read<C>(
+        rpc: &RpcClient<RpcService, C>,
         hash: Hash,
     ) -> anyhow::Result<Self>
     where
-        C: quic_rpc::ServiceConnection<S>,
-        S: quic_rpc::Service,
+        C: Connector<RpcService>,
     {
         Self::from_rpc_read_at(rpc, hash, 0, ReadAtLen::All).await
     }
 
-    async fn from_rpc_read_at<C, S>(
-        rpc: &RpcClient<crate::rpc::proto::RpcService, C, S>,
+    async fn from_rpc_read_at<C>(
+        rpc: &RpcClient<RpcService, C>,
         hash: Hash,
         offset: u64,
         len: ReadAtLen,
     ) -> anyhow::Result<Self>
     where
-        C: quic_rpc::ServiceConnection<S>,
-        S: quic_rpc::Service,
+        C: Connector<RpcService>,
     {
         let stream = rpc
             .server_streaming(ReadAtRequest { hash, offset, len })
@@ -999,20 +993,17 @@ mod tests {
         use std::{path::Path, sync::Arc};
 
         use iroh_net::{NodeAddr, NodeId};
-        use quic_rpc::client::BoxedServiceConnection;
+        use quic_rpc::transport::{Connector, Listener};
         use tokio_util::task::AbortOnDropHandle;
 
+        use super::RpcService;
         use crate::{
             provider::{CustomEventSender, EventSender},
             rpc::client::{blobs, tags},
             util::local_pool::LocalPool,
         };
 
-        type RpcClient = quic_rpc::RpcClient<
-            crate::rpc::proto::RpcService,
-            BoxedServiceConnection<crate::rpc::proto::RpcService>,
-            crate::rpc::proto::RpcService,
-        >;
+        type RpcClient = quic_rpc::RpcClient<RpcService>;
 
         /// An iroh node that just has the blobs transport
         #[derive(Debug)]
@@ -1129,10 +1120,9 @@ mod tests {
             let router = router.spawn().await?;
 
             // Setup RPC
-            let (internal_rpc, controller) =
-                quic_rpc::transport::flume::service_connection::<crate::rpc::proto::RpcService>(32);
-            let controller = quic_rpc::transport::boxed::Connection::new(controller);
-            let internal_rpc = quic_rpc::transport::boxed::ServerEndpoint::new(internal_rpc);
+            let (internal_rpc, controller) = quic_rpc::transport::flume::channel(32);
+            let controller = controller.boxed();
+            let internal_rpc = internal_rpc.boxed();
             let internal_rpc = quic_rpc::RpcServer::new(internal_rpc);
 
             let rpc_server_task: tokio::task::JoinHandle<()> = tokio::task::spawn(async move {

src/rpc/client/blobs.rs

@@ -9,7 +9,7 @@ use bytes::Bytes;
 use futures_buffered::BufferedStreamExt;
 use futures_lite::StreamExt;
 use futures_util::{sink::Buffer, FutureExt, SinkExt, Stream};
-use quic_rpc::{client::UpdateSink, RpcClient};
+use quic_rpc::{client::UpdateSink, Connector, RpcClient};
 use tokio::io::AsyncRead;
 use tokio_util::io::ReaderStream;
 use tracing::{debug, warn};
@@ -25,6 +25,7 @@ use crate::{
             BatchAddStreamUpdate, BatchCreateTempTagRequest, BatchUpdate,
         },
         tags::{self, SyncMode},
+        RpcService,
     },
     store::ImportMode,
     util::{SetTagOption, TagDrop},
@@ -33,19 +34,17 @@ use crate::{
 
 /// A scope in which blobs can be added.
 #[derive(derive_more::Debug)]
-struct BatchInner<C, S>
+struct BatchInner<C>
 where
-    C: quic_rpc::ServiceConnection<S>,
-    S: quic_rpc::Service,
+    C: Connector<RpcService>,
 {
     /// The id of the scope.
     batch: BatchId,
     /// The rpc client.
-    rpc: RpcClient<crate::rpc::proto::RpcService, C, S>,
+    rpc: RpcClient<RpcService, C>,
     /// The stream to send drop
     #[debug(skip)]
-    updates:
-        Mutex<Buffer<UpdateSink<S, C, BatchUpdate, crate::rpc::proto::RpcService>, BatchUpdate>>,
+    updates: Mutex<Buffer<UpdateSink<C, BatchUpdate>, BatchUpdate>>,
 }
 
 /// A batch for write operations.
@@ -55,15 +54,13 @@ where
 /// It is not a transaction, so things in a batch are not atomic. Also, there is
 /// no isolation between batches.
 #[derive(derive_more::Debug)]
-pub struct Batch<C, S>(Arc<BatchInner<C, S>>)
+pub struct Batch<C>(Arc<BatchInner<C>>)
 where
-    C: quic_rpc::ServiceConnection<S>,
-    S: quic_rpc::Service;
+    C: Connector<RpcService>;
 
-impl<C, S> TagDrop for BatchInner<C, S>
+impl<C> TagDrop for BatchInner<C>
 where
-    C: quic_rpc::ServiceConnection<S>,
-    S: quic_rpc::Service,
+    C: Connector<RpcService>,
 {
     fn on_drop(&self, content: &HashAndFormat) {
         let mut updates = self.updates.lock().unwrap();
@@ -131,15 +128,14 @@ impl Default for AddReaderOpts {
     }
 }
 
-impl<C, S> Batch<C, S>
+impl<C> Batch<C>
 where
-    C: quic_rpc::ServiceConnection<S>,
-    S: quic_rpc::Service,
+    C: Connector<RpcService>,
 {
     pub(super) fn new(
         batch: BatchId,
-        rpc: RpcClient<crate::rpc::proto::RpcService, C, S>,
-        updates: UpdateSink<S, C, BatchUpdate, crate::rpc::proto::RpcService>,
+        rpc: RpcClient<RpcService, C>,
+        updates: UpdateSink<C, BatchUpdate>,
         buffer_size: usize,
     ) -> Self {
         let updates = updates.buffer(buffer_size);

src/rpc/client/blobs/batch.rs

@@ -12,31 +12,30 @@
 //! [`Client::delete`] can be used to delete a tag.
 use anyhow::Result;
 use futures_lite::{Stream, StreamExt};
-use quic_rpc::{client::BoxedServiceConnection, RpcClient};
+use quic_rpc::{client::BoxedConnector, Connector, RpcClient};
 use serde::{Deserialize, Serialize};
 
 use crate::{
-    rpc::proto::tags::{DeleteRequest, ListRequest},
+    rpc::proto::{
+        tags::{DeleteRequest, ListRequest},
+        RpcService,
+    },
     BlobFormat, Hash, Tag,
 };
 
 /// Iroh tags client.
 #[derive(Debug, Clone)]
 #[repr(transparent)]
-pub struct Client<
-    C = BoxedServiceConnection<crate::rpc::proto::RpcService>,
-    S = crate::rpc::proto::RpcService,
-> {
-    pub(super) rpc: RpcClient<crate::rpc::proto::RpcService, C, S>,
+pub struct Client<C = BoxedConnector<RpcService>> {
+    pub(super) rpc: RpcClient<RpcService, C>,
 }
 
-impl<C, S> Client<C, S>
+impl<C> Client<C>
 where
-    C: quic_rpc::ServiceConnection<S>,
-    S: quic_rpc::Service,
+    C: Connector<RpcService>,
 {
     /// Creates a new client
-    pub fn new(rpc: RpcClient<crate::rpc::proto::RpcService, C, S>) -> Self {
+    pub fn new(rpc: RpcClient<RpcService, C>) -> Self {
         Self { rpc }
     }