feat: create importer online component

.github/workflows/e2e-run-with-importer.yml → .github/workflows/e2e-importer.yml

@@ -1,4 +1,4 @@
-name: E2E Run With Importer
+name: E2E Importer
 
 on:
   pull_request:
@@ -26,10 +26,10 @@ on:
       - 'Cargo.toml'
 
 jobs:
-  run_with_importer_test:
+  importer_test:
     strategy:
       fail-fast: false
-    name: E2E Run With Importer on BRLCToken
+    name: E2E Importer on BRLCToken
     runs-on: ubuntu-latest
     timeout-minutes: 45
 

Cargo.toml

@@ -155,14 +155,6 @@ path = "src/bin/rpc_downloader.rs"
 name = "importer-offline"
 path = "src/bin/importer_offline.rs"
 
-[[bin]]
-name = "importer-online"
-path = "src/bin/importer_online.rs"
-
-[[bin]]
-name = "run-with-importer"
-path = "src/bin/run_with_importer.rs"
-
 # ------------------------------------------------------------------------------
 # Features
 # ------------------------------------------------------------------------------

e2e/cloudwalk-contracts/integration/test/helpers/rpc.ts

@@ -44,7 +44,7 @@ export let ETHERJS = new JsonRpcProvider(providerUrl, undefined);
 
 export function updateProviderUrl(providerName: string) {
     switch (providerName) {
-        case "run-with-importer":
+        case "importer":
             providerUrl = "http://localhost:3001?app=e2e";
             break;
         case "stratus":

e2e/cloudwalk-contracts/integration/test/importer.test.ts

@@ -13,7 +13,7 @@ import {
     updateProviderUrl,
 } from "./helpers/rpc";
 
-describe("Run With Importer integration test", function () {
+describe("Importer integration test", function () {
     before(async function () {
         await setDeployer();
     });
@@ -92,7 +92,7 @@ describe("Run With Importer integration test", function () {
                 }
             });
 
-            it(`${params.name}: Validate transaction mined delay between Stratus and Run With Importer`, async function () {
+            it(`${params.name}: Validate transaction mined delay between Stratus and Importer`, async function () {
                 // Get Stratus timestamps
                 updateProviderUrl("stratus");
                 const stratusTimestamps = await Promise.all(
@@ -103,9 +103,9 @@ describe("Run With Importer integration test", function () {
                     }),
                 );
 
-                // Get Run With Importer timestamps
-                updateProviderUrl("run-with-importer");
-                const runWithImporterTimestamps = await Promise.all(
+                // Get Importer timestamps
+                updateProviderUrl("importer");
+                const importerTimestamps = await Promise.all(
                     txHashList.map(async (txHash) => {
                         const receipt = await sendWithRetry("eth_getTransactionReceipt", [txHash], 20);
                         const block = await sendWithRetry("eth_getBlockByNumber", [receipt.blockNumber, false]);
@@ -116,17 +116,17 @@ describe("Run With Importer integration test", function () {
                 // Total time it took for Stratus to process all the blocks containing transactions
                 const stratusProcessingTime = stratusTimestamps[stratusTimestamps.length - 1] - stratusTimestamps[0];
 
-                // Total time it took for Run With Importer to process all the blocks containing transactions
-                const runWithImporterProcessingTime =
-                    runWithImporterTimestamps[runWithImporterTimestamps.length - 1] - runWithImporterTimestamps[0];
+                // Total time it took for Importer to process all the blocks containing transactions
+                const importerProcessingTime =
+                    importerTimestamps[importerTimestamps.length - 1] - importerTimestamps[0];
 
                 console.log(`          ✔ Number of transactions sent: ${txHashList.length}`);
                 console.log(
-                    `          ✔ Stratus processing time: ${stratusProcessingTime}s | Run With Importer processing time: ${runWithImporterProcessingTime}s`,
+                    `          ✔ Stratus processing time: ${stratusProcessingTime}s | Importer processing time: ${importerProcessingTime}s`,
                 );
             });
 
-            it(`${params.name}: Validate all transactions were imported from Stratus to Run With Importer`, async function () {
+            it(`${params.name}: Validate all transactions were imported from Stratus to Importer`, async function () {
                 // Get Stratus transaction receipts
                 updateProviderUrl("stratus");
                 const stratusReceipts = await Promise.all(
@@ -136,9 +136,9 @@ describe("Run With Importer integration test", function () {
                     }),
                 );
 
-                // Get Run With Importer transaction receipts
-                updateProviderUrl("run-with-importer");
-                const runWithImporterReceipts = await Promise.all(
+                // Get Importer transaction receipts
+                updateProviderUrl("importer");
+                const importerReceipts = await Promise.all(
                     txHashList.map(async (txHash) => {
                         const receipt = await sendWithRetry("eth_getTransactionReceipt", [txHash]);
                         return receipt;
@@ -148,11 +148,11 @@ describe("Run With Importer integration test", function () {
                 // Assert that all transactions were imported
                 for (let i = 0; i < txHashList.length; i++) {
                     expect(stratusReceipts[i]).to.exist;
-                    expect(runWithImporterReceipts[i]).to.exist;
+                    expect(importerReceipts[i]).to.exist;
                 }
             });
 
-            it(`${params.name}: Validate each transaction was imported into the same block between Stratus and Run With Importer`, async function () {
+            it(`${params.name}: Validate each transaction was imported into the same block between Stratus and Importer`, async function () {
                 // Get Stratus block numbers
                 updateProviderUrl("stratus");
                 const stratusBlockNumbers = await Promise.all(
@@ -162,38 +162,38 @@ describe("Run With Importer integration test", function () {
                     }),
                 );
 
-                // Get Run With Importer block numbers
-                updateProviderUrl("run-with-importer");
-                const runWithImporterBlockNumbers = await Promise.all(
+                // Get Importer block numbers
+                updateProviderUrl("importer");
+                const importerBlockNumbers = await Promise.all(
                     txHashList.map(async (txHash) => {
                         const receipt = await sendWithRetry("eth_getTransactionReceipt", [txHash], 20);
                         return receipt.blockNumber;
                     }),
                 );
 
-                // Assert that each transaction fell into the same block between Stratus and Run With Importer
+                // Assert that each transaction fell into the same block between Stratus and Importer
                 for (let i = 0; i < txHashList.length; i++) {
                     expect(
                         stratusBlockNumbers[i],
-                        `Transaction ${txHashList[i]} did not fall into the same block between Stratus and Run With Importer`,
-                    ).to.equal(runWithImporterBlockNumbers[i]);
+                        `Transaction ${txHashList[i]} did not fall into the same block between Stratus and Importer`,
+                    ).to.equal(importerBlockNumbers[i]);
                 }
             });
 
-            it(`${params.name}: Validate balances between Stratus and Run With Importer`, async function () {
+            it(`${params.name}: Validate balances between Stratus and Importer`, async function () {
                 for (let i = 0; i < wallets.length; i++) {
                     // Get Stratus balance
                     updateProviderUrl("stratus");
                     const stratusBalance = await brlcToken.balanceOf(wallets[i].address);
 
-                    // Get Run With Importer balance
-                    updateProviderUrl("run-with-importer");
-                    const runWithImporterBalance = await brlcToken.balanceOf(wallets[i].address);
+                    // Get Importer balance
+                    updateProviderUrl("importer");
+                    const importerBalance = await brlcToken.balanceOf(wallets[i].address);
 
                     // Assert that the balances are equal
                     expect(stratusBalance).to.equal(
-                        runWithImporterBalance,
-                        `Wallet ${wallets[i].address} balances are not equal between Stratus and Run With Importer`,
+                        importerBalance,
+                        `Wallet ${wallets[i].address} balances are not equal between Stratus and Importer`,
                     );
                 }
                 updateProviderUrl("stratus");

justfile

@@ -312,22 +312,22 @@ e2e-importer-online:
 e2e-importer-online-up:
     #!/bin/bash
 
-    # Build Stratus and Run With Importer binaries
-    echo "Building Stratus and Run With Importer binaries"
-    cargo build --release --bin stratus --bin run-with-importer --features dev
+    # Build Stratus binary
+    echo "Building Stratus binary"
+    cargo build --release --bin stratus --features dev
 
     mkdir e2e_logs
 
-    # Start Stratus binary
-    RUST_LOG=info cargo run --release --bin stratus --features dev -- --block-mode 1s --enable-genesis --enable-test-accounts --perm-storage=rocks --rocks-path-prefix=temp_3000 --tokio-console-address=0.0.0.0:6668 --metrics-exporter-address=0.0.0.0:9000 -a 0.0.0.0:3000 > e2e_logs/stratus.log &
+    # Start Stratus with leader flag
+    RUST_LOG=info cargo run --release --bin stratus --features dev -- --mode leader --block-mode 1s --enable-genesis --enable-test-accounts --perm-storage=rocks --rocks-path-prefix=temp_3000 --tokio-console-address=0.0.0.0:6668 --metrics-exporter-address=0.0.0.0:9000 -a 0.0.0.0:3000 > e2e_logs/stratus.log &
 
-    # Wait for Stratus to start
+    # Wait for Stratus with leader flag to start
     wait-service --tcp 0.0.0.0:3000 -t {{ wait_service_timeout }} -- echo
 
-    # Start Run With Importer binary
-    RUST_LOG=info cargo run --release --bin run-with-importer --features dev -- --block-mode 1s --enable-test-accounts --perm-storage=rocks --rocks-path-prefix=temp_3001 --tokio-console-address=0.0.0.0:6669 --metrics-exporter-address=0.0.0.0:9001 -a 0.0.0.0:3001 -r http://0.0.0.0:3000/ -w ws://0.0.0.0:3000/ > e2e_logs/run_with_importer.log &
+    # Start Stratus with follower flag
+    RUST_LOG=info cargo run --release --bin stratus --features dev -- --mode follower --enable-test-accounts --perm-storage=rocks --rocks-path-prefix=temp_3001 --tokio-console-address=0.0.0.0:6669 --metrics-exporter-address=0.0.0.0:9001 -a 0.0.0.0:3001 -r http://0.0.0.0:3000/ -w ws://0.0.0.0:3000/ > e2e_logs/importer.log &
 
-    # Wait for Run With Importer to start
+    # Wait for Stratus with follower flag to start
     wait-service --tcp 0.0.0.0:3001 -t {{ wait_service_timeout }} -- echo
 
     if [ -d e2e/cloudwalk-contracts ]; then
@@ -349,12 +349,8 @@ e2e-importer-online-up:
 e2e-importer-online-down:
     #!/bin/bash
 
-    # Kill run-with-importer
-    killport 3001
-    run_with_importer_pid=$(pgrep -f 'run-with-importer')
-    kill $run_with_importer_pid
-
     # Kill Stratus
+    killport 3001
     killport 3000
     stratus_pid=$(pgrep -f 'stratus')
     kill $stratus_pid

src/bin/rpc_downloader.rs

@@ -30,7 +30,7 @@ async fn run(config: RpcDownloaderConfig) -> anyhow::Result<()> {
     let _timer = DropTimer::start("rpc-downloader");
 
     let rpc_storage = config.rpc_storage.init().await?;
-    let chain = Arc::new(BlockchainClient::new_http(&config.external_rpc, config.external_rpc_timeout).await?);
+    let chain = Arc::new(BlockchainClient::new_http(Some(&config.external_rpc), config.external_rpc_timeout).await?);
 
     let block_end = match config.block_end {
         Some(end) => BlockNumber::from(end),

src/config.rs

@@ -14,6 +14,7 @@ use tokio::runtime::Builder;
 use tokio::runtime::Runtime;
 
 use crate::eth::executor::ExecutorConfig;
+use crate::eth::importer::ImporterConfig;
 use crate::eth::miner::MinerConfig;
 use crate::eth::primitives::Address;
 use crate::eth::rpc::RpcServerConfig;
@@ -63,6 +64,10 @@ pub struct CommonConfig {
     #[arg(long = "env", env = "ENV", default_value = "local")]
     pub env: Environment,
 
+    /// Stratus mode.
+    #[arg(long = "mode", env = "MODE", default_value = "leader")] // should we leave a default value?
+    pub mode: StratusMode,
+
     /// Number of threads to execute global async tasks.
     #[arg(long = "async-threads", env = "ASYNC_THREADS", default_value = "10")]
     pub num_async_threads: usize,
@@ -169,6 +174,9 @@ pub struct StratusConfig {
     #[clap(flatten)]
     pub miner: MinerConfig,
 
+    #[clap(flatten)]
+    pub importer: ImporterConfig,
+
     #[deref]
     #[clap(flatten)]
     pub common: CommonConfig,
@@ -453,6 +461,33 @@ impl FromStr for Environment {
     }
 }
 
+// -----------------------------------------------------------------------------
+// Enum: Stratus Mode
+// -----------------------------------------------------------------------------
+#[derive(DebugAsJson, strum::Display, strum::VariantNames, Clone, Copy, Parser, serde::Serialize)]
+pub enum StratusMode {
+    #[serde(rename = "leader")]
+    #[strum(to_string = "leader")]
+    Leader,
+
+    #[serde(rename = "follower")]
+    #[strum(to_string = "follower")]
+    Follower,
+}
+
+impl FromStr for StratusMode {
+    type Err = anyhow::Error;
+
+    fn from_str(s: &str) -> anyhow::Result<Self, Self::Err> {
+        let s = s.trim().to_lowercase();
+        match s.as_ref() {
+            "leader" => Ok(Self::Leader),
+            "follower" => Ok(Self::Follower),
+            s => Err(anyhow!("unknown stratus mode: \"{}\" - valid values are {:?}", s, StratusMode::VARIANTS)),
+        }
+    }
+}
+
 // -----------------------------------------------------------------------------
 // Enum: ValidatorMethodConfig
 // -----------------------------------------------------------------------------