feat(consensus): non-blocking proposal handling

starkware-libs · Nov 5, 2024 · 316e4d2 · 316e4d2
1 parent 52a20af
commit 316e4d2
Show file tree

Hide file tree

Showing 5 changed files with 283 additions and 139 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/crates/sequencing/papyrus_consensus/Cargo.toml b/crates/sequencing/papyrus_consensus/Cargo.toml
@@ -28,6 +28,7 @@ tokio.workspace = true
 tracing.workspace = true
 
 [dev-dependencies]
+enum-as-inner = "0.6.1"
 mockall.workspace = true
 papyrus_network = { workspace = true, features = ["testing"] }
 papyrus_network_types = { workspace = true, features = ["testing"] }

diff --git a/crates/sequencing/papyrus_consensus/src/manager.rs b/crates/sequencing/papyrus_consensus/src/manager.rs
@@ -18,7 +18,7 @@ use starknet_api::core::ContractAddress;
 use tracing::{debug, info, instrument};
 
 use crate::config::TimeoutsConfig;
-use crate::single_height_consensus::{ShcReturn, ShcTask, SingleHeightConsensus};
+use crate::single_height_consensus::{ShcReturn, SingleHeightConsensus};
 use crate::types::{
     BroadcastConsensusMessageChannel,
     ConsensusContext,
@@ -125,13 +125,13 @@ impl MultiHeightManager {
             validators,
             self.timeouts.clone(),
         );
-        let mut shc_tasks = FuturesUnordered::new();
+        let mut shc_events = FuturesUnordered::new();
 
         match shc.start(context).await? {
             ShcReturn::Decision(decision) => return Ok(decision),
             ShcReturn::Tasks(tasks) => {
                 for task in tasks {
-                    shc_tasks.push(create_task_handler(task));
+                    shc_events.push(task.run());
                 }
             }
         }
@@ -142,16 +142,16 @@ impl MultiHeightManager {
                 message = next_message(&mut current_height_messages, broadcast_channels) => {
                     self.handle_message(context, height, &mut shc, message?).await?
                 },
-                Some(shc_task) = shc_tasks.next() => {
-                    shc.handle_event(context, shc_task.event).await?
+                Some(shc_event) = shc_events.next() => {
+                    shc.handle_event(context, shc_event).await?
                 },
             };
 
             match shc_return {
                 ShcReturn::Decision(decision) => return Ok(decision),
                 ShcReturn::Tasks(tasks) => {
                     for task in tasks {
-                        shc_tasks.push(create_task_handler(task));
+                        shc_events.push(task.run());
                     }
                 }
             }
@@ -278,8 +278,3 @@ where
         }
     }
 }
-
-async fn create_task_handler(task: ShcTask) -> ShcTask {
-    tokio::time::sleep(task.duration).await;
-    task
-}