apollo_propeller: add PreReconstruction phase and shard collection

[sirandreww-starkware]

Note

Medium Risk
Introduces new state and concurrency around shard collection and reconstruction triggering, which could cause subtle lifecycle/ordering bugs even though the reconstruction result is not yet consumed.

Overview
Adds an explicit ReconstructionPhase::PreReconstruction state to MessageProcessor to collect validated shard units, track whether the node’s own shard index has been seen, and capture the message signature.

Once tree_manager.should_build() is satisfied, the processor now kicks off a background reconstruction via spawn_reconstruction_task, storing a pending_reconstruction oneshot receiver (result handling is not yet implemented).

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

[reviewable-StarkWare]

This change is 

[sirandreww-starkware]

Warning

This pull request is not mergeable via GitHub because a downstack PR is open. Once all requirements are satisfied, merge this PR as a stack on Graphite.
Learn more

• 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 👈 (View in Graphite)
• 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
• 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 2 potential issues.

[guy-starkware]

@guy-starkware reviewed all commit messages and made 3 comments.
Reviewable status: 0 of 1 files reviewed, 5 unresolved discussions (waiting on @noamsp-starkware, @ShahakShama, and @sirandreww-starkware).

---

crates/apollo_propeller/src/message_processor.rs line 102 at r1 (raw file):

        let mut pending_reconstruction: Option<oneshot::Receiver<ReconstructionResult>> = None;

        // State machine: PreReconstruction -> PostReconstruction

This comment doesn't look like it fits the code under it. Maybe I'm missing something?

---

crates/apollo_propeller/src/message_processor.rs line 224 at r1 (raw file):

    ) {
        // Store the signature from the first unit we receive
        if signature.is_none() && !unit.signature().is_empty() {

What happens if there are units without signatures? (either one such unit or all of them)

---

crates/apollo_propeller/src/message_processor.rs line 239 at r1 (raw file):

        tracing::trace!("[MSG_PROC] Starting reconstruction with {} shards", received_shards.len());

        let shards = received_shards.clone();

why not send the shards in by value? Why clone?

[guy-starkware]

Reviewable status: 0 of 1 files reviewed, 5 unresolved discussions (waiting on noamsp-starkware, ShahakShama, and sirandreww-starkware).

[guy-starkware]

@guy-starkware reviewed 1 file.
Reviewable status: all files reviewed, 5 unresolved discussions (waiting on noamsp-starkware, ShahakShama, and sirandreww-starkware).

[sirandreww-starkware]

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

---

crates/apollo_propeller/src/message_processor.rs line 102 at r1 (raw file):

Previously, guy-starkware wrote…

This comment doesn't look like it fits the code under it. Maybe I'm missing something?

You're not, mistake in split, will be solved in the next PR

---

crates/apollo_propeller/src/message_processor.rs line 224 at r1 (raw file):

Previously, guy-starkware wrote…

What happens if there are units without signatures? (either one such unit or all of them)

Added a comment to explain this

---

crates/apollo_propeller/src/message_processor.rs line 239 at r1 (raw file):

Previously, guy-starkware wrote…

why not send the shards in by value? Why clone?

you're right, using std::mem::take here

[guy-starkware]

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

---

crates/apollo_propeller/src/message_processor.rs line 224 at r1 (raw file):

Previously, sirandreww-starkware (Andrew Luka) wrote…

Added a comment to explain this

Is there a check somewhere that all signatures are the same for all units?

[ShahakShama]

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

---

crates/apollo_propeller/src/message_processor.rs line 136 at r2 (raw file):

                // the validator permanently lost.
                Ok(result) = async {
                    pending_validation.as_mut().unwrap().await

instead of unwrap and if is_some, do here unwrap_or(pending())