DA compression (#1609)

Related #1605. VM PR
FuelLabs/fuel-vm#670.

This PR adds DA compression crate for Fuel blocks, performed upon block
creation. The compressed blocks are stored into the offchain database
and can be fetched using the GraphQL API.

## Note for reviewers

To keep this reasonably compact, decompression support is not included
in this PR, and will be done as a follow-up. As a result, the full data
roundtrip testing is not part of this PR. There's no proof here that
compression of full blocks is reversible.

## TODO
#### Features
- [x] Temporal registry db support
- [x] Optimize temporal registry eviction implementation
- [x] Implement TxId ↔ TxPointer lookups
- [x] Integrate with the block committer (GraphQL interface, probably)
#### Tests
- [x] compressed blocks are available from non-block-producer nodes
- [ ] e2e test for the full decompression cycle (moved to a follow-up)


## Follow-up issues
- Sync the node from L1:
#2208
- Decompression roudntrip tests
#2238
- Figure out which cache eviction algorithm/behavior is wanted:
#2231
- Figure out if we need to remove the compressed blocks from the db
after a while
- Merkle roots for fraud proofs
#2232

---------

Co-authored-by: Aaryamann Challani <43716372+rymnc@users.noreply.github.com>
Co-authored-by: Green Baneling <XgreenX9999@gmail.com>
Co-authored-by: Rafał Chabowski <88321181+rafal-ch@users.noreply.github.com>

CHANGELOG.md

@@ -7,6 +7,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 ## [Unreleased]
 
 ### Added
+- [1609](https://github.com/FuelLabs/fuel-core/pull/1609): Add DA compression support. Compressed blocks are stored in the offchain database when blocks are produced, and can be fetched using the GraphQL API.
 - [2290](https://github.com/FuelLabs/fuel-core/pull/2290): Added a new CLI argument `--graphql-max-directives`. The default value is `10`.
 - [2195](https://github.com/FuelLabs/fuel-core/pull/2195): Added enforcement of the limit on the size of the L2 transactions per block according to the `block_transaction_size_limit` parameter.
 - [2131](https://github.com/FuelLabs/fuel-core/pull/2131): Add flow in TxPool in order to ask to newly connected peers to share their transaction pool

Cargo.toml

@@ -8,6 +8,7 @@ members = [
   "bin/keygen",
   "crates/chain-config",
   "crates/client",
+  "crates/compression",
   "crates/database",
   "crates/fuel-core",
   "crates/fuel-gas-price-algorithm",
@@ -62,6 +63,7 @@ fuel-core-keygen = { version = "0.36.0", path = "./crates/keygen" }
 fuel-core-keygen-bin = { version = "0.36.0", path = "./bin/keygen" }
 fuel-core-chain-config = { version = "0.36.0", path = "./crates/chain-config", default-features = false }
 fuel-core-client = { version = "0.36.0", path = "./crates/client" }
+fuel-core-compression = { version = "0.36.0", path = "./crates/compression" }
 fuel-core-database = { version = "0.36.0", path = "./crates/database" }
 fuel-core-metrics = { version = "0.36.0", path = "./crates/metrics" }
 fuel-core-services = { version = "0.36.0", path = "./crates/services" }
@@ -131,6 +133,7 @@ test-strategy = "0.3"
 parquet = { version = "49.0", default-features = false }
 rayon = "1.10.0"
 bytes = "1.5.0"
+paste = "1.0"
 pretty_assertions = "1.4.0"
 proptest = "1.1"
 pin-project-lite = "0.2"

bin/fuel-core/Cargo.toml

@@ -27,6 +27,7 @@ dirs = "4.0"
 dotenvy = { version = "0.15", optional = true }
 fuel-core = { workspace = true, features = ["wasm-executor"] }
 fuel-core-chain-config = { workspace = true }
+fuel-core-compression = { workspace = true }
 fuel-core-poa = { workspace = true }
 fuel-core-types = { workspace = true, features = ["std"] }
 hex = { workspace = true }

bin/fuel-core/src/cli/run.rs

@@ -20,7 +20,10 @@ use fuel_core::{
         CombinedDatabase,
         CombinedDatabaseConfig,
     },
-    fuel_core_graphql_api::ServiceConfig as GraphQLConfig,
+    fuel_core_graphql_api::{
+        worker_service::DaCompressionConfig,
+        ServiceConfig as GraphQLConfig,
+    },
     producer::Config as ProducerConfig,
     service::{
         config::Trigger,
@@ -190,6 +193,11 @@ pub struct Command {
     #[cfg(feature = "aws-kms")]
     pub consensus_aws_kms: Option<String>,
 
+    /// If given, the node will produce and store da-compressed blocks
+    /// with the given retention time.
+    #[arg(long = "da-compression", env)]
+    pub da_compression: Option<humantime::Duration>,
+
     /// A new block is produced instantly when transactions are available.
     #[clap(flatten)]
     pub poa_trigger: PoATriggerArgs,
@@ -272,6 +280,7 @@ impl Command {
             consensus_key,
             #[cfg(feature = "aws-kms")]
             consensus_aws_kms,
+            da_compression,
             poa_trigger,
             predefined_blocks_path,
             coinbase_recipient,
@@ -418,6 +427,15 @@ impl Command {
         let block_importer =
             fuel_core::service::config::fuel_core_importer::Config::new();
 
+        let da_compression = match da_compression {
+            Some(retention) => {
+                DaCompressionConfig::Enabled(fuel_core_compression::Config {
+                    temporal_registry_retention: retention.into(),
+                })
+            }
+            None => DaCompressionConfig::Disabled,
+        };
+
         let TxPoolArgs {
             tx_pool_ttl,
             tx_max_number,
@@ -476,6 +494,7 @@ impl Command {
             min_gas_price,
             gas_price_threshold_percent,
             block_importer,
+            da_compression,
             #[cfg(feature = "relayer")]
             relayer: relayer_cfg,
             #[cfg(feature = "p2p")]

crates/client/assets/schema.sdl

@@ -284,6 +284,10 @@ enum ContractParametersVersion {
 	V1
 }
 
+type DaCompressedBlock {
+	bytes: HexString!
+}
+
 union DependentCost = LightOperation | HeavyOperation
 
 type DryRunFailureStatus {
@@ -942,6 +946,12 @@ type Query {
 		"""
 		excludedIds: ExcludeInput
 	): [[CoinType!]!]!
+	daCompressedBlock(
+		"""
+		Height of the block
+		"""
+		height: U32!
+	): DaCompressedBlock
 	contract(
 		"""
 		ID of the Contract

crates/client/src/client.rs

@@ -81,6 +81,7 @@ use schema::{
     block::BlockByIdArgs,
     coins::CoinByIdArgs,
     contract::ContractByIdArgs,
+    da_compressed::DaCompressedBlockByHeightArgs,
     tx::{
         TxArg,
         TxIdArgs,
@@ -880,6 +881,23 @@ impl FuelClient {
         Ok(block)
     }
 
+    pub async fn da_compressed_block(
+        &self,
+        height: BlockHeight,
+    ) -> io::Result<Option<Vec<u8>>> {
+        let query = schema::da_compressed::DaCompressedBlockByHeightQuery::build(
+            DaCompressedBlockByHeightArgs {
+                height: U32(height.into()),
+            },
+        );
+
+        Ok(self
+            .query(query)
+            .await?
+            .da_compressed_block
+            .map(|b| b.bytes.into()))
+    }
+
     /// Retrieve a blob by its ID
     pub async fn blob(&self, id: BlobId) -> io::Result<Option<types::Blob>> {
         let query = schema::blob::BlobByIdQuery::build(BlobByIdArgs { id: id.into() });

crates/client/src/client/schema.rs

@@ -32,6 +32,7 @@ pub mod block;
 pub mod chain;
 pub mod coins;
 pub mod contract;
+pub mod da_compressed;
 pub mod message;
 pub mod node_info;
 pub mod upgrades;

crates/client/src/client/schema/da_compressed.rs

@@ -0,0 +1,44 @@
+use crate::client::schema::{
+    schema,
+    U32,
+};
+
+use super::HexString;
+
+#[derive(cynic::QueryVariables, Debug)]
+pub struct DaCompressedBlockByHeightArgs {
+    pub height: U32,
+}
+
+#[derive(cynic::QueryFragment, Clone, Debug)]
+#[cynic(
+    schema_path = "./assets/schema.sdl",
+    graphql_type = "Query",
+    variables = "DaCompressedBlockByHeightArgs"
+)]
+pub struct DaCompressedBlockByHeightQuery {
+    #[arguments(height: $height)]
+    pub da_compressed_block: Option<DaCompressedBlock>,
+}
+
+/// Block with transaction ids
+#[derive(cynic::QueryFragment, Clone, Debug)]
+#[cynic(schema_path = "./assets/schema.sdl")]
+pub struct DaCompressedBlock {
+    pub bytes: HexString,
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn block_by_height_query_gql_output() {
+        use cynic::QueryBuilder;
+        let operation =
+            DaCompressedBlockByHeightQuery::build(DaCompressedBlockByHeightArgs {
+                height: U32(0),
+            });
+        insta::assert_snapshot!(operation.query)
+    }
+}

crates/client/src/client/schema/snapshots/fuel_core_client__client__schema__da_compressed__tests__block_by_height_query_gql_output.snap

@@ -0,0 +1,9 @@
+---
+source: crates/client/src/client/schema/da_compressed.rs
+expression: operation.query
+---
+query($height: U32!) {
+  daCompressedBlock(height: $height) {
+    bytes
+  }
+}

crates/compression/Cargo.toml

@@ -0,0 +1,44 @@
+[package]
+name = "fuel-core-compression"
+version = { workspace = true }
+authors = { workspace = true }
+categories = ["cryptography::cryptocurrencies"]
+edition = { workspace = true }
+homepage = { workspace = true }
+keywords = [
+  "blockchain",
+  "cryptocurrencies",
+  "fuel-core",
+  "fuel-client",
+  "fuel-compression",
+]
+license = { workspace = true }
+repository = { workspace = true }
+description = "Compression and decompression of Fuel blocks for DA storage."
+
+[dependencies]
+anyhow = { workspace = true }
+fuel-core-types = { workspace = true, features = [
+  "alloc",
+  "serde",
+  "da-compression",
+] }
+paste = { workspace = true }
+rand = { workspace = true, optional = true }
+serde = { version = "1.0", features = ["derive"] }
+strum = { workspace = true }
+strum_macros = { workspace = true }
+
+[dev-dependencies]
+fuel-core-compression = { path = ".", features = ["test-helpers"] }
+postcard = { version = "1.0", features = ["use-std"] }
+proptest = { workspace = true }
+tokio = { workspace = true, features = ["macros", "rt-multi-thread"] }
+
+[features]
+test-helpers = [
+  "dep:rand",
+  "fuel-core-types/test-helpers",
+  "fuel-core-types/random",
+  "fuel-core-types/std",
+]

crates/compression/README.md

@@ -0,0 +1,27 @@
+# Compression and decompression of transactions for the DA layer
+
+## Compressed block header
+
+Each compressed block begins with a version field, so that it's possible to change the format later.
+
+## Temporal registry
+
+This crate provides offchain registries for different types such as `AssetId`, `ContractId`, scripts, and predicates. Each registry is a key-value store with three-byte key. The registries are essentially compression caches. The three byte key allows cache size of 16 million values before reregistering the older values.
+
+The registries allow replacing repeated objects with their respective keys, so if an object
+is used multiple times in a short interval (couple of months, maybe), then the full value
+exists on only a single uncompressed block.
+
+### Fraud proofs
+
+Compressed block will contain a merkle root over all compression smts, followed by newly registered values along with their keys. Using an SMT provides flexibility around the algorithm we use to define keys without knowing how exactly values were chosen to be registered.
+
+Each registry also uses an SMT. Since the keys are three bytes long, the depth of the SMT is capped at 24 levels. 
+
+## Compression of `UtxoIds`
+
+Since each `UtxoId` only appears once, there's no point in registering them. Instead, they are replaced with `TxPointer` and output index, which are still unique.
+
+### Fraud proofs
+
+During fraud proofs we need to use the `prev_root` to prove that the referenced block height is part of the chain.