fix: updates, added test

Cargo.lock

@@ -2871,7 +2871,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "4ee796ad498c8d9a1d68e477df8f754ed784ef875de1414ebdaf169f70a6a784"
 
 [[package]]
-name = "indexer"
+name = "indexer_core"
 version = "0.1.0"
 dependencies = [
  "anyhow",
@@ -2882,6 +2882,7 @@ dependencies = [
  "log",
  "nomos-core",
  "serde",
+ "serde_json",
  "tokio",
  "url",
 ]
@@ -2944,6 +2945,7 @@ dependencies = [
  "env_logger",
  "futures",
  "hex",
+ "indexer_core",
  "key_protocol",
  "log",
  "nssa",

Cargo.toml

@@ -20,7 +20,7 @@ members = [
   "examples/program_deployment/methods",
   "examples/program_deployment/methods/guest",
   "bedrock_client",
-  "indexer",
+  "indexer_core",
 ]
 
 [workspace.dependencies]
@@ -36,7 +36,7 @@ sequencer_runner = { path = "sequencer_runner" }
 wallet = { path = "wallet" }
 test_program_methods = { path = "test_program_methods" }
 bedrock_client = { path = "bedrock_client" }
-indexer = { path = "indexer" }
+indexer_core = { path = "indexer_core" }
 
 tokio = { version = "1.28.2", features = [
   "net",

indexer_core/Cargo.toml

@@ -1,5 +1,5 @@
 [package]
-name = "indexer"
+name = "indexer_core"
 version = "0.1.0"
 edition = "2024"
 
@@ -15,3 +15,4 @@ borsh.workspace = true
 futures.workspace = true
 url.workspace = true
 nomos-core.workspace = true
+serde_json.workspace = true

indexer_core/src/config.rs

@@ -1,3 +1,6 @@
+use std::{fs::File, io::BufReader, path::Path};
+
+use anyhow::Result;
 use nomos_core::mantle::ops::channel::ChannelId;
 use serde::{Deserialize, Serialize};
 
@@ -18,3 +21,12 @@ pub struct IndexerConfig {
     pub sequencer_client_config: ClientConfig,
     pub channel_id: ChannelId,
 }
+
+impl IndexerConfig {
+    pub fn from_path(config_home: &Path) -> Result<IndexerConfig> {
+        let file = File::open(config_home)?;
+        let reader = BufReader::new(file);
+
+        Ok(serde_json::from_reader(reader)?)
+    }
+}

integration_tests/Cargo.toml

@@ -11,6 +11,7 @@ sequencer_runner.workspace = true
 wallet.workspace = true
 common.workspace = true
 key_protocol.workspace = true
+indexer_core.workspace = true
 
 anyhow.workspace = true
 env_logger.workspace = true

integration_tests/configs/indexer/indexer_config.json

@@ -0,0 +1,16 @@
+{
+    "bedrock_client_config": {
+        "addr": "http://127.0.0.1:8080",
+        "auth": [
+            "user",
+            null
+        ]
+    },
+    "channel_id": "0101010101010101010101010101010101010101010101010101010101010101",
+    "max_retries": 10,
+    "resubscribe_interval_millis": 1000,
+    "sequencer_client_config": {
+        "addr": "will_be_replaced_in_runtime"
+    },
+    "start_delay_millis": 100
+}

integration_tests/configs/sequencer/bedrock_local_attached/sequencer_config.json

@@ -159,11 +159,6 @@
         "channel_id": "0101010101010101010101010101010101010101010101010101010101010101",
         "node_url": "http://127.0.0.1:8080",
         "user": "user",
-        "password": null,
-        "indexer_config": {
-            "resubscribe_interval_millis": 1000,
-            "start_delay_millis": 1000,
-            "max_retries": 10
-        }
+        "password": null
     }
 }

integration_tests/src/lib.rs

@@ -10,6 +10,7 @@ use common::{
     transaction::{EncodedTransaction, NSSATransaction},
 };
 use futures::FutureExt as _;
+use indexer_core::{IndexerCore, config::IndexerConfig};
 use log::debug;
 use nssa::PrivacyPreservingTransaction;
 use nssa_core::Commitment;
@@ -38,6 +39,7 @@ static LOGGER: LazyLock<()> = LazyLock::new(env_logger::init);
 pub struct TestContext {
     sequencer_server_handle: ServerHandle,
     sequencer_loop_handle: JoinHandle<Result<()>>,
+    indexer_loop_handle: Option<JoinHandle<Result<()>>>,
     sequencer_client: SequencerClient,
     wallet: WalletCore,
     _temp_sequencer_dir: TempDir,
@@ -68,7 +70,13 @@ impl TestContext {
         let sequencer_config = SequencerConfig::from_path(&sequencer_config_path)
             .context("Failed to create sequencer config from file")?;
 
-        Self::new_with_sequencer_config(sequencer_config).await
+        let indexer_config_path =
+            PathBuf::from(manifest_dir).join("configs/indexer/indexer_config.json");
+
+        let indexer_config = IndexerConfig::from_path(&indexer_config_path)
+            .context("Failed to create indexer config from file")?;
+
+        Self::new_with_sequencer_and_indexer_configs(sequencer_config, indexer_config).await
     }
 
     /// Create new test context with custom sequencer config.
@@ -105,6 +113,60 @@ impl TestContext {
         Ok(Self {
             sequencer_server_handle,
             sequencer_loop_handle,
+            indexer_loop_handle: None,
+            sequencer_client,
+            wallet,
+            _temp_sequencer_dir: temp_sequencer_dir,
+            _temp_wallet_dir: temp_wallet_dir,
+        })
+    }
+
+    /// Create new test context with custom sequencer and indexer configs.
+    ///
+    /// `home` and `port` fields of the provided config will be overridden to meet tests parallelism
+    /// requirements.
+    pub async fn new_with_sequencer_and_indexer_configs(
+        sequencer_config: SequencerConfig,
+        mut indexer_config: IndexerConfig,
+    ) -> Result<Self> {
+        // Ensure logger is initialized only once
+        *LOGGER;
+
+        debug!("Test context setup");
+
+        let (sequencer_server_handle, sequencer_addr, sequencer_loop_handle, temp_sequencer_dir) =
+            Self::setup_sequencer(sequencer_config)
+                .await
+                .context("Failed to setup sequencer")?;
+
+        // Convert 0.0.0.0 to 127.0.0.1 for client connections
+        // When binding to port 0, the server binds to 0.0.0.0:<random_port>
+        // but clients need to connect to 127.0.0.1:<port> to work reliably
+        let sequencer_addr = if sequencer_addr.ip().is_unspecified() {
+            format!("http://127.0.0.1:{}", sequencer_addr.port())
+        } else {
+            format!("http://{sequencer_addr}")
+        };
+
+        let (wallet, temp_wallet_dir) = Self::setup_wallet(sequencer_addr.clone())
+            .await
+            .context("Failed to setup wallet")?;
+
+        let sequencer_client = SequencerClient::new(sequencer_addr.clone())
+            .context("Failed to create sequencer client")?;
+
+        indexer_config.sequencer_client_config.addr = sequencer_addr;
+
+        let indexer_core = IndexerCore::new(indexer_config)?;
+
+        let indexer_loop_handle = Some(tokio::spawn(async move {
+            indexer_core.subscribe_parse_block_stream().await
+        }));
+
+        Ok(Self {
+            sequencer_server_handle,
+            sequencer_loop_handle,
+            indexer_loop_handle,
             sequencer_client,
             wallet,
             _temp_sequencer_dir: temp_sequencer_dir,
@@ -193,6 +255,7 @@ impl Drop for TestContext {
         let Self {
             sequencer_server_handle,
             sequencer_loop_handle,
+            indexer_loop_handle,
             sequencer_client: _,
             wallet: _,
             _temp_sequencer_dir,
@@ -200,6 +263,9 @@ impl Drop for TestContext {
         } = self;
 
         sequencer_loop_handle.abort();
+        if let Some(indexer_loop_handle) = indexer_loop_handle {
+            indexer_loop_handle.abort();
+        }
 
         // Can't wait here as Drop can't be async, but anyway stop signal should be sent
         sequencer_server_handle.stop(true).now_or_never();

integration_tests/tests/indexer.rs

@@ -1,34 +1,23 @@
-// use anyhow::Result;
-// use integration_tests::TestContext;
-// use log::info;
-// use tokio::test;
+use anyhow::Result;
+use integration_tests::TestContext;
+use log::info;
+use tokio::test;
 
-// ToDo: Uncomment when indexer RPC is available
-//#[ignore = "needs complicated setup"]
-//#[test]
+#[ignore = "needs complicated setup"]
+#[test]
 // To run this test properly, you need nomos node running in the background.
 // For instructions in building nomos node, refer to [this](https://github.com/logos-blockchain/logos-blockchain?tab=readme-ov-file#running-a-logos-blockchain-node).
 //
 // Recommended to run node locally from build binary.
-// async fn indexer_run_local_node() -> Result<()> {
-//     let ctx = TestContext::new_bedrock_local_attached().await?;
+async fn indexer_run_local_node() -> Result<()> {
+    let _ctx = TestContext::new_bedrock_local_attached().await?;
 
-//     info!("Let's observe behaviour");
+    info!("Let's observe behaviour");
 
-//     tokio::time::sleep(std::time::Duration::from_secs(180)).await;
+    tokio::time::sleep(std::time::Duration::from_secs(180)).await;
 
-//     let gen_id = ctx
-//         .sequencer_client()
-//         .get_last_seen_l2_block_at_indexer()
-//         .await
-//         .unwrap()
-//         .last_block
-//         .unwrap();
+    // No way to check state of indexer now
+    // When it will be a service, then it will become possible.
 
-//     // Checking, that some blocks are landed on bedrock
-//     assert!(gen_id > 0);
-
-//     info!("Last seen L2 block at indexer is {gen_id}");
-
-//     Ok(())
-// }
+    Ok(())
+}