ENG-326: Full Node Commitments

CODEOWNERS

@@ -0,0 +1,2 @@
+# global
+*   @l-monninger @andyjsbell @0xmovses @mzabaluev

Cargo.toml

@@ -14,6 +14,8 @@ members = [
 
     "protocol-units/mempool/*",
 
+    "protocol-units/settlement/mcr/client",
+
     "util/*",
     "util/buildtime/buildtime-helpers",
     "util/buildtime/buildtime-macros",
@@ -61,6 +63,8 @@ m1-da-light-node-grpc = { path = "protocol-units/da/m1/light-node-grpc" }
 m1-da-light-node-util = { path = "protocol-units/da/m1/util" }
 m1-da-light-node-verifier = { path = "protocol-units/da/m1/light-node-verifier" }
 m1-da-light-node-client = { path = "protocol-units/da/m1/light-node-client" }
+## settlement
+mcr-settlement-client = { path = "protocol-units/settlement/mcr/client" }
 ## mempool
 mempool-util = { path = "protocol-units/mempool/util" }
 move-rocks = { path = "protocol-units/mempool/move-rocks" }

PR.md

@@ -0,0 +1,25 @@
+# Summary
+- **Status**: `draft` <!-- `review`, `suspended` -->
+- **RFCs**: [Link to RFC](#./link/to/rfc), [Link to RFC](#./link/to/rfc), or $\emptyset$.
+- **Categories**: any of `protocol-units`, `networks`, `scripts`, `util`, `cicd`, or `misc`.
+
+<!--
+Add your summary text here. 
+ -->
+
+# Changelog
+
+<!-- 
+Describe your changes. List roughly in order of importance.
+-->
+
+# Testing
+
+<!--
+Describe your Test Plan and explain added or modified test components.
+-->
+
+# Outstanding issues
+<!--
+List any outstanding issues that need to be addressed in future PRs, but which do not block merging this PR.
+-->

networks/monza/monza-full-node/src/partial.rs

@@ -41,16 +41,18 @@ impl <T : MonzaExecutor + Send + Sync + Clone>MonzaPartialNode<T> {
         }
     }
 
-    pub async fn bind_transaction_channel(&mut self) -> Result<(), anyhow::Error> {
-        self.executor.set_tx_channel(self.transaction_sender.clone()).await?;
-        Ok(())
-    }
-
-    pub async fn bound(executor : T, light_node_client: LightNodeServiceClient<tonic::transport::Channel>) -> Result<Self, anyhow::Error> {
-        let mut node = Self::new(executor, light_node_client);
-        node.bind_transaction_channel().await?;
-        Ok(node)
-    }
+	fn bind_transaction_channel(&mut self) {
+		self.executor.set_tx_channel(self.transaction_sender.clone());
+	}
+
+	pub fn bound(
+		executor: T,
+		light_node_client: LightNodeServiceClient<tonic::transport::Channel>,
+	) -> Result<Self, anyhow::Error> {
+		let mut node = Self::new(executor, light_node_client);
+		node.bind_transaction_channel();
+		Ok(node)
+	}
 
     pub async fn tick_write_transactions_to_da(&self) -> Result<(), anyhow::Error> {
 
@@ -166,7 +168,7 @@ impl <T : MonzaExecutor + Send + Sync + Clone>MonzaPartialNode<T> {
             );
             let block_id = executable_block.block_id;
             self.executor.execute_block(
-                &FinalityMode::Opt,
+                FinalityMode::Opt,
                 executable_block
             ).await?;
 
@@ -225,7 +227,7 @@ impl MonzaPartialNode<MonzaExecutorV1> {
         let executor = MonzaExecutorV1::try_from_env(tx).await.context(
             "Failed to get executor from environment"
         )?;
-        Self::bound(executor, light_node_client).await
+        Self::bound(executor, light_node_client)
     }
 
 }

networks/suzuka/suzuka-full-node/Cargo.toml

@@ -14,6 +14,7 @@ rust-version = { workspace = true }
 [dependencies]
 suzuka-executor = { workspace = true }
 m1-da-light-node-client = { workspace = true }
+mcr-settlement-client = { workspace = true }
 async-channel = { workspace = true }
 serde_json = { workspace = true }
 anyhow = { workspace = true }

networks/suzuka/suzuka-full-node/src/partial.rs

@@ -17,39 +17,54 @@ use async_channel::{Sender, Receiver};
 use sha2::Digest;
 use crate::*;
 use tokio_stream::StreamExt;
+use tokio::sync::mpsc::{self, error::TrySendError};
 use tokio::sync::RwLock;
-use movement_types::Block;
-
+use movement_types::{Block, BlockCommitment};
+use mcr_settlement_client::{McrSettlementClient, McrSettlementClientOperations};
 
 #[derive(Clone)]
-pub struct SuzukaPartialNode<T : SuzukaExecutor + Send + Sync + Clone> {
+pub struct SuzukaPartialNode<T, C>
+{
     executor: T,
     transaction_sender : Sender<SignedTransaction>,
     pub transaction_receiver : Receiver<SignedTransaction>,
     light_node_client: Arc<RwLock<LightNodeServiceClient<tonic::transport::Channel>>>,
+    settlement_client: Arc<C>,
 }
 
-impl <T : SuzukaExecutor + Send + Sync + Clone>SuzukaPartialNode<T> {
-
-    pub fn new(executor : T, light_node_client: LightNodeServiceClient<tonic::transport::Channel>) -> Self {
+impl<T, C> SuzukaPartialNode<T, C>
+where
+    T: SuzukaExecutor + Send + Sync,
+    C: McrSettlementClientOperations + Send + Sync + 'static,
+{
+
+    pub fn new(
+        executor: T,
+        light_node_client: LightNodeServiceClient<tonic::transport::Channel>,
+        settlement_client: C,
+    ) -> Self {
         let (transaction_sender, transaction_receiver) = async_channel::unbounded();
         Self {
             executor : executor,
             transaction_sender,
             transaction_receiver,
             light_node_client : Arc::new(RwLock::new(light_node_client)),
+            settlement_client: Arc::new(settlement_client)
         }
     }
 
-    pub async fn bind_transaction_channel(&mut self) -> Result<(), anyhow::Error> {
-        self.executor.set_tx_channel(self.transaction_sender.clone()).await?;
-        Ok(())
-    }
-
-    pub async fn bound(executor : T, light_node_client: LightNodeServiceClient<tonic::transport::Channel>) -> Result<Self, anyhow::Error> {
-        let mut node = Self::new(executor, light_node_client);
-        node.bind_transaction_channel().await?;
-        Ok(node)
+	fn bind_transaction_channel(&mut self) {
+		self.executor.set_tx_channel(self.transaction_sender.clone());
+	}
+
+	pub fn bound(
+		executor: T,
+		light_node_client: LightNodeServiceClient<tonic::transport::Channel>,
+        settlement_client: C,
+	) -> Result<Self, anyhow::Error> {
+		let mut node = Self::new(executor, light_node_client, settlement_client);
+		node.bind_transaction_channel();
+		Ok(node)
     }
 
     pub async fn tick_write_transactions_to_da(&self) -> Result<(), anyhow::Error> {
@@ -108,9 +123,11 @@ impl <T : SuzukaExecutor + Send + Sync + Clone>SuzukaPartialNode<T> {
     // receive transactions from the transaction channel and send them to be executed
     // ! This assumes the m1 da light node is running sequencer mode
     pub async fn read_blocks_from_da(&self) -> Result<(), anyhow::Error> {
-        
+
         let block_head_height = self.executor.get_block_head_height().await?;
-
+
+        let (sender, receiver) = mpsc::channel(16);
+
         let mut stream = {
             let client_ptr = self.light_node_client.clone();
             let mut light_node_client =  client_ptr.write().await;
@@ -121,6 +138,14 @@ impl <T : SuzukaExecutor + Send + Sync + Clone>SuzukaPartialNode<T> {
             ).await?
         }.into_inner();
 
+        let settlement_client = self.settlement_client.clone();
+        // TODO: consume the commitment stream to finalize blocks
+        let commitment_stream = settlement_client.stream_block_commitments().await?;
+
+        tokio::spawn(async move {
+            process_commitments(receiver, settlement_client).await;
+        });
+
         while let Some(blob) = stream.next().await {
 
             println!("Stream hot!");
@@ -165,13 +190,22 @@ impl <T : SuzukaExecutor + Send + Sync + Clone>SuzukaPartialNode<T> {
                 block
             );
             let block_id = executable_block.block_id;
-            self.executor.execute_block(
-                &FinalityMode::Opt,
+            let commitment = self.executor.execute_block(
+                FinalityMode::Opt,
                 executable_block
             ).await?;
 
             println!("Executed block: {:?}", block_id);
 
+            match sender.try_send(commitment) {
+                Ok(_) => {},
+                Err(TrySendError::Closed(_)) => {
+                    break;
+                },
+                Err(TrySendError::Full(_commitment)) => {
+                    println!("Commitment channel full, dropping commitment");
+                }
+            }
         }
 
         Ok(())
@@ -180,7 +214,25 @@ impl <T : SuzukaExecutor + Send + Sync + Clone>SuzukaPartialNode<T> {
 
 }
 
-impl <T : SuzukaExecutor + Send + Sync + Clone>SuzukaFullNode for SuzukaPartialNode<T> {
+async fn process_commitments<C>(
+    mut receiver: mpsc::Receiver<BlockCommitment>,
+    settlement_client: Arc<C>,
+) -> Result<(), anyhow::Error>
+where
+    C: McrSettlementClientOperations,
+{
+    while let Some(commitment) = receiver.recv().await {
+        println!("Got commitment: {:?}", commitment);
+        settlement_client.post_block_commitment(commitment).await?;
+    }
+    Ok(())
+}
+
+impl<T, C> SuzukaFullNode for SuzukaPartialNode<T, C>
+where
+    T: SuzukaExecutor + Send + Sync,
+    C: McrSettlementClientOperations + Send + Sync + 'static,
+{
 
         /// Runs the services until crash or shutdown.
         async fn run_services(&self) -> Result<(), anyhow::Error> {
@@ -217,15 +269,16 @@ impl <T : SuzukaExecutor + Send + Sync + Clone>SuzukaFullNode for SuzukaPartialN
 
 }
 
-impl SuzukaPartialNode<SuzukaExecutorV1> {
+impl SuzukaPartialNode<SuzukaExecutorV1, McrSettlementClient> {
 
     pub async fn try_from_env() -> Result<Self, anyhow::Error> {
         let (tx, _) = async_channel::unbounded();
         let light_node_client = LightNodeServiceClient::connect("http://[::1]:30730").await?;
         let executor = SuzukaExecutorV1::try_from_env(tx).await.context(
             "Failed to get executor from environment"
         )?;
-        Self::bound(executor, light_node_client).await
+        let settlement_client = McrSettlementClient::new();
+        Self::bound(executor, light_node_client, settlement_client)
     }
 
 }

protocol-units/execution/maptos/opt-executor/Cargo.toml

@@ -33,6 +33,8 @@ tokio = { workspace = true }
 rand = { workspace = true }
 rand_core = { workspace = true }
 bcs = { workspace = true}
+futures = { workspace = true }
+async-channel = { workspace = true }
 
 aptos-vm = { workspace = true }
 aptos-config = { workspace = true }
@@ -58,9 +60,8 @@ aptos-mempool = { workspace = true }
 aptos-temppath = { workspace = true }
 aptos-faucet-core = { workspace = true }
 aptos-cached-packages = { workspace = true }
-futures = { workspace = true }
-async-channel = { workspace = true }
 maptos-execution-util = { workspace = true }
+movement-types = { workspace = true }
 
 dirs = { workspace = true }
 tempfile = { workspace = true }