store nonce & message hash

crates/katana/core/src/service/block_producer.rs

@@ -13,9 +13,11 @@ use katana_pool::validation::stateful::TxValidator;
 use katana_primitives::block::{BlockHashOrNumber, ExecutableBlock, PartialHeader};
 use katana_primitives::receipt::Receipt;
 use katana_primitives::trace::TxExecInfo;
-use katana_primitives::transaction::{ExecutableTxWithHash, TxHash, TxWithHash};
+use katana_primitives::transaction::{ExecutableTxWithHash, TxHash, TxWithHash, Tx};
 use katana_primitives::version::CURRENT_STARKNET_VERSION;
+use katana_primitives::FieldElement;
 use katana_provider::error::ProviderError;
+use katana_provider::traits::messaging::MessagingProvider;
 use katana_provider::traits::block::{BlockHashProvider, BlockNumberProvider};
 use katana_provider::traits::env::BlockEnvProvider;
 use katana_provider::traits::state::StateFactoryProvider;
@@ -296,9 +298,11 @@ impl<EF: ExecutorFactory> IntervalBlockProducer<EF> {
     fn execute_transactions(
         executor: PendingExecutor,
         transactions: Vec<ExecutableTxWithHash>,
+        backend: Arc<Backend<EF>>,
     ) -> TxExecutionResult {
-        let executor = &mut executor.write();
+        let provider = backend.blockchain.provider();
 
+        let executor = &mut executor.write();
         let new_txs_count = transactions.len();
         executor.execute_transactions(transactions)?;
 
@@ -309,13 +313,43 @@ impl<EF: ExecutorFactory> IntervalBlockProducer<EF> {
         let results = txs
             .iter()
             .skip(total_txs - new_txs_count)
-            .filter_map(|(tx, res)| match res {
-                ExecutionResult::Failed { .. } => None,
-                ExecutionResult::Success { receipt, trace, .. } => Some(TxWithOutcome {
-                    tx: tx.clone(),
-                    receipt: receipt.clone(),
-                    exec_info: trace.clone(),
-                }),
+            .filter_map(|(tx, res)| {
+                let tx_ref: &Tx = &tx.transaction;
+
+                trace!(target: LOG_TARGET, "Executed transaction: {:?}", tx);
+                let _ = match tx_ref {
+                    Tx::L1Handler(l1_tx) => {
+                        // get stored nonce from message hash
+                        let message_hash_bytes = l1_tx.message_hash;
+                        let message_hash_bytes: [u8; 32] = *message_hash_bytes;
+                        match FieldElement::from_bytes_be(&message_hash_bytes) {
+                            Ok(message_hash) => {
+                                match provider.get_nonce_from_message_hash(message_hash) {
+                                    Ok(Some(nonce)) => provider.set_gather_message_nonce(nonce),
+                                    Ok(None) => {
+                                        Ok(())
+                                    },
+                                    Err(_e) => {
+                                        Ok(())
+                                    }
+                                }
+                            },
+                            Err(_e) => {
+                                Ok(())
+                            }
+                        }
+                    },
+                    _ => Ok({})
+                };
+
+                match res {
+                    ExecutionResult::Failed { .. } => None,
+                    ExecutionResult::Success { receipt, trace, .. } => Some(TxWithOutcome {
+                        tx: tx.clone(),
+                        receipt: receipt.clone(),
+                        exec_info: trace.clone(),
+                    }),
+                }
             })
             .collect::<Vec<TxWithOutcome>>();
 
@@ -399,10 +433,11 @@ impl<EF: ExecutorFactory> Stream for IntervalBlockProducer<EF> {
 
                 let transactions: Vec<ExecutableTxWithHash> =
                     std::mem::take(&mut pin.queued).into_iter().flatten().collect();
+                let backend = pin.backend.clone();
 
                 let fut = pin
                     .blocking_task_spawner
-                    .spawn(|| Self::execute_transactions(executor, transactions));
+                    .spawn(|| Self::execute_transactions(executor, transactions, backend));
 
                 pin.ongoing_execution = Some(Box::pin(fut));
             }

crates/katana/core/src/service/messaging/service.rs

@@ -213,6 +213,7 @@ impl<EF: ExecutorFactory> Stream for MessagingService<EF> {
                     pin.backend.clone(),
                     pin.gather_from_block,
                     pin.max_block,
+                    pin.chunk_size,
                 )));
             }
 

crates/katana/core/src/service/messaging/starknet.rs

@@ -204,10 +204,14 @@ impl Messenger for StarknetMessaging {
                     event = ?e,
                     "Converting event into L1HandlerTx."
                 );
-
-                if let Ok(tx) = l1_handler_tx_from_event(e, chain_id) {
-                    l1_handler_txs.push(tx)
-                }
+                block_events.iter().for_each(|e| {
+                    if let Ok(tx) = l1_handler_tx_from_event(e, chain_id) {
+                        let last_processed_nonce = self.provider.get_gather_message_nonce().unwrap_or(0.into());
+                        if tx.nonce > last_processed_nonce {
+                            l1_handler_txs.push(tx)
+                        }
+                    }
+                })
             });
 
         Ok((to_block, l1_handler_txs))
@@ -236,6 +240,10 @@ impl Messenger for StarknetMessaging {
         }
 
         self.send_hashes(hashes.clone()).await?;
+        for (index, hash) in hashes.iter().enumerate() {
+            self.send_hashes(std::slice::from_ref(hash)).await?;
+            self.provider.set_send_from_index(*hash, index as u64).await?;
+        }
 
         Ok(hashes)
     }

crates/katana/primitives/src/contract.rs

@@ -14,6 +14,9 @@ pub type StorageValue = Felt;
 /// Represents the type for a contract nonce.
 pub type Nonce = Felt;
 
+/// Represents the type for a message hash.
+pub type MessageHash = FieldElement;
+
 /// Represents a contract address.
 #[derive(Default, Clone, Copy, Eq, PartialEq, PartialOrd, Ord, Hash, Debug, Deref)]
 #[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]

crates/katana/storage/db/Cargo.toml

@@ -21,6 +21,8 @@ tempfile = { workspace = true, optional = true }
 thiserror.workspace = true
 tracing.workspace = true
 
+alloy-primitives.workspace = true
+
 # codecs
 [dependencies.postcard]
 default-features = false

crates/katana/storage/db/src/tables.rs

@@ -1,6 +1,6 @@
 use katana_primitives::block::{BlockHash, BlockNumber, FinalityStatus, Header};
 use katana_primitives::class::{ClassHash, CompiledClass, CompiledClassHash, FlattenedSierraClass};
-use katana_primitives::contract::{ContractAddress, GenericContractInfo, StorageKey};
+use katana_primitives::contract::{ContractAddress, GenericContractInfo, Nonce, StorageKey};
 use katana_primitives::receipt::Receipt;
 use katana_primitives::trace::TxExecInfo;
 use katana_primitives::transaction::{Tx, TxHash, TxNumber};
@@ -44,7 +44,7 @@ pub enum TableType {
     DupSort,
 }
 
-pub const NUM_TABLES: usize = 24;
+pub const NUM_TABLES: usize = 27;
 
 /// Macro to declare `libmdbx` tables.
 #[macro_export]
@@ -168,7 +168,10 @@ define_tables_enum! {[
     (ClassChangeHistory, TableType::DupSort),
     (StorageChangeHistory, TableType::DupSort),
     (StorageChangeSet, TableType::Table),
-    (MessagingInfo, TableType::Table)
+    (MessagingInfo, TableType::Table),
+    (MessagingNonceInfo, TableType::Table),
+    (MessagingMessageNonceMapping, TableType::Table),
+    (MessagingIndexInfo, TableType::Table),
 ]}
 
 tables! {
@@ -227,7 +230,17 @@ tables! {
     StorageChangeHistory: (BlockNumber, ContractStorageKey) => ContractStorageEntry,
 
     /// Stores the block number related to messaging service
-    MessagingInfo: (u64) => BlockNumber
+    MessagingInfo: (u64) => BlockNumber,
+
+    /// Stores the nonce related to messaging service
+    MessagingNonceInfo: (u64) => Nonce,
+
+    /// Map a message hash to a message nonce
+    MessagingMessageNonceMapping: (TxHash) => Nonce,
+
+    /// Stores the index of the messaging service
+    MessagingIndexInfo: (u64) => u64,
+
 }
 
 #[cfg(test)]
@@ -262,6 +275,9 @@ mod tests {
         assert_eq!(Tables::ALL[21].name(), StorageChangeHistory::NAME);
         assert_eq!(Tables::ALL[22].name(), StorageChangeSet::NAME);
         assert_eq!(Tables::ALL[23].name(), MessagingInfo::NAME);
+        assert_eq!(Tables::ALL[24].name(), MessagingNonceInfo::NAME);
+        assert_eq!(Tables::ALL[25].name(), MessagingMessageNonceMapping::NAME);
+        assert_eq!(Tables::ALL[26].name(), MessagingMessageNonceMapping::NAME);
 
         assert_eq!(Tables::Headers.table_type(), TableType::Table);
         assert_eq!(Tables::BlockHashes.table_type(), TableType::Table);
@@ -287,6 +303,9 @@ mod tests {
         assert_eq!(Tables::StorageChangeHistory.table_type(), TableType::DupSort);
         assert_eq!(Tables::StorageChangeSet.table_type(), TableType::Table);
         assert_eq!(Tables::MessagingInfo.table_type(), TableType::Table);
+        assert_eq!(Tables::MessagingNonceInfo.table_type(), TableType::Table);
+        assert_eq!(Tables::MessagingMessageNonMapping.table_type(), TableType::Table);
+        assert_eq!(Tables::MessagingIndexInfo.table_type(), TableType::Table);
     }
 
     use katana_primitives::block::{BlockHash, BlockNumber, FinalityStatus, Header};

crates/katana/storage/provider/src/lib.rs

@@ -6,7 +6,7 @@ use katana_primitives::block::{
     SealedBlockWithStatus,
 };
 use katana_primitives::class::{ClassHash, CompiledClass, CompiledClassHash, FlattenedSierraClass};
-use katana_primitives::contract::{ContractAddress, StorageKey, StorageValue};
+use katana_primitives::contract::{ContractAddress, MessageHash, Nonce, StorageKey, StorageValue};
 use katana_primitives::env::BlockEnv;
 use katana_primitives::receipt::Receipt;
 use katana_primitives::state::{StateUpdates, StateUpdatesWithDeclaredClasses};
@@ -401,4 +401,28 @@ where
     fn set_gather_from_block(&self, gather_from_block: BlockNumber) -> ProviderResult<()> {
         self.provider.set_gather_from_block(gather_from_block)
     }
+
+    fn get_gather_message_nonce(&self) -> ProviderResult<Option<Nonce>> {
+        self.provider.get_gather_message_nonce()
+    }
+
+    fn set_gather_message_nonce(&self, nonce: Nonce) -> ProviderResult<()> {
+        self.provider.set_gather_message_nonce(nonce)
+    }
+
+    fn get_nonce_from_message_hash(&self, message_hash: MessageHash) -> ProviderResult<Option<Nonce>> {
+        self.provider.get_nonce_from_message_hash(message_hash)
+    }
+
+    fn set_nonce_from_message_hash(&self, message_hash: MessageHash, nonce: Nonce) -> ProviderResult<()> {
+        self.provider.set_nonce_from_message_hash(message_hash, nonce)
+    }
+
+    fn get_send_from_index(&self) -> ProviderResult<Option<u64>> {
+        self.provider.get_send_from_index()
+    }
+
+    fn set_send_from_index(&self, _send_from_index: u64) -> ProviderResult<()> {
+        self.provider.get_send_from_index(send_from_index)
+    }
 }

crates/katana/storage/provider/src/providers/db/mod.rs

@@ -21,7 +21,7 @@ use katana_primitives::block::{
 };
 use katana_primitives::class::{ClassHash, CompiledClassHash};
 use katana_primitives::contract::{
-    ContractAddress, GenericContractInfo, Nonce, StorageKey, StorageValue,
+    ContractAddress, GenericContractInfo, MessageHash, Nonce, StorageKey, StorageValue,
 };
 use katana_primitives::env::BlockEnv;
 use katana_primitives::receipt::Receipt;
@@ -36,7 +36,7 @@ use crate::traits::block::{
     HeaderProvider,
 };
 use crate::traits::env::BlockEnvProvider;
-use crate::traits::messaging::{MessagingProvider, GATHER_FROM_BLOCK_KEY, SEND_FROM_BLOCK_KEY};
+use crate::traits::messaging::{MessagingProvider, GATHER_FROM_BLOCK_KEY, SEND_FROM_BLOCK_KEY, GATHER_FROM_NONCE_KEY, SEND_FROM_INDEX_KEY};
 use crate::traits::state::{StateFactoryProvider, StateProvider, StateRootProvider};
 use crate::traits::state_update::StateUpdateProvider;
 use crate::traits::transaction::{
@@ -791,6 +791,48 @@ impl MessagingProvider for DbProvider {
             Ok(())
         })?
     }
+
+    fn get_gather_message_nonce(&self) -> ProviderResult<Option<Nonce>> {
+        let db_tx = self.0.tx()?;
+        let nonce = db_tx.get::<tables::MessagingNonceInfo>(GATHER_FROM_NONCE_KEY)?;
+        db_tx.commit()?;
+        Ok(nonce)
+    }
+
+    fn set_gather_message_nonce(&self, nonce: Nonce) -> ProviderResult<()> {
+        self.0.update(|db_tx| {
+            db_tx.put::<tables::MessagingNonceInfo>(GATHER_FROM_NONCE_KEY, nonce)?;
+            Ok(())
+        })?
+    }
+
+    fn get_nonce_from_message_hash(&self, message_hash: MessageHash) -> ProviderResult<Option<Nonce>> {
+        let db_tx = self.0.tx()?;
+        let nonce = db_tx.get::<tables::MessagingMessageNonceMapping>(message_hash)?;
+        db_tx.commit()?;
+        Ok(nonce)
+    }
+
+    fn set_nonce_from_message_hash(&self, message_hash: MessageHash, nonce: Nonce) -> ProviderResult<()> {
+        self.0.update(|db_tx| {
+            db_tx.put::<tables::MessagingMessageNonceMapping>(message_hash, nonce)?;
+            Ok(())
+        })?
+    }
+
+    fn get_send_from_index(&self) -> ProviderResult<Option<u64>> {
+        let db_tx = self.0.tx()?;
+        let index = db_tx.get::<tables::MessagingIndexInfo>(SEND_FROM_INDEX_KEY)?;
+        db_tx.commit()?;
+        Ok(index)
+    }
+
+    fn set_send_from_index(&self, send_from_index: u64) -> ProviderResult<()> {
+        self.0.update(|db_tx| {
+            db_tx.put::<tables::MessagingIndexInfo>(SEND_FROM_INDEX_KEY, send_from_index)?;
+            Ok(())
+        })?
+    }
 }
 
 #[cfg(test)]

crates/katana/storage/provider/src/providers/fork/mod.rs

@@ -10,7 +10,7 @@ use katana_primitives::block::{
     SealedBlockWithStatus,
 };
 use katana_primitives::class::{ClassHash, CompiledClass, CompiledClassHash, FlattenedSierraClass};
-use katana_primitives::contract::ContractAddress;
+use katana_primitives::contract::{ContractAddress, MessageHash, Nonce};
 use katana_primitives::env::BlockEnv;
 use katana_primitives::receipt::Receipt;
 use katana_primitives::state::{StateUpdates, StateUpdatesWithDeclaredClasses};
@@ -599,4 +599,29 @@ impl MessagingProvider for ForkedProvider {
     fn set_gather_from_block(&self, _gather_from_block: BlockNumber) -> ProviderResult<()> {
         Ok(())
     }
+
+    fn get_gather_message_nonce(&self) -> ProviderResult<Option<Nonce>> {
+        Ok(None)
+    }
+
+    fn set_gather_message_nonce(&self, _nonce: Nonce) -> ProviderResult<()> {
+        Ok(())
+    }
+
+    fn get_nonce_from_message_hash(&self, _message_hash: MessageHash) -> ProviderResult<Option<Nonce>> {
+        Ok(None)
+    }
+
+    fn set_nonce_from_message_hash(&self, _message_hash: MessageHash, _nonce: Nonce) -> ProviderResult<()> {
+        Ok(())
+    }
+
+    fn get_send_from_index(&self) -> ProviderResult<Option<u64>> {
+        Ok(None)
+    }
+
+    fn set_send_from_index(&self, _send_from_index: u64) -> ProviderResult<()> {
+        Ok(())
+    }
+
 }

crates/katana/storage/provider/src/providers/in_memory/cache.rs

@@ -3,6 +3,7 @@ use std::sync::Arc;
 
 use katana_db::models::block::StoredBlockBodyIndices;
 use katana_primitives::block::{BlockHash, BlockNumber, FinalityStatus, Header};
+use katana_primitives::contract::Nonce;
 use katana_primitives::class::{ClassHash, CompiledClass, CompiledClassHash, FlattenedSierraClass};
 use katana_primitives::contract::{ContractAddress, GenericContractInfo, StorageKey, StorageValue};
 use katana_primitives::receipt::Receipt;
@@ -89,6 +90,9 @@ pub struct CacheDb<Db> {
     pub(crate) transaction_numbers: HashMap<TxHash, TxNumber>,
     pub(crate) transaction_block: HashMap<TxNumber, BlockNumber>,
     pub(crate) messaging_info: HashMap<u64, BlockNumber>,
+    pub(crate) messaging_nonce_info: HashMap<u64, Nonce>,
+    pub(crate) messaging_message_nonce_mapping: HashMap<Nonce, Nonce>,
+    pub(crate) messaging_index_info: HashMap<u64, u64>,
 }
 
 impl<Db> CacheStateDb<Db> {
@@ -122,6 +126,9 @@ impl<Db> CacheDb<Db> {
             latest_block_hash: Default::default(),
             latest_block_number: Default::default(),
             messaging_info: Default::default(),
+            messaging_nonce_info: Default::default(),
+            messaging_message_nonce_mapping: Default::default(),
+            messaging_index_info: Default::default(),
         }
     }
 }