feat: send proof though msgio crate (#254)

* chore: batch propagation

Signed-off-by: Sacha Froment <[email protected]>

* chore: --

Signed-off-by: Sacha Froment <[email protected]>

* feat: add simple runner

Signed-off-by: Sacha Froment <[email protected]>

* fix: run need to be mutable

* fix: msgio reorganize

* feat: use commiter in sequencer

* chore: format

Signed-off-by: Sacha Froment <[email protected]>

* feat: make prover send proof though Q

* mod: rm unnecessary filed in message header && change some coment

* mod: submitter use enveloped message

* mod: chagne committer use enveloped msg && change tests prover context

* fix: just fix

* mod: modification deprecated funtion && just fix && cargo clippy && cargo test

* fix: copilot feedback

* chore: remove useless comment

Signed-off-by: Sacha Froment <[email protected]>

* mod: change keccak computaion to use util function

* fix: dep order

* chore: just fix

Signed-off-by: Sacha Froment <[email protected]>

* fix: copilot

---------

Signed-off-by: Sacha Froment <[email protected]>
Co-authored-by: Sacha Froment <[email protected]>

Author: Giwook-Han

.gitignore

@@ -28,3 +28,5 @@ Cargo.toml.bak
 
 bitcoin/regtest
 bitcoin/testnet4
+
+.mojave

Cargo.lock

@@ -11,6 +11,7 @@ documentation = { workspace = true }
 [dependencies]
 mojave-msgio = { workspace = true }
 mojave-task = { workspace = true }
+mojave-utils = { workspace = true }
 
 ethrex-common = { workspace = true }
 ethrex-p2p = { workspace = true }
@@ -20,6 +21,7 @@ bitcoin = { workspace = true, features = ["std", "secp-recovery"] }
 bytes = { workspace = true }
 
 bitcoincore-rpc = { workspace = true }
+hex = { workspace = true }
 rand = { workspace = true }
 # TODO: make sure when we update bitcoin that this is still needed
 # this is mandatory to handle 2 version of secp256k1 as bitcoin lib await this version right now

crates/batch-submitter/Cargo.toml

@@ -3,12 +3,13 @@ use ethrex_common::types::batch::Batch;
 use ethrex_p2p::{
     network::P2PContext,
     rlpx::{
-        Message,
+        message::Message as RlpxMessage,
         mojave::messages::{MojaveBatch, MojaveMessage},
     },
 };
-use mojave_msgio::types::Publisher;
+use mojave_msgio::types::{self, Publisher};
 use mojave_task::Service;
+use mojave_utils::hash;
 use tokio::sync::broadcast;
 
 use crate::error::{Error, Result};
@@ -50,14 +51,28 @@ where
 
         self.commit_next_batch_to_l1(batch.clone())?;
 
-        let data = bincode::serialize(&batch)?;
+        // didn't check about dedup here
+        let msg_id = hex::encode(hash::compute_keccak(&batch.number.to_le_bytes()));
+
+        let msg = types::Message {
+            header: types::MessageHeader {
+                version: 1,
+                kind: types::MessageKind::BatchSubmit,
+                message_id: msg_id,
+                // Only one message is sent per batch, so sequence number is always 1.
+                seq: 1,
+            },
+            body: &batch,
+        };
+
+        let data = bincode::serialize(&msg)?;
         let data = Bytes::from(data);
         self.queue.publish(data).await?;
 
         self.p2p_context
-            .broadcast_mojave_message(Message::Mojave(MojaveMessage::Batch(MojaveBatch::new(
-                batch,
-            ))))?;
+            .broadcast_mojave_message(RlpxMessage::Mojave(MojaveMessage::Batch(
+                MojaveBatch::new(batch),
+            )))?;
 
         Ok(())
     }

crates/batch-submitter/src/committer.rs

@@ -38,6 +38,10 @@ impl JobId {
     pub fn is_empty(&self) -> bool {
         self.0.is_empty()
     }
+
+    pub fn as_str(&self) -> &str {
+        self.0.as_str()
+    }
 }
 
 impl Borrow<str> for JobId {

crates/client/src/types.rs

@@ -12,6 +12,7 @@ documentation = { workspace = true }
 [dependencies]
 async-trait = { workspace = true }
 bytes = { workspace = true }
+serde = { workspace = true }
 thiserror = { workspace = true }
 
 [dev-dependencies]

crates/msgio/Cargo.toml

@@ -1,9 +1,30 @@
 use async_trait::async_trait;
 use bytes::Bytes;
+use serde::{Deserialize, Serialize};
 
 use crate::error::Result;
 
 #[async_trait]
 pub trait Publisher: Send + Sync + 'static {
     async fn publish(&self, msg: Bytes) -> Result<()>;
 }
+
+#[derive(Clone, Copy, Debug, Serialize, Deserialize)]
+pub enum MessageKind {
+    ProofResponse,
+    BatchSubmit,
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct MessageHeader {
+    pub version: u8,
+    pub kind: MessageKind,
+    pub message_id: String,
+    pub seq: u64,
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct Message<T> {
+    pub header: MessageHeader,
+    pub body: T,
+}

crates/msgio/src/types.rs

@@ -10,6 +10,7 @@ documentation = { workspace = true }
 
 [dependencies]
 mojave-client = { workspace = true }
+mojave-msgio = { workspace = true }
 mojave-rpc-core = { workspace = true }
 mojave-rpc-macros = { workspace = true }
 mojave-rpc-server = { workspace = true }
@@ -25,7 +26,6 @@ hex = { workspace = true }
 reqwest = { workspace = true }
 serde = { workspace = true, features = ["derive"] }
 serde_json = { workspace = true }
-tiny-keccak = { workspace = true }
 tokio = { workspace = true }
 tracing = { workspace = true }
 

crates/prover/Cargo.toml

@@ -5,8 +5,11 @@ use crate::{
 use mojave_rpc_server::{RpcRegistry, RpcService};
 use mojave_utils::rpc::error::{Error, Result};
 
-use std::sync::Arc;
-use tokio::{net::TcpListener, sync::mpsc};
+use std::{collections::HashSet, sync::Arc};
+use tokio::{
+    net::TcpListener,
+    sync::{Mutex, mpsc},
+};
 use tracing::info;
 
 pub async fn start_api(
@@ -16,10 +19,18 @@ pub async fn start_api(
     queue_capacity: usize,
 ) -> Result<()> {
     let (job_sender, job_receiver) = mpsc::channel::<JobRecord>(queue_capacity);
+    // use dummy publisher for now
+    let publisher = Arc::new(
+        mojave_msgio::dummy::Dummy::new()
+            .await
+            .map_err(|e| Error::Internal(e.to_string()))?,
+    );
     let context = Arc::new(ProverRpcContext {
         aligned_mode,
         job_store: JobStore::default(),
         sender: job_sender,
+        publisher,
+        sent_ids: Mutex::new(HashSet::new()),
     });
     tracing::info!(aligned_mode = %aligned_mode, "Prover RPC context initialized");
 

crates/prover/src/rpc/api.rs

@@ -1,9 +1,14 @@
-use tokio::sync::mpsc;
+use std::{collections::HashSet, sync::Arc};
+
+use mojave_msgio::types::Publisher;
+use tokio::sync::{Mutex, mpsc};
 
 use crate::{job::JobStore, rpc::types::JobRecord};
 
 pub struct ProverRpcContext {
     pub aligned_mode: bool,
     pub job_store: JobStore,
     pub sender: mpsc::Sender<JobRecord>,
+    pub publisher: Arc<dyn Publisher>,
+    pub sent_ids: Mutex<HashSet<String>>,
 }