Merge branch 'master' into 2657-scalars-HexString-cow

grandima · web-flow · commit c203c6ba916c · 2025-06-19T13:37:16.000+02:00
diff --git a/.changes/added/2959.md b/.changes/added/2959.md
@@ -0,0 +1 @@
+Add `Registrations` table to the compression database. This is merkleized.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/README.md b/README.md
@@ -15,7 +15,7 @@ Fuel client implementation.
 |----------|---------|
 | Fuel Ignition | 0.43.2 |
 | Testnet | 0.43.2 |
-| Devnet | 0.43.2 |
+| Devnet | 0.44.0 |
 
 ## Contributing
 
diff --git a/crates/services/compression/src/errors.rs b/crates/services/compression/src/errors.rs
@@ -12,6 +12,9 @@ pub enum CompressionError {
     /// Failed to read compressed block from storage
     #[error("failed to write compressed block to storage: `{0}`")]
     FailedToWriteCompressedBlock(StorageError),
+    /// Failed to write registrations to storage
+    #[error("failed to write registrations to storage: `{0}`")]
+    FailedToWriteRegistrations(StorageError),
     /// Failed to get the size of compressed block
     #[error("failed to get size of compressed block: `{0}`")]
     FailedToGetCompressedBlockSize(StorageError),
diff --git a/crates/services/compression/src/ports/compression_storage.rs b/crates/services/compression/src/ports/compression_storage.rs
@@ -1,7 +1,9 @@
 use crate::{
     errors::CompressionError,
-    storage,
-    storage::CompressedBlocks,
+    storage::{
+        self,
+        CompressedBlocks,
+    },
 };
 use fuel_core_storage::{
     self,
@@ -65,6 +67,16 @@ where
             .insert(&height, compressed_block)
             .map_err(CompressionError::FailedToWriteCompressedBlock)?;
 
+        #[cfg(feature = "fault-proving")]
+        self.storage_as_mut::<crate::storage::registrations::Registrations>()
+            .insert(
+                &height,
+                fuel_core_compression::VersionedBlockPayload::registrations(
+                    compressed_block,
+                ),
+            )
+            .map_err(CompressionError::FailedToWriteRegistrations)?;
+
         // this should not hit the db, we get it from the transaction
         let size = StorageSize::<CompressedBlocks>::size_of_value(self, &height)
             .map_err(CompressionError::FailedToGetCompressedBlockSize)?;
diff --git a/crates/services/compression/src/service.rs b/crates/services/compression/src/service.rs
@@ -763,4 +763,43 @@ mod tests {
         assert!(maybe_block.is_some());
         service.shutdown().await.unwrap();
     }
+
+    #[cfg(feature = "fault-proving")]
+    #[tokio::test]
+    async fn compression_service__writes_to_registrations_table() {
+        // given: we create the compression service
+        let block_with_metadata = BlockWithMetadata::default();
+        let block_source = MockBlockSource::new(vec![block_with_metadata]);
+        let storage = test_storage();
+        let config_provider = MockConfigProvider::default();
+        let canonical_height_provider = MockCanonicalHeightProvider::default();
+
+        let uninit_service = UninitializedCompressionService::new(
+            block_source,
+            storage,
+            config_provider.config(),
+            canonical_height_provider,
+        );
+        let sync_observer = uninit_service.shared_data();
+
+        // when: the compression service is started
+        let mut service = uninit_service
+            .into_task(&Default::default(), ())
+            .await
+            .unwrap();
+        let _ = service.run(&mut StateWatcher::started()).await;
+
+        // then: we ensure we can get the registrations per block + root of that table
+        let target_block_height = 0;
+        sync_observer
+            .await_synced_until(&target_block_height)
+            .await
+            .unwrap();
+        let maybe_registrations = service
+            .storage
+            .storage_as_ref::<storage::registrations::Registrations>()
+            .get(&target_block_height.into())
+            .unwrap();
+        assert!(maybe_registrations.is_some());
+    }
 }
diff --git a/crates/services/compression/src/storage.rs b/crates/services/compression/src/storage.rs
@@ -20,6 +20,8 @@ pub mod predicate_code;
 pub mod registry_index;
 pub mod script_code;
 // TODO: https://github.com/FuelLabs/fuel-core/issues/2842
+#[cfg(feature = "fault-proving")]
+pub mod registrations;
 pub mod timestamps;
 
 /// Merkleized Address table type alias
@@ -48,3 +50,7 @@ pub type Timestamps = Merkleized<timestamps::Timestamps>;
 
 /// Re-export to match api
 pub use compressed_blocks::CompressedBlocks;
+
+/// Merkleized Registrations table type alias
+#[cfg(feature = "fault-proving")]
+pub type Registrations = Merkleized<registrations::Registrations>;
diff --git a/crates/services/compression/src/storage/column.rs b/crates/services/compression/src/storage/column.rs
@@ -40,6 +40,9 @@ pub enum CompressionColumn {
     EvictorCache = 7,
     /// Keeps track of timestamps, will be removed eventually, see [`Timestamps`](crate::storage::timestamps::Timestamps)
     Timestamps = 8,
+    #[cfg(feature = "fault-proving")]
+    /// Keeps track of registrations per table associated with a compressed block, see [`Registrations`](crate::storage::registrations::Registrations)
+    Registrations = 9,
 }
 
 impl AsU32 for CompressionColumn {
diff --git a/crates/services/compression/src/storage/registrations.rs b/crates/services/compression/src/storage/registrations.rs
@@ -0,0 +1,54 @@
+//! Registrations table.
+//! Maps from block height -> registrations made for compressing that block
+
+use fuel_core_storage::{
+    Mappable,
+    blueprint::plain::Plain,
+    codec::postcard::Postcard,
+    merkle::sparse::MerkleizedTableColumn,
+    structured_storage::TableWithBlueprint,
+};
+
+use super::column::{
+    CompressionColumn,
+    MerkleizedColumnOf,
+};
+
+/// Table that indexes the registrations.
+pub struct Registrations;
+
+impl MerkleizedTableColumn for Registrations {
+    type TableColumn = CompressionColumn;
+
+    fn table_column() -> Self::TableColumn {
+        Self::TableColumn::Registrations
+    }
+}
+
+impl Mappable for Registrations {
+    type Key = Self::OwnedKey;
+    type OwnedKey = fuel_core_types::fuel_types::BlockHeight;
+    type Value = Self::OwnedValue;
+    type OwnedValue = fuel_core_compression::registry::RegistrationsPerTable;
+}
+
+impl TableWithBlueprint for Registrations {
+    type Blueprint = Plain<Postcard, Postcard>;
+    type Column = MerkleizedColumnOf<Self>;
+
+    fn column() -> Self::Column {
+        Self::Column::TableColumn(Self::table_column())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fuel_core_storage::basic_storage_tests!(
+        Registrations,
+        <Registrations as Mappable>::Key::default(),
+        <Registrations as Mappable>::Value::default(),
+        <Registrations as Mappable>::Value::default()
+    );
+}
diff --git a/crates/services/parallel-executor/Cargo.toml b/crates/services/parallel-executor/Cargo.toml
@@ -14,7 +14,15 @@ description = "Fuel Block Parallel Executor"
 wasm-executor = ["fuel-core-upgradable-executor/wasm-executor"]
 
 [dependencies]
+fuel-core-executor = { workspace = true, features = ["std"] }
 fuel-core-storage = { workspace = true, features = ["std"] }
 fuel-core-types = { workspace = true, features = ["std"] }
 fuel-core-upgradable-executor = { workspace = true, features = ["std"] }
-tokio = { workspace = true, features = ["rt-multi-thread"] }
+futures = { workspace = true, features = ["std"] }
+tokio = { workspace = true, features = ["rt-multi-thread", "sync", "macros"] }
+
+[dev-dependencies]
+anyhow = { workspace = true }
+fuel-core-storage = { workspace = true, features = ["test-helpers"] }
+fuel-core-types = { workspace = true, features = ["test-helpers", "serde"] }
+rand = { workspace = true }
diff --git a/crates/services/parallel-executor/src/executor.rs b/crates/services/parallel-executor/src/executor.rs
@@ -1,4 +1,7 @@
-use crate::config::Config;
+use crate::{
+    config::Config,
+    ports::TransactionsSource,
+};
 use fuel_core_storage::transactional::Changes;
 use fuel_core_types::{
     blockchain::block::Block,
@@ -14,10 +17,7 @@ use fuel_core_types::{
         },
     },
 };
-use fuel_core_upgradable_executor::{
-    executor::Executor as UpgradableExecutor,
-    native_executor::ports::TransactionsSource,
-};
+use fuel_core_upgradable_executor::executor::Executor as UpgradableExecutor;
 use std::{
     num::NonZeroUsize,
     sync::{
@@ -77,7 +77,7 @@ impl<S, R> Executor<S, R> {
 
 impl<S, R> Executor<S, R> {
     /// Produces the block and returns the result of the execution without committing the changes.
-    pub fn produce_without_commit_with_source<TxSource>(
+    pub async fn produce_without_commit_with_source<TxSource>(
         &self,
         _components: Components<TxSource>,
     ) -> ExecutorResult<Uncommitted<ExecutionResult, Changes>>
diff --git a/crates/services/parallel-executor/src/lib.rs b/crates/services/parallel-executor/src/lib.rs
@@ -1,2 +1,7 @@
 pub mod config;
 pub mod executor;
+pub mod once_transaction_source;
+pub mod ports;
+
+#[cfg(test)]
+mod tests;
diff --git a/crates/services/parallel-executor/src/once_transaction_source.rs b/crates/services/parallel-executor/src/once_transaction_source.rs
@@ -0,0 +1,76 @@
+use std::sync::Mutex;
+
+use fuel_core_executor::ports::{
+    MaybeCheckedTransaction,
+    TransactionsSource as ExecutorTransactionsSource,
+};
+use fuel_core_types::fuel_vm::checked_transaction::CheckedTransaction;
+
+use crate::ports::{
+    TransactionFiltered,
+    TransactionSourceExecutableTransactions,
+    TransactionsSource,
+};
+
+pub struct OnceTransactionsSource {
+    transactions: Mutex<Vec<CheckedTransaction>>,
+    consensus_parameters_version: u32,
+}
+
+impl OnceTransactionsSource {
+    pub fn new(
+        transactions: Vec<CheckedTransaction>,
+        consensus_parameters_version: u32,
+    ) -> Self {
+        Self {
+            transactions: Mutex::new(transactions),
+            consensus_parameters_version,
+        }
+    }
+}
+
+impl ExecutorTransactionsSource for OnceTransactionsSource {
+    fn next(
+        &self,
+        _gas_limit: u64,
+        transactions_limit: u16,
+        _block_transaction_size_limit: u32,
+    ) -> Vec<fuel_core_executor::ports::MaybeCheckedTransaction> {
+        let mut transactions = self.transactions.lock().expect("Mutex poisoned");
+        // Avoid panicking if we request more transactions than there are in the vector
+        let transactions_limit = (transactions_limit as usize).min(transactions.len());
+        transactions
+            .drain(..transactions_limit)
+            .map(|tx| {
+                MaybeCheckedTransaction::CheckedTransaction(
+                    tx,
+                    self.consensus_parameters_version,
+                )
+            })
+            .collect()
+    }
+}
+
+impl TransactionsSource for OnceTransactionsSource {
+    fn get_executable_transactions(
+        &mut self,
+        _gas_limit: u64,
+        tx_count_limit: u16,
+        _block_transaction_size_limit: u32,
+        filter: crate::ports::Filter,
+    ) -> TransactionSourceExecutableTransactions {
+        let mut transactions = self.transactions.lock().expect("Mutex poisoned");
+        // Avoid panicking if we request more transactions than there are in the vector
+        let transactions_limit = (tx_count_limit as usize).min(transactions.len());
+        let txs = transactions.drain(..transactions_limit).collect();
+        TransactionSourceExecutableTransactions {
+            transactions: txs,
+            filtered: TransactionFiltered::NotFiltered,
+            filter,
+        }
+    }
+    fn get_new_transactions_notifier(&mut self) -> tokio::sync::Notify {
+        // This is a one-time source, so we don't need to notify about new transactions
+        tokio::sync::Notify::new()
+    }
+}
diff --git a/crates/services/parallel-executor/src/ports.rs b/crates/services/parallel-executor/src/ports.rs
@@ -0,0 +1,76 @@
+use std::collections::HashSet;
+
+use fuel_core_storage::Result as StorageResult;
+use fuel_core_types::{
+    blockchain::primitives::DaBlockHeight,
+    entities::coins::coin::CompressedCoin,
+    fuel_tx::{
+        ConsensusParameters,
+        ContractId,
+        UtxoId,
+    },
+    fuel_types::BlockHeight,
+    fuel_vm::checked_transaction::CheckedTransaction,
+};
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum TransactionFiltered {
+    /// Some transactions were filtered out and so could be fetched in the future
+    Filtered,
+    /// No transactions were filtered out
+    NotFiltered,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct Filter {
+    /// The set of contract IDs to filter out
+    pub excluded_contract_ids: HashSet<ContractId>,
+}
+
+impl Filter {
+    pub fn new(excluded_contract_ids: HashSet<ContractId>) -> Self {
+        Self {
+            excluded_contract_ids,
+        }
+    }
+}
+
+pub struct TransactionSourceExecutableTransactions {
+    /// The transactions that can be executed
+    pub transactions: Vec<CheckedTransaction>,
+    /// Indicates whether some transactions were filtered out based on the filter
+    pub filtered: TransactionFiltered,
+    /// The filter used to fetch these transactions
+    pub filter: Filter,
+}
+
+pub trait TransactionsSource {
+    /// Returns the a batch of transactions to satisfy the given parameters
+    fn get_executable_transactions(
+        &mut self,
+        gas_limit: u64,
+        tx_count_limit: u16,
+        block_transaction_size_limit: u32,
+        filter: Filter,
+    ) -> TransactionSourceExecutableTransactions;
+
+    /// Returns a notification receiver for new transactions
+    fn get_new_transactions_notifier(&mut self) -> tokio::sync::Notify;
+}
+
+pub trait Storage {
+    /// Get a coin by a UTXO
+    fn get_coin(&self, utxo: &UtxoId) -> StorageResult<Option<CompressedCoin>>;
+
+    /// Get the DA block height based on provided height
+    fn get_da_height_by_l2_height(
+        &self,
+        block_height: &BlockHeight,
+    ) -> StorageResult<Option<DaBlockHeight>>;
+
+    /// Get consensus parameters based on a version
+    fn get_consensus_parameters(
+        &self,
+        consensus_parameters_version: u32,
+    ) -> StorageResult<ConsensusParameters>;
+}
diff --git a/crates/services/parallel-executor/src/tests/mocks.rs b/crates/services/parallel-executor/src/tests/mocks.rs
diff --git a/crates/services/parallel-executor/src/tests/mod.rs b/crates/services/parallel-executor/src/tests/mod.rs
diff --git a/crates/services/parallel-executor/src/tests/tests_executor.rs b/crates/services/parallel-executor/src/tests/tests_executor.rs

Original file line number	Diff line number	Diff line change
`@@ -0,0 +1 @@`
	`1`	+Add `Registrations` table to the compression database. This is merkleized.
Original file line number	Diff line number	Diff line change
`@@ -40,6 +40,9 @@ pub enum CompressionColumn {`
`40`	`40`	`EvictorCache = 7,`
`41`	`41`	/// Keeps track of timestamps, will be removed eventually, see [`Timestamps`](crate::storage::timestamps::Timestamps)
`42`	`42`	`Timestamps = 8,`
	`43`	`+ #[cfg(feature = "fault-proving")]`
	`44`	+ /// Keeps track of registrations per table associated with a compressed block, see [`Registrations`](crate::storage::registrations::Registrations)
	`45`	`+ Registrations = 9,`
`43`	`46`	`}`
`44`	`47`
`45`	`48`	`impl AsU32 for CompressionColumn {`