refactor: use a separate queue for inbound disco packets from relays

[Frando]

Description

We have reports that the queue of inbound packets from the relay becomes full (i.e. a node is not polling the endpoint fast enough to process all incoming packets from a relay). When this happens, we currently lose not only data packets but also disco packets. This is bad because it can break (and not only stall) connectivity.
This PR adds a separate queue for disco packets coming from relays by testing if a packet is a disco packet or not already in the ActiveRelayActor instead of within the magicsocket. Disco packets are put on a separate queue, and processed in the relay actor.

Breaking Changes

Notes & open questions

Change checklist

• Self-review.
• Documentation updates following the style guide, if relevant.
• Tests if relevant.
• All breaking changes documented.
  • List all breaking changes in the above "Breaking Changes" section.
  • Open an issue or PR on any number0 repos that are affected by this breaking change. Give guidance on how the updates should be handled or do the actual updates themselves. The major ones are:
    • quic-rpc
    • iroh-gossip
    • iroh-blobs
    • dumbpipe
    • sendme

[github-actions]

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

Last updated: 2025-05-16T07:57:13Z

[github-actions]

Netsim report & logs for this PR have been generated and is available at: LOGS
This report will remain available for 3 days.

Last updated for commit: 918dbc4

[flub]

I would be tempted to push this relay_disco_recv all the way into ActiveRelayActor and then it can queue this message and stop reading from the relay channel instead of having to drop it.

[Frando]

This is already in the ActiveRelayActor! So all it would take is making ActiveRelayActor::handle_relay_msg an async fn, and using send(message).await here I think? It is already only called from async fns, and that should then add backpressure I think.

[Frando]

I.e., is this what you meant?

apply backpressure on relay disco recv

diff --git a/iroh/src/magicsock/relay_actor.rs b/iroh/src/magicsock/relay_actor.rs
index 46b6e4d4c..a3479da42 100644
--- a/iroh/src/magicsock/relay_actor.rs
+++ b/iroh/src/magicsock/relay_actor.rs
@@ -619,5 +619,5 @@ impl ActiveRelayActor {
                     match msg {
                         Ok(msg) => {
-                            self.handle_relay_msg(msg, &mut state);
+                            self.handle_relay_msg(msg, &mut state).await;
                             // reset the ping timer, we have just received a message
                             ping_interval.reset();
@@ -642,5 +642,5 @@ impl ActiveRelayActor {
     }
 
-    fn handle_relay_msg(&mut self, msg: ReceivedMessage, state: &mut ConnectedRelayState) {
+    async fn handle_relay_msg(&mut self, msg: ReceivedMessage, state: &mut ConnectedRelayState) {
         match msg {
             ReceivedMessage::ReceivedPacket {
@@ -677,5 +677,5 @@ impl ActiveRelayActor {
                                 relay_remote_node_id: datagram.src,
                             };
-                            if let Err(err) = self.relay_disco_recv.try_send(message) {
+                            if let Err(err) = self.relay_disco_recv.send(message).await {
                                 warn!("Dropping received relay disco packet: {err:#}");
                             }
@@ -775,5 +775,5 @@ impl ActiveRelayActor {
                     };
                     match msg {
-                        Ok(msg) => self.handle_relay_msg(msg, state),
+                        Ok(msg) => self.handle_relay_msg(msg, state).await,
                         Err(err) => break Err(anyhow!("Client stream read error: {err:#}")),
                     }

[Frando]

Pushed that change. Tests seem fine with it!

[flub]

Ah, github was giving me the wrong context and I thought this was the RelayActor...

Conceptually the "queue the message" is the same as my "we can't do async stuf" comment above. If the self.relay_disco_recv.try_send fails somehow the message needs to be stored on the actor and then the main loop should stop polling for further messages from the relay until it was finally able to send this. The RelayActor and ActiveRelayActor already go to some length to do this and it has worked very well. Exactly where the state should be stored and how to keep yet another case of this readable and maintainable is another question. E.g. in the RelayActor stores a datagram_send_fut "sending future" (the MaybeFuture) and if that's set then don't poll something. The ActiveRelayActor has various versions of the run loops. That may not work so well if it also has to do this.

[flub]

We can not make this async. Actors can not do async stuff in their main loop. We violate this in plenty of places, but this actor has finally been cleaned up. We should not regress this.

[Frando]

Right. But how would you apply backpressure then?

[Frando]

Removed that commit again.

[flub]

Right. But how would you apply backpressure then?

For completeness answering that here as well (it is mentioned above): by keeping the blocked message around in the actor state somehow and not polling for new messages from the relay while you have this pending message.

Of course in this case it means you're backpressuring the entire relay TCP stream while you only really want to backpressure disco messages. But to really do something about this you'd have to make two tcp streams to the relay and... lots of things we don't want to do.

[flub]

nit: "actor closed" is sufficient since you already have an info span with the actor name. But I don't mind if you do this either.

[Frando]

@flub I pushed a commit that should apply backpressure onto the TCP stream to the relay, without ever awaiting in the actor loop. The code is a bit complicated, but I think it is correct. I'd be glad for a review and if this is roughly what you had in mind as well. I can then try to make this code a bit simpler maybe.

[flub]

Shouldn't this live on ConnectedRelayState?

[Frando]

I moved this into a new struct together with the queue senders in the latest commit, but still kept it on the ActiveRelayActor. My reasoning: If the connection to a relay breaks, but we still have pending received items queued, we can still try pushing those to the magicsock.

[flub]

We use .expect() for cases like this, I thought .unwrap() was denied by some linter even?

[flub]

Oh yeah, that's complex. Though I agree that it looks about right. I also suspect your suggestion that this could be simplified a bit might be right.

I'm still on the fence whether this change is worth it though. If we do this, the dropping becomes less of an issue in the receiving magicsock. We'd then be dropping these in the relay server, which hides the problem entirely from the users.

To make this now work better we need to make the relay server behave more like a router. It can't backpressure this up to the sender, because the sender can be sending to multiple destinations from it's single TCP stream. What the relay server can do is start marking ECN codepoints on packets when queues it travels through start filling up. Quinn will then notice those and start to back off before there is full congestion. Given that DISCO is really relatively few packets, the fact they're not congestion controlled is then probably not an issue. So only if we start doing this will we start to get any benefit from this.

---

Point is, this is a bigger project. And if we do it we need to write this up in a tracking issue and make an informed decision that we'll do this and get to the finish line. Technically I'd be great to have the relay server mark ECN points though. I'd love it and think it'll make a real impact.

[Frando]

I pushed two commits:

• A fix to the changes to the RelayDatagramRecvQueue: We need to store a list of send wakers, because this can be sent to from multiple ActiveRelayActors
• Some cleanups to the new code added to the active relay actor to make this a bit nicer