apollo_propeller: add unit validation receiving and spawning

[sirandreww-starkware]

Note

Medium Risk
Introduces new concurrency and thread-pool usage in the message processing/validation path; incorrect handling of the pending validation result or validator handoff could lead to stalled processing or subtle race/logic bugs.

Overview
apollo_propeller now depends on rayon and uses it in MessageProcessor::run to offload CPU-bound shard validation (signature/merkle proof checks) from the Tokio runtime.

MessageProcessor instantiates a per-message UnitValidator, receives (PeerId, PropellerUnit) from unit_rx, and spawns validation work onto the Rayon thread pool, tracking the in-flight job via a oneshot channel (pending_validation) to prevent concurrent validations.

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

[reviewable-StarkWare]

This change is 

[github-actions]

Artifacts upload workflows:

• Started at 2026-01-26T14:10:10Z
• Started at 2026-02-04T11:54:30Z
• Started at 2026-02-04T11:54:29Z
• Started at 2026-02-08T15:07:11Z
• Started at 2026-02-09T14:28:07Z
• Started at 2026-02-19T10:16:48Z
• Started at 2026-02-19T15:36:46Z
• Started at 2026-02-19T15:36:47Z
• Started at 2026-02-19T16:02:15Z
• Started at 2026-02-22T07:37:07Z
• Started at 2026-02-22T09:15:16Z

[sirandreww-starkware]

• apollo_propeller: add channel closure handling #12011
• apollo_propeller: add PostReconstruction phase and threshold checking #12010
• apollo_propeller: add reconstruction result handling #12009
• apollo_propeller: add PreReconstruction phase and shard collection #12008
• apollo_propeller: add reconstruction types and spawn task #12007
• apollo_propeller: add validated unit handling and shard broadcasting #12006
• apollo_propeller: add validation result handling #12005
• apollo_propeller: add unit validation receiving and spawning #12004 👈 (View in Graphite)
• apollo_propeller: add MessageProcessor scaffolding #12003 : 1 other dependent PR (#12780 )
• apollo_propeller: added shard index utility functions to schedule manager #12002
• apollo_propeller: add sharding module with tests #11944 : 1 other dependent PR (#12001 )
• apollo_propeller: add message padding utilities #11943
• apollo_propeller: add behaviour public API stubs #11942
• apollo_propeller: add minimal config #11196 : 1 other dependent PR (#11197 )
• apollo_propeller: add behaviour skeleton #11195
• apollo_propeller: add Event enum to types #11194
• apollo_propeller: add dial upgrade error handling to handler #11074
• apollo_propeller: add rate-limited queue length warnings #11073
• apollo_propeller: add concurrent streams support to handler #11072
• apollo_propeller: implement outbound substream with message batching #11071
• apollo_propeller: add handler helper methods #11070
• main-v0.14.2

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

[cursor]

Cursor Bugbot has reviewed your changes and found 1 potential issue.

[ShahakShama]

@ShahakShama reviewed 3 files and all commit messages, and made 6 comments.
Reviewable status: all files reviewed, 7 unresolved discussions (waiting on @dan, @noamsp-starkware, and @sirandreww-starkware).

---

Cargo.toml line 323 at r3 (raw file):

rand_chacha = "0.3.1"
rand_distr = "0.4.3"
rayon = "1.11.0"

@dan-starkware do you approve?

---

crates/apollo_propeller/src/message_processor.rs line 78 at r3 (raw file):

        loop {
            tokio::select! {
                _ = sleep_until(deadline) => {

Use tokio timeout instead of select and sleep

---

crates/apollo_propeller/src/message_processor.rs line 87 at r3 (raw file):

                    let (result_tx, result_rx) = oneshot::channel();
                    let mut validator_moved = validator.take().unwrap();

Why is this unwrap safe? (I think it's not)

---

crates/apollo_propeller/src/message_processor.rs line 89 at r3 (raw file):

                    let mut validator_moved = validator.take().unwrap();

                    rayon::spawn(move || {

Why do you use rayon and not tokio? Explain both to me and in a comment

---

crates/apollo_propeller/src/message_processor.rs line 89 at r3 (raw file):

                    let mut validator_moved = validator.take().unwrap();

                    rayon::spawn(move || {

Shouldn't you store the task handle and poll it

---

crates/apollo_propeller/src/message_processor.rs line 94 at r3 (raw file):

                    });

                    pending_validation = Some(result_rx);

What do you do with pending_validation?

[guy-starkware]

@guy-starkware reviewed all commit messages.
Reviewable status: all files reviewed, 6 unresolved discussions (waiting on dan, noamsp-starkware, and sirandreww-starkware).

[sirandreww-starkware]

@sirandreww-starkware made 6 comments.
Reviewable status: 1 of 4 files reviewed, 6 unresolved discussions (waiting on dan, guy-starkware, noamsp-starkware, and ShahakShama).

---

Cargo.toml line 323 at r3 (raw file):

Previously, ShahakShama wrote…

@dan-starkware do you approve?

sent him a slack message, waiting for his approval

---

crates/apollo_propeller/src/message_processor.rs line 78 at r3 (raw file):

Previously, ShahakShama wrote…

Use tokio timeout instead of select and sleep

addressed this in a previous PR

---

crates/apollo_propeller/src/message_processor.rs line 87 at r3 (raw file):

Previously, ShahakShama wrote…

Why is this unwrap safe? (I think it's not)

added a comment as to why

---

crates/apollo_propeller/src/message_processor.rs line 89 at r3 (raw file):

Previously, ShahakShama wrote…

Why do you use rayon and not tokio? Explain both to me and in a comment

added a comment, let's talk f2f about this

---

crates/apollo_propeller/src/message_processor.rs line 89 at r3 (raw file):

Previously, ShahakShama wrote…

Shouldn't you store the task handle and poll it

Both options here tokio::spawn_blocking and rayon::spawn give the future to a runtime and it starts running, we use the oneshot to get a response and the task is short lived. So we do not use the handler at all.

---

crates/apollo_propeller/src/message_processor.rs line 94 at r3 (raw file):

Previously, ShahakShama wrote…

What do you do with pending_validation?

In the next PR I read from it in this same select and I handle a validated unit.

[ShahakShama]

@ShahakShama reviewed 3 files and all commit messages, made 5 comments, and resolved 2 discussions.
Reviewable status: all files reviewed, 7 unresolved discussions (waiting on dan, noamsp-starkware, and sirandreww-starkware).

---

crates/apollo_propeller/src/message_processor.rs line 87 at r3 (raw file):

Previously, sirandreww-starkware (Andrew Luka) wrote…

added a comment as to why

I don't like this. Let's put a TODO to see if we can avoid this expect

---

crates/apollo_propeller/src/message_processor.rs line 89 at r3 (raw file):

Previously, sirandreww-starkware (Andrew Luka) wrote…

Both options here tokio::spawn_blocking and rayon::spawn give the future to a runtime and it starts running, we use the oneshot to get a response and the task is short lived. So we do not use the handler at all.

How will you know if the handler panicked?
And in general we have a convention not to leave detached tasks

---

crates/apollo_propeller/src/message_processor.rs line 61 at r5 (raw file):

            Arc::clone(&self.tree_manager),
        ));
        let mut pending_validation: Option<oneshot::Receiver<ValidationResult>> = None;

Consider storing all pending validations in a FuturesUnordered instead

---

crates/apollo_propeller/src/message_processor.rs line 70 at r5 (raw file):

                }

                Some((sender, unit)) = self.unit_rx.recv(), if pending_validation.is_none() => {

What if pending validation is some? I don't know this syntax but it looks like you're taking an item from the stream and discarding it

---

crates/apollo_propeller/src/message_processor.rs line 93 at r5 (raw file):

                    });

                    pending_validation = Some(result_rx);

Why not just handle the validation here?

[ShahakShama]

@ShahakShama made 4 comments and resolved 2 discussions.
Reviewable status: all files reviewed, 5 unresolved discussions (waiting on dan, noamsp-starkware, and sirandreww-starkware).

---

crates/apollo_propeller/src/message_processor.rs line 87 at r3 (raw file):

Previously, ShahakShama wrote…

I don't like this. Let's put a TODO to see if we can avoid this expect

Unite both options into one enum

---

crates/apollo_propeller/src/message_processor.rs line 89 at r3 (raw file):

Previously, sirandreww-starkware (Andrew Luka) wrote…

added a comment, let's talk f2f about this

Add a TODO to investigate why rayon is better and if that is still true after improvements

---

crates/apollo_propeller/src/message_processor.rs line 89 at r3 (raw file):

Previously, ShahakShama wrote…

How will you know if the handler panicked?
And in general we have a convention not to leave detached tasks

Add a comment that it's ok to drop the handle because we know the status through the oneshot channel

---

crates/apollo_propeller/src/message_processor.rs line 70 at r5 (raw file):

Previously, ShahakShama wrote…

What if pending validation is some? I don't know this syntax but it looks like you're taking an item from the stream and discarding it

Add a comment what this does