feat: ExEx proof of concept (#5)

* feat(exex): ExEx v0.0.000001

* feat(exex):

* feat(exex): Running minimal

* feat(exex):

* feat(exex):

* feat(exex):

* feat(exex): Arch

* feat(exex): Full node exex

* feat(exex): Updated structs before context

* feat(exex): lint

.github/workflows/karnot-ci.yml

@@ -2,7 +2,7 @@ on:
   workflow_dispatch:
     inputs:
       environment:
-        description: 'Environment to deploy'
+        description: "Environment to deploy"
         required: true
         type: choice
         options:
@@ -14,14 +14,14 @@ jobs:
       - uses: actions/checkout@v3
       - name: Debug Information
         run: |
-            echo "Selected environment: ${{ github.event.inputs.environment }}"
-            echo "Current directory: $(pwd)"
-            echo "Directory contents:"
-            ls -R
+          echo "Selected environment: ${{ github.event.inputs.environment }}"
+          echo "Current directory: $(pwd)"
+          echo "Directory contents:"
+          ls -R
       - name: Run Karnot Cloud Pro
         uses: karnotxyz/kcloud-pro-github-action@main
         with:
-          input_file: './infra/pragma/pragma-env.yaml' # Specify your file
+          input_file: "./infra/pragma/pragma-env.yaml" # Specify your file
           environment: ${{ github.event.inputs.environment }}
           KARNOT_CLOUD_URL: ${{ secrets.KARNOT_CLOUD_URL }}
-          KARNOT_CLOUD_TOKEN: ${{ secrets.KARNOT_CLOUD_TOKEN }}
+          KARNOT_CLOUD_TOKEN: ${{ secrets.KARNOT_CLOUD_TOKEN }}

.github/workflows/release.yml

@@ -26,4 +26,4 @@ jobs:
           commit: ${{ github.sha }}
           draft: false
           prerelease: false
-          generateReleaseNotes: true
+          generateReleaseNotes: true

CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ## Next release
 
+- feat: Madara ExExs proof of concept
 - feat(cli): launcher script and release workflows
 - fix: cleaned cli settings for sequencer, devnet and full
 - feat: move to karnot runner

Cargo.toml

@@ -19,6 +19,7 @@ members = [
   "crates/primitives/state_update",
   "crates/primitives/chain_config",
   "crates/primitives/utils",
+  "crates/primitives/exex",
   "crates/proc-macros",
   "crates/tests",
 ]
@@ -43,6 +44,7 @@ default-members = [
   "crates/primitives/state_update",
   "crates/primitives/chain_config",
   "crates/primitives/utils",
+  "crates/primitives/exex",
   "crates/proc-macros",
   "crates/tests",
 ]
@@ -92,6 +94,7 @@ mp-receipt = { path = "crates/primitives/receipt", default-features = false }
 mp-state-update = { path = "crates/primitives/state_update", default-features = false }
 mp-utils = { path = "crates/primitives/utils", default-features = false }
 mp-chain-config = { path = "crates/primitives/chain_config", default-features = false }
+mp-exex = { path = "crates/primitives/exex", default-features = false }
 
 # Madara client
 mc-telemetry = { path = "crates/client/telemetry" }
@@ -176,6 +179,8 @@ env_logger = "0.11.3"
 mockall = "0.13.0"
 serial_test = "3.1.1"
 itertools = "0.13.0"
+tokio-stream = "0.1.16"
+tokio-util = "0.7.12"
 
 [patch.crates-io]
 starknet-core = { git = "https://github.com/kasarlabs/starknet-rs.git", branch = "fork" }

configs/presets/devnet.yaml

@@ -9,8 +9,8 @@ versioned_constants:
   "0.13.2": "crates/primitives/chain_config/resources/versioned_constants_13_2.json"
 eth_core_contract_address: "0xE2Bb56ee936fd6433DC0F6e7e3b8365C906AA057"
 latest_protocol_version: "0.13.2"
-block_time: "30s"
-pending_block_update_time: "2s"
+block_time: "1s"
+pending_block_update_time: "500ms"
 execution_batch_size: 16
 bouncer_config:
   block_max_capacity:

crates/client/devnet/src/lib.rs

@@ -331,6 +331,7 @@ mod tests {
             Arc::clone(&importer),
             Arc::clone(&mempool),
             Arc::clone(&l1_data_provider),
+            Option::None,
         )
         .unwrap();
 

crates/client/mempool/Cargo.toml

@@ -37,6 +37,7 @@ mp-block.workspace = true
 mp-chain-config.workspace = true
 mp-class.workspace = true
 mp-convert.workspace = true
+mp-exex.workspace = true
 mp-receipt.workspace = true
 mp-state-update.workspace = true
 mp-transactions.workspace = true

crates/client/mempool/src/block_production.rs

@@ -1,5 +1,6 @@
 // TODO: Move this into its own crate.
 
+use anyhow::Context;
 use blockifier::blockifier::transaction_executor::{TransactionExecutor, VisitedSegmentsMapping};
 use blockifier::bouncer::{Bouncer, BouncerWeights, BuiltinCount};
 use blockifier::state::cached_state::CommitmentStateDiff;
@@ -13,13 +14,15 @@ use mc_exec::{BlockifierStateAdapter, ExecutionContext};
 use mp_block::{BlockId, BlockTag, MadaraPendingBlock};
 use mp_class::ConvertedClass;
 use mp_convert::ToFelt;
+use mp_exex::{ExExManagerHandle, ExExNotification};
 use mp_receipt::from_blockifier_execution_info;
 use mp_state_update::{
     ContractStorageDiffItem, DeclaredClassItem, DeployedContractItem, NonceUpdate, ReplacedClassItem, StateDiff,
     StorageEntry,
 };
 use mp_transactions::TransactionWithHash;
 use mp_utils::graceful_shutdown;
+use starknet_api::block::BlockNumber;
 use starknet_types_core::felt::Felt;
 use std::collections::VecDeque;
 use std::mem;
@@ -176,6 +179,7 @@ pub struct BlockProductionTask<Mempool: MempoolProvider> {
     pub(crate) executor: TransactionExecutor<BlockifierStateAdapter>,
     l1_data_provider: Arc<dyn L1DataProvider>,
     current_pending_tick: usize,
+    exex_manager: Option<ExExManagerHandle>,
 }
 
 impl<Mempool: MempoolProvider> BlockProductionTask<Mempool> {
@@ -189,6 +193,7 @@ impl<Mempool: MempoolProvider> BlockProductionTask<Mempool> {
         importer: Arc<BlockImporter>,
         mempool: Arc<Mempool>,
         l1_data_provider: Arc<dyn L1DataProvider>,
+        exex_manager: Option<ExExManagerHandle>,
     ) -> Result<Self, Error> {
         let parent_block_hash = backend
             .get_block_hash(&BlockId::Tag(BlockTag::Latest))?
@@ -218,6 +223,7 @@ impl<Mempool: MempoolProvider> BlockProductionTask<Mempool> {
             block: pending_block,
             declared_classes: vec![],
             l1_data_provider,
+            exex_manager,
         })
     }
 
@@ -396,7 +402,6 @@ impl<Mempool: MempoolProvider> BlockProductionTask<Mempool> {
             self.continue_block(self.backend.chain_config().bouncer_config.block_max_capacity)?;
 
         // Convert the pending block to a closed block and save to db.
-
         let parent_block_hash = Felt::ZERO; // temp parent block hash
         let new_empty_block = MadaraPendingBlock::new_empty(make_pending_header(
             parent_block_hash,
@@ -427,6 +432,7 @@ impl<Mempool: MempoolProvider> BlockProductionTask<Mempool> {
         self.current_pending_tick = 0;
 
         log::info!("⛏️  Closed block #{} with {} transactions - {:?}", block_n, n_txs, start_time.elapsed());
+        let _ = self.notify_exexs(block_n).context("Sending notification to ExExs");
 
         Ok(())
     }
@@ -479,4 +485,14 @@ impl<Mempool: MempoolProvider> BlockProductionTask<Mempool> {
     fn block_n(&self) -> u64 {
         self.executor.block_context.block_info().block_number.0
     }
+
+    /// Sends a notification to the ExExs that a block has been closed.
+    fn notify_exexs(&self, block_n: u64) -> anyhow::Result<()> {
+        let Some(manager) = self.exex_manager.as_ref() else {
+            return Ok(());
+        };
+
+        let notification = ExExNotification::BlockClosed { new: BlockNumber(block_n) };
+        manager.send(notification).map_err(|e| anyhow::anyhow!("Could not send ExEx notification: {}", e))
+    }
 }

crates/client/sync/Cargo.toml

@@ -26,6 +26,7 @@ mp-block = { workspace = true }
 mp-chain-config = { workspace = true }
 mp-class = { workspace = true }
 mp-convert = { workspace = true }
+mp-exex = { workspace = true }
 mp-receipt = { workspace = true }
 mp-transactions = { workspace = true }
 mp-utils = { workspace = true }

crates/client/sync/src/l2.rs

@@ -10,7 +10,10 @@ use mc_block_import::{
 use mc_db::MadaraBackend;
 use mc_db::MadaraStorageError;
 use mc_telemetry::{TelemetryHandle, VerbosityLevel};
+use mp_exex::ExExManagerHandle;
+use mp_exex::ExExNotification;
 use mp_utils::{channel_wait_or_graceful_shutdown, wait_or_graceful_shutdown, PerfStopwatch};
+use starknet_api::block::BlockNumber;
 use starknet_api::core::ChainId;
 use starknet_providers::{ProviderError, SequencerGatewayProvider};
 use starknet_types_core::felt::Felt;
@@ -41,6 +44,16 @@ pub struct L2StateUpdate {
     pub block_hash: Felt,
 }
 
+/// Sends a notification to the ExExs that a block has been imported.
+fn notify_exexs(exex_manager: &Option<ExExManagerHandle>, block_n: u64) -> anyhow::Result<()> {
+    let Some(manager) = exex_manager.as_ref() else {
+        return Ok(());
+    };
+
+    let notification = ExExNotification::BlockClosed { new: BlockNumber(block_n) };
+    manager.send(notification).map_err(|e| anyhow::anyhow!("Could not send ExEx notification: {}", e))
+}
+
 #[allow(clippy::too_many_arguments)]
 async fn l2_verify_and_apply_task(
     backend: Arc<MadaraBackend>,
@@ -49,6 +62,7 @@ async fn l2_verify_and_apply_task(
     validation: BlockValidationContext,
     backup_every_n_blocks: Option<u64>,
     telemetry: TelemetryHandle,
+    exex_manager: Option<ExExManagerHandle>,
 ) -> anyhow::Result<()> {
     while let Some(block) = channel_wait_or_graceful_shutdown(pin!(updates_receiver.recv())).await {
         let BlockImportResult { header, block_hash } = block_import.verify_apply(block, validation.clone()).await?;
@@ -66,6 +80,8 @@ async fn l2_verify_and_apply_task(
             header.global_state_root
         );
 
+        notify_exexs(&exex_manager, header.block_number)?;
+
         telemetry.send(
             VerbosityLevel::Info,
             serde_json::json!({
@@ -185,6 +201,7 @@ pub async fn sync(
     chain_id: ChainId,
     telemetry: TelemetryHandle,
     block_importer: Arc<BlockImporter>,
+    exex_manager: Option<ExExManagerHandle>,
 ) -> anyhow::Result<()> {
     let (fetch_stream_sender, fetch_stream_receiver) = mpsc::channel(8);
     let (block_conv_sender, block_conv_receiver) = mpsc::channel(4);
@@ -231,6 +248,7 @@ pub async fn sync(
         validation.clone(),
         config.backup_every_n_blocks,
         telemetry,
+        exex_manager,
     ));
     join_set.spawn(l2_pending_block_task(
         Arc::clone(backend),

crates/client/sync/src/lib.rs

@@ -5,6 +5,7 @@ use mc_block_import::BlockImporter;
 use mc_db::MadaraBackend;
 use mc_telemetry::TelemetryHandle;
 use mp_convert::ToFelt;
+use mp_exex::ExExManagerHandle;
 use starknet_providers::SequencerGatewayProvider;
 use std::{sync::Arc, time::Duration};
 
@@ -23,6 +24,7 @@ pub async fn sync(
     backup_every_n_blocks: Option<u64>,
     telemetry: TelemetryHandle,
     pending_block_poll_interval: Duration,
+    exex_manager: Option<ExExManagerHandle>,
 ) -> anyhow::Result<()> {
     let (starting_block, ignore_block_order) = if let Some(starting_block) = starting_block {
         log::warn!("Forcing unordered state. This will most probably break your database.");
@@ -65,6 +67,7 @@ pub async fn sync(
         backend.chain_config().chain_id.clone(),
         telemetry,
         block_importer,
+        exex_manager,
     )
     .await?;
 

crates/node/Cargo.toml

@@ -30,6 +30,7 @@ mc-telemetry = { workspace = true }
 mp-block = { workspace = true }
 mp-chain-config = { workspace = true }
 mp-convert = { workspace = true }
+mp-exex = { workspace = true }
 mp-utils = { workspace = true }
 
 # Starknet