refactor(iroh): Modularize protocol (#2454)

## Description

Use a newly created macro crate
https://github.com/n0-computer/nested-enum-utils to allow for from/to
conversions for deeply nested enums.

This allows us to split the protocol enum into multiple parts without
changing anything about the basic concept of how things are handled.

## Breaking Changes

Weirdly, I think there should be none since the protocol is private.

## Notes & open questions

Obvious next steps would be
- [x] define the subsystem specific messages in submodules
- [x] publish the macro crate
- [x] (maybe) split up the handling of the different subsystems
- [x] (maybe) remove the subsystem prefixes from the messages

Not so obvious next steps (maybe next PR)
- Allow moving the subsystem protocols to different crates. This is
difficult because you can not set the target when generating the
conversions.

## Change checklist

- [x] Self-review.
- [x] Documentation updates if relevant.
- [x] Tests if relevant.
- [x] All breaking changes documented.

Author: rklaehn

Cargo.lock

@@ -31,6 +31,7 @@ iroh-base = { version = "0.19.0", path = "../iroh-base", features = ["key"] }
 iroh-io = { version = "0.6.0", features = ["stats"] }
 iroh-metrics = { version = "0.19.0", path = "../iroh-metrics", optional = true }
 iroh-net = { version = "0.19.0", path = "../iroh-net" }
+nested_enum_utils = "0.1.0"
 num_cpus = { version = "1.15.0" }
 portable-atomic = "1"
 iroh-docs = { version = "0.19.0", path = "../iroh-docs" }

iroh/Cargo.toml

@@ -5,9 +5,9 @@ use futures_lite::{stream::StreamExt, Stream};
 use iroh_docs::{Author, AuthorId};
 use ref_cast::RefCast;
 
-use crate::rpc_protocol::{
-    AuthorCreateRequest, AuthorDeleteRequest, AuthorExportRequest, AuthorGetDefaultRequest,
-    AuthorImportRequest, AuthorListRequest, AuthorSetDefaultRequest,
+use crate::rpc_protocol::authors::{
+    CreateRequest, DeleteRequest, ExportRequest, GetDefaultRequest, ImportRequest, ListRequest,
+    SetDefaultRequest,
 };
 
 use super::{flatten, RpcClient};
@@ -27,7 +27,7 @@ impl Client {
     ///
     /// If you need only a single author, use [`Self::default`].
     pub async fn create(&self) -> Result<AuthorId> {
-        let res = self.rpc.rpc(AuthorCreateRequest).await??;
+        let res = self.rpc.rpc(CreateRequest).await??;
         Ok(res.author_id)
     }
 
@@ -38,7 +38,7 @@ impl Client {
     ///
     /// The default author can be set with [`Self::set_default`].
     pub async fn default(&self) -> Result<AuthorId> {
-        let res = self.rpc.rpc(AuthorGetDefaultRequest).await??;
+        let res = self.rpc.rpc(GetDefaultRequest).await??;
         Ok(res.author_id)
     }
 
@@ -49,31 +49,29 @@ impl Client {
     /// On a persistent node, the author id will be saved to a file in the data directory and
     /// reloaded after a restart.
     pub async fn set_default(&self, author_id: AuthorId) -> Result<()> {
-        self.rpc
-            .rpc(AuthorSetDefaultRequest { author_id })
-            .await??;
+        self.rpc.rpc(SetDefaultRequest { author_id }).await??;
         Ok(())
     }
 
     /// List document authors for which we have a secret key.
     pub async fn list(&self) -> Result<impl Stream<Item = Result<AuthorId>>> {
-        let stream = self.rpc.server_streaming(AuthorListRequest {}).await?;
+        let stream = self.rpc.server_streaming(ListRequest {}).await?;
         Ok(flatten(stream).map(|res| res.map(|res| res.author_id)))
     }
 
     /// Export the given author.
     ///
     /// Warning: This contains sensitive data.
     pub async fn export(&self, author: AuthorId) -> Result<Option<Author>> {
-        let res = self.rpc.rpc(AuthorExportRequest { author }).await??;
+        let res = self.rpc.rpc(ExportRequest { author }).await??;
         Ok(res.author)
     }
 
     /// Import the given author.
     ///
     /// Warning: This contains sensitive data.
     pub async fn import(&self, author: Author) -> Result<()> {
-        self.rpc.rpc(AuthorImportRequest { author }).await??;
+        self.rpc.rpc(ImportRequest { author }).await??;
         Ok(())
     }
 
@@ -83,7 +81,7 @@ impl Client {
     ///
     /// Returns an error if attempting to delete the default author.
     pub async fn delete(&self, author: AuthorId) -> Result<()> {
-        self.rpc.rpc(AuthorDeleteRequest { author }).await??;
+        self.rpc.rpc(DeleteRequest { author }).await??;
         Ok(())
     }
 }

iroh/src/client/authors.rs

@@ -20,6 +20,7 @@ use iroh_blobs::{
     format::collection::{Collection, SimpleStore},
     get::db::DownloadProgress as BytesDownloadProgress,
     store::{ConsistencyCheckProgress, ExportFormat, ExportMode, ValidateProgress},
+    util::SetTagOption,
     BlobFormat, Hash, Tag,
 };
 use iroh_net::NodeAddr;
@@ -31,12 +32,13 @@ use tokio::io::{AsyncRead, AsyncReadExt, ReadBuf};
 use tokio_util::io::{ReaderStream, StreamReader};
 use tracing::warn;
 
-use crate::rpc_protocol::{
-    BlobAddPathRequest, BlobAddStreamRequest, BlobAddStreamUpdate, BlobConsistencyCheckRequest,
-    BlobDeleteBlobRequest, BlobDownloadRequest, BlobExportRequest, BlobListIncompleteRequest,
-    BlobListRequest, BlobReadAtRequest, BlobReadAtResponse, BlobValidateRequest,
-    CreateCollectionRequest, CreateCollectionResponse, NodeStatusRequest, SetTagOption,
+use crate::rpc_protocol::blobs::{
+    AddPathRequest, AddStreamRequest, AddStreamUpdate, ConsistencyCheckRequest,
+    CreateCollectionRequest, CreateCollectionResponse, DeleteRequest, DownloadRequest,
+    ExportRequest, ListIncompleteRequest, ListRequest, ReadAtRequest, ReadAtResponse,
+    ValidateRequest,
 };
+use crate::rpc_protocol::node::StatusRequest;
 
 use super::{flatten, tags, Iroh, RpcClient};
 
@@ -110,7 +112,7 @@ impl Client {
     ) -> Result<AddProgress> {
         let stream = self
             .rpc
-            .server_streaming(BlobAddPathRequest {
+            .server_streaming(AddPathRequest {
                 path,
                 in_place,
                 tag,
@@ -158,12 +160,12 @@ impl Client {
         input: impl Stream<Item = io::Result<Bytes>> + Send + Unpin + 'static,
         tag: SetTagOption,
     ) -> anyhow::Result<AddProgress> {
-        let (mut sink, progress) = self.rpc.bidi(BlobAddStreamRequest { tag }).await?;
+        let (mut sink, progress) = self.rpc.bidi(AddStreamRequest { tag }).await?;
         let mut input = input.map(|chunk| match chunk {
-            Ok(chunk) => Ok(BlobAddStreamUpdate::Chunk(chunk)),
+            Ok(chunk) => Ok(AddStreamUpdate::Chunk(chunk)),
             Err(err) => {
                 warn!("Abort send, reason: failed to read from source stream: {err:?}");
-                Ok(BlobAddStreamUpdate::Abort)
+                Ok(AddStreamUpdate::Abort)
             }
         });
         tokio::spawn(async move {
@@ -205,7 +207,7 @@ impl Client {
     ) -> Result<impl Stream<Item = Result<ValidateProgress>>> {
         let stream = self
             .rpc
-            .server_streaming(BlobValidateRequest { repair })
+            .server_streaming(ValidateRequest { repair })
             .await?;
         Ok(stream.map(|res| res.map_err(anyhow::Error::from)))
     }
@@ -219,7 +221,7 @@ impl Client {
     ) -> Result<impl Stream<Item = Result<ConsistencyCheckProgress>>> {
         let stream = self
             .rpc
-            .server_streaming(BlobConsistencyCheckRequest { repair })
+            .server_streaming(ConsistencyCheckRequest { repair })
             .await?;
         Ok(stream.map(|r| r.map_err(anyhow::Error::from)))
     }
@@ -266,7 +268,7 @@ impl Client {
         } = opts;
         let stream = self
             .rpc
-            .server_streaming(BlobDownloadRequest {
+            .server_streaming(DownloadRequest {
                 hash,
                 format,
                 nodes,
@@ -295,7 +297,7 @@ impl Client {
         format: ExportFormat,
         mode: ExportMode,
     ) -> Result<ExportProgress> {
-        let req = BlobExportRequest {
+        let req = ExportRequest {
             hash,
             path: destination,
             format,
@@ -309,13 +311,13 @@ impl Client {
 
     /// List all complete blobs.
     pub async fn list(&self) -> Result<impl Stream<Item = Result<BlobInfo>>> {
-        let stream = self.rpc.server_streaming(BlobListRequest).await?;
+        let stream = self.rpc.server_streaming(ListRequest).await?;
         Ok(flatten(stream))
     }
 
     /// List all incomplete (partial) blobs.
     pub async fn list_incomplete(&self) -> Result<impl Stream<Item = Result<IncompleteBlobInfo>>> {
-        let stream = self.rpc.server_streaming(BlobListIncompleteRequest).await?;
+        let stream = self.rpc.server_streaming(ListIncompleteRequest).await?;
         Ok(flatten(stream))
     }
 
@@ -354,7 +356,7 @@ impl Client {
 
     /// Delete a blob.
     pub async fn delete_blob(&self, hash: Hash) -> Result<()> {
-        self.rpc.rpc(BlobDeleteBlobRequest { hash }).await??;
+        self.rpc.rpc(DeleteRequest { hash }).await??;
         Ok(())
     }
 
@@ -365,7 +367,7 @@ impl Client {
         blob_format: BlobFormat,
         addr_options: AddrInfoOptions,
     ) -> Result<BlobTicket> {
-        let mut addr = self.rpc.rpc(NodeStatusRequest).await??.addr;
+        let mut addr = self.rpc.rpc(StatusRequest).await??.addr;
         addr.apply_options(addr_options);
         let ticket = BlobTicket::new(addr, hash, blob_format).expect("correct ticket");
 
@@ -791,19 +793,19 @@ impl Reader {
         len: Option<usize>,
     ) -> anyhow::Result<Self> {
         let stream = rpc
-            .server_streaming(BlobReadAtRequest { hash, offset, len })
+            .server_streaming(ReadAtRequest { hash, offset, len })
             .await?;
         let mut stream = flatten(stream);
 
         let (size, is_complete) = match stream.next().await {
-            Some(Ok(BlobReadAtResponse::Entry { size, is_complete })) => (size, is_complete),
+            Some(Ok(ReadAtResponse::Entry { size, is_complete })) => (size, is_complete),
             Some(Err(err)) => return Err(err),
             Some(Ok(_)) => return Err(anyhow!("Expected header frame, but got data frame")),
             None => return Err(anyhow!("Expected header frame, but RPC stream was dropped")),
         };
 
         let stream = stream.map(|item| match item {
-            Ok(BlobReadAtResponse::Data { chunk }) => Ok(chunk),
+            Ok(ReadAtResponse::Data { chunk }) => Ok(chunk),
             Ok(_) => Err(io::Error::new(io::ErrorKind::Other, "Expected data frame")),
             Err(err) => Err(io::Error::new(io::ErrorKind::Other, format!("{err}"))),
         });