feat(starfish): send full block

.github/CODEOWNERS

@@ -85,6 +85,7 @@
 /crates/prometheus-closure-metric/ @iotaledger/node
 /crates/shared-crypto/ @iotaledger/core-protocol
 /crates/simulacrum/ @iotaledger/vm-language
+/crates/starfish/ @iotaledger/consensus
 /crates/telemetry-subscribers/ @iotaledger/core-protocol
 /crates/transaction-fuzzer/ @iotaledger/node @iotaledger/vm-language
 /crates/test-cluster/ @iotaledger/infrastructure

.github/crates-filters.yml

@@ -203,8 +203,16 @@ iota-move-natives-latest:
 iota-verifier-latest:
   - "iota-execution/latest/iota-verifier/**"
 
-# Consensus
+# Mysticeti Consensus
 consensus-config:
   - "consensus/config/**"
 consensus-core:
   - "consensus/core/**"
+
+# Starfish Consensus
+starfish-config:
+  - "crates/starfish/config/**"
+starfish-core:
+  - "crates/starfish/core/**"
+starfish-simtests:
+  - "crates/starfish/simtests/**"

.gitignore

@@ -16,6 +16,7 @@ build/
 !crates/iota-single-node-benchmark/tests/data/package_publish_from_bytecode/package_b/build/
 storage/
 !consensus/core/src/storage/
+!crates/starfish/core/src/storage/
 !crates/iota-types/src/storage/
 
 # Move-related files

Cargo.toml

@@ -153,6 +153,9 @@ members = [
   "crates/prometheus-closure-metric",
   "crates/shared-crypto",
   "crates/simulacrum",
+  "crates/starfish/config",
+  "crates/starfish/core",
+  "crates/starfish/simtests",
   "crates/telemetry-subscribers",
   "crates/test-cluster",
   "crates/transaction-fuzzer",
@@ -470,6 +473,9 @@ move-vm-types = { path = "external-crates/move/crates/move-vm-types" }
 prometheus-closure-metric = { path = "crates/prometheus-closure-metric" }
 shared-crypto = { path = "crates/shared-crypto" }
 simulacrum = { path = "crates/simulacrum" }
+starfish-config = { path = "crates/starfish/config" }
+starfish-core = { path = "crates/starfish/core" }
+starfish-simtests = { path = "crates/starfish/simtests" }
 telemetry-subscribers = { path = "crates/telemetry-subscribers" }
 test-cluster = { path = "crates/test-cluster" }
 transaction-fuzzer = { path = "crates/transaction-fuzzer" }

consensus/core/src/linearizer.rs

@@ -159,7 +159,7 @@ impl Linearizer {
         let mut to_commit = Vec::new();
 
         // The new logic will perform the recursion without stopping at the highest
-        // round round that has been committed per authority. Instead it will
+        // round that has been committed per authority. Instead it will
         // allow to commit blocks that are lower than the highest committed round for an
         // authority but higher than gc_round.
         if context.protocol_config.consensus_linearize_subdag_v2() {

crates/iota-config/src/node.rs

@@ -465,6 +465,8 @@ impl NodeConfig {
 pub enum ConsensusProtocol {
     #[serde(rename = "mysticeti")]
     Mysticeti,
+    #[serde(rename = "starfish")]
+    Starfish,
 }
 
 #[derive(Debug, Clone, Deserialize, Serialize)]

crates/iota-core/Cargo.toml

@@ -89,6 +89,8 @@ move-core-types.workspace = true
 move-package.workspace = true
 move-symbol-pool.workspace = true
 shared-crypto.workspace = true
+starfish-config.workspace = true
+starfish-core.workspace = true
 telemetry-subscribers.workspace = true
 typed-store.workspace = true
 typed-store-derive.workspace = true

crates/iota-core/src/consensus_handler.rs

@@ -492,6 +492,51 @@ impl Drop for MysticetiConsensusHandler {
     }
 }
 
+/// Consensus handler used by Starfish.
+/// During initialization, the sender is passed into Starfish which can send
+/// consensus output to the channel.
+pub struct StarfishConsensusHandler {
+    handle: Option<tokio::task::JoinHandle<()>>,
+}
+
+impl StarfishConsensusHandler {
+    pub fn new(
+        mut consensus_handler: ConsensusHandler<CheckpointService>,
+        mut receiver: UnboundedReceiver<starfish_core::CommittedSubDag>,
+        commit_consumer_monitor: Arc<starfish_core::CommitConsumerMonitor>,
+    ) -> Self {
+        let handle = spawn_monitored_task!(async move {
+            // TODO: pause when execution is overloaded, so consensus can detect the
+            // backpressure.
+            while let Some(consensus_output) = receiver.recv().await {
+                let commit_index = consensus_output.commit_ref.index;
+                consensus_handler
+                    .handle_consensus_output(consensus_output)
+                    .await;
+                commit_consumer_monitor.set_highest_handled_commit(commit_index);
+            }
+        });
+        Self {
+            handle: Some(handle),
+        }
+    }
+
+    pub async fn abort(&mut self) {
+        if let Some(handle) = self.handle.take() {
+            handle.abort();
+            let _ = handle.await;
+        }
+    }
+}
+
+impl Drop for StarfishConsensusHandler {
+    fn drop(&mut self) {
+        if let Some(handle) = self.handle.take() {
+            handle.abort();
+        }
+    }
+}
+
 impl<C> ConsensusHandler<C> {
     fn authenticator_state_update_transaction(
         &self,