Add tests that use the rpc client with quinn.

Some errors don't happen if you use the mem transport.
Also fix bug where add_bytes fails if the bytes are larger than the max
frame size.

Author: rklaehn

Cargo.lock

@@ -184,4 +184,4 @@ incremental = false
 [patch.crates-io]
 iroh-base = { git = "https://github.com/n0-computer/iroh" }
 iroh = { git = "https://github.com/n0-computer/iroh" }
-quic-rpc = { git = "https://github.com/n0-computer/quic-rpc" }
+quic-rpc = { path = "../quic-rpc" }

Cargo.toml

@@ -268,10 +268,10 @@ where
             }
         });
         tokio::spawn(async move {
-            // TODO: Is it important to catch this error? It should also result in an error on the
-            // response stream. If we deem it important, we could one-shot send it into the
-            // BlobAddProgress and return from there. Not sure.
             if let Err(err) = sink.send_all(&mut input).await {
+                // if we get an error in send_all due to the connection being closed, this will just fail again.
+                // if we get an error due to something else (serialization or size limit), tell the remote to abort.
+                sink.send(AddStreamUpdate::Abort).await.ok();
                 warn!("Failed to send input stream to remote: {err:?}");
             }
         });
@@ -281,7 +281,7 @@ where
 
     /// Write a blob by passing bytes.
     pub async fn add_bytes(&self, bytes: impl Into<Bytes>) -> anyhow::Result<AddOutcome> {
-        let input = futures_lite::stream::once(Ok(bytes.into()));
+        let input = chunked_bytes_stream(bytes.into(), 1024 * 64).map(Ok);
         self.add_stream(input, SetTagOption::Auto).await?.await
     }
 
@@ -291,7 +291,7 @@ where
         bytes: impl Into<Bytes>,
         name: impl Into<Tag>,
     ) -> anyhow::Result<AddOutcome> {
-        let input = futures_lite::stream::once(Ok(bytes.into()));
+        let input = chunked_bytes_stream(bytes.into(), 1024 * 64).map(Ok);
         self.add_stream(input, SetTagOption::Named(name.into()))
             .await?
             .await
@@ -987,6 +987,12 @@ pub struct DownloadOptions {
     pub mode: DownloadMode,
 }
 
+fn chunked_bytes_stream(mut b: Bytes, c: usize) -> impl Stream<Item = Bytes> {
+    futures_lite::stream::iter(std::iter::from_fn(move || {
+        Some(b.split_to(b.len().min(c))).filter(|x| !x.is_empty())
+    }))
+}
+
 #[cfg(test)]
 mod tests {
     use std::{path::Path, time::Duration};

src/rpc/client/blobs.rs

@@ -131,10 +131,23 @@ async fn node_and_client() -> TestResult<(Node, BlobsClient, TempDir)> {
 #[tokio::test]
 async fn quinn_rpc_smoke() -> TestResult<()> {
     let _ = tracing_subscriber::fmt::try_init();
-    let (node, client, _testdir) = node_and_client().await?;
-    println!("Made a client");
-    let hash = client.add_bytes(b"hello".to_vec()).await?;
-    println!("Hash: {:?}", hash);
-    drop(node);
+    let (_node, client, _testdir) = node_and_client().await?;
+    let data = b"hello";
+    let hash = client.add_bytes(data.to_vec()).await?.hash;
+    assert_eq!(hash, iroh_blobs::Hash::new(data));
+    let data2 = client.read_to_bytes(hash).await?;
+    assert_eq!(data, &data2[..]);
+    Ok(())
+}
+
+#[tokio::test]
+async fn quinn_rpc_large() -> TestResult<()> {
+    let _ = tracing_subscriber::fmt::try_init();
+    let (_node, client, _testdir) = node_and_client().await?;
+    let data = vec![0; 1024 * 1024 * 16];
+    let hash = client.add_bytes(data.clone()).await?.hash;
+    assert_eq!(hash, iroh_blobs::Hash::new(&data));
+    let data2 = client.read_to_bytes(hash).await?;
+    assert_eq!(data, &data2[..]);
     Ok(())
 }