WIP: Rust Indexer/Operator/Relayer

Cargo.toml

@@ -5,12 +5,39 @@ resolver = "2"
 members = [
     "indexer",
     "contracts/evm/test/ffi/bls-utils",
+    "relayer-rs",
+    "operator-rs",
+    "core-rs",
 ]
 
 [workspace.package]
 version = "0.0.1"
 edition = "2021"
 repository = "https://github.com/NethermindEth/near-sffl"
 
+[workspace.dependencies]
+alloy = { version = "0.4.2", features = ["full", "node-bindings", "rpc-types-debug", "rpc-types-trace", "json-rpc", "rpc-client", "serde"] }
+alloy-primitives = "0.8.3"
+alloy-rpc-client = "0.4.2"
+alloy-rpc-types = "0.4.2"
+alloy-transport-ws = "0.4.2"
+alloy-rlp = "0.3.8"
+tokio = { version = "1.0", features = ["full"] }
+prometheus = "0.13"
+serde = { version = "1.0", features = ["derive"] }
+serde_json = "1.0"
+anyhow = "1.0"
+tracing = "0.1"
+tracing-subscriber = "0.3"
+futures-util = "0.3.30"
+eyre = "0.6"
+bincode = "1.3"
+serde_yaml = "0.9"
+clap = "4.5.18"
+tempfile = "3.10.0"
+async-trait = "0.1"
+log = "0.4"
+core-rs = { path = "core-rs" }
+
 [patch.crates-io]
 parity-secp256k1 = { git = 'https://github.com/paritytech/rust-secp256k1.git' }

Makefile

@@ -103,6 +103,10 @@ start-indexer: ##
 	cargo run -p indexer --release -- --home-dir ~/.near/localnet init --chain-id localnet
 	cargo run -p indexer --release -- --home-dir ~/.near/localnet run --da-contract-ids da.test.near --rollup-ids 2 --rmq-address "amqp://127.0.0.1:5672"
 
+start-indexer-with-fastnear: ##
+	cargo run --features use_fastnear -p indexer --release -- --home-dir ~/.near/localnet init --chain-id localnet 
+	cargo run --features use_fastnear -p indexer --release -- --home-dir ~/.near/localnet run --da-contract-ids da.test.near --rollup-ids 2 --rmq-address "amqp://127.0.0.1:5672"
+
 start-test-relayer: ##
 	CGO_LDFLAGS="-L ./relayer/libs ${CGO_LDFLAGS}" go run relayer/cmd/main.go run-args --rpc-url ws://127.0.0.1:8546 --da-account-id da.test.near --key-path ~/.near-credentials/localnet/da.test.near.json
 

core-rs/Cargo.toml

@@ -0,0 +1,23 @@
+[package]
+name = "core-rs"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+tokio = { workspace = true }
+async-trait = { workspace = true }
+thiserror = "1.0"
+anyhow = { workspace = true }
+
+alloy = { workspace = true }
+mockall = "0.13.0"
+
+serde = { workspace = true }
+serde_json = { workspace = true }
+sha2 = "0.10.6"
+
+log = { workspace = true }
+futures-util = { workspace = true }
+
+[dev-dependencies]
+tokio-test = "0.4"

core-rs/src/lib.rs

@@ -0,0 +1,6 @@
+// pub mod chainio;
+// pub mod config;
+// pub mod metricable;
+// pub mod smt;
+// pub mod types;
+pub mod safeclient;

core-rs/src/safeclient/client.rs

@@ -0,0 +1,156 @@
+use alloy::eips::BlockNumberOrTag;
+use alloy::primitives::{B256, U256};
+use alloy::providers::{Provider, ProviderBuilder, RootProvider, WsConnect};
+use alloy::pubsub::PubSubFrontend;
+use alloy::rpc::types::{Block, Filter, Header, Log, Transaction, TransactionReceipt};
+use anyhow::Result;
+use async_trait::async_trait;
+use std::sync::Arc;
+use std::time::Duration;
+use tokio::sync::broadcast;
+use futures_util::StreamExt;
+use log::{error, warn};
+
+
+#[async_trait]
+pub trait SafeClient: Send + Sync {
+    async fn block_number(&self) -> Result<U256>;
+    async fn get_block(&self, block_number: BlockNumberOrTag) -> Result<Option<Block>>;
+    async fn get_transaction(&self, tx_hash: B256) -> Result<Option<Transaction>>;
+    async fn get_transaction_receipt(&self, tx_hash: B256) -> Result<Option<TransactionReceipt>>;
+    async fn get_logs(&self, filter: Filter) -> Result<Vec<Log>>;
+    async fn subscribe_logs(&self, filter: Filter) -> Result<broadcast::Receiver<Log>>;
+    async fn subscribe_new_heads(&self) -> Result<broadcast::Receiver<Header>>;
+    fn close(&self);
+}
+
+pub struct SafeEthClient {
+    provider: Arc<RootProvider<PubSubFrontend>>,
+    log_resub_interval: Duration,
+    header_timeout: Duration,
+    block_chunk_size: u64,
+    block_max_range: u64,
+    close_sender: broadcast::Sender<()>,
+}
+
+pub struct SafeEthClientOptions {
+    pub log_resub_interval: Duration,
+    pub header_timeout: Duration,
+    pub block_chunk_size: u64,
+    pub block_max_range: u64,
+}
+
+impl Default for SafeEthClientOptions {
+    fn default() -> Self {
+        Self {
+            log_resub_interval: Duration::from_secs(300),
+            header_timeout: Duration::from_secs(30),
+            block_chunk_size: 100,
+            block_max_range: 100,
+        }
+    }
+}
+
+impl SafeEthClient {
+    pub async fn new(ws_url: &str, options: SafeEthClientOptions) -> Result<Self> {
+        let ws = WsConnect::new(ws_url);
+        let provider = ProviderBuilder::new().on_ws(ws).await?;
+        let (close_sender, _) = broadcast::channel(1);
+
+        Ok(Self {
+            provider: Arc::new(provider),
+            log_resub_interval: options.log_resub_interval,
+            header_timeout: options.header_timeout,
+            block_chunk_size: options.block_chunk_size,
+            block_max_range: options.block_max_range,
+            close_sender,
+        })
+    }
+}
+
+#[async_trait]
+impl SafeClient for SafeEthClient {
+    async fn block_number(&self) -> Result<U256> {
+        Ok(U256::from(self.provider.get_block_number().await?))
+    }
+
+    async fn get_block(&self, block_number: BlockNumberOrTag) -> Result<Option<Block>> {
+        Ok(self.provider.get_block(alloy::eips::BlockId::Number(block_number), alloy::rpc::types::BlockTransactionsKind::Hashes).await?)
+    }
+
+    async fn get_transaction(&self, tx_hash: B256) -> Result<Option<Transaction>> {
+        Ok(self.provider.get_transaction_by_hash(tx_hash).await?)
+    }
+
+    async fn get_transaction_receipt(&self, tx_hash: B256) -> Result<Option<TransactionReceipt>> {
+        Ok(self.provider.get_transaction_receipt(tx_hash).await?)
+    }
+
+    async fn get_logs(&self, filter: Filter) -> Result<Vec<Log>> {
+        Ok(self.provider.get_logs(&filter).await?)
+    }
+
+    async fn subscribe_logs(&self, filter: Filter) -> Result<broadcast::Receiver<Log>> {
+        let (tx, rx) = broadcast::channel(100);
+        let log_resub_interval = self.log_resub_interval;
+        let subscription = self.provider.subscribe_logs(&filter).await?;
+        let mut stream = subscription.into_stream();
+
+        tokio::spawn(async move {
+            loop {
+                tokio::select! {
+                    Some(log) = stream.next() => {
+                        if tx.send(log).is_err() {
+                            error!("Error sending log: channel closed");
+                            break;
+                        }
+                    }
+                    _ = tokio::time::sleep(log_resub_interval) => {
+                        warn!("Timeout waiting for new log");
+                        break;
+                    }
+                    else => {
+                        error!("Log stream ended unexpectedly");
+                        break;
+                    }
+                }
+            }
+        });
+
+        Ok(rx)
+    }
+
+    async fn subscribe_new_heads(&self) -> Result<broadcast::Receiver<Header>> {
+        let (tx, rx) = broadcast::channel(100);
+        let header_timeout = self.header_timeout;
+        let subscription = self.provider.subscribe_blocks().await?;
+        let mut stream = subscription.into_stream();
+
+        tokio::spawn(async move {
+            loop {
+                tokio::select! {
+                    Some(block) = stream.next() => {
+                        if tx.send(block.header).is_err() {
+                            error!("Error sending header: channel closed");
+                            break;
+                        }
+                    }
+                    _ = tokio::time::sleep(header_timeout) => {
+                        warn!("Timeout waiting for new header");
+                        break;
+                    }
+                    else => {
+                        error!("Header stream ended unexpectedly");
+                        break;
+                    }
+                }
+            }
+        });
+
+        Ok(rx)
+    }
+
+    fn close(&self) {
+        let _ = self.close_sender.send(());
+    }
+}

core-rs/src/safeclient/mod.rs

@@ -0,0 +1,5 @@
+mod client;
+mod utils;
+
+pub use client::{SafeClient, SafeEthClient, SafeEthClientOptions};
+pub use utils::*;

core-rs/src/safeclient/utils.rs

@@ -0,0 +1,26 @@
+use alloy::{primitives::B256, rpc::types::Log};
+use sha2::{Digest, Sha256};
+
+pub fn hash_log(log: &Log) -> B256 {
+    let mut hasher = Sha256::new();
+
+    // Hash the log fields
+    hasher.update(log.address().as_slice());
+    for topic in log.topics() {
+        hasher.update(topic.as_slice());
+    }
+    hasher.update(log.data().data.clone());
+
+    // Hash additional block and tx info
+    if let Some(block_hash) = log.block_hash {
+        hasher.update(block_hash.as_slice());
+    }
+    if let Some(transaction_hash) = log.transaction_hash {
+        hasher.update(transaction_hash.as_slice());
+    }
+    if let Some(log_index) = log.log_index {
+        hasher.update(&log_index.to_be_bytes());
+    }
+
+    B256::from_slice(&hasher.finalize())
+}

indexer/Cargo.toml

@@ -9,29 +9,33 @@ edition = "2021"
 actix = "0.13.1"
 actix-web = "4.5.1"
 futures = "0.3.5"
-tokio = { version = "1.28.2", features = ["sync", "time"] }
+tokio = { workspace = true }
 deadpool = "0.10.0"
 lapin = "2.3.1"
 deadpool-lapin = "0.11.0"
 tokio-executor-trait = "2.1.0"
 tokio-reactor-trait = "1.1.0"
-prometheus = "0.13.3"
+prometheus = { workspace = true }
 
-clap = { version = "4.4.11", features = ["color", "derive", "env"] }
+clap = { workspace = true }
 openssl-probe = "0.1.4"
-serde = { version = "1", features = ["derive"] }
-serde_json = "1.0.68"
+serde = { workspace = true }
+serde_json = { workspace = true }
 
-tracing = { version = "0.1.36", features = ["std"] }
+tracing = { workspace = true }
 thiserror = "1.0.56"
-anyhow = "1.0.79"
+anyhow = { workspace = true }
 
 near-indexer = { git = "https://github.com/near/nearcore", rev = "b3d767e7664d8e123a35313ccc66c8ac1afb2058" }
 near-client = { git = "https://github.com/near/nearcore", rev = "b3d767e7664d8e123a35313ccc66c8ac1afb2058" }
 near-o11y = { git = "https://github.com/near/nearcore", rev = "b3d767e7664d8e123a35313ccc66c8ac1afb2058" }
 near-client-primitives = { git = "https://github.com/near/nearcore", rev = "b3d767e7664d8e123a35313ccc66c8ac1afb2058" }
+reqwest = { version = "0.12.7", features = ["json"] }
 borsh = { version = "1.0.0", features = ["derive", "rc"] }
-serde_yaml = "0.9.34"
+serde_yaml = { workspace = true }
 
 [dev-dependencies]
 near-crypto = { git = "https://github.com/near/nearcore", rev = "b3d767e7664d8e123a35313ccc66c8ac1afb2058" }
+
+[features]
+use_fastnear = []

indexer/Dockerfile

@@ -14,11 +14,11 @@ RUN apt-get update -qq && \
         llvm \
         clang
 
-COPY ./indexer/Cargo.toml .
+COPY ./Cargo.toml .
 RUN mkdir ./src && echo "fn main() {}" > ./src/main.rs
 RUN cargo build --release -p indexer
 
-COPY ./indexer .
+COPY . .
 RUN touch ./src/main.rs
 
 RUN cargo build --release -p indexer
@@ -29,7 +29,7 @@ WORKDIR /indexer-app
 RUN apt update && apt install -yy openssl ca-certificates jq curl
 
 COPY --from=builder /tmp/indexer/target/release/indexer .
-COPY ./indexer/entrypoint.sh ./entrypoint.sh
+COPY entrypoint.sh ./entrypoint.sh
 RUN chmod +x ./entrypoint.sh
 
 EXPOSE 3030

indexer/FastIndexer.dockerfile

@@ -0,0 +1,47 @@
+FROM rust:1.79 AS builder
+WORKDIR /tmp/indexer
+
+# Copy from nearcore:
+# https://github.com/near/nearcore/blob/master/Dockerfile
+RUN apt-get update -qq && \
+    apt-get install -y \
+        git \
+        cmake \
+        g++ \
+        pkg-config \
+        libssl-dev \
+        curl \
+        llvm \
+        clang
+
+COPY ./Cargo.toml .
+RUN mkdir ./src && echo "fn main() {}" > ./src/main.rs
+RUN cargo build --release -p indexer --features use_fastnear
+
+COPY . .
+RUN touch ./src/main.rs
+
+RUN cargo build --release -p indexer --features use_fastnear
+
+FROM debian:bookworm-slim as runtime
+WORKDIR /indexer-app
+
+RUN apt update && apt install -yy openssl ca-certificates jq curl
+
+COPY --from=builder /tmp/indexer/target/release/indexer .
+COPY entrypoint.sh ./entrypoint.sh
+RUN chmod +x ./entrypoint.sh
+
+EXPOSE 3030
+
+HEALTHCHECK --interval=20s --timeout=30s --retries=10000 \
+  CMD (curl -f -s -X POST -H "Content-Type: application/json" \
+    -d '{"jsonrpc":"2.0","method":"block","params":{"finality":"optimistic"},"id":"dontcare"}' \
+    http://localhost:3030 | \
+  jq -es 'if . == [] then null else .[] | (now - (.result.header.timestamp / 1000000000)) < 10 end') && \
+  (curl -f -s -X POST -H "Content-Type: application/json" \
+    -d '{"jsonrpc":"2.0","method":"status","params":[],"id":"dontcare"}' \
+    http://localhost:3030 | \
+  jq -es 'if . == [] then null else .[] | .result.sync_info.syncing == false end')
+
+ENTRYPOINT [ "./entrypoint.sh" ]