apollo_propeller: add RegisterHandler command and inbound unit SelectAll

[sirandreww-starkware]

Note

Medium Risk
Changes the engine’s async event loop to poll additional inbound streams, which can affect message ingestion ordering/backpressure and shutdown behavior if not wired correctly.

Overview
Adds a new engine command, EngineCommand::RegisterHandler, that lets the behaviour register a per-connection bounded Receiver<PropellerUnit> with the engine.

The engine now maintains a SelectAll of tagged inbound streams and extends its main tokio::select! loop to consume (PeerId, PropellerUnit) items from these receivers, routing them through the existing handle_unit path alongside the existing HandleHandlerOutput mechanism.

^{Written by Cursor Bugbot for commit 82c4842. This will update automatically on new commits. Configure here.}

[reviewable-StarkWare]

This change is 

[sirandreww-starkware]

• apollo_propeller: route inbound units through bounded channel #13350 : 2 dependent PRs (#13582 , #14533 )
• apollo_propeller: buffer unsent units from partial batches #13349
• apollo_propeller: wire bounded unit channels between handler and engine #13348
• apollo_propeller: add RegisterHandler command and inbound unit SelectAll #13347 👈 (View in Graphite)
• apollo_propeller: add inbound_channel_capacity config field #13346
• main

This stack of pull requests is managed by Graphite. Learn more about stacking.

[guy-starkware]

@guy-starkware reviewed 1 file and all commit messages, and made 1 comment.
Reviewable status: all files reviewed, 1 unresolved discussion (waiting on ShahakShama and sirandreww-starkware).

---

crates/apollo_propeller/src/engine.rs line 32 at r1 (raw file):

/// A stream of `(PeerId, PropellerUnit)` from a single connection handler's bounded channel.
type InboundUnitStream = Pin<Box<dyn futures::Stream<Item = (PeerId, PropellerUnit)> + Send>>;

I thought each connection is to a specific PeerId. Wouldn't that just make each PeerId that comes out this stream the same peer id? Maybe we don't need to send it?

[cursor]

PR Summary

Medium Risk
Touches the engine’s core select loop and inbound unit delivery; behavior is localized but duplicate delivery is possible if both RegisterHandler and HandleHandlerOutput feed units for the same peer.

Overview
Adds a second inbound path for propeller units: connection handlers can hand the engine a bounded futures::channel::mpsc::Receiver<PropellerUnit> via new EngineCommand::RegisterHandler, instead of only delivering units through HandleHandlerOutput.

The engine keeps a SelectAll of per-connection streams tagged with (PeerId, PropellerUnit) and polls inbound_unit_receivers.next() in its main tokio::select! loop, feeding the same handle_unit as before. HandleHandlerOutput / send errors are unchanged in this diff.

^{Reviewed by Cursor Bugbot for commit 1c30253. Bugbot is set up for automated code reviews on this repo. Configure here.}

[github-actions]

There hasn't been any activity on this pull request recently, and in order to prioritize active work, it has been marked as stale.
This PR will be closed and locked in 7 days if no further activity occurs.
Thank you for your contributions!

[ShahakShama]

@ShahakShama reviewed 1 file and all commit messages.
Reviewable status: all files reviewed, 1 unresolved discussion (waiting on sirandreww-starkware).

[ShahakShama]

@ShahakShama made 1 comment.
Reviewable status: all files reviewed, 1 unresolved discussion (waiting on sirandreww-starkware).

ooo

[sirandreww-starkware]

@sirandreww-starkware reviewed 2 files and all commit messages, and made 1 comment.
Reviewable status: all files reviewed, 1 unresolved discussion (waiting on sirandreww-starkware).

[sirandreww-starkware]

@sirandreww-starkware resolved 1 discussion and dismissed @guy-starkware from a discussion.
Reviewable status: complete! all files reviewed, all discussions resolved (waiting on sirandreww-starkware).