feat: abstract over stream types on provide and get side #147

rklaehn · 2025-09-11T11:27:36Z

Description

This PR introduces abstract versions of iroh::endpoint::{SendStream, RecvStream} and modifies the provide side and get side implementation to use these abstract streams instead of directly using iroh::endpoint streams.

This is necessary for wrapping the streams into a transformation such as compression, see the discussion in n0-computer/sendme#93 .

The compression example shows how streams can be wrapped into compression/decompression to create a derived protocol with a different ALPN that is identical to the blobs protocol except for compression.

Breaking Changes

iroh::endpoint::SendStream and iroh::endpoint::RecvStream are replaced with the traits iroh_blobs::util::SendStream and iroh_blobs::util::RecvStream in the get FSM and in the provider side API.

Notes & open questions

Change checklist

Self-review.
Documentation updates following the style guide, if relevant.
Tests if relevant.
All breaking changes documented.

… notifications/requests for each event type.

This shows how to limit serving content in various ways - by node id - by content hash - throttling - limiting max number of connections

so the other side can know if reconnecting is OK

Also make the whole get fsm generic so it can be used with an arbitrary stream, not just a quinn/iroh RecvStream.

github-actions · 2025-09-11T11:29:34Z

Documentation for this PR has been generated and is available at: https://n0-computer.github.io/iroh-blobs/pr/147/docs/iroh_blobs/

Last updated: 2025-09-17T09:42:39Z

We only need async-compression as a dev dep for the compression example!

examples/compression.rs

dignifiedquire · 2025-09-17T08:21:06Z

src/util/stream.rs

+    /// Send bytes to the stream. This takes a `Bytes` because iroh can directly use them.
+    fn send_bytes(&mut self, bytes: Bytes) -> impl Future<Output = io::Result<()>> + Send;
+    /// Send that sends a fixed sized buffer.
+    fn send<const L: usize>(


why not make it implement tokio::io::Write instead?

I tried doing a

impl<T: SendStream> tokio::io::AsyncWrite for T

But then you get a collision when you do the reference stuff

impl<T: SendStream> SendStream for &mut T

So you can either make this trait convenient to use on its own, or make it convenient to adapt to the tokio world, but not both.

dignifiedquire · 2025-09-17T08:21:27Z

src/util/stream.rs

+/// An abstract `iroh::endpoint::RecvStream`.
+pub trait RecvStream: Send {
+    /// Receive up to `len` bytes from the stream, directly into a `Bytes`.
+    fn recv_bytes(&mut self, len: usize) -> impl Future<Output = io::Result<Bytes>> + Send;


why not make it implement tokio::io::Read instead? and provide this as a helper method?

Same as above, but most notably I want the ability to return a Bytes without copying, and if you add a helper method you have to copy.

iroh::endpoint::RecvStream has fns that produce a Bytes, blobs needs Bytes, but if you go through tokio::io::AsyncRead you work with &mut [u8] and lose the Bytes / must copy. Unless there is a trick I am not aware of.

https://docs.rs/tokio/latest/tokio/io/trait.AsyncRead.html uses ReadBuf which I believe can actually avoid this issue?

but it's tricky and involves unsafe in some cases I think

Yeah, I would rather have Bytes -> Bytes instead of Bytes -> complex rube goldberg machine involving ReadBuf and unsafe -> Bytes...

src/util/stream.rs

Frando · 2025-09-17T08:15:21Z

src/util/stream.rs

+    /// Send bytes to the stream. This takes a `Bytes` because iroh can directly use them.
+    fn send_bytes(&mut self, bytes: Bytes) -> impl Future<Output = io::Result<()>> + Send;
+    /// Send that sends a fixed sized buffer.
+    fn send<const L: usize>(


Why the const L instead of taking &[u8]?

Since these traits are hopeless, they will never be dynable, I thought I might as well use a const generic since it is convenient. But sure, could change this.

In case of the recv version the code just reads nicer if you return a fixed size slice instead of taking a &mut [u8], and for returning a fixed size slice you don't have to allocate.

src/util/stream.rs

Frando · 2025-09-17T08:26:14Z

src/util/stream.rs

+    /// Note that this is different from `recv_bytes`, which will return fewer bytes if the stream ends.
+    fn recv_bytes_exact(&mut self, len: usize) -> impl Future<Output = io::Result<Bytes>> + Send;
+    /// Receive exactly `L` bytes from the stream, directly into a `[u8; L]`.
+    fn recv<const L: usize>(&mut self) -> impl Future<Output = io::Result<[u8; L]>> + Send;


Maybe call this recv_exact?

src/util/stream.rs

Frando · 2025-09-17T08:32:54Z

src/util/stream.rs

+        Ok((value, data.len()))
+    }
+
+    async fn read_length_prefixed<T: DeserializeOwned>(


Yeah these are great. We should move all this with some cleanup and better docs to an iroh-util crate I think. It will be useful for many people working with iroh send/recv streams even independently of the trait abstraction.

I am somewhat hesitant to have yet another crate, but I guess maybe we have to do it even before iroh 1.0?

hopefully this makes it more clear what they are for!

dignifiedquire · 2025-09-17T10:45:49Z

examples/compression.rs

@@ -72,7 +73,7 @@ mod lz4 {
        }
    }

-    impl SendStreamSpecific for SendStream {
+    impl AsyncWriteSendStreamExtra for SendStream {


maybe AsyncWriteSendStreamExt ? for Extension?

Hm, ...Ext traits for me are specifically for the "extension method" hack.

trait FooExt { ... }

impl<T: Foo> FooExt for T {}

rklaehn and others added 30 commits August 28, 2025 14:54

WIP make provider events a proper irpc protocol and allow configuring…

e5ac6dd

… notifications/requests for each event type.

Add transfer_completed and transfer_aborted fn.

d17c6f6

Nicer proto

b23995e

Update tests

a78c212

tests pass

df1e1ef

Everything works

a9ac8e5

minimize diff and required changes

1e4a581

clippy

6449930

Footgun protection

b26aefb

Add limit example

6d86e4f

This shows how to limit serving content in various ways - by node id - by content hash - throttling - limiting max number of connections

Add len to notify_payload_write

4b87b6d

clippy

f992a44

nicer connection counter

4bddf77

Add docs for the limit example.

33333a9

refactor: make limits example more DRY

9a62a58

Make sure to send a proper reset code when resetting a connection

071db5e

so the other side can know if reconnecting is OK

deny

2d72de0

Use async syntax for implementing ProtocolHandler

2dac46c

Use irpc::channel::SendError as default sink error.

a67d787

fixup

546f57e

Remove map_err that isn't needed anymore

f399e2b

Refactor the GetError to be just a list of things that can go wrong.

3f0a661

Also make the whole get fsm generic so it can be used with an arbitrary stream, not just a quinn/iroh RecvStream.

silence some of the tests

2f9ebd5

Genericize provider side a bit

d764dc0

Refactor error and make get and provide side generic

4e8387a

Add example how to add compression to the entire blobs protocol.

4c4a5e7

Working adapters

f3d02e7

compression example works again

41284d2

Generic receive into store

349c36b

More moving stuff around

bc159ca

clippy

1390954

n0bot bot added this to iroh Sep 11, 2025

github-project-automation bot moved this to 🏗 In progress in iroh Sep 11, 2025

Base automatically changed from provider-events-refactor to main September 11, 2025 15:01

Merge branch 'main' into newtype-it

4ddc137

rklaehn mentioned this pull request Sep 16, 2025

feat: Generic fsm and provide side #145

Closed

4 tasks

rklaehn changed the title ~~Newtype it~~ feat: abstract over stream types on provide and get side Sep 16, 2025

rklaehn marked this pull request as ready for review September 16, 2025 11:39

rklaehn mentioned this pull request Sep 17, 2025

Add zstd file compression support n0-computer/sendme#93

Closed

Remove async-compression dep on compile

3dc6d97

We only need async-compression as a dev dep for the compression example!

dignifiedquire reviewed Sep 17, 2025

View reviewed changes

examples/compression.rs Outdated Show resolved Hide resolved

dignifiedquire reviewed Sep 17, 2025

View reviewed changes

src/util/stream.rs Outdated Show resolved Hide resolved

Frando reviewed Sep 17, 2025

View reviewed changes

rklaehn mentioned this pull request Sep 17, 2025

feat: compress entire blobs protocol using zstd or lz4 n0-computer/sendme#106

Open

rklaehn added 2 commits September 17, 2025 12:28

PR review: added cancellation safety note and id()

6ceacfd

PR review: rename the weirdly named ...Specific traits

845e01e

hopefully this makes it more clear what they are for!

dignifiedquire reviewed Sep 17, 2025

View reviewed changes

rklaehn requested review from Frando and dignifiedquire September 17, 2025 13:22

feat: abstract over stream types on provide and get side #147

Are you sure you want to change the base?

feat: abstract over stream types on provide and get side #147

Conversation

rklaehn commented Sep 11, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Description

Breaking Changes

Notes & open questions

Change checklist

Uh oh!

github-actions bot commented Sep 11, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

rklaehn Sep 17, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

rklaehn Sep 17, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

rklaehn commented Sep 11, 2025 •

edited

Loading

github-actions bot commented Sep 11, 2025 •

edited

Loading

rklaehn Sep 17, 2025 •

edited

Loading

rklaehn Sep 17, 2025 •

edited

Loading