fault_proving(global_roots): service definition and initial tests

CHANGELOG.md

@@ -14,6 +14,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 - [2150](https://github.com/FuelLabs/fuel-core/pull/2150): Upgraded `libp2p` to `0.54.1` and introduced `ConnectionLimiter` to limit pending incoming/outgoing connections.
 - [2491](https://github.com/FuelLabs/fuel-core/pull/2491): Storage read replays of historical blocks for execution tracing. Only available behind `--historical-execution` flag.
 - [2666](https://github.com/FuelLabs/fuel-core/pull/2666): Added two new CLI arguments to control the GraphQL queries consistency: `--graphql-block-height-tolerance` (default: `10`) and `--graphql-block-height-min-timeout` (default: `30s`). If a request requires a specific block height and the node is slightly behind, it will wait instead of failing.
+- [2722](https://github.com/FuelLabs/fuel-core/pull/2722): Service definition for state root service.
 
 ### Fixed
 - [2646](https://github.com/FuelLabs/fuel-core/pull/2646): Improved performance of fetching block height by caching it when the view is created.

Cargo.toml

@@ -10,11 +10,12 @@ members = [
   "crates/client",
   "crates/compression",
   "crates/database",
-  "crates/fraud_proofs/global_merkle_root/storage",
   "crates/fuel-core",
   "crates/fuel-gas-price-algorithm",
   "crates/keygen",
   "crates/metrics",
+  "crates/proof_system/global_merkle_root/service",
+  "crates/proof_system/global_merkle_root/storage",
   "crates/services",
   "crates/services/consensus_module",
   "crates/services/consensus_module/bft",
@@ -96,6 +97,10 @@ fuel-core-wasm-executor = { version = "0.41.7", path = "./crates/services/upgrad
 fuel-core-xtask = { version = "0.0.0", path = "./xtask" }
 fuel-gas-price-algorithm = { version = "0.41.7", path = "crates/fuel-gas-price-algorithm" }
 
+# Proof system
+fuel-core-global-merkle-root-storage = { version = "0.41.6", path = "./crates/proof_system/global_merkle_root/storage" }
+fuel-core-global-merkle-root-service = { version = "0.41.6", path = "./crates/proof_system/global_merkle_root/service" }
+
 # Fuel dependencies
 fuel-vm-private = { version = "0.59.2", package = "fuel-vm", default-features = false }
 

crates/proof_system/global_merkle_root/service/Cargo.toml

@@ -0,0 +1,44 @@
+[package]
+authors = { workspace = true }
+categories = ["cryptography::cryptocurrencies"]
+description = "Service definition for the global merkle root service."
+edition = { workspace = true }
+homepage = { workspace = true }
+keywords = ["blockchain", "cryptocurrencies", "fuel-client", "fuel-core"]
+license = { workspace = true }
+name = "fuel-core-global-merkle-root-service"
+repository = { workspace = true }
+version = { workspace = true }
+
+[dependencies]
+anyhow = { workspace = true }
+async-trait = { workspace = true }
+fuel-core-global-merkle-root-storage = { workspace = true }
+fuel-core-services = { workspace = true }
+fuel-core-storage = { workspace = true, features = ["alloc"] }
+fuel-core-types = { workspace = true, default-features = false, features = [
+  "serde",
+  "alloc",
+] }
+tokio = { workspace = true }
+tracing = { workspace = true }
+
+[dev-dependencies]
+derive_more = { workspace = true }
+fuel-core-global-merkle-root-storage = { workspace = true, features = [
+  "test-helpers",
+] }
+fuel-core-storage = { workspace = true, features = ["alloc", "test-helpers"] }
+fuel-core-types = { workspace = true, default-features = false, features = [
+  "serde",
+  "random",
+  "test-helpers",
+] }
+rand = { workspace = true }
+
+[features]
+fault-proving = [
+  "fuel-core-types/fault-proving",
+  "fuel-core-storage/fault-proving",
+  "fuel-core-global-merkle-root-storage/fault-proving",
+]

crates/proof_system/global_merkle_root/service/src/lib.rs

@@ -0,0 +1,16 @@
+//! The state root service
+
+#![deny(clippy::arithmetic_side_effects)]
+#![deny(clippy::cast_possible_truncation)]
+#![deny(unused_crate_dependencies)]
+#![deny(missing_docs)]
+#![deny(warnings)]
+
+/// Port definitions
+pub mod ports;
+
+/// Service definition
+pub mod service;
+
+#[cfg(test)]
+mod tests;

crates/proof_system/global_merkle_root/service/src/ports.rs

@@ -0,0 +1,22 @@
+use std::future::Future;
+
+use fuel_core_global_merkle_root_storage::column::Column;
+use fuel_core_storage::{
+    kv_store::KeyValueInspect,
+    transactional::Modifiable,
+};
+use fuel_core_types::blockchain::block::Block;
+
+/// A stream of blocks
+pub trait BlockStream {
+    /// Error type
+    type Error;
+
+    /// Get the next block
+    fn next(&mut self) -> impl Future<Output = Result<Block, Self::Error>> + Send;
+}
+
+/// The storage requirements for the merkle root service
+pub trait ServiceStorage: KeyValueInspect<Column = Column> + Modifiable {}
+
+impl<T> ServiceStorage for T where T: KeyValueInspect<Column = Column> + Modifiable {}

crates/proof_system/global_merkle_root/service/src/service.rs

@@ -0,0 +1,138 @@
+use std::ops::Deref;
+
+use fuel_core_global_merkle_root_storage::update::UpdateMerkleizedTables;
+use fuel_core_services::{
+    RunnableService,
+    RunnableTask,
+    ServiceRunner,
+    StateWatcher,
+    TaskNextAction,
+};
+use fuel_core_storage::transactional::WriteTransaction;
+use fuel_core_types::fuel_types::ChainId;
+
+use crate::ports::{
+    BlockStream,
+    ServiceStorage,
+};
+
+/// Service definition as a thin wrapper over `ServiceRunner`.
+pub struct Service<B, S>(ServiceRunner<UpdateMerkleRootTask<B, S>>)
+where
+    B: BlockStream + Send + 'static,
+    B::Error: std::error::Error + Send + Sync + 'static,
+    S: ServiceStorage + Send + 'static;
+
+impl<B, S> Service<B, S>
+where
+    B: BlockStream + Send + 'static,
+    B::Error: std::error::Error + Send + Sync + 'static,
+    S: ServiceStorage + Send + 'static,
+{
+    /// Construct a new service.
+    pub fn new(chain_id: ChainId, storage: S, blocks: B) -> Self {
+        Self(ServiceRunner::new(UpdateMerkleRootTask::new(
+            chain_id, storage, blocks,
+        )))
+    }
+}
+
+impl<B, S> Deref for Service<B, S>
+where
+    B: BlockStream + Send + 'static,
+    B::Error: std::error::Error + Send + Sync + 'static,
+    S: ServiceStorage + Send + 'static,
+{
+    type Target = ServiceRunner<UpdateMerkleRootTask<B, S>>;
+
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}
+
+/// The inner task definition. Holds the state of the service.
+pub struct UpdateMerkleRootTask<BlockStream, Storage> {
+    chain_id: ChainId,
+    storage: Storage,
+    blocks: BlockStream,
+}
+
+impl<B, S> UpdateMerkleRootTask<B, S> {
+    /// Create a new task
+    pub fn new(chain_id: ChainId, storage: S, blocks: B) -> Self {
+        Self {
+            chain_id,
+            storage,
+            blocks,
+        }
+    }
+}
+
+#[async_trait::async_trait]
+impl<B, S> RunnableService for UpdateMerkleRootTask<B, S>
+where
+    B: BlockStream + Send,
+    B::Error: std::error::Error + Send + Sync + 'static,
+    S: ServiceStorage + Send,
+{
+    const NAME: &'static str = "MerkleRootService";
+
+    type SharedData = ();
+
+    type Task = Self;
+
+    type TaskParams = ();
+
+    fn shared_data(&self) -> Self::SharedData {}
+
+    async fn into_task(
+        self,
+        _state_watcher: &StateWatcher,
+        _params: Self::TaskParams,
+    ) -> anyhow::Result<Self::Task> {
+        Ok(self)
+    }
+}
+
+impl<B, S> RunnableTask for UpdateMerkleRootTask<B, S>
+where
+    B: BlockStream + Send,
+    B::Error: std::error::Error + Send + Sync + 'static,
+    S: ServiceStorage + Send,
+{
+    #[tracing::instrument(skip(self, watcher))]
+    async fn run(&mut self, watcher: &mut StateWatcher) -> TaskNextAction {
+        tokio::select! {
+            biased;
+            _ = watcher.while_started() => {
+                TaskNextAction::Stop
+            }
+            _ = self.process_next_block() => {
+                TaskNextAction::Continue
+            }
+        }
+    }
+
+    async fn shutdown(self) -> anyhow::Result<()> {
+        Ok(())
+    }
+}
+
+impl<B, S> UpdateMerkleRootTask<B, S>
+where
+    B: BlockStream,
+    B::Error: std::error::Error + Send + Sync + 'static,
+    S: ServiceStorage,
+{
+    #[tracing::instrument(skip(self))]
+    async fn process_next_block(&mut self) -> anyhow::Result<()> {
+        let block = self.blocks.next().await?;
+        let mut tx = self.storage.write_transaction();
+
+        tx.update_merkleized_tables(self.chain_id, &block)?;
+
+        tx.commit()?;
+
+        Ok(())
+    }
+}