apollo_propeller: add MessageProcessor scaffolding

Author: sirandreww-starkware

Cargo.lock

@@ -19,6 +19,7 @@ prost.workspace = true
 reed-solomon-simd.workspace = true
 sha2.workspace = true
 thiserror.workspace = true
+tokio.workspace = true
 tracing.workspace = true
 unsigned-varint.workspace = true
 

crates/apollo_propeller/Cargo.toml

@@ -9,6 +9,7 @@ pub mod padding;
 mod padding_test;
 pub mod protocol;
 // TODO(AndrewL): Consider renaming this to `erasure_coding` or `error_correction_code`.
+pub mod message_processor;
 pub mod reed_solomon;
 #[cfg(test)]
 mod reed_solomon_test;

crates/apollo_propeller/src/lib.rs

@@ -0,0 +1,98 @@
+use std::ops::ControlFlow;
+use std::sync::Arc;
+use std::time::Duration;
+
+use libp2p::identity::{PeerId, PublicKey};
+use tokio::sync::mpsc;
+use tokio::time::sleep_until;
+use tracing::{debug, trace};
+
+use crate::tree::PropellerScheduleManager;
+use crate::types::{Channel, Event, MessageRoot};
+use crate::unit::PropellerUnit;
+use crate::ShardIndex;
+
+pub type UnitToValidate = (PeerId, PropellerUnit);
+
+#[derive(Debug)]
+pub enum EventStateManagerToEngine {
+    BehaviourEvent(Event),
+    Finalized { channel: Channel, publisher: PeerId, message_root: MessageRoot },
+    BroadcastUnit { unit: PropellerUnit, peers: Vec<PeerId> },
+}
+
+/// Message processor that handles validation and state management for a single message.
+pub struct MessageProcessor {
+    pub channel: Channel,
+    pub publisher: PeerId,
+    pub message_root: MessageRoot,
+    pub my_shard_index: ShardIndex,
+
+    pub publisher_public_key: PublicKey,
+    pub tree_manager: Arc<PropellerScheduleManager>,
+    pub local_peer_id: PeerId,
+
+    // Unbounded because these bridge sync -> async contexts and shard messages from the network
+    // must not be dropped or delayed.
+    pub unit_rx: mpsc::UnboundedReceiver<UnitToValidate>,
+    pub engine_tx: mpsc::UnboundedSender<EventStateManagerToEngine>,
+
+    pub timeout: Duration,
+}
+
+impl MessageProcessor {
+    pub async fn run(mut self) {
+        debug!(
+            "[MSG_PROC] Started for channel={:?} publisher={:?} root={:?}",
+            self.channel, self.publisher, self.message_root
+        );
+
+        // Local state variables
+        let deadline = tokio::time::Instant::now() + self.timeout;
+
+        // TODO(AndrewL): remove this
+        #[allow(clippy::never_loop)]
+        loop {
+            tokio::select! {
+                _ = sleep_until(deadline) => {
+                    let _ = self.emit_timeout_and_finalize().await;
+                    break;
+                }
+            }
+        }
+
+        debug!(
+            "[MSG_PROC] Stopped for channel={:?} publisher={:?} root={:?}",
+            self.channel, self.publisher, self.message_root
+        );
+    }
+
+    async fn emit_timeout_and_finalize(&mut self) -> ControlFlow<()> {
+        trace!(
+            "[MSG_PROC] Timeout reached for channel={:?} publisher={:?} root={:?}",
+            self.channel,
+            self.publisher,
+            self.message_root
+        );
+
+        self.emit_and_finalize(Event::MessageTimeout {
+            channel: self.channel,
+            publisher: self.publisher,
+            message_root: self.message_root,
+        })
+    }
+
+    fn emit_and_finalize(&self, event: Event) -> ControlFlow<()> {
+        self.engine_tx
+            .send(EventStateManagerToEngine::BehaviourEvent(event))
+            .expect("Engine task has exited");
+        self.engine_tx
+            .send(EventStateManagerToEngine::Finalized {
+                channel: self.channel,
+                publisher: self.publisher,
+                message_root: self.message_root,
+            })
+            .expect("Engine task has exited");
+        ControlFlow::Break(())
+    }
+}