fix(repo): S3 payload and encoding (#366)

* refactor(repo): Change from S3 specific to a Storage trait

* refactor(repo): use displaydoc instead of error

* fix(data-parser): adjust encode to support internal messages

* fix(repo): s3 being saved and parsed corretly

* fix(repo): final adjustments

* build(repo): fix lint

* build(repo): fix chart

* build(repo): fix tests

* fix(storage): add retry mechanism

* build(repo): fix lint

* fix(repo): small adjustments

.env.sample

@@ -9,6 +9,7 @@ AWS_ENDPOINT_URL=http://localhost:4566
 AWS_REGION=us-east-1
 AWS_S3_ENABLED=false
 AWS_S3_BUCKET_NAME=fuel-streams-local
+STORAGE_MAX_RETRIES=5
 
 # NATS Configuration
 NATS_URL=nats://localhost:4222

.github/workflows/ci.yaml

@@ -222,7 +222,7 @@ jobs:
       AWS_SECRET_ACCESS_KEY: test
       AWS_REGION: us-east-1
       AWS_ENDPOINT_URL: http://localhost:4566
-      AWS_S3_BUCKET_NAME: fuel-streams-test
+      AWS_S3_BUCKET_NAME: fuel-streams-local
     strategy:
       fail-fast: false
       matrix:
@@ -247,16 +247,16 @@ jobs:
           tool: cargo-nextest
           locked: true
 
-      - name: Start Nats
+      - name: Start Docker
         run: |
-          make start-nats
+          make start-docker
 
       - name: Run tests
         run: make test PACKAGE=${{ matrix.package }} PROFILE=ci
 
-      - name: Stop Nats
+      - name: Stop Docker
         if: always()
-        run: make stop-nats
+        run: make stop-docker
 
   build:
     needs: install-deps

Cargo.toml

@@ -26,8 +26,6 @@ actix-web = "4.9"
 anyhow = "1.0"
 async-nats = "0.38"
 async-trait = "0.1"
-assert_matches = "1.5.0"
-bytes = "1.9"
 chrono = { version = "0.4", features = ["serde"] }
 clap = { version = "4.5", features = ["derive", "env"] }
 dotenvy = "0.15"
@@ -49,9 +47,6 @@ fuel-core-storage = { version = "0.40.2" }
 fuel-core-types = { version = "0.40.2", default-features = false, features = ["std", "serde"] }
 fuel-core-services = { version = "0.40.2", default-features = false, features = ["test-helpers"] }
 futures-util = "0.3"
-itertools = "0.13"
-mockall = "0.13"
-mockall_double = "0.3.1"
 hex = "0.4"
 pretty_assertions = "1.4"
 num_cpus = "1.16"
@@ -63,7 +58,6 @@ sha2 = "0.10"
 strum = "0.26"
 strum_macros = "0.26"
 tokio = { version = "1.41", features = ["full"] }
-tokio-stream = "0.1.16"
 tracing = "0.1"
 tracing-subscriber = "0.3"
 tracing-actix-web = "0.7"
@@ -78,7 +72,6 @@ fuel-streams-storage = { version = "0.0.13", path = "crates/fuel-streams-storage
 fuel-streams-executors = { version = "0.0.13", path = "crates/fuel-streams-executors" }
 subject-derive = { version = "0.0.13", path = "crates/fuel-streams-macros/subject-derive" }
 sv-publisher = { version = "0.0.13", path = "crates/sv-publisher" }
-sv-consumer = { version = "0.0.13", path = "crates/sv-consumer" }
 sv-webserver = { version = "0.0.13", path = "crates/sv-webserver" }
 
 # Workspace projects

benches/data-parser/Cargo.toml

@@ -34,6 +34,7 @@ path = "benches/deserialize_decompress.rs"
 fuel-core-types = { workspace = true }
 fuel-data-parser = { workspace = true, features = ["test-helpers", "bench-helpers"] }
 rand = { workspace = true }
+serde = { workspace = true }
 strum = { workspace = true }
 tokio = { workspace = true }
 

benches/data-parser/benches/deserialize_decompress.rs

@@ -1,6 +1,5 @@
 use criterion::{black_box, criterion_group, criterion_main, Criterion};
-use data_parser::generate_test_block;
-use fuel_core_types::{blockchain::block::Block, fuel_tx::Transaction};
+use data_parser::{generate_test_block, TestBlock};
 use fuel_data_parser::{
     DataParser,
     SerializationType,
@@ -56,7 +55,7 @@ fn bench_decompress_deserialize(c: &mut Criterion) {
 
             b.to_async(&runtime).iter(|| async {
                 let deserialized_and_decompressed = data_parser
-                    .decode::<Block<Transaction>>(serialized_and_compressed)
+                    .decode::<TestBlock>(serialized_and_compressed)
                     .await
                     .expect("decompresison and deserialization");
 

benches/data-parser/src/lib.rs

@@ -8,9 +8,17 @@ use fuel_core_types::{
     fuel_types::BlockHeight,
     tai64::Tai64,
 };
+use fuel_data_parser::{DataEncoder, DataParserError};
 use rand::Rng;
 
-pub fn generate_test_block() -> Block<Transaction> {
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
+pub struct TestBlock(Block<Transaction>);
+
+impl DataEncoder for TestBlock {
+    type Err = DataParserError;
+}
+
+pub fn generate_test_block() -> TestBlock {
     let mut rng = rand::thread_rng();
     let block_height: u32 = rng.gen_range(1..100);
     let block_txs: u32 = rng.gen_range(1..100);
@@ -40,7 +48,7 @@ pub fn generate_test_block() -> Block<Transaction> {
     block
         .header_mut()
         .set_transaction_root(Bytes32::new(tx_root));
-    block
+    TestBlock(block)
 }
 
 pub fn generate_test_tx() -> Transaction {

cluster/charts/fuel-streams/Chart.yaml

@@ -2,7 +2,7 @@ apiVersion: v2
 appVersion: "1.0"
 description: A Helm chart for Kubernetes
 name: fuel-streams
-version: 0.7.2
+version: 0.7.3
 dependencies:
   - name: nats
     version: 1.2.8

cluster/charts/fuel-streams/values.yaml

@@ -221,11 +221,6 @@ consumer:
         podValue: 4
         periodSeconds: 15
 
-  env:
-    PORT: 8080
-    PUBLISHER_MAX_THREADS: "32"
-    NATS_URL: "fuel-streams-nats-publisher:4222"
-
 # -------------------------------------------------------------------------------------------------
 # Consumer configuration
 # -------------------------------------------------------------------------------------------------

cluster/docker/docker-compose.yml

@@ -54,10 +54,10 @@ services:
     environment:
       - SERVICES=s3 # Enable just S3 service
       - DEBUG=1
-      - AWS_ACCESS_KEY_ID=${AWS_ACCESS_KEY_ID}
-      - AWS_SECRET_ACCESS_KEY=${AWS_SECRET_ACCESS_KEY}
-      - DEFAULT_REGION=${AWS_REGION}
-      - DEFAULT_BUCKETS=${AWS_S3_BUCKET_NAME}
+      - AWS_ACCESS_KEY_ID=${AWS_ACCESS_KEY_ID:-test}
+      - AWS_SECRET_ACCESS_KEY=${AWS_SECRET_ACCESS_KEY:-test}
+      - DEFAULT_REGION=${AWS_REGION:-us-east-1}
+      - DEFAULT_BUCKETS=${AWS_S3_BUCKET_NAME:-fuel-streams-local}
     volumes:
       - ./localstack-data:/var/lib/localstack
       - /var/run/docker.sock:/var/run/docker.sock

cluster/docker/init-localstack.sh

@@ -3,6 +3,6 @@ set -e
 
 echo "Creating S3 bucket in LocalStack..."
 
-BUCKET_NAME=${AWS_S3_BUCKET_NAME:-fuel-streams-test}
+BUCKET_NAME=${AWS_S3_BUCKET_NAME:-fuel-streams-local}
 awslocal s3 mb "s3://${BUCKET_NAME}"
 echo "Bucket created: ${BUCKET_NAME}"