From 956dc6278ad6c8581b3ce4177afa39467f89e6bd Mon Sep 17 00:00:00 2001 From: erhant Date: Fri, 19 Jun 2026 12:29:19 +0300 Subject: [PATCH 01/14] fix!(indexer_ffi): fix 3 existing issues, refactor the runtime handling, rm unused cbindgen.toml --- .gitignore | 8 + Cargo.lock | 8 +- .../tests/indexer_ffi_block_batching.rs | 26 +- .../tests/indexer_ffi_helpers/mod.rs | 11 +- .../tests/indexer_test_run_ffi.rs | 15 +- lez/indexer/ffi/Cargo.toml | 12 +- lez/indexer/ffi/build.rs | 1 + lez/indexer/ffi/cbindgen.toml | 2 - lez/indexer/ffi/indexer_ffi.h | 150 +++++++----- lez/indexer/ffi/src/api/client.rs | 36 --- lez/indexer/ffi/src/api/lifecycle.rs | 57 +++-- lez/indexer/ffi/src/api/logging.rs | 14 ++ lez/indexer/ffi/src/api/mod.rs | 2 +- lez/indexer/ffi/src/api/query.rs | 222 ++++++++++-------- lez/indexer/ffi/src/api/types/account.rs | 19 +- lez/indexer/ffi/src/api/types/block.rs | 73 +++--- lez/indexer/ffi/src/api/types/transaction.rs | 91 ++++--- lez/indexer/ffi/src/client.rs | 33 --- lez/indexer/ffi/src/indexer.rs | 102 ++++---- lez/indexer/ffi/src/lib.rs | 1 - 20 files changed, 456 insertions(+), 427 deletions(-) delete mode 100644 lez/indexer/ffi/cbindgen.toml delete mode 100644 lez/indexer/ffi/src/api/client.rs create mode 100644 lez/indexer/ffi/src/api/logging.rs delete mode 100644 lez/indexer/ffi/src/client.rs diff --git a/.gitignore b/.gitignore index 4605856c..8041f277 100644 --- a/.gitignore +++ b/.gitignore @@ -1,17 +1,25 @@ .gitconfig + res/ target/ deps/ data/ + .idea/ .vscode/ + rocksdb sequencer/service/data/ storage.json + result + wallet-ffi/wallet_ffi.h bedrock_signing_key integration_tests/configs/debug/ venv/ + keycard_wallet/python/__pycache__/ keycard_wallet/python/keycard-py/ + +.DS_Store diff --git a/Cargo.lock b/Cargo.lock index ee5c97f2..9af3ff27 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3823,16 +3823,14 @@ dependencies = [ name = "indexer_ffi" version = "0.1.0" dependencies = [ - "anyhow", "cbindgen", - "indexer_service", + "env_logger", + "futures", + "indexer_core", "indexer_service_protocol", - "indexer_service_rpc", - "jsonrpsee", "lee", "log", "tokio", - "url", ] [[package]] diff --git a/integration_tests/tests/indexer_ffi_block_batching.rs b/integration_tests/tests/indexer_ffi_block_batching.rs index eeef276d..52f68027 100644 --- a/integration_tests/tests/indexer_ffi_block_batching.rs +++ b/integration_tests/tests/indexer_ffi_block_batching.rs @@ -5,7 +5,7 @@ )] use anyhow::Result; -use indexer_ffi::{Runtime, api::types::FfiOption}; +use indexer_ffi::api::types::FfiOption; use integration_tests::L2_TO_L1_TIMEOUT; use log::info; @@ -14,21 +14,21 @@ mod indexer_ffi_helpers; #[test] fn indexer_ffi_block_batching() -> Result<()> { - let (ctx, indexer_ffi, _indexer_dir) = indexer_ffi_helpers::setup()?; + // `_ctx` keeps the bedrock/sequencer harness (and its runtime) alive for the + // duration of the test; the indexer was started on that runtime. + let (_ctx, indexer_ffi, _indexer_dir) = indexer_ffi_helpers::setup()?; // WAIT info!("Waiting for indexer to parse blocks"); std::thread::sleep(L2_TO_L1_TIMEOUT); - // Safety: ctx runtime is valid for the lifetime of the returned Runtime - let runtime = unsafe { Runtime::from_borrowed(ctx.runtime()) }; - let last_block_indexer_ffi_res = unsafe { - indexer_ffi_helpers::query_last_block(&raw const runtime, &raw const indexer_ffi) - }; + let last_block_indexer_ffi_res = + unsafe { indexer_ffi_helpers::query_last_block(&raw const indexer_ffi) }; assert!(last_block_indexer_ffi_res.error.is_ok()); + assert!(last_block_indexer_ffi_res.is_some); - let last_block_indexer = unsafe { *last_block_indexer_ffi_res.value }; + let last_block_indexer = last_block_indexer_ffi_res.block_id; info!("Last block on indexer FFI now is {last_block_indexer}"); @@ -37,14 +37,8 @@ fn indexer_ffi_block_batching() -> Result<()> { let before_ffi = FfiOption::::from_none(); let limit = 100; - let block_batch_ffi_res = unsafe { - indexer_ffi_helpers::query_block_vec( - &raw const runtime, - &raw const indexer_ffi, - before_ffi, - limit, - ) - }; + let block_batch_ffi_res = + unsafe { indexer_ffi_helpers::query_block_vec(&raw const indexer_ffi, before_ffi, limit) }; assert!(block_batch_ffi_res.error.is_ok()); diff --git a/integration_tests/tests/indexer_ffi_helpers/mod.rs b/integration_tests/tests/indexer_ffi_helpers/mod.rs index c3c3caff..35bf7ef3 100644 --- a/integration_tests/tests/indexer_ffi_helpers/mod.rs +++ b/integration_tests/tests/indexer_ffi_helpers/mod.rs @@ -13,6 +13,7 @@ use indexer_ffi::{ api::{ PointerResult, lifecycle::InitializedIndexerServiceFFIResult, + query::LastBlockIdResult, types::{FfiAccountId, FfiOption, FfiVec, account::FfiAccount, block::FfiBlock}, }, }; @@ -20,20 +21,15 @@ use integration_tests::{BlockingTestContext, TestContext}; use tempfile::TempDir; unsafe extern "C" { - pub unsafe fn query_last_block( - runtime: *const Runtime, - indexer: *const IndexerServiceFFI, - ) -> PointerResult; + pub unsafe fn query_last_block(indexer: *const IndexerServiceFFI) -> LastBlockIdResult; pub unsafe fn query_block_vec( - runtime: *const Runtime, indexer: *const IndexerServiceFFI, before: FfiOption, limit: u64, ) -> PointerResult, OperationStatus>; pub unsafe fn query_account( - runtime: *const Runtime, indexer: *const IndexerServiceFFI, account_id: FfiAccountId, ) -> PointerResult; @@ -41,7 +37,6 @@ unsafe extern "C" { pub unsafe fn start_indexer( runtime: *const Runtime, config_path: *const c_char, - port: u16, ) -> InitializedIndexerServiceFFIResult; } @@ -69,7 +64,7 @@ pub fn setup_indexer_ffi( let res = // SAFETY: lib function ensures validity of value. - unsafe { start_indexer(std::ptr::from_ref(runtime), CString::new(config_path.to_str().unwrap())?.as_ptr(), 0) }; + unsafe { start_indexer(std::ptr::from_ref(runtime), CString::new(config_path.to_str().unwrap())?.as_ptr()) }; if res.error.is_error() { anyhow::bail!("Indexer FFI error {:?}", res.error); diff --git a/integration_tests/tests/indexer_test_run_ffi.rs b/integration_tests/tests/indexer_test_run_ffi.rs index 2c7c2103..0a5f4133 100644 --- a/integration_tests/tests/indexer_test_run_ffi.rs +++ b/integration_tests/tests/indexer_test_run_ffi.rs @@ -5,7 +5,6 @@ )] use anyhow::Result; -use indexer_ffi::Runtime; use integration_tests::L2_TO_L1_TIMEOUT; use log::info; @@ -14,20 +13,20 @@ mod indexer_ffi_helpers; #[test] fn indexer_test_run_ffi() -> Result<()> { - let (ctx, indexer_ffi, _indexer_dir) = indexer_ffi_helpers::setup()?; + // `_ctx` keeps the bedrock/sequencer harness (and its runtime) alive for the + // duration of the test; the indexer was started on that runtime. + let (_ctx, indexer_ffi, _indexer_dir) = indexer_ffi_helpers::setup()?; // RUN OBSERVATION std::thread::sleep(L2_TO_L1_TIMEOUT); - // Safety: ctx runtime is valid for the lifetime of the returned Runtime - let runtime = unsafe { Runtime::from_borrowed(ctx.runtime()) }; - let last_block_indexer_ffi_res = unsafe { - indexer_ffi_helpers::query_last_block(&raw const runtime, &raw const indexer_ffi) - }; + let last_block_indexer_ffi_res = + unsafe { indexer_ffi_helpers::query_last_block(&raw const indexer_ffi) }; assert!(last_block_indexer_ffi_res.error.is_ok()); + assert!(last_block_indexer_ffi_res.is_some); - let last_block_indexer_ffi = unsafe { *last_block_indexer_ffi_res.value }; + let last_block_indexer_ffi = last_block_indexer_ffi_res.block_id; info!("Last block on indexer FFI now is {last_block_indexer_ffi}"); diff --git a/lez/indexer/ffi/Cargo.toml b/lez/indexer/ffi/Cargo.toml index 66f6a518..50748d56 100644 --- a/lez/indexer/ffi/Cargo.toml +++ b/lez/indexer/ffi/Cargo.toml @@ -6,15 +6,15 @@ version = "0.1.0" [dependencies] lee.workspace = true -indexer_service.workspace = true -indexer_service_rpc = { workspace = true, features = ["client"] } -indexer_service_protocol.workspace = true +indexer_core.workspace = true +# `convert` brings in the LEE <-> protocol `From`/`TryFrom` impls the query +# marshalling relies on (previously pulled in transitively via `indexer_service`). +indexer_service_protocol = { workspace = true, features = ["convert"] } -url.workspace = true +env_logger.workspace = true log = { workspace = true } tokio = { features = ["rt-multi-thread"], workspace = true } -jsonrpsee.workspace = true -anyhow.workspace = true +futures.workspace = true [build-dependencies] cbindgen = "0.29" diff --git a/lez/indexer/ffi/build.rs b/lez/indexer/ffi/build.rs index 92c95407..349ff38d 100644 --- a/lez/indexer/ffi/build.rs +++ b/lez/indexer/ffi/build.rs @@ -6,6 +6,7 @@ fn main() { cbindgen::Builder::new() .with_crate(crate_dir) .with_language(cbindgen::Language::C) + .with_pragma_once(true) .generate() .expect("Unable to generate bindings") .write_to_file("indexer_ffi.h"); diff --git a/lez/indexer/ffi/cbindgen.toml b/lez/indexer/ffi/cbindgen.toml deleted file mode 100644 index 79f622b7..00000000 --- a/lez/indexer/ffi/cbindgen.toml +++ /dev/null @@ -1,2 +0,0 @@ -language = "C" # For increased compatibility -no_includes = true diff --git a/lez/indexer/ffi/indexer_ffi.h b/lez/indexer/ffi/indexer_ffi.h index 84aaeae7..06953e39 100644 --- a/lez/indexer/ffi/indexer_ffi.h +++ b/lez/indexer/ffi/indexer_ffi.h @@ -1,3 +1,5 @@ +#pragma once + #include #include #include @@ -22,11 +24,19 @@ typedef enum FfiBedrockStatus { Finalized, } FfiBedrockStatus; -typedef struct Option_u64 Option_u64; - +/** + * FFI-owned indexer. + * + * Has three fields behind `c_void` (so that cbindgen never needs to see their Rust layout): + * - An [`IndexerCore`] used to answer queries + * - The background task [`JoinHandle`] that drives ingestion (consuming the block stream so the + * store stays populated) + * - A [`Handle`] to the runtime they live on. + */ typedef struct IndexerServiceFFI { - void *indexer_handle; - void *indexer_client; + void *core; + void *ingest_handle; + void *runtime_handle; } IndexerServiceFFI; /** @@ -83,15 +93,18 @@ typedef struct PointerResult_Runtime__OperationStatus { } PointerResult_Runtime__OperationStatus; /** - * Simple wrapper around a pointer to a value or an error. + * Result of [`query_last_block`], returned **inline** (no heap allocation, so + * there is no corresponding `free_*` to call). * - * Pointer is not guaranteed. You should check the error field before - * dereferencing the pointer. + * `block_id` is only meaningful when `error` is `Ok` *and* `is_some` is + * `true`. An `Ok` result with `is_some == false` means the indexer has no + * finalized block yet (an empty chain) — which is distinct from an error. */ -typedef struct PointerResult_Option_u64_____OperationStatus { - struct Option_u64 *value; +typedef struct LastBlockIdResult { + uint64_t block_id; + bool is_some; enum OperationStatus error; -} PointerResult_Option_u64_____OperationStatus; +} LastBlockIdResult; typedef uint64_t FfiBlockId; @@ -411,7 +424,6 @@ typedef struct PointerResult_FfiVec_FfiTransaction_____OperationStatus { * # Arguments * * - `config_path`: A pointer to a string representing the path to the configuration file. - * - `port`: Number representing a port, on which indexers RPC will start. * * # Returns * @@ -424,8 +436,7 @@ typedef struct PointerResult_FfiVec_FfiTransaction_____OperationStatus { * - `config_path` is a valid pointer to a null-terminated C string. */ InitializedIndexerServiceFFIResult start_indexer(const struct Runtime *runtime, - const char *config_path, - uint16_t port); + const char *config_path); /** * Creates a new [`tokio::runtime::Runtime`]. @@ -452,6 +463,20 @@ struct PointerResult_Runtime__OperationStatus new_runtime(void); */ enum OperationStatus stop_indexer(struct IndexerServiceFFI *indexer); +/** + * Initializes the FFI's logger. + * + * Wires up `env_logger`, so the library's `log::*` output is controlled by the + * `RUST_LOG` environment variable (e.g. `RUST_LOG=info`). Without this, the + * FFI's log calls go nowhere — and since failures are otherwise reported only + * as numeric [`OperationStatus`](crate::errors::OperationStatus) codes, there + * is no other way to see *why* a call failed. + * + * Safe to call multiple times and from any consumer: if a global logger is + * already set, the call is a no-op. + */ +void init_logger(void); + /** * # Safety * It's up to the caller to pass a proper pointer, if somehow from c/c++ side @@ -469,16 +494,15 @@ void free_cstring(char *block); * * # Returns * - * A `PointerResult, OperationStatus>` indicating success or failure. + * A [`LastBlockIdResult`] indicating success or failure. The block id is + * returned inline; nothing needs to be freed. * * # Safety * * The caller must ensure that: - * - `runtime` is a valid pointer to a [`Runtime`] instance. * - `indexer` is a valid pointer to a [`IndexerServiceFFI`] instance. */ -struct PointerResult_Option_u64_____OperationStatus query_last_block(const struct Runtime *runtime, - const struct IndexerServiceFFI *indexer); +struct LastBlockIdResult query_last_block(const struct IndexerServiceFFI *indexer); /** * Query the block by id from indexer. @@ -495,15 +519,13 @@ struct PointerResult_Option_u64_____OperationStatus query_last_block(const struc * # Safety * * The caller must ensure that: - * - `runtime` is a valid pointer to a [`Runtime`] instance. * - `indexer` is a valid pointer to a [`IndexerServiceFFI`] instance. */ -struct PointerResult_FfiBlockOpt__OperationStatus query_block(const struct Runtime *runtime, - const struct IndexerServiceFFI *indexer, +struct PointerResult_FfiBlockOpt__OperationStatus query_block(const struct IndexerServiceFFI *indexer, FfiBlockId block_id); /** - * Query the block by id from indexer. + * Query the block by hash from indexer. * * # Arguments * @@ -517,11 +539,9 @@ struct PointerResult_FfiBlockOpt__OperationStatus query_block(const struct Runti * # Safety * * The caller must ensure that: - * - `runtime` is a valid pointer to a [`Runtime`] instance. * - `indexer` is a valid pointer to a [`IndexerServiceFFI`] instance. */ -struct PointerResult_FfiBlockOpt__OperationStatus query_block_by_hash(const struct Runtime *runtime, - const struct IndexerServiceFFI *indexer, +struct PointerResult_FfiBlockOpt__OperationStatus query_block_by_hash(const struct IndexerServiceFFI *indexer, FfiHashType hash); /** @@ -539,15 +559,13 @@ struct PointerResult_FfiBlockOpt__OperationStatus query_block_by_hash(const stru * # Safety * * The caller must ensure that: - * - `runtime` is a valid pointer to a [`Runtime`] instance. * - `indexer` is a valid pointer to a [`IndexerServiceFFI`] instance. */ -struct PointerResult_FfiAccount__OperationStatus query_account(const struct Runtime *runtime, - const struct IndexerServiceFFI *indexer, +struct PointerResult_FfiAccount__OperationStatus query_account(const struct IndexerServiceFFI *indexer, FfiAccountId account_id); /** - * Query the trasnaction by hash from indexer. + * Query the transaction by hash from indexer. * * # Arguments * @@ -562,10 +580,8 @@ struct PointerResult_FfiAccount__OperationStatus query_account(const struct Runt * * The caller must ensure that: * - `indexer` is a valid pointer to a [`IndexerServiceFFI`] instance. - * - `runtime` is a valid pointer to a [`Runtime`] instance. */ -struct PointerResult_FfiOption_FfiTransaction_____OperationStatus query_transaction(const struct Runtime *runtime, - const struct IndexerServiceFFI *indexer, +struct PointerResult_FfiOption_FfiTransaction_____OperationStatus query_transaction(const struct IndexerServiceFFI *indexer, FfiHashType hash); /** @@ -585,10 +601,8 @@ struct PointerResult_FfiOption_FfiTransaction_____OperationStatus query_transact * * The caller must ensure that: * - `indexer` is a valid pointer to a [`IndexerServiceFFI`] instance. - * - `runtime` is a valid pointer to a [`Runtime`] instance. */ -struct PointerResult_FfiVec_FfiBlock_____OperationStatus query_block_vec(const struct Runtime *runtime, - const struct IndexerServiceFFI *indexer, +struct PointerResult_FfiVec_FfiBlock_____OperationStatus query_block_vec(const struct IndexerServiceFFI *indexer, struct FfiOption_u64 before, uint64_t limit); @@ -604,16 +618,14 @@ struct PointerResult_FfiVec_FfiBlock_____OperationStatus query_block_vec(const s * * # Returns * - * A `PointerResult, OperationStatus>` indicating success or failure. + * A `PointerResult, OperationStatus>` indicating success or failure. * * # Safety * * The caller must ensure that: * - `indexer` is a valid pointer to a [`IndexerServiceFFI`] instance. - * - `runtime` is a valid pointer to a [`Runtime`] instance. */ -struct PointerResult_FfiVec_FfiTransaction_____OperationStatus query_transactions_by_account(const struct Runtime *runtime, - const struct IndexerServiceFFI *indexer, +struct PointerResult_FfiVec_FfiTransaction_____OperationStatus query_transactions_by_account(const struct IndexerServiceFFI *indexer, FfiAccountId account_id, uint64_t offset, uint64_t limit); @@ -621,9 +633,14 @@ struct PointerResult_FfiVec_FfiTransaction_____OperationStatus query_transaction /** * Frees the resources associated with the given ffi account. * + * Takes ownership of the whole allocation produced by a `query_*` call: the + * outer `Box` (the `PointerResult.value` pointer) *and* its inner + * data buffer. Passing the struct by value previously freed only the inner + * buffer and leaked the outer box. + * * # Arguments * - * - `val`: An instance of `FfiAccount`. + * - `val`: The `*mut FfiAccount` returned in `PointerResult.value`. * * # Returns * @@ -632,12 +649,18 @@ struct PointerResult_FfiVec_FfiTransaction_____OperationStatus query_transaction * # Safety * * The caller must ensure that: - * - `val` is a valid instance of `FfiAccount`. + * - `val` is a pointer to an `FfiAccount` produced by this library and not yet freed. */ -void free_ffi_account(struct FfiAccount val); +void free_ffi_account(struct FfiAccount *val); /** - * Frees the resources associated with the given ffi block. + * Frees the resources owned by an `FfiBlock` value. + * + * This frees the block's transaction bodies (the only heap-owning field); the + * header/status fields are `Copy`. It operates on the struct by value because + * it is an element-level helper, used both for the vector path + * ([`free_ffi_block_vec`]) and the optional path ([`free_ffi_block_opt`]) — in + * neither case is an `FfiBlock` itself wrapped in its own outer box. * * # Arguments * @@ -650,16 +673,20 @@ void free_ffi_account(struct FfiAccount val); * # Safety * * The caller must ensure that: - * - `val` is a valid instance of `FfiBlock`. + * - `val` is a valid instance of `FfiBlock` produced by this library and not yet freed. */ void free_ffi_block(struct FfiBlock val); /** * Frees the resources associated with the given ffi block option. * + * Takes ownership of the whole allocation produced by a `query_*` call: the + * outer `Box` (the `PointerResult.value` pointer), the inner + * `Box` (when present), and that block's transaction bodies. + * * # Arguments * - * - `val`: An instance of `FfiBlockOpt`. + * - `val`: The `*mut FfiBlockOpt` returned in `PointerResult.value`. * * # Returns * @@ -668,16 +695,20 @@ void free_ffi_block(struct FfiBlock val); * # Safety * * The caller must ensure that: - * - `val` is a valid instance of `FfiBlockOpt`. + * - `val` is a pointer to an `FfiBlockOpt` produced by this library and not yet freed. */ -void free_ffi_block_opt(FfiBlockOpt val); +void free_ffi_block_opt(FfiBlockOpt *val); /** * Frees the resources associated with the given ffi block vector. * + * Takes ownership of the whole allocation produced by a `query_*` call: the + * outer `Box>` (the `PointerResult.value` pointer), the + * vector's backing buffer, and every block within it. + * * # Arguments * - * - `val`: An instance of `FfiVec`. + * - `val`: The `*mut FfiVec` returned in `PointerResult.value`. * * # Returns * @@ -686,9 +717,9 @@ void free_ffi_block_opt(FfiBlockOpt val); * # Safety * * The caller must ensure that: - * - `val` is a valid instance of `FfiVec`. + * - `val` is a pointer to an `FfiVec` produced by this library and not yet freed. */ -void free_ffi_block_vec(struct FfiVec_FfiBlock val); +void free_ffi_block_vec(struct FfiVec_FfiBlock *val); /** * Frees the resources associated with the given ffi transaction. @@ -711,9 +742,13 @@ void free_ffi_transaction(struct FfiTransaction val); /** * Frees the resources associated with the given ffi transaction option. * + * Takes ownership of the whole allocation produced by a `query_*` call: the + * outer `Box>` (the `PointerResult.value` pointer), + * the inner `Box` (when present), and its body. + * * # Arguments * - * - `val`: An instance of `FfiOption`. + * - `val`: The `*mut FfiOption` returned in `PointerResult.value`. * * # Returns * @@ -722,16 +757,21 @@ void free_ffi_transaction(struct FfiTransaction val); * # Safety * * The caller must ensure that: - * - `val` is a valid instance of `FfiOption`. + * - `val` is a pointer to an `FfiOption` produced by this library and not yet + * freed. */ -void free_ffi_transaction_opt(struct FfiOption_FfiTransaction val); +void free_ffi_transaction_opt(struct FfiOption_FfiTransaction *val); /** * Frees the resources associated with the given vector of ffi transactions. * + * Takes ownership of the whole allocation produced by a `query_*` call: the + * outer `Box>` (the `PointerResult.value` pointer), the + * vector's backing buffer, and every transaction within it. + * * # Arguments * - * - `val`: An instance of `FfiVec`. + * - `val`: The `*mut FfiVec` returned in `PointerResult.value`. * * # Returns * @@ -740,9 +780,9 @@ void free_ffi_transaction_opt(struct FfiOption_FfiTransaction val); * # Safety * * The caller must ensure that: - * - `val` is a valid instance of `FfiVec`. + * - `val` is a pointer to an `FfiVec` produced by this library and not yet freed. */ -void free_ffi_transaction_vec(struct FfiVec_FfiTransaction val); +void free_ffi_transaction_vec(struct FfiVec_FfiTransaction *val); bool is_ok(const enum OperationStatus *self); diff --git a/lez/indexer/ffi/src/api/client.rs b/lez/indexer/ffi/src/api/client.rs deleted file mode 100644 index 825a57de..00000000 --- a/lez/indexer/ffi/src/api/client.rs +++ /dev/null @@ -1,36 +0,0 @@ -use std::net::SocketAddr; - -use url::Url; - -use crate::OperationStatus; - -#[derive(Debug, Clone, Copy)] -pub enum UrlProtocol { - Http, - Ws, -} - -impl std::fmt::Display for UrlProtocol { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - match self { - Self::Http => write!(f, "http"), - Self::Ws => write!(f, "ws"), - } - } -} - -pub(crate) fn addr_to_url(protocol: UrlProtocol, addr: SocketAddr) -> Result { - // 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: - // but clients need to connect to 127.0.0.1: to work reliably - let url_string = if addr.ip().is_unspecified() { - format!("{protocol}://127.0.0.1:{}", addr.port()) - } else { - format!("{protocol}://{addr}") - }; - - url_string.parse().map_err(|e| { - log::error!("Could not parse indexer url: {e}"); - OperationStatus::InitializationError - }) -} diff --git a/lez/indexer/ffi/src/api/lifecycle.rs b/lez/indexer/ffi/src/api/lifecycle.rs index d124901f..bedebb2b 100644 --- a/lez/indexer/ffi/src/api/lifecycle.rs +++ b/lez/indexer/ffi/src/api/lifecycle.rs @@ -1,14 +1,9 @@ use std::{ffi::c_char, path::PathBuf}; -use crate::{ - IndexerServiceFFI, Runtime, - api::{ - PointerResult, - client::{UrlProtocol, addr_to_url}, - }, - client::{IndexerClient, IndexerClientTrait as _}, - errors::OperationStatus, -}; +use futures::StreamExt as _; +use indexer_core::{IndexerCore, config::IndexerConfig}; + +use crate::{IndexerServiceFFI, Runtime, api::PointerResult, errors::OperationStatus}; pub type InitializedIndexerServiceFFIResult = PointerResult; @@ -18,7 +13,6 @@ pub type InitializedIndexerServiceFFIResult = PointerResult InitializedIndexerServiceFFIResult { // SAFETY: The caller must ensure the validness of the `runtime` and `config_path` pointers. - unsafe { setup_indexer(runtime, config_path, port) }.map_or_else( + unsafe { setup_indexer(runtime, config_path) }.map_or_else( InitializedIndexerServiceFFIResult::from_error, InitializedIndexerServiceFFIResult::from_value, ) @@ -57,7 +50,6 @@ pub extern "C" fn new_runtime() -> PointerResult { /// # Arguments /// /// - `config_path`: A pointer to a string representing the path to the configuration file. -/// - `port`: Number representing a port, on which indexers RPC will start. /// /// # Returns /// @@ -71,7 +63,6 @@ pub extern "C" fn new_runtime() -> PointerResult { unsafe fn setup_indexer( runtime: *const Runtime, config_path: *const c_char, - port: u16, ) -> Result { let user_config_path = PathBuf::from( unsafe { std::ffi::CStr::from_ptr(config_path) } @@ -81,7 +72,7 @@ unsafe fn setup_indexer( OperationStatus::InitializationError })?, ); - let config = indexer_service::IndexerConfig::from_path(&user_config_path).map_err(|e| { + let config = IndexerConfig::from_path(&user_config_path).map_err(|e| { log::error!("Failed to read config: {e}"); OperationStatus::InitializationError })?; @@ -90,22 +81,30 @@ unsafe fn setup_indexer( // `tokio::runtime::Runtime` instance. let runtime = unsafe { &*runtime }; - let indexer_handle = runtime - .block_on(indexer_service::run_server(config, port)) - .map_err(|e| { - log::error!("Could not start indexer service: {e}"); - OperationStatus::InitializationError - })?; + let core = IndexerCore::new(config).map_err(|e| { + log::error!("Could not initialize indexer core: {e}"); + OperationStatus::InitializationError + })?; - let indexer_url = addr_to_url(UrlProtocol::Ws, indexer_handle.addr())?; - let indexer_client = runtime - .block_on(IndexerClient::new(&indexer_url)) - .map_err(|e| { - log::error!("Could not start indexer client: {e}"); - OperationStatus::InitializationError - })?; + // The block stream writes each parsed block into the store as a side effect + // of being polled, so we spawn a task that simply drains it. There are no + // subscribers — queries read the store directly via `core()`. + let ingest_core = core.clone(); + let ingest_handle = runtime.spawn(async move { + let mut block_stream = std::pin::pin!(ingest_core.subscribe_parse_block_stream()); + while let Some(result) = block_stream.next().await { + if let Err(e) = result { + log::error!("Indexer ingestion error: {e:#}"); + } + } + log::warn!("Indexer block stream ended"); + }); - Ok(IndexerServiceFFI::new(indexer_handle, indexer_client)) + Ok(IndexerServiceFFI::new( + core, + ingest_handle, + runtime.handle().clone(), + )) } /// Stops and frees the resources associated with the given indexer service. diff --git a/lez/indexer/ffi/src/api/logging.rs b/lez/indexer/ffi/src/api/logging.rs new file mode 100644 index 00000000..207d4b61 --- /dev/null +++ b/lez/indexer/ffi/src/api/logging.rs @@ -0,0 +1,14 @@ +/// Initializes the FFI's logger. +/// +/// Wires up `env_logger`, so the library's `log::*` output is controlled by the +/// `RUST_LOG` environment variable (e.g. `RUST_LOG=info`). Without this, the +/// FFI's log calls go nowhere — and since failures are otherwise reported only +/// as numeric [`OperationStatus`](crate::errors::OperationStatus) codes, there +/// is no other way to see *why* a call failed. +/// +/// Safe to call multiple times and from any consumer: if a global logger is +/// already set, the call is a no-op. +#[unsafe(no_mangle)] +pub extern "C" fn init_logger() { + let _ignore_me = env_logger::try_init(); +} diff --git a/lez/indexer/ffi/src/api/mod.rs b/lez/indexer/ffi/src/api/mod.rs index ea2b91d7..2e4be797 100644 --- a/lez/indexer/ffi/src/api/mod.rs +++ b/lez/indexer/ffi/src/api/mod.rs @@ -1,7 +1,7 @@ pub use result::PointerResult; -pub mod client; pub mod lifecycle; +pub mod logging; pub mod memory; pub mod query; pub mod result; diff --git a/lez/indexer/ffi/src/api/query.rs b/lez/indexer/ffi/src/api/query.rs index f10de598..548deabc 100644 --- a/lez/indexer/ffi/src/api/query.rs +++ b/lez/indexer/ffi/src/api/query.rs @@ -1,8 +1,7 @@ -use indexer_service_protocol::{AccountId, HashType}; -use indexer_service_rpc::RpcClient as _; +use indexer_service_protocol::AccountId; use crate::{ - IndexerServiceFFI, Runtime, + IndexerServiceFFI, api::{ PointerResult, types::{ @@ -15,6 +14,45 @@ use crate::{ errors::OperationStatus, }; +/// Result of [`query_last_block`], returned **inline** (no heap allocation, so +/// there is no corresponding `free_*` to call). +/// +/// `block_id` is only meaningful when `error` is `Ok` *and* `is_some` is +/// `true`. An `Ok` result with `is_some == false` means the indexer has no +/// finalized block yet (an empty chain) — which is distinct from an error. +#[repr(C)] +pub struct LastBlockIdResult { + pub block_id: u64, + pub is_some: bool, + pub error: OperationStatus, +} + +impl LastBlockIdResult { + const fn error(error: OperationStatus) -> Self { + Self { + block_id: 0, + is_some: false, + error, + } + } + + const fn none() -> Self { + Self { + block_id: 0, + is_some: false, + error: OperationStatus::Ok, + } + } + + const fn some(block_id: u64) -> Self { + Self { + block_id, + is_some: true, + error: OperationStatus::Ok, + } + } +} + /// Query the last block id from indexer. /// /// # Arguments @@ -23,34 +61,29 @@ use crate::{ /// /// # Returns /// -/// A `PointerResult, OperationStatus>` indicating success or failure. +/// A [`LastBlockIdResult`] indicating success or failure. The block id is +/// returned inline; nothing needs to be freed. /// /// # Safety /// /// The caller must ensure that: -/// - `runtime` is a valid pointer to a [`Runtime`] instance. /// - `indexer` is a valid pointer to a [`IndexerServiceFFI`] instance. #[unsafe(no_mangle)] -pub unsafe extern "C" fn query_last_block( - runtime: *const Runtime, - indexer: *const IndexerServiceFFI, -) -> PointerResult, OperationStatus> { +pub unsafe extern "C" fn query_last_block(indexer: *const IndexerServiceFFI) -> LastBlockIdResult { if indexer.is_null() { log::error!("Attempted to query a null indexer pointer. This is a bug. Aborting."); - return PointerResult::from_error(OperationStatus::NullPointer); + return LastBlockIdResult::error(OperationStatus::NullPointer); } let indexer = unsafe { &*indexer }; - let client = indexer.client(); - let runtime = unsafe { &*runtime }; - - runtime - .block_on(client.get_last_finalized_block_id()) - .map_or_else( - |_| PointerResult::from_error(OperationStatus::ClientError), - PointerResult::from_value, - ) + indexer.core().store.get_last_block_id().map_or_else( + |e| { + log::error!("Failed to query last block id: {e:#}"); + LastBlockIdResult::error(OperationStatus::ClientError) + }, + |opt| opt.map_or_else(LastBlockIdResult::none, LastBlockIdResult::some), + ) } /// Query the block by id from indexer. @@ -67,11 +100,9 @@ pub unsafe extern "C" fn query_last_block( /// # Safety /// /// The caller must ensure that: -/// - `runtime` is a valid pointer to a [`Runtime`] instance. /// - `indexer` is a valid pointer to a [`IndexerServiceFFI`] instance. #[unsafe(no_mangle)] pub unsafe extern "C" fn query_block( - runtime: *const Runtime, indexer: *const IndexerServiceFFI, block_id: FfiBlockId, ) -> PointerResult { @@ -82,24 +113,23 @@ pub unsafe extern "C" fn query_block( let indexer = unsafe { &*indexer }; - let client = indexer.client(); - let runtime = unsafe { &*runtime }; + indexer.core().store.get_block_at_id(block_id).map_or_else( + |e| { + log::error!("Failed to query block by id: {e:#}"); + PointerResult::from_error(OperationStatus::ClientError) + }, + |block_opt| { + let block_ffi = block_opt.map_or_else(FfiBlockOpt::from_none, |block| { + let block: indexer_service_protocol::Block = block.into(); + FfiBlockOpt::from_value(block.into()) + }); - runtime - .block_on(client.get_block_by_id(block_id)) - .map_or_else( - |_| PointerResult::from_error(OperationStatus::ClientError), - |block_opt| { - let block_ffi = block_opt.map_or_else(FfiBlockOpt::from_none, |block| { - FfiBlockOpt::from_value(block.into()) - }); - - PointerResult::from_value(block_ffi) - }, - ) + PointerResult::from_value(block_ffi) + }, + ) } -/// Query the block by id from indexer. +/// Query the block by hash from indexer. /// /// # Arguments /// @@ -113,11 +143,9 @@ pub unsafe extern "C" fn query_block( /// # Safety /// /// The caller must ensure that: -/// - `runtime` is a valid pointer to a [`Runtime`] instance. /// - `indexer` is a valid pointer to a [`IndexerServiceFFI`] instance. #[unsafe(no_mangle)] pub unsafe extern "C" fn query_block_by_hash( - runtime: *const Runtime, indexer: *const IndexerServiceFFI, hash: FfiHashType, ) -> PointerResult { @@ -128,15 +156,18 @@ pub unsafe extern "C" fn query_block_by_hash( let indexer = unsafe { &*indexer }; - let client = indexer.client(); - let runtime = unsafe { &*runtime }; - - runtime - .block_on(client.get_block_by_hash(HashType(hash.data))) + indexer + .core() + .store + .get_block_by_hash(hash.data) .map_or_else( - |_| PointerResult::from_error(OperationStatus::ClientError), + |e| { + log::error!("Failed to query block by hash: {e:#}"); + PointerResult::from_error(OperationStatus::ClientError) + }, |block_opt| { let block_ffi = block_opt.map_or_else(FfiBlockOpt::from_none, |block| { + let block: indexer_service_protocol::Block = block.into(); FfiBlockOpt::from_value(block.into()) }); @@ -159,11 +190,9 @@ pub unsafe extern "C" fn query_block_by_hash( /// # Safety /// /// The caller must ensure that: -/// - `runtime` is a valid pointer to a [`Runtime`] instance. /// - `indexer` is a valid pointer to a [`IndexerServiceFFI`] instance. #[unsafe(no_mangle)] pub unsafe extern "C" fn query_account( - runtime: *const Runtime, indexer: *const IndexerServiceFFI, account_id: FfiAccountId, ) -> PointerResult { @@ -174,23 +203,29 @@ pub unsafe extern "C" fn query_account( let indexer = unsafe { &*indexer }; - let client = indexer.client(); - let runtime = unsafe { &*runtime }; - - runtime - .block_on(client.get_account(AccountId { - value: account_id.data, - })) + // `account_current_state` is the only async store call; drive it on the + // runtime the indexer was started on. + let account_id = AccountId { + value: account_id.data, + }; + indexer + .runtime_handle() + .block_on( + indexer + .core() + .store + .account_current_state(&account_id.into()), + ) .map_or_else( - |_| PointerResult::from_error(OperationStatus::ClientError), - |acc| { - let acc_lee: lee::Account = acc.try_into().expect("Source is in blocks, must fit"); - PointerResult::from_value(acc_lee.into()) + |e| { + log::error!("Failed to query account: {e:#}"); + PointerResult::from_error(OperationStatus::ClientError) }, + |account| PointerResult::from_value(account.into()), ) } -/// Query the trasnaction by hash from indexer. +/// Query the transaction by hash from indexer. /// /// # Arguments /// @@ -205,10 +240,8 @@ pub unsafe extern "C" fn query_account( /// /// The caller must ensure that: /// - `indexer` is a valid pointer to a [`IndexerServiceFFI`] instance. -/// - `runtime` is a valid pointer to a [`Runtime`] instance. #[unsafe(no_mangle)] pub unsafe extern "C" fn query_transaction( - runtime: *const Runtime, indexer: *const IndexerServiceFFI, hash: FfiHashType, ) -> PointerResult, OperationStatus> { @@ -219,15 +252,18 @@ pub unsafe extern "C" fn query_transaction( let indexer = unsafe { &*indexer }; - let client = indexer.client(); - let runtime = unsafe { &*runtime }; - - runtime - .block_on(client.get_transaction(HashType(hash.data))) + indexer + .core() + .store + .get_transaction_by_hash(hash.data) .map_or_else( - |_| PointerResult::from_error(OperationStatus::ClientError), + |e| { + log::error!("Failed to query transaction: {e:#}"); + PointerResult::from_error(OperationStatus::ClientError) + }, |tx_opt| { let tx_ffi = tx_opt.map_or_else(FfiOption::::from_none, |tx| { + let tx: indexer_service_protocol::Transaction = tx.into(); FfiOption::::from_value(tx.into()) }); @@ -252,10 +288,8 @@ pub unsafe extern "C" fn query_transaction( /// /// The caller must ensure that: /// - `indexer` is a valid pointer to a [`IndexerServiceFFI`] instance. -/// - `runtime` is a valid pointer to a [`Runtime`] instance. #[unsafe(no_mangle)] pub unsafe extern "C" fn query_block_vec( - runtime: *const Runtime, indexer: *const IndexerServiceFFI, before: FfiOption, limit: u64, @@ -267,21 +301,26 @@ pub unsafe extern "C" fn query_block_vec( let indexer = unsafe { &*indexer }; - let client = indexer.client(); - let runtime = unsafe { &*runtime }; - let before_std = before.is_some.then(|| unsafe { *before.value }); - runtime - .block_on(client.get_blocks(before_std, limit)) + indexer + .core() + .store + .get_block_batch(before_std, limit) .map_or_else( - |_| PointerResult::from_error(OperationStatus::ClientError), + |e| { + log::error!("Failed to query block batch: {e:#}"); + PointerResult::from_error(OperationStatus::ClientError) + }, |block_vec| { PointerResult::from_value( block_vec .into_iter() - .map(Into::into) - .collect::>() + .map(|block| { + let block: indexer_service_protocol::Block = block.into(); + block.into() + }) + .collect::>() .into(), ) }, @@ -299,16 +338,14 @@ pub unsafe extern "C" fn query_block_vec( /// /// # Returns /// -/// A `PointerResult, OperationStatus>` indicating success or failure. +/// A `PointerResult, OperationStatus>` indicating success or failure. /// /// # Safety /// /// The caller must ensure that: /// - `indexer` is a valid pointer to a [`IndexerServiceFFI`] instance. -/// - `runtime` is a valid pointer to a [`Runtime`] instance. #[unsafe(no_mangle)] pub unsafe extern "C" fn query_transactions_by_account( - runtime: *const Runtime, indexer: *const IndexerServiceFFI, account_id: FfiAccountId, offset: u64, @@ -321,25 +358,24 @@ pub unsafe extern "C" fn query_transactions_by_account( let indexer = unsafe { &*indexer }; - let client = indexer.client(); - let runtime = unsafe { &*runtime }; - - runtime - .block_on(client.get_transactions_by_account( - AccountId { - value: account_id.data, - }, - offset, - limit, - )) + indexer + .core() + .store + .get_transactions_by_account(account_id.data, offset, limit) .map_or_else( - |_| PointerResult::from_error(OperationStatus::ClientError), + |e| { + log::error!("Failed to query transactions by account: {e:#}"); + PointerResult::from_error(OperationStatus::ClientError) + }, |tx_vec| { PointerResult::from_value( tx_vec .into_iter() - .map(Into::into) - .collect::>() + .map(|tx| { + let tx: indexer_service_protocol::Transaction = tx.into(); + tx.into() + }) + .collect::>() .into(), ) }, diff --git a/lez/indexer/ffi/src/api/types/account.rs b/lez/indexer/ffi/src/api/types/account.rs index 2309b84b..f2eb8e58 100644 --- a/lez/indexer/ffi/src/api/types/account.rs +++ b/lez/indexer/ffi/src/api/types/account.rs @@ -100,9 +100,14 @@ impl From<&FfiAccount> for indexer_service_protocol::Account { /// Frees the resources associated with the given ffi account. /// +/// Takes ownership of the whole allocation produced by a `query_*` call: the +/// outer `Box` (the `PointerResult.value` pointer) *and* its inner +/// data buffer. Passing the struct by value previously freed only the inner +/// buffer and leaked the outer box. +/// /// # Arguments /// -/// - `val`: An instance of `FfiAccount`. +/// - `val`: The `*mut FfiAccount` returned in `PointerResult.value`. /// /// # Returns /// @@ -111,9 +116,15 @@ impl From<&FfiAccount> for indexer_service_protocol::Account { /// # Safety /// /// The caller must ensure that: -/// - `val` is a valid instance of `FfiAccount`. +/// - `val` is a pointer to an `FfiAccount` produced by this library and not yet freed. #[unsafe(no_mangle)] -pub unsafe extern "C" fn free_ffi_account(val: FfiAccount) { - let orig_val: indexer_service_protocol::Account = val.into(); +pub unsafe extern "C" fn free_ffi_account(val: *mut FfiAccount) { + if val.is_null() { + log::error!("Trying to free a null pointer. Exiting"); + return; + } + // Reclaim the outer box, then convert to drop the inner data buffer. + let boxed = unsafe { Box::from_raw(val) }; + let orig_val: indexer_service_protocol::Account = (*boxed).into(); drop(orig_val); } diff --git a/lez/indexer/ffi/src/api/types/block.rs b/lez/indexer/ffi/src/api/types/block.rs index e7ae0760..b652a7fe 100644 --- a/lez/indexer/ffi/src/api/types/block.rs +++ b/lez/indexer/ffi/src/api/types/block.rs @@ -2,7 +2,7 @@ use indexer_service_protocol::{BedrockStatus, Block, BlockHeader, HashType, Sign use crate::api::types::{ FfiBlockId, FfiHashType, FfiOption, FfiSignature, FfiTimestamp, FfiVec, - transaction::free_ffi_transaction_vec, vectors::FfiBlockBody, + transaction::free_transaction_vec_value, vectors::FfiBlockBody, }; #[repr(C)] @@ -91,7 +91,13 @@ impl From for BedrockStatus { } } -/// Frees the resources associated with the given ffi block. +/// Frees the resources owned by an `FfiBlock` value. +/// +/// This frees the block's transaction bodies (the only heap-owning field); the +/// header/status fields are `Copy`. It operates on the struct by value because +/// it is an element-level helper, used both for the vector path +/// ([`free_ffi_block_vec`]) and the optional path ([`free_ffi_block_opt`]) — in +/// neither case is an `FfiBlock` itself wrapped in its own outer box. /// /// # Arguments /// @@ -104,7 +110,7 @@ impl From for BedrockStatus { /// # Safety /// /// The caller must ensure that: -/// - `val` is a valid instance of `FfiBlock`. +/// - `val` is a valid instance of `FfiBlock` produced by this library and not yet freed. #[unsafe(no_mangle)] pub unsafe extern "C" fn free_ffi_block(val: FfiBlock) { // We don't really need all the casts, but just in case @@ -121,16 +127,18 @@ pub unsafe extern "C" fn free_ffi_block(val: FfiBlock) { #[expect(clippy::let_underscore_must_use, reason = "No use for this Copy type")] let _: BedrockStatus = val.bedrock_status.into(); - unsafe { - free_ffi_transaction_vec(ffi_tx_ffi_vec); - }; + free_transaction_vec_value(ffi_tx_ffi_vec); } /// Frees the resources associated with the given ffi block option. /// +/// Takes ownership of the whole allocation produced by a `query_*` call: the +/// outer `Box` (the `PointerResult.value` pointer), the inner +/// `Box` (when present), and that block's transaction bodies. +/// /// # Arguments /// -/// - `val`: An instance of `FfiBlockOpt`. +/// - `val`: The `*mut FfiBlockOpt` returned in `PointerResult.value`. /// /// # Returns /// @@ -139,37 +147,32 @@ pub unsafe extern "C" fn free_ffi_block(val: FfiBlock) { /// # Safety /// /// The caller must ensure that: -/// - `val` is a valid instance of `FfiBlockOpt`. +/// - `val` is a pointer to an `FfiBlockOpt` produced by this library and not yet freed. #[unsafe(no_mangle)] -pub unsafe extern "C" fn free_ffi_block_opt(val: FfiBlockOpt) { - if val.is_some { - let value = unsafe { Box::from_raw(val.value) }; - - // We don't really need all the casts, but just in case - // All except `ffi_tx_ffi_vec` is Copy types, so no need for Drop - let _ = BlockHeader { - block_id: value.header.block_id, - prev_block_hash: HashType(value.header.prev_block_hash.data), - hash: HashType(value.header.hash.data), - timestamp: value.header.timestamp, - signature: Signature(value.header.signature.data), - }; - let ffi_tx_ffi_vec = value.body; - - #[expect(clippy::let_underscore_must_use, reason = "No use for this Copy type")] - let _: BedrockStatus = value.bedrock_status.into(); - +pub unsafe extern "C" fn free_ffi_block_opt(val: *mut FfiBlockOpt) { + if val.is_null() { + log::error!("Trying to free a null pointer. Exiting"); + return; + } + // Reclaim the outer box, then the inner block box (if any). + let opt = unsafe { Box::from_raw(val) }; + if opt.is_some { + let block = unsafe { Box::from_raw(opt.value) }; unsafe { - free_ffi_transaction_vec(ffi_tx_ffi_vec); - }; + free_ffi_block(*block); + } } } /// Frees the resources associated with the given ffi block vector. /// +/// Takes ownership of the whole allocation produced by a `query_*` call: the +/// outer `Box>` (the `PointerResult.value` pointer), the +/// vector's backing buffer, and every block within it. +/// /// # Arguments /// -/// - `val`: An instance of `FfiVec`. +/// - `val`: The `*mut FfiVec` returned in `PointerResult.value`. /// /// # Returns /// @@ -178,10 +181,16 @@ pub unsafe extern "C" fn free_ffi_block_opt(val: FfiBlockOpt) { /// # Safety /// /// The caller must ensure that: -/// - `val` is a valid instance of `FfiVec`. +/// - `val` is a pointer to an `FfiVec` produced by this library and not yet freed. #[unsafe(no_mangle)] -pub unsafe extern "C" fn free_ffi_block_vec(val: FfiVec) { - let ffi_block_std_vec: Vec<_> = val.into(); +pub unsafe extern "C" fn free_ffi_block_vec(val: *mut FfiVec) { + if val.is_null() { + log::error!("Trying to free a null pointer. Exiting"); + return; + } + // Reclaim the outer box, then the backing buffer and each block. + let boxed = unsafe { Box::from_raw(val) }; + let ffi_block_std_vec: Vec<_> = (*boxed).into(); for block in ffi_block_std_vec { unsafe { free_ffi_block(block); diff --git a/lez/indexer/ffi/src/api/types/transaction.rs b/lez/indexer/ffi/src/api/types/transaction.rs index ca733ed3..d5cb9035 100644 --- a/lez/indexer/ffi/src/api/types/transaction.rs +++ b/lez/indexer/ffi/src/api/types/transaction.rs @@ -463,9 +463,13 @@ pub unsafe extern "C" fn free_ffi_transaction(val: FfiTransaction) { /// Frees the resources associated with the given ffi transaction option. /// +/// Takes ownership of the whole allocation produced by a `query_*` call: the +/// outer `Box>` (the `PointerResult.value` pointer), +/// the inner `Box` (when present), and its body. +/// /// # Arguments /// -/// - `val`: An instance of `FfiOption`. +/// - `val`: The `*mut FfiOption` returned in `PointerResult.value`. /// /// # Returns /// @@ -474,48 +478,32 @@ pub unsafe extern "C" fn free_ffi_transaction(val: FfiTransaction) { /// # Safety /// /// The caller must ensure that: -/// - `val` is a valid instance of `FfiOption`. +/// - `val` is a pointer to an `FfiOption` produced by this library and not yet +/// freed. #[unsafe(no_mangle)] -pub unsafe extern "C" fn free_ffi_transaction_opt(val: FfiOption) { - if val.is_some { - let value = unsafe { Box::from_raw(val.value) }; - - match value.kind { - FfiTransactionKind::Public => { - let body = unsafe { Box::from_raw(value.body.public_body) }; - let std_body: PublicTransaction = body.into(); - drop(std_body); - } - FfiTransactionKind::Private => { - let body = unsafe { Box::from_raw(value.body.private_body) }; - let std_body: PrivacyPreservingTransaction = body.into(); - drop(std_body); - } - FfiTransactionKind::ProgramDeploy => { - let body = unsafe { Box::from_raw(value.body.program_deployment_body) }; - let std_body: ProgramDeploymentTransaction = body.into(); - drop(std_body); - } +pub unsafe extern "C" fn free_ffi_transaction_opt(val: *mut FfiOption) { + if val.is_null() { + log::error!("Trying to free a null pointer. Exiting"); + return; + } + // Reclaim the outer box, then the inner transaction box (if any). + let opt = unsafe { Box::from_raw(val) }; + if opt.is_some { + let tx = unsafe { Box::from_raw(opt.value) }; + unsafe { + free_ffi_transaction(*tx); } } } -/// Frees the resources associated with the given vector of ffi transactions. +/// Frees the resources owned by an `FfiVec` value (the backing +/// buffer and each transaction), without owning an outer box. /// -/// # Arguments -/// -/// - `val`: An instance of `FfiVec`. -/// -/// # Returns -/// -/// void. -/// -/// # Safety -/// -/// The caller must ensure that: -/// - `val` is a valid instance of `FfiVec`. -#[unsafe(no_mangle)] -pub unsafe extern "C" fn free_ffi_transaction_vec(val: FfiVec) { +/// This is the element-level helper shared by the block free path +/// ([`crate::api::types::block::free_ffi_block`], whose body is a transaction +/// vector held by value) and the public [`free_ffi_transaction_vec`] entry +/// point (which first reclaims the outer box). +pub(crate) fn free_transaction_vec_value(val: FfiVec) { let ffi_tx_std_vec: Vec<_> = val.into(); for tx in ffi_tx_std_vec { unsafe { @@ -524,6 +512,35 @@ pub unsafe extern "C" fn free_ffi_transaction_vec(val: FfiVec) { } } +/// Frees the resources associated with the given vector of ffi transactions. +/// +/// Takes ownership of the whole allocation produced by a `query_*` call: the +/// outer `Box>` (the `PointerResult.value` pointer), the +/// vector's backing buffer, and every transaction within it. +/// +/// # Arguments +/// +/// - `val`: The `*mut FfiVec` returned in `PointerResult.value`. +/// +/// # Returns +/// +/// void. +/// +/// # Safety +/// +/// The caller must ensure that: +/// - `val` is a pointer to an `FfiVec` produced by this library and not yet freed. +#[unsafe(no_mangle)] +pub unsafe extern "C" fn free_ffi_transaction_vec(val: *mut FfiVec) { + if val.is_null() { + log::error!("Trying to free a null pointer. Exiting"); + return; + } + // Reclaim the outer box, then the backing buffer and each transaction. + let boxed = unsafe { Box::from_raw(val) }; + free_transaction_vec_value(*boxed); +} + fn cast_validity_window(window: ValidityWindow) -> [u64; 2] { [ window.0.0.unwrap_or_default(), diff --git a/lez/indexer/ffi/src/client.rs b/lez/indexer/ffi/src/client.rs deleted file mode 100644 index f05b350e..00000000 --- a/lez/indexer/ffi/src/client.rs +++ /dev/null @@ -1,33 +0,0 @@ -use std::{ops::Deref, sync::Arc}; - -use anyhow::{Context as _, Result}; -use log::info; -pub use url::Url; - -pub trait IndexerClientTrait: Clone { - async fn new(indexer_url: &Url) -> Result; -} - -#[derive(Clone)] -pub struct IndexerClient(Arc); - -impl IndexerClientTrait for IndexerClient { - async fn new(indexer_url: &Url) -> Result { - info!("Connecting to Indexer at {indexer_url}"); - - let client = jsonrpsee::ws_client::WsClientBuilder::default() - .build(indexer_url) - .await - .context("Failed to create websocket client")?; - - Ok(Self(Arc::new(client))) - } -} - -impl Deref for IndexerClient { - type Target = jsonrpsee::ws_client::WsClient; - - fn deref(&self) -> &Self::Target { - &self.0 - } -} diff --git a/lez/indexer/ffi/src/indexer.rs b/lez/indexer/ffi/src/indexer.rs index e8707697..327773b3 100644 --- a/lez/indexer/ffi/src/indexer.rs +++ b/lez/indexer/ffi/src/indexer.rs @@ -1,95 +1,75 @@ -use std::{ffi::c_void, net::SocketAddr}; +use std::ffi::c_void; -use indexer_service::IndexerHandle; - -use crate::client::IndexerClient; +use indexer_core::IndexerCore; +use tokio::{runtime::Handle, task::JoinHandle}; +/// FFI-owned indexer. +/// +/// Has three fields behind `c_void` (so that cbindgen never needs to see their Rust layout): +/// - An [`IndexerCore`] used to answer queries +/// - The background task [`JoinHandle`] that drives ingestion (consuming the block stream so the +/// store stays populated) +/// - A [`Handle`] to the runtime they live on. #[repr(C)] pub struct IndexerServiceFFI { - indexer_handle: *mut c_void, - indexer_client: *mut c_void, + core: *mut c_void, + ingest_handle: *mut c_void, + runtime_handle: *mut c_void, } impl IndexerServiceFFI { #[must_use] - pub fn new( - indexer_handle: indexer_service::IndexerHandle, - indexer_client: IndexerClient, - ) -> Self { + pub fn new(core: IndexerCore, ingest_handle: JoinHandle<()>, runtime_handle: Handle) -> Self { Self { - // Box the complex types and convert to opaque pointers - indexer_handle: Box::into_raw(Box::new(indexer_handle)).cast::(), - indexer_client: Box::into_raw(Box::new(indexer_client)).cast::(), + core: Box::into_raw(Box::new(core)).cast::(), + ingest_handle: Box::into_raw(Box::new(ingest_handle)).cast::(), + runtime_handle: Box::into_raw(Box::new(runtime_handle)).cast::(), } } - /// Helper to take ownership back. + /// Borrow the [`IndexerCore`] to run a query against its store. #[must_use] - pub fn into_parts(mut self) -> (Box, Box) { - let Self { - indexer_handle, - indexer_client, - } = &mut self; - - let indexer_handle_boxed = unsafe { Box::from_raw(indexer_handle.cast::()) }; - let indexer_client_boxed = unsafe { Box::from_raw(indexer_client.cast::()) }; - - // Assigning nulls to prevent double free on drop, since ownership is transferred to caller - *indexer_handle = std::ptr::null_mut(); - *indexer_client = std::ptr::null_mut(); - - (indexer_handle_boxed, indexer_client_boxed) - } - - /// Helper to get indexer handle addr. - #[must_use] - pub const fn addr(&self) -> SocketAddr { - let indexer_handle = unsafe { - self.indexer_handle - .cast::() - .as_ref() - .expect("Indexer Handle must be non-null pointer") - }; - - indexer_handle.addr() - } - - /// Helper to get indexer handle ref. - #[must_use] - pub const fn handle(&self) -> &IndexerHandle { + pub const fn core(&self) -> &IndexerCore { unsafe { - self.indexer_handle - .cast::() + self.core + .cast::() .as_ref() - .expect("Indexer Handle must be non-null pointer") + .expect("IndexerCore must be a non-null pointer") } } - /// Helper to get indexer client ref. + /// Borrow the runtime handle to `block_on` an async store query. #[must_use] - pub const fn client(&self) -> &IndexerClient { + pub const fn runtime_handle(&self) -> &Handle { unsafe { - self.indexer_client - .cast::() + self.runtime_handle + .cast::() .as_ref() - .expect("Indexer Client must be non-null pointer") + .expect("Runtime handle must be a non-null pointer") } } } -// Implement Drop to prevent memory leaks +// Implement Drop to stop ingestion and free the boxed resources. impl Drop for IndexerServiceFFI { fn drop(&mut self) { let Self { - indexer_handle, - indexer_client, + core, + ingest_handle, + runtime_handle, } = self; - if !indexer_handle.is_null() { - drop(unsafe { Box::from_raw(indexer_handle.cast::()) }); + if !ingest_handle.is_null() { + // Stop the background ingestion task before tearing down the core. + let handle = unsafe { Box::from_raw(ingest_handle.cast::>()) }; + handle.abort(); + drop(handle); } - if !indexer_client.is_null() { - drop(unsafe { Box::from_raw(indexer_client.cast::()) }); + if !core.is_null() { + drop(unsafe { Box::from_raw(core.cast::()) }); + } + if !runtime_handle.is_null() { + drop(unsafe { Box::from_raw(runtime_handle.cast::()) }); } } } diff --git a/lez/indexer/ffi/src/lib.rs b/lez/indexer/ffi/src/lib.rs index 9e34b111..0ca197c7 100644 --- a/lez/indexer/ffi/src/lib.rs +++ b/lez/indexer/ffi/src/lib.rs @@ -5,7 +5,6 @@ pub use indexer::IndexerServiceFFI; pub use runtime::Runtime; pub mod api; -mod client; mod errors; mod indexer; mod runtime; From 4a3fa1d4bef8992892fb3cfa24481b39d466e0aa Mon Sep 17 00:00:00 2001 From: erhant Date: Fri, 19 Jun 2026 13:41:36 +0300 Subject: [PATCH 02/14] refactor: logger now only enables for indexer crates + takes in a loglevel option --- .../tests/indexer_ffi_state_consistency.rs | 5 --- ...dexer_ffi_state_consistency_with_labels.rs | 4 -- lez/indexer/ffi/indexer_ffi.h | 18 ++++----- lez/indexer/ffi/src/api/logging.rs | 39 ++++++++++++++----- 4 files changed, 38 insertions(+), 28 deletions(-) diff --git a/integration_tests/tests/indexer_ffi_state_consistency.rs b/integration_tests/tests/indexer_ffi_state_consistency.rs index f84a3790..0a41c68c 100644 --- a/integration_tests/tests/indexer_ffi_state_consistency.rs +++ b/integration_tests/tests/indexer_ffi_state_consistency.rs @@ -8,7 +8,6 @@ use std::time::Duration; use anyhow::{Context as _, Result}; -use indexer_ffi::Runtime; use indexer_service_protocol::Account; use integration_tests::{ L2_TO_L1_TIMEOUT, TIME_TO_WAIT_FOR_BLOCK_SECONDS, private_mention, public_mention, @@ -102,11 +101,8 @@ fn indexer_ffi_state_consistency() -> Result<()> { info!("Waiting for indexer to parse blocks"); std::thread::sleep(L2_TO_L1_TIMEOUT); - // Safety: ctx runtime is valid for the lifetime of the returned Runtime - let runtime = unsafe { Runtime::from_borrowed(ctx.runtime()) }; let acc1_ind_state_ffi = unsafe { indexer_ffi_helpers::query_account( - &raw const runtime, &raw const indexer_ffi, (&ctx.ctx().existing_public_accounts()[0]).into(), ) @@ -119,7 +115,6 @@ fn indexer_ffi_state_consistency() -> Result<()> { let acc2_ind_state_ffi = unsafe { indexer_ffi_helpers::query_account( - &raw const runtime, &raw const indexer_ffi, (&ctx.ctx().existing_public_accounts()[1]).into(), ) diff --git a/integration_tests/tests/indexer_ffi_state_consistency_with_labels.rs b/integration_tests/tests/indexer_ffi_state_consistency_with_labels.rs index 34d5a4d7..fbc0b422 100644 --- a/integration_tests/tests/indexer_ffi_state_consistency_with_labels.rs +++ b/integration_tests/tests/indexer_ffi_state_consistency_with_labels.rs @@ -8,7 +8,6 @@ use std::time::Duration; use anyhow::Result; -use indexer_ffi::Runtime; use indexer_service_protocol::Account; use integration_tests::{L2_TO_L1_TIMEOUT, TIME_TO_WAIT_FOR_BLOCK_SECONDS, public_mention}; use log::info; @@ -75,11 +74,8 @@ fn indexer_ffi_state_consistency_with_labels() -> Result<()> { info!("Waiting for indexer to parse blocks"); std::thread::sleep(L2_TO_L1_TIMEOUT); - // Safety: ctx runtime is valid for the lifetime of the returned Runtime - let runtime = unsafe { Runtime::from_borrowed(ctx.runtime()) }; let acc1_ind_state_ffi = unsafe { indexer_ffi_helpers::query_account( - &raw const runtime, &raw const indexer_ffi, (&ctx.ctx().existing_public_accounts()[0]).into(), ) diff --git a/lez/indexer/ffi/indexer_ffi.h b/lez/indexer/ffi/indexer_ffi.h index 06953e39..04d33b33 100644 --- a/lez/indexer/ffi/indexer_ffi.h +++ b/lez/indexer/ffi/indexer_ffi.h @@ -464,18 +464,18 @@ struct PointerResult_Runtime__OperationStatus new_runtime(void); enum OperationStatus stop_indexer(struct IndexerServiceFFI *indexer); /** - * Initializes the FFI's logger. + * Initializes logging for the indexer at `level`. * - * Wires up `env_logger`, so the library's `log::*` output is controlled by the - * `RUST_LOG` environment variable (e.g. `RUST_LOG=info`). Without this, the - * FFI's log calls go nowhere — and since failures are otherwise reported only - * as numeric [`OperationStatus`](crate::errors::OperationStatus) codes, there - * is no other way to see *why* a call failed. + * - `level` is a null-terminated string (`off`/`error`/`warn`/`info`/`debug`/ `trace`, + * case-insensitive); null or unparseable falls back to `info`. * - * Safe to call multiple times and from any consumer: if a global logger is - * already set, the call is a no-op. + * Only the `indexer_ffi` and `indexer_core` targets are enabled! + * + * # Safety + * - `level` must be a valid null-terminated C string, or null. + * - First call to this function wins; subsequent calls are no-ops. */ -void init_logger(void); +void init_logger(const char *level); /** * # Safety diff --git a/lez/indexer/ffi/src/api/logging.rs b/lez/indexer/ffi/src/api/logging.rs index 207d4b61..34bc5f76 100644 --- a/lez/indexer/ffi/src/api/logging.rs +++ b/lez/indexer/ffi/src/api/logging.rs @@ -1,14 +1,33 @@ -/// Initializes the FFI's logger. +use std::ffi::{CStr, c_char}; + +use log::LevelFilter; + +/// Initializes logging for the indexer at `level`. /// -/// Wires up `env_logger`, so the library's `log::*` output is controlled by the -/// `RUST_LOG` environment variable (e.g. `RUST_LOG=info`). Without this, the -/// FFI's log calls go nowhere — and since failures are otherwise reported only -/// as numeric [`OperationStatus`](crate::errors::OperationStatus) codes, there -/// is no other way to see *why* a call failed. +/// - `level` is a null-terminated string (`off`/`error`/`warn`/`info`/`debug`/ `trace`, +/// case-insensitive); null or unparseable falls back to `info`. /// -/// Safe to call multiple times and from any consumer: if a global logger is -/// already set, the call is a no-op. +/// Only the `indexer_ffi` and `indexer_core` targets are enabled! +/// +/// # Safety +/// - `level` must be a valid null-terminated C string, or null. +/// - First call to this function wins; subsequent calls are no-ops. #[unsafe(no_mangle)] -pub extern "C" fn init_logger() { - let _ignore_me = env_logger::try_init(); +pub unsafe extern "C" fn init_logger(level: *const c_char) { + let level = if level.is_null() { + LevelFilter::Info + } else { + unsafe { CStr::from_ptr(level) } + .to_str() + .ok() + .and_then(|s| s.parse().ok()) + .unwrap_or(LevelFilter::Info) + }; + + env_logger::Builder::new() + .filter_level(LevelFilter::Off) + .filter_module("indexer_ffi", level) + .filter_module("indexer_core", level) + .try_init() + .ok(); } From 0db82d2344e3e46fb17164e85281c10a4b1857f2 Mon Sep 17 00:00:00 2001 From: erhant Date: Fri, 19 Jun 2026 14:25:45 +0300 Subject: [PATCH 03/14] refactor: better runtime handling + logger lint fix --- lez/indexer/ffi/indexer_ffi.h | 25 +++++------------- lez/indexer/ffi/src/api/lifecycle.rs | 38 ++++++++++++++-------------- lez/indexer/ffi/src/api/logging.rs | 5 ++-- lez/indexer/ffi/src/api/query.rs | 2 +- lez/indexer/ffi/src/indexer.rs | 32 +++++++++++++---------- 5 files changed, 47 insertions(+), 55 deletions(-) diff --git a/lez/indexer/ffi/indexer_ffi.h b/lez/indexer/ffi/indexer_ffi.h index 04d33b33..4bd2d549 100644 --- a/lez/indexer/ffi/indexer_ffi.h +++ b/lez/indexer/ffi/indexer_ffi.h @@ -31,12 +31,13 @@ typedef enum FfiBedrockStatus { * - An [`IndexerCore`] used to answer queries * - The background task [`JoinHandle`] that drives ingestion (consuming the block stream so the * store stays populated) - * - A [`Handle`] to the runtime they live on. + * - The [`Runtime`] they run on. It owns the underlying tokio runtime when we created it (and + * drops it on teardown) and merely borrows it when the caller supplied one. */ typedef struct IndexerServiceFFI { void *core; void *ingest_handle; - void *runtime_handle; + void *runtime; } IndexerServiceFFI; /** @@ -81,17 +82,6 @@ typedef struct Runtime { struct Pointer_Runtime inner; } Runtime; -/** - * Simple wrapper around a pointer to a value or an error. - * - * Pointer is not guaranteed. You should check the error field before - * dereferencing the pointer. - */ -typedef struct PointerResult_Runtime__OperationStatus { - struct Runtime *value; - enum OperationStatus error; -} PointerResult_Runtime__OperationStatus; - /** * Result of [`query_last_block`], returned **inline** (no heap allocation, so * there is no corresponding `free_*` to call). @@ -423,6 +413,8 @@ typedef struct PointerResult_FfiVec_FfiTransaction_____OperationStatus { * * # Arguments * + * - `runtime`: A runtime for the indexer to run on, or null to have the indexer create and own + * one. * - `config_path`: A pointer to a string representing the path to the configuration file. * * # Returns @@ -432,17 +424,12 @@ typedef struct PointerResult_FfiVec_FfiTransaction_____OperationStatus { * * # Safety * The caller must ensure that: - * - `runtime` is a valid pointer to a `tokio::runtime::Runtime` instance. + * - `runtime` is either null or a valid pointer to a [`Runtime`] that outlives the indexer. * - `config_path` is a valid pointer to a null-terminated C string. */ InitializedIndexerServiceFFIResult start_indexer(const struct Runtime *runtime, const char *config_path); -/** - * Creates a new [`tokio::runtime::Runtime`]. - */ -struct PointerResult_Runtime__OperationStatus new_runtime(void); - /** * Stops and frees the resources associated with the given indexer service. * diff --git a/lez/indexer/ffi/src/api/lifecycle.rs b/lez/indexer/ffi/src/api/lifecycle.rs index bedebb2b..07b2a9be 100644 --- a/lez/indexer/ffi/src/api/lifecycle.rs +++ b/lez/indexer/ffi/src/api/lifecycle.rs @@ -12,6 +12,8 @@ pub type InitializedIndexerServiceFFIResult = PointerResult PointerResult { - Runtime::new().map_or_else( - |_e| PointerResult::from_error(OperationStatus::InitializationError), - PointerResult::from_value, - ) -} - /// Initializes and starts an indexer based on the provided /// configuration file path. /// /// # Arguments /// +/// - `runtime`: A runtime for the indexer to run on, or null to create and own one. /// - `config_path`: A pointer to a string representing the path to the configuration file. /// /// # Returns @@ -58,7 +52,7 @@ pub extern "C" fn new_runtime() -> PointerResult { /// /// # Safety /// The caller must ensure that: -/// - `runtime` is a valid pointer to a `tokio::runtime::Runtime` instance. +/// - `runtime` is either null or a valid pointer to a [`Runtime`] that outlives the indexer. /// - `config_path` is a valid pointer to a null-terminated C string. unsafe fn setup_indexer( runtime: *const Runtime, @@ -77,9 +71,19 @@ unsafe fn setup_indexer( OperationStatus::InitializationError })?; - // SAFETY: The caller must ensure that `runtime` is a valid pointer to a - // `tokio::runtime::Runtime` instance. - let runtime = unsafe { &*runtime }; + // Use the caller's runtime if one was supplied, otherwise create (and own) + // our own. The `Runtime` wrapper drops the underlying tokio runtime only + // when we own it; a borrowed one is left to its external owner. + let runtime = if runtime.is_null() { + Runtime::new().map_err(|e| { + log::error!("Could not create tokio runtime: {e}"); + OperationStatus::InitializationError + })? + } else { + // SAFETY: the caller guarantees `runtime` is valid and outlives the indexer. + let caller = unsafe { &*runtime }; + unsafe { Runtime::from_borrowed(caller.as_ref()) } + }; let core = IndexerCore::new(config).map_err(|e| { log::error!("Could not initialize indexer core: {e}"); @@ -100,11 +104,7 @@ unsafe fn setup_indexer( log::warn!("Indexer block stream ended"); }); - Ok(IndexerServiceFFI::new( - core, - ingest_handle, - runtime.handle().clone(), - )) + Ok(IndexerServiceFFI::new(core, ingest_handle, runtime)) } /// Stops and frees the resources associated with the given indexer service. diff --git a/lez/indexer/ffi/src/api/logging.rs b/lez/indexer/ffi/src/api/logging.rs index 34bc5f76..06c41688 100644 --- a/lez/indexer/ffi/src/api/logging.rs +++ b/lez/indexer/ffi/src/api/logging.rs @@ -24,10 +24,9 @@ pub unsafe extern "C" fn init_logger(level: *const c_char) { .unwrap_or(LevelFilter::Info) }; - env_logger::Builder::new() + let _dontcare = env_logger::Builder::new() .filter_level(LevelFilter::Off) .filter_module("indexer_ffi", level) .filter_module("indexer_core", level) - .try_init() - .ok(); + .try_init(); } diff --git a/lez/indexer/ffi/src/api/query.rs b/lez/indexer/ffi/src/api/query.rs index 548deabc..23233a76 100644 --- a/lez/indexer/ffi/src/api/query.rs +++ b/lez/indexer/ffi/src/api/query.rs @@ -209,7 +209,7 @@ pub unsafe extern "C" fn query_account( value: account_id.data, }; indexer - .runtime_handle() + .runtime() .block_on( indexer .core() diff --git a/lez/indexer/ffi/src/indexer.rs b/lez/indexer/ffi/src/indexer.rs index 327773b3..619373e6 100644 --- a/lez/indexer/ffi/src/indexer.rs +++ b/lez/indexer/ffi/src/indexer.rs @@ -1,7 +1,9 @@ use std::ffi::c_void; use indexer_core::IndexerCore; -use tokio::{runtime::Handle, task::JoinHandle}; +use tokio::task::JoinHandle; + +use crate::Runtime; /// FFI-owned indexer. /// @@ -9,21 +11,22 @@ use tokio::{runtime::Handle, task::JoinHandle}; /// - An [`IndexerCore`] used to answer queries /// - The background task [`JoinHandle`] that drives ingestion (consuming the block stream so the /// store stays populated) -/// - A [`Handle`] to the runtime they live on. +/// - The [`Runtime`] they run on. It owns the underlying tokio runtime when we created it (and +/// drops it on teardown) and merely borrows it when the caller supplied one. #[repr(C)] pub struct IndexerServiceFFI { core: *mut c_void, ingest_handle: *mut c_void, - runtime_handle: *mut c_void, + runtime: *mut c_void, } impl IndexerServiceFFI { #[must_use] - pub fn new(core: IndexerCore, ingest_handle: JoinHandle<()>, runtime_handle: Handle) -> Self { + pub fn new(core: IndexerCore, ingest_handle: JoinHandle<()>, runtime: Runtime) -> Self { Self { core: Box::into_raw(Box::new(core)).cast::(), ingest_handle: Box::into_raw(Box::new(ingest_handle)).cast::(), - runtime_handle: Box::into_raw(Box::new(runtime_handle)).cast::(), + runtime: Box::into_raw(Box::new(runtime)).cast::(), } } @@ -38,14 +41,14 @@ impl IndexerServiceFFI { } } - /// Borrow the runtime handle to `block_on` an async store query. + /// Borrow the runtime to `block_on` an async store query. #[must_use] - pub const fn runtime_handle(&self) -> &Handle { + pub const fn runtime(&self) -> &Runtime { unsafe { - self.runtime_handle - .cast::() + self.runtime + .cast::() .as_ref() - .expect("Runtime handle must be a non-null pointer") + .expect("Runtime must be a non-null pointer") } } } @@ -56,7 +59,7 @@ impl Drop for IndexerServiceFFI { let Self { core, ingest_handle, - runtime_handle, + runtime, } = self; if !ingest_handle.is_null() { @@ -68,8 +71,11 @@ impl Drop for IndexerServiceFFI { if !core.is_null() { drop(unsafe { Box::from_raw(core.cast::()) }); } - if !runtime_handle.is_null() { - drop(unsafe { Box::from_raw(runtime_handle.cast::()) }); + // Dropping the `Runtime` shuts down the tokio runtime only if we own it + // (a borrowed one is left for its external owner). Done last, and from + // the consumer thread, so it never drops from within a runtime worker. + if !runtime.is_null() { + drop(unsafe { Box::from_raw(runtime.cast::()) }); } } } From ea13ef10d8506597bd945c56ad0a7cdd60af735b Mon Sep 17 00:00:00 2001 From: erhant Date: Fri, 19 Jun 2026 16:27:01 +0300 Subject: [PATCH 04/14] refactor: drastically shorten (480s to 160s) the test time by adding block `wait` helper --- .../tests/indexer_ffi_block_batching.rs | 15 +++-------- .../tests/indexer_ffi_helpers/mod.rs | 25 +++++++++++++++++++ .../tests/indexer_test_run_ffi.rs | 16 +++--------- 3 files changed, 33 insertions(+), 23 deletions(-) diff --git a/integration_tests/tests/indexer_ffi_block_batching.rs b/integration_tests/tests/indexer_ffi_block_batching.rs index 52f68027..c244fbb0 100644 --- a/integration_tests/tests/indexer_ffi_block_batching.rs +++ b/integration_tests/tests/indexer_ffi_block_batching.rs @@ -6,7 +6,6 @@ use anyhow::Result; use indexer_ffi::api::types::FfiOption; -use integration_tests::L2_TO_L1_TIMEOUT; use log::info; #[path = "indexer_ffi_helpers/mod.rs"] @@ -18,17 +17,11 @@ fn indexer_ffi_block_batching() -> Result<()> { // duration of the test; the indexer was started on that runtime. let (_ctx, indexer_ffi, _indexer_dir) = indexer_ffi_helpers::setup()?; - // WAIT + // WAIT: poll until the indexer has finalized at least two blocks (so the + // chain-consistency check below verifies at least one block link), returning + // early instead of sleeping for the full timeout. info!("Waiting for indexer to parse blocks"); - std::thread::sleep(L2_TO_L1_TIMEOUT); - - let last_block_indexer_ffi_res = - unsafe { indexer_ffi_helpers::query_last_block(&raw const indexer_ffi) }; - - assert!(last_block_indexer_ffi_res.error.is_ok()); - assert!(last_block_indexer_ffi_res.is_some); - - let last_block_indexer = last_block_indexer_ffi_res.block_id; + let last_block_indexer = indexer_ffi_helpers::wait_for_indexer_ffi_block(&indexer_ffi, 2)?; info!("Last block on indexer FFI now is {last_block_indexer}"); diff --git a/integration_tests/tests/indexer_ffi_helpers/mod.rs b/integration_tests/tests/indexer_ffi_helpers/mod.rs index 35bf7ef3..73d7bf57 100644 --- a/integration_tests/tests/indexer_ffi_helpers/mod.rs +++ b/integration_tests/tests/indexer_ffi_helpers/mod.rs @@ -84,3 +84,28 @@ pub fn setup() -> Result<(BlockingTestContext, IndexerServiceFFI, TempDir)> { let (indexer_ffi, indexer_dir) = setup_indexer_ffi(&runtime, ctx.ctx().bedrock_addr())?; Ok((ctx, indexer_ffi, indexer_dir)) } + +/// Poll the indexer FFI until its last finalized block id reaches `min_block_id` +/// or until [`integration_tests::L2_TO_L1_TIMEOUT`] elapses. +/// +/// This avoids blindly sleeping for the full timeout: the indexer typically +/// catches up in a fraction of that time, so we return as soon as it does and +/// only use the timeout as a ceiling. Returns the last observed block id. +pub fn wait_for_indexer_ffi_block(indexer: &IndexerServiceFFI, min_block_id: u64) -> Result { + let start = std::time::Instant::now(); + loop { + // SAFETY: `indexer` is a valid reference for the duration of the call. + let res = unsafe { query_last_block(std::ptr::from_ref(indexer)) }; + if res.error.is_ok() && res.is_some && res.block_id >= min_block_id { + return Ok(res.block_id); + } + if start.elapsed() >= integration_tests::L2_TO_L1_TIMEOUT { + anyhow::bail!( + "Indexer FFI did not reach block {min_block_id} within {:?}. Last observed block id: {}", + integration_tests::L2_TO_L1_TIMEOUT, + res.block_id + ); + } + std::thread::sleep(std::time::Duration::from_secs(2)); + } +} diff --git a/integration_tests/tests/indexer_test_run_ffi.rs b/integration_tests/tests/indexer_test_run_ffi.rs index 0a5f4133..56945d30 100644 --- a/integration_tests/tests/indexer_test_run_ffi.rs +++ b/integration_tests/tests/indexer_test_run_ffi.rs @@ -1,11 +1,9 @@ #![expect( clippy::tests_outside_test_module, - clippy::undocumented_unsafe_blocks, reason = "We don't care about these in tests" )] use anyhow::Result; -use integration_tests::L2_TO_L1_TIMEOUT; use log::info; #[path = "indexer_ffi_helpers/mod.rs"] @@ -17,16 +15,10 @@ fn indexer_test_run_ffi() -> Result<()> { // duration of the test; the indexer was started on that runtime. let (_ctx, indexer_ffi, _indexer_dir) = indexer_ffi_helpers::setup()?; - // RUN OBSERVATION - std::thread::sleep(L2_TO_L1_TIMEOUT); - - let last_block_indexer_ffi_res = - unsafe { indexer_ffi_helpers::query_last_block(&raw const indexer_ffi) }; - - assert!(last_block_indexer_ffi_res.error.is_ok()); - assert!(last_block_indexer_ffi_res.is_some); - - let last_block_indexer_ffi = last_block_indexer_ffi_res.block_id; + // RUN OBSERVATION: poll until the indexer has finalized at least one block, + // returning early instead of sleeping for the full timeout. + let last_block_indexer_ffi = + indexer_ffi_helpers::wait_for_indexer_ffi_block(&indexer_ffi, 1)?; info!("Last block on indexer FFI now is {last_block_indexer_ffi}"); From f3134cde5876d43f401849495cc1d2ca6ac471ed Mon Sep 17 00:00:00 2001 From: erhant Date: Fri, 19 Jun 2026 16:30:38 +0300 Subject: [PATCH 05/14] chore: fmt and lint --- integration_tests/tests/indexer_test_run_ffi.rs | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/integration_tests/tests/indexer_test_run_ffi.rs b/integration_tests/tests/indexer_test_run_ffi.rs index 56945d30..e37b619e 100644 --- a/integration_tests/tests/indexer_test_run_ffi.rs +++ b/integration_tests/tests/indexer_test_run_ffi.rs @@ -17,8 +17,7 @@ fn indexer_test_run_ffi() -> Result<()> { // RUN OBSERVATION: poll until the indexer has finalized at least one block, // returning early instead of sleeping for the full timeout. - let last_block_indexer_ffi = - indexer_ffi_helpers::wait_for_indexer_ffi_block(&indexer_ffi, 1)?; + let last_block_indexer_ffi = indexer_ffi_helpers::wait_for_indexer_ffi_block(&indexer_ffi, 1)?; info!("Last block on indexer FFI now is {last_block_indexer_ffi}"); From 0c52ec969548b277038624ff9405e4e3b44a14c5 Mon Sep 17 00:00:00 2001 From: erhant Date: Fri, 19 Jun 2026 18:50:02 +0300 Subject: [PATCH 06/14] fix(indexer): stop FFI integration tests segfaulting on query_account MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The indexer FFI test helper borrowed `ctx.runtime()` via `Runtime::from_borrowed` and then moved `ctx` (and its by-value `tokio::runtime::Runtime` field) into the tuple returned from `setup()`. That move relocates the runtime, leaving the raw pointer the indexer stored dangling. Sync queries never touch the runtime, so they passed; `query_account` is the only path that `block_on`s it, so it dereferenced freed stack memory → SIGSEGV (hence only the two `indexer_ffi_state_consistency*` tests crashed). Pass a null runtime so the FFI creates and owns its own — the same lifetime path the production module uses (`start_indexer(nullptr, …)`) — instead of borrowing a runtime whose address isn't stable across the move. --- .../tests/indexer_ffi_helpers/mod.rs | 18 +++++++++--------- 1 file changed, 9 insertions(+), 9 deletions(-) diff --git a/integration_tests/tests/indexer_ffi_helpers/mod.rs b/integration_tests/tests/indexer_ffi_helpers/mod.rs index 73d7bf57..e9527269 100644 --- a/integration_tests/tests/indexer_ffi_helpers/mod.rs +++ b/integration_tests/tests/indexer_ffi_helpers/mod.rs @@ -40,10 +40,7 @@ unsafe extern "C" { ) -> InitializedIndexerServiceFFIResult; } -pub fn setup_indexer_ffi( - runtime: &Runtime, - bedrock_addr: SocketAddr, -) -> Result<(IndexerServiceFFI, TempDir)> { +pub fn setup_indexer_ffi(bedrock_addr: SocketAddr) -> Result<(IndexerServiceFFI, TempDir)> { let temp_indexer_dir = tempfile::tempdir().context("Failed to create temp dir for indexer home")?; @@ -63,8 +60,9 @@ pub fn setup_indexer_ffi( file.flush()?; let res = - // SAFETY: lib function ensures validity of value. - unsafe { start_indexer(std::ptr::from_ref(runtime), CString::new(config_path.to_str().unwrap())?.as_ptr()) }; + // SAFETY: null runtime → the FFI creates and owns its own tokio runtime, + // so there is no external runtime whose address we must keep stable. + unsafe { start_indexer(std::ptr::null(), CString::new(config_path.to_str().unwrap())?.as_ptr()) }; if res.error.is_error() { anyhow::bail!("Indexer FFI error {:?}", res.error); @@ -79,9 +77,11 @@ pub fn setup_indexer_ffi( pub fn setup() -> Result<(BlockingTestContext, IndexerServiceFFI, TempDir)> { let ctx = TestContext::builder().disable_indexer().build_blocking()?; - // Safety: ctx runtime is valid for the lifetime of the returned Runtime - let runtime = unsafe { Runtime::from_borrowed(ctx.runtime()) }; - let (indexer_ffi, indexer_dir) = setup_indexer_ffi(&runtime, ctx.ctx().bedrock_addr())?; + // Don't borrow `ctx.runtime()`: `ctx` (and its by-value tokio runtime) is + // moved into the returned tuple, which would leave any pointer into it + // dangling. Pass a null runtime so the FFI owns its own — the same path the + // production module uses. + let (indexer_ffi, indexer_dir) = setup_indexer_ffi(ctx.ctx().bedrock_addr())?; Ok((ctx, indexer_ffi, indexer_dir)) } From 1c1e80f6462231317c74c769af2b67d3bee15df4 Mon Sep 17 00:00:00 2001 From: erhant Date: Fri, 19 Jun 2026 23:18:41 +0300 Subject: [PATCH 07/14] feat(indexer)!: make storage location caller-driven, not config-driven MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The indexer's storage location was the `home` field of IndexerConfig, used only to derive the RocksDB dir. Defaulting to "." meant it landed in the process CWD — fine for the standalone service, but wrong when the indexer runs embedded in a logos_host subprocess (RocksDB ended up in an arbitrary/unwritable dir). Storage location is an operational concern the host should own, not something baked into a user-editable config. Remove `home` from IndexerConfig and pass the storage directory explicitly: - core: `IndexerCore::new(config, storage_dir)` derives `/rocksdb`. - ffi: `start_indexer(runtime, config_path, storage_dir)`; null/empty storage_dir falls back to ".". Lets a host (e.g. a Logos module's instance persistence path) own where state lives. - service: `run_server(config, storage_dir, port)` + a `--data-dir` flag (default ".") on the binary, preserving current behaviour. - drop `home` from the committed indexer config JSONs and the test fixtures. BREAKING CHANGE: `start_indexer` gains a `storage_dir` parameter and IndexerConfig no longer has a `home` field. --- .../tests/indexer_ffi_helpers/mod.rs | 13 +++++--- .../docker-all-in-one/indexer_config.json | 1 - lez/indexer/core/src/config.rs | 9 +---- lez/indexer/core/src/lib.rs | 6 ++-- lez/indexer/ffi/indexer_ffi.h | 7 +++- lez/indexer/ffi/src/api/lifecycle.rs | 33 +++++++++++++++++-- .../service/configs/debug/indexer_config.json | 1 - lez/indexer/service/src/lib.rs | 14 +++++--- lez/indexer/service/src/main.rs | 11 +++++-- lez/indexer/service/src/service.rs | 6 ++-- test_fixtures/src/config.rs | 3 +- test_fixtures/src/setup.rs | 6 ++-- 12 files changed, 73 insertions(+), 37 deletions(-) diff --git a/integration_tests/tests/indexer_ffi_helpers/mod.rs b/integration_tests/tests/indexer_ffi_helpers/mod.rs index e9527269..170102fd 100644 --- a/integration_tests/tests/indexer_ffi_helpers/mod.rs +++ b/integration_tests/tests/indexer_ffi_helpers/mod.rs @@ -37,6 +37,7 @@ unsafe extern "C" { pub unsafe fn start_indexer( runtime: *const Runtime, config_path: *const c_char, + storage_dir: *const c_char, ) -> InitializedIndexerServiceFFIResult; } @@ -49,9 +50,8 @@ pub fn setup_indexer_ffi(bedrock_addr: SocketAddr) -> Result<(IndexerServiceFFI, temp_indexer_dir.path().display() ); - let indexer_config = - integration_tests::config::indexer_config(bedrock_addr, temp_indexer_dir.path().to_owned()) - .context("Failed to create Indexer config")?; + let indexer_config = integration_tests::config::indexer_config(bedrock_addr) + .context("Failed to create Indexer config")?; let config_json = serde_json::to_vec(&indexer_config)?; let config_path = temp_indexer_dir.path().join("indexer_config.json"); @@ -59,10 +59,13 @@ pub fn setup_indexer_ffi(bedrock_addr: SocketAddr) -> Result<(IndexerServiceFFI, file.write_all(&config_json)?; file.flush()?; + let config_path_c = CString::new(config_path.to_str().unwrap())?; + let storage_dir_c = CString::new(temp_indexer_dir.path().to_str().unwrap())?; let res = // SAFETY: null runtime → the FFI creates and owns its own tokio runtime, - // so there is no external runtime whose address we must keep stable. - unsafe { start_indexer(std::ptr::null(), CString::new(config_path.to_str().unwrap())?.as_ptr()) }; + // so there is no external runtime whose address we must keep stable. The + // temp dir is the indexer's storage location. + unsafe { start_indexer(std::ptr::null(), config_path_c.as_ptr(), storage_dir_c.as_ptr()) }; if res.error.is_error() { anyhow::bail!("Indexer FFI error {:?}", res.error); diff --git a/lez/configs/docker-all-in-one/indexer_config.json b/lez/configs/docker-all-in-one/indexer_config.json index f2005ff5..c1ff65b0 100644 --- a/lez/configs/docker-all-in-one/indexer_config.json +++ b/lez/configs/docker-all-in-one/indexer_config.json @@ -1,5 +1,4 @@ { - "home": "./indexer/service", "consensus_info_polling_interval": "1s", "bedrock_config": { "addr": "http://logos-blockchain-node-0:18080" diff --git a/lez/indexer/core/src/config.rs b/lez/indexer/core/src/config.rs index 6a019828..cb7f3dfe 100644 --- a/lez/indexer/core/src/config.rs +++ b/lez/indexer/core/src/config.rs @@ -1,9 +1,4 @@ -use std::{ - fs::File, - io::BufReader, - path::{Path, PathBuf}, - time::Duration, -}; +use std::{fs::File, io::BufReader, path::Path, time::Duration}; use anyhow::{Context as _, Result}; use common::config::BasicAuth; @@ -21,8 +16,6 @@ pub struct ClientConfig { #[derive(Debug, Clone, Serialize, Deserialize)] pub struct IndexerConfig { - /// Home dir of indexer storage. - pub home: PathBuf, #[serde(with = "humantime_serde")] pub consensus_info_polling_interval: Duration, pub bedrock_config: ClientConfig, diff --git a/lez/indexer/core/src/lib.rs b/lez/indexer/core/src/lib.rs index b0416905..e05fdd1f 100644 --- a/lez/indexer/core/src/lib.rs +++ b/lez/indexer/core/src/lib.rs @@ -1,4 +1,4 @@ -use std::sync::Arc; +use std::{path::Path, sync::Arc}; use anyhow::Result; use common::block::Block; @@ -23,8 +23,8 @@ pub struct IndexerCore { } impl IndexerCore { - pub fn new(config: IndexerConfig) -> Result { - let home = config.home.join("rocksdb"); + pub fn new(config: IndexerConfig, storage_dir: &Path) -> Result { + let home = storage_dir.join("rocksdb"); let basic_auth = config.bedrock_config.auth.clone().map(Into::into); let node = NodeHttpClient::new( diff --git a/lez/indexer/ffi/indexer_ffi.h b/lez/indexer/ffi/indexer_ffi.h index 4bd2d549..eb3bf117 100644 --- a/lez/indexer/ffi/indexer_ffi.h +++ b/lez/indexer/ffi/indexer_ffi.h @@ -416,6 +416,9 @@ typedef struct PointerResult_FfiVec_FfiTransaction_____OperationStatus { * - `runtime`: A runtime for the indexer to run on, or null to have the indexer create and own * one. * - `config_path`: A pointer to a string representing the path to the configuration file. + * - `storage_dir`: A pointer to a string naming the directory under which the indexer stores its + * state (`RocksDB`), or null/empty to use the current directory. The host (e.g. a Logos module's + * instance persistence path) owns this location. * * # Returns * @@ -426,9 +429,11 @@ typedef struct PointerResult_FfiVec_FfiTransaction_____OperationStatus { * The caller must ensure that: * - `runtime` is either null or a valid pointer to a [`Runtime`] that outlives the indexer. * - `config_path` is a valid pointer to a null-terminated C string. + * - `storage_dir` is either null or a valid pointer to a null-terminated C string. */ InitializedIndexerServiceFFIResult start_indexer(const struct Runtime *runtime, - const char *config_path); + const char *config_path, + const char *storage_dir); /** * Stops and frees the resources associated with the given indexer service. diff --git a/lez/indexer/ffi/src/api/lifecycle.rs b/lez/indexer/ffi/src/api/lifecycle.rs index 07b2a9be..f668f3ee 100644 --- a/lez/indexer/ffi/src/api/lifecycle.rs +++ b/lez/indexer/ffi/src/api/lifecycle.rs @@ -15,6 +15,9 @@ pub type InitializedIndexerServiceFFIResult = PointerResult InitializedIndexerServiceFFIResult { - // SAFETY: The caller must ensure the validness of the `runtime` and `config_path` pointers. - unsafe { setup_indexer(runtime, config_path) }.map_or_else( + // SAFETY: The caller must ensure the validness of the pointer arguments. + unsafe { setup_indexer(runtime, config_path, storage_dir) }.map_or_else( InitializedIndexerServiceFFIResult::from_error, InitializedIndexerServiceFFIResult::from_value, ) @@ -44,6 +49,7 @@ pub unsafe extern "C" fn start_indexer( /// /// - `runtime`: A runtime for the indexer to run on, or null to create and own one. /// - `config_path`: A pointer to a string representing the path to the configuration file. +/// - `storage_dir`: A pointer to a string naming the storage directory, or null/empty for `.`. /// /// # Returns /// @@ -54,9 +60,11 @@ pub unsafe extern "C" fn start_indexer( /// The caller must ensure that: /// - `runtime` is either null or a valid pointer to a [`Runtime`] that outlives the indexer. /// - `config_path` is a valid pointer to a null-terminated C string. +/// - `storage_dir` is either null or a valid pointer to a null-terminated C string. unsafe fn setup_indexer( runtime: *const Runtime, config_path: *const c_char, + storage_dir: *const c_char, ) -> Result { let user_config_path = PathBuf::from( unsafe { std::ffi::CStr::from_ptr(config_path) } @@ -71,6 +79,25 @@ unsafe fn setup_indexer( OperationStatus::InitializationError })?; + // The host owns where state lives. An empty/null `storage_dir` falls back to + // the current directory (matches the standalone service's `--data-dir` + // default), but a Logos module passes its instance persistence path. + let storage_dir = if storage_dir.is_null() { + PathBuf::from(".") + } else { + let storage_dir = unsafe { std::ffi::CStr::from_ptr(storage_dir) } + .to_str() + .map_err(|e| { + log::error!("Could not convert the storage dir to string: {e}"); + OperationStatus::InitializationError + })?; + if storage_dir.is_empty() { + PathBuf::from(".") + } else { + PathBuf::from(storage_dir) + } + }; + // Use the caller's runtime if one was supplied, otherwise create (and own) // our own. The `Runtime` wrapper drops the underlying tokio runtime only // when we own it; a borrowed one is left to its external owner. @@ -85,7 +112,7 @@ unsafe fn setup_indexer( unsafe { Runtime::from_borrowed(caller.as_ref()) } }; - let core = IndexerCore::new(config).map_err(|e| { + let core = IndexerCore::new(config, &storage_dir).map_err(|e| { log::error!("Could not initialize indexer core: {e}"); OperationStatus::InitializationError })?; diff --git a/lez/indexer/service/configs/debug/indexer_config.json b/lez/indexer/service/configs/debug/indexer_config.json index a3ad0b6a..47b1629e 100644 --- a/lez/indexer/service/configs/debug/indexer_config.json +++ b/lez/indexer/service/configs/debug/indexer_config.json @@ -1,5 +1,4 @@ { - "home": ".", "consensus_info_polling_interval": "1s", "bedrock_config": { "addr": "http://localhost:8080" diff --git a/lez/indexer/service/src/lib.rs b/lez/indexer/service/src/lib.rs index b0a6e516..b1c57163 100644 --- a/lez/indexer/service/src/lib.rs +++ b/lez/indexer/service/src/lib.rs @@ -1,4 +1,4 @@ -use std::net::SocketAddr; +use std::{net::SocketAddr, path::Path}; use anyhow::{Context as _, Result}; pub use indexer_core::config::*; @@ -65,9 +65,13 @@ impl Drop for IndexerHandle { } } -pub async fn run_server(config: IndexerConfig, port: u16) -> Result { +pub async fn run_server( + config: IndexerConfig, + storage_dir: &Path, + port: u16, +) -> Result { #[cfg(feature = "mock-responses")] - let _ = config; + let _ = (config, storage_dir); let server = Server::builder() .build(SocketAddr::from(([0, 0, 0, 0], port))) @@ -82,8 +86,8 @@ pub async fn run_server(config: IndexerConfig, port: u16) -> Result Result<()> { env_logger::init(); - let Args { config_path, port } = Args::parse(); + let Args { + config_path, + port, + data_dir, + } = Args::parse(); let cancellation_token = listen_for_shutdown_signal(); let config = indexer_service::IndexerConfig::from_path(&config_path)?; - let indexer_handle = indexer_service::run_server(config, port).await?; + let indexer_handle = indexer_service::run_server(config, data_dir.as_path(), port).await?; tokio::select! { () = cancellation_token.cancelled() => { diff --git a/lez/indexer/service/src/service.rs b/lez/indexer/service/src/service.rs index a959b80c..7a8ed90f 100644 --- a/lez/indexer/service/src/service.rs +++ b/lez/indexer/service/src/service.rs @@ -1,4 +1,4 @@ -use std::{pin::pin, sync::Arc}; +use std::{path::Path, pin::pin, sync::Arc}; use anyhow::{Context as _, Result, bail}; use arc_swap::ArcSwap; @@ -19,8 +19,8 @@ pub struct IndexerService { } impl IndexerService { - pub fn new(config: IndexerConfig) -> Result { - let indexer = IndexerCore::new(config)?; + pub fn new(config: IndexerConfig, storage_dir: &Path) -> Result { + let indexer = IndexerCore::new(config, storage_dir)?; let subscription_service = SubscriptionService::spawn_new(indexer.clone()); Ok(Self { diff --git a/test_fixtures/src/config.rs b/test_fixtures/src/config.rs index 73cd775b..8684dd4d 100644 --- a/test_fixtures/src/config.rs +++ b/test_fixtures/src/config.rs @@ -163,9 +163,8 @@ pub fn wallet_config(sequencer_addr: SocketAddr) -> Result { }) } -pub fn indexer_config(bedrock_addr: SocketAddr, home: PathBuf) -> Result { +pub fn indexer_config(bedrock_addr: SocketAddr) -> Result { Ok(IndexerConfig { - home, consensus_info_polling_interval: Duration::from_secs(1), bedrock_config: ClientConfig { addr: addr_to_url(UrlProtocol::Http, bedrock_addr) diff --git a/test_fixtures/src/setup.rs b/test_fixtures/src/setup.rs index 7eb0e1fd..6559192f 100644 --- a/test_fixtures/src/setup.rs +++ b/test_fixtures/src/setup.rs @@ -98,10 +98,10 @@ pub async fn setup_indexer(bedrock_addr: SocketAddr) -> Result<(IndexerHandle, T temp_indexer_dir.path().display() ); - let indexer_config = config::indexer_config(bedrock_addr, temp_indexer_dir.path().to_owned()) - .context("Failed to create Indexer config")?; + let indexer_config = + config::indexer_config(bedrock_addr).context("Failed to create Indexer config")?; - indexer_service::run_server(indexer_config, 0) + indexer_service::run_server(indexer_config, temp_indexer_dir.path(), 0) .await .context("Failed to run Indexer Service") .map(|handle| (handle, temp_indexer_dir)) From c0fbaaf08e94b7cd046caa876c65590d2a6de620 Mon Sep 17 00:00:00 2001 From: erhant Date: Mon, 22 Jun 2026 13:16:35 +0300 Subject: [PATCH 08/14] refactor: use channel id for rocksdb filename [skip ci] --- Justfile | 4 ++-- lez/indexer/core/src/lib.rs | 4 +++- 2 files changed, 5 insertions(+), 3 deletions(-) diff --git a/Justfile b/Justfile index 87cab5b7..690bdc1d 100644 --- a/Justfile +++ b/Justfile @@ -93,7 +93,7 @@ clean: @echo "🧹 Cleaning run artifacts" rm -rf lez/sequencer/service/bedrock_signing_key rm -rf lez/sequencer/service/rocksdb - rm -rf lez/indexer/service/rocksdb + rm -rf lez/indexer/service/rocksdb* rm -rf lez/wallet/configs/debug/storage.json - rm -rf rocksdb + rm -rf rocksdb* cd bedrock && docker compose down -v diff --git a/lez/indexer/core/src/lib.rs b/lez/indexer/core/src/lib.rs index e05fdd1f..d8890af2 100644 --- a/lez/indexer/core/src/lib.rs +++ b/lez/indexer/core/src/lib.rs @@ -24,7 +24,9 @@ pub struct IndexerCore { impl IndexerCore { pub fn new(config: IndexerConfig, storage_dir: &Path) -> Result { - let home = storage_dir.join("rocksdb"); + // Namespace the DB by channel so indexers on different channels can + // share a storage dir without their RocksDB state colliding. + let home = storage_dir.join(format!("rocksdb-{}", config.channel_id)); let basic_auth = config.bedrock_config.auth.clone().map(Into::into); let node = NodeHttpClient::new( From c57bf16d1545aa9a116f2383f814c50a37e7b005 Mon Sep 17 00:00:00 2001 From: erhant Date: Mon, 22 Jun 2026 18:36:29 +0300 Subject: [PATCH 09/14] feat: added status reporter for indexer (for UI) --- Cargo.lock | 1 + lez/indexer/core/src/lib.rs | 59 ++++++++++++++++++++++++++++++-- lez/indexer/core/src/status.rs | 41 ++++++++++++++++++++++ lez/indexer/ffi/Cargo.toml | 2 ++ lez/indexer/ffi/indexer_ffi.h | 25 ++++++++++++++ lez/indexer/ffi/src/api/query.rs | 52 ++++++++++++++++++++++++++++ 6 files changed, 177 insertions(+), 3 deletions(-) create mode 100644 lez/indexer/core/src/status.rs diff --git a/Cargo.lock b/Cargo.lock index 9af3ff27..318b47e6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3830,6 +3830,7 @@ dependencies = [ "indexer_service_protocol", "lee", "log", + "serde_json", "tokio", ] diff --git a/lez/indexer/core/src/lib.rs b/lez/indexer/core/src/lib.rs index d8890af2..deb4d7af 100644 --- a/lez/indexer/core/src/lib.rs +++ b/lez/indexer/core/src/lib.rs @@ -1,4 +1,7 @@ -use std::{path::Path, sync::Arc}; +use std::{ + path::Path, + sync::{Arc, Mutex}, +}; use anyhow::Result; use common::block::Block; @@ -10,16 +13,23 @@ use logos_blockchain_zone_sdk::{ CommonHttpClient, ZoneMessage, adapter::NodeHttpClient, indexer::ZoneIndexer, }; -use crate::{block_store::IndexerStore, config::IndexerConfig}; +use crate::{ + block_store::IndexerStore, + config::IndexerConfig, + status::{IndexerStatus, SyncState}, +}; pub mod block_store; pub mod config; +pub mod status; #[derive(Clone)] pub struct IndexerCore { pub zone_indexer: Arc>, pub config: IndexerConfig, pub store: IndexerStore, + /// Live ingestion status; updated by the ingest stream, read by `status`. + pub status: Arc>, } impl IndexerCore { @@ -39,9 +49,31 @@ impl IndexerCore { zone_indexer: Arc::new(zone_indexer), config, store: IndexerStore::open_db(&home)?, + status: Arc::new(Mutex::new(IndexerStatus::starting())), }) } + /// Snapshot of the current ingestion status (sync state + indexed tip). + /// + /// Combines the ingest loop's live state with the L2 tip read fresh from the + /// store, so callers (FFI/RPC) can tell "catching up" from "failed". + #[must_use] + pub fn status(&self) -> IndexerStatus { + let mut snapshot = self + .status + .lock() + .expect("indexer status mutex poisoned") + .clone(); + snapshot.indexed_block_id = self.store.get_last_block_id().ok().flatten(); + snapshot + } + + /// Apply a short, non-blocking update to the shared status. + fn set_status(&self, update: impl FnOnce(&mut IndexerStatus)) { + let mut status = self.status.lock().expect("indexer status mutex poisoned"); + update(&mut status); + } + pub fn subscribe_parse_block_stream(&self) -> impl futures::Stream> + '_ { let poll_interval = self.config.consensus_info_polling_interval; let initial_cursor = self @@ -62,14 +94,34 @@ impl IndexerCore { let stream = match self.zone_indexer.next_messages(cursor).await { Ok(s) => s, Err(err) => { + // `next_messages` reads L1 consensus info internally, so + // this also covers an unreachable/misconfigured L1 node. error!("Failed to start zone-sdk next_messages stream: {err}"); + self.set_status(|s| { + s.state = SyncState::Error; + s.last_error = Some(format!("cannot reach L1 / read channel: {err}")); + }); tokio::time::sleep(poll_interval).await; continue; } }; let mut stream = std::pin::pin!(stream); + // Flip to Syncing on the first message of this cycle (not merely on + // a successful poll) so the steady-state CaughtUp status doesn't + // flicker. Until then the state stays Starting (cold-start scan of + // empty L1 history) or CaughtUp (idle). + let mut announced_syncing = false; + while let Some((msg, slot)) = stream.next().await { + if !announced_syncing { + self.set_status(|s| { + s.state = SyncState::Syncing; + s.last_error = None; + }); + announced_syncing = true; + } + let zone_block = match msg { ZoneMessage::Block(b) => b, // Non-block messages don't carry a cursor position; the @@ -107,7 +159,8 @@ impl IndexerCore { yield Ok(block); } - // Stream ended (caught up to LIB). Sleep then poll again. + // Stream drained: caught up to LIB as of this cycle. Sleep then poll again. + self.set_status(|s| s.state = SyncState::CaughtUp); tokio::time::sleep(poll_interval).await; } } diff --git a/lez/indexer/core/src/status.rs b/lez/indexer/core/src/status.rs new file mode 100644 index 00000000..a8a8b845 --- /dev/null +++ b/lez/indexer/core/src/status.rs @@ -0,0 +1,41 @@ +use serde::Serialize; + +/// Coarse lifecycle state of the indexer's ingestion loop, so a client can tell +/// "still catching up" apart from "something went wrong". +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum SyncState { + /// Booted; no ingestion cycle has run yet. + Starting, + /// Streaming finalized messages toward the L1 frontier. + Syncing, + /// Drained the stream up to LIB; idle until new blocks finalize. + CaughtUp, + /// The last cycle failed (e.g. the L1 node is unreachable). See `last_error`. + Error, +} + +/// Snapshot of the indexer's sync status. +/// +/// `state` is the coarse phase and `last_error` carries the reason when it is +/// `Error`. `indexed_block_id` is the L2 tip, filled from the store at read time +/// (so it is always `None` in the value held behind the status mutex — see +/// `IndexerCore::status`). +#[derive(Debug, Clone, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct IndexerStatus { + pub state: SyncState, + pub indexed_block_id: Option, + pub last_error: Option, +} + +impl IndexerStatus { + /// Initial status before any ingestion cycle has run. + pub(crate) const fn starting() -> Self { + Self { + state: SyncState::Starting, + indexed_block_id: None, + last_error: None, + } + } +} diff --git a/lez/indexer/ffi/Cargo.toml b/lez/indexer/ffi/Cargo.toml index 50748d56..331a996a 100644 --- a/lez/indexer/ffi/Cargo.toml +++ b/lez/indexer/ffi/Cargo.toml @@ -15,6 +15,8 @@ env_logger.workspace = true log = { workspace = true } tokio = { features = ["rt-multi-thread"], workspace = true } futures.workspace = true +# Serializes the indexer status snapshot to JSON for `query_status`. +serde_json.workspace = true [build-dependencies] cbindgen = "0.29" diff --git a/lez/indexer/ffi/indexer_ffi.h b/lez/indexer/ffi/indexer_ffi.h index eb3bf117..ae61eecb 100644 --- a/lez/indexer/ffi/indexer_ffi.h +++ b/lez/indexer/ffi/indexer_ffi.h @@ -496,6 +496,31 @@ void free_cstring(char *block); */ struct LastBlockIdResult query_last_block(const struct IndexerServiceFFI *indexer); +/** + * Query the indexer's current sync status as a JSON C-string. + * + * The JSON schema is owned by `indexer_core` (`IndexerStatus`): an object with + * `state` (`starting`/`syncing`/`caught_up`/`error`), `indexedBlockId`, and + * `lastError`. Lets a client distinguish "still catching up" from "something + * went wrong". + * + * # Arguments + * + * - `indexer`: A pointer to the [`IndexerServiceFFI`] instance to be queried. + * + * # Returns + * + * A heap-allocated, null-terminated JSON string that the caller MUST free with + * `free_cstring`. Returns null on error (null `indexer` pointer or a + * serialization failure). + * + * # Safety + * + * The caller must ensure that: + * - `indexer` is a valid pointer to a [`IndexerServiceFFI`] instance. + */ +char *query_status(const struct IndexerServiceFFI *indexer); + /** * Query the block by id from indexer. * diff --git a/lez/indexer/ffi/src/api/query.rs b/lez/indexer/ffi/src/api/query.rs index 23233a76..1943f6d4 100644 --- a/lez/indexer/ffi/src/api/query.rs +++ b/lez/indexer/ffi/src/api/query.rs @@ -1,3 +1,5 @@ +use std::ffi::{CString, c_char}; + use indexer_service_protocol::AccountId; use crate::{ @@ -86,6 +88,56 @@ pub unsafe extern "C" fn query_last_block(indexer: *const IndexerServiceFFI) -> ) } +/// Query the indexer's current sync status as a JSON C-string. +/// +/// The JSON schema is owned by `indexer_core` (`IndexerStatus`): an object with +/// `state` (`starting`/`syncing`/`caught_up`/`error`), `indexedBlockId`, and +/// `lastError`. Lets a client distinguish "still catching up" from "something +/// went wrong". +/// +/// # Arguments +/// +/// - `indexer`: A pointer to the [`IndexerServiceFFI`] instance to be queried. +/// +/// # Returns +/// +/// A heap-allocated, null-terminated JSON string that the caller MUST free with +/// `free_cstring`. Returns null on error (null `indexer` pointer or a +/// serialization failure). +/// +/// # Safety +/// +/// The caller must ensure that: +/// - `indexer` is a valid pointer to a [`IndexerServiceFFI`] instance. +#[unsafe(no_mangle)] +pub unsafe extern "C" fn query_status(indexer: *const IndexerServiceFFI) -> *mut c_char { + if indexer.is_null() { + log::error!( + "Attempted to query status on a null indexer pointer. This is a bug. Aborting." + ); + return std::ptr::null_mut(); + } + + let indexer = unsafe { &*indexer }; + let status = indexer.core().status(); + + let json = match serde_json::to_string(&status) { + Ok(json) => json, + Err(e) => { + log::error!("Failed to serialize indexer status: {e}"); + return std::ptr::null_mut(); + } + }; + + CString::new(json).map_or_else( + |e| { + log::error!("Indexer status JSON contained an interior nul byte: {e}"); + std::ptr::null_mut() + }, + CString::into_raw, + ) +} + /// Query the block by id from indexer. /// /// # Arguments From 0bb7b30d6369c7cc6acdfb5d02dde479c4dafedb Mon Sep 17 00:00:00 2001 From: erhant Date: Tue, 23 Jun 2026 10:45:13 +0300 Subject: [PATCH 10/14] fix: rm comments, add `cpp_compat` to cbindgen --- lez/indexer/ffi/Cargo.toml | 3 --- lez/indexer/ffi/build.rs | 1 + lez/indexer/ffi/indexer_ffi.h | 8 ++++++++ 3 files changed, 9 insertions(+), 3 deletions(-) diff --git a/lez/indexer/ffi/Cargo.toml b/lez/indexer/ffi/Cargo.toml index 331a996a..a1615b75 100644 --- a/lez/indexer/ffi/Cargo.toml +++ b/lez/indexer/ffi/Cargo.toml @@ -7,15 +7,12 @@ version = "0.1.0" [dependencies] lee.workspace = true indexer_core.workspace = true -# `convert` brings in the LEE <-> protocol `From`/`TryFrom` impls the query -# marshalling relies on (previously pulled in transitively via `indexer_service`). indexer_service_protocol = { workspace = true, features = ["convert"] } env_logger.workspace = true log = { workspace = true } tokio = { features = ["rt-multi-thread"], workspace = true } futures.workspace = true -# Serializes the indexer status snapshot to JSON for `query_status`. serde_json.workspace = true [build-dependencies] diff --git a/lez/indexer/ffi/build.rs b/lez/indexer/ffi/build.rs index 349ff38d..4d2faae8 100644 --- a/lez/indexer/ffi/build.rs +++ b/lez/indexer/ffi/build.rs @@ -6,6 +6,7 @@ fn main() { cbindgen::Builder::new() .with_crate(crate_dir) .with_language(cbindgen::Language::C) + .with_cpp_compat(true) .with_pragma_once(true) .generate() .expect("Unable to generate bindings") diff --git a/lez/indexer/ffi/indexer_ffi.h b/lez/indexer/ffi/indexer_ffi.h index ae61eecb..569a26a3 100644 --- a/lez/indexer/ffi/indexer_ffi.h +++ b/lez/indexer/ffi/indexer_ffi.h @@ -407,6 +407,10 @@ typedef struct PointerResult_FfiVec_FfiTransaction_____OperationStatus { enum OperationStatus error; } PointerResult_FfiVec_FfiTransaction_____OperationStatus; +#ifdef __cplusplus +extern "C" { +#endif // __cplusplus + /** * Creates and starts an indexer based on the provided * configuration file path. @@ -804,3 +808,7 @@ void free_ffi_transaction_vec(struct FfiVec_FfiTransaction *val); bool is_ok(const enum OperationStatus *self); bool is_error(const enum OperationStatus *self); + +#ifdef __cplusplus +} // extern "C" +#endif // __cplusplus From 5d51e0b59cf07925e36c55bdc9204d578714e2f8 Mon Sep 17 00:00:00 2001 From: erhant Date: Tue, 23 Jun 2026 11:09:49 +0300 Subject: [PATCH 11/14] fix(indexer): use `/var/lib/indexer_service` volume path in `Dockerfile` + add `*` wildcard for rocksdb ignore --- .gitignore | 2 +- lez/indexer/service/Dockerfile | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/.gitignore b/.gitignore index 8041f277..4c1a18c8 100644 --- a/.gitignore +++ b/.gitignore @@ -8,7 +8,7 @@ data/ .idea/ .vscode/ -rocksdb +rocksdb* sequencer/service/data/ storage.json diff --git a/lez/indexer/service/Dockerfile b/lez/indexer/service/Dockerfile index 2a7e829a..7499875e 100644 --- a/lez/indexer/service/Dockerfile +++ b/lez/indexer/service/Dockerfile @@ -71,4 +71,4 @@ ENV RUST_LOG=info USER indexer_service_user WORKDIR /indexer_service -CMD ["indexer_service", "/etc/indexer_service/indexer_config.json"] +CMD ["indexer_service", "/etc/indexer_service/indexer_config.json", "--data-dir", "/var/lib/indexer_service"] From 21959374004e8a8748339963a8486af1b007e98d Mon Sep 17 00:00:00 2001 From: erhant Date: Tue, 23 Jun 2026 11:24:54 +0300 Subject: [PATCH 12/14] fix(indexer_ffi): use already FFI-safe `Runtime` --- lez/indexer/ffi/indexer_ffi.h | 57 +++++++++++++++++----------------- lez/indexer/ffi/src/indexer.rs | 43 +++++++++---------------- 2 files changed, 42 insertions(+), 58 deletions(-) diff --git a/lez/indexer/ffi/indexer_ffi.h b/lez/indexer/ffi/indexer_ffi.h index 569a26a3..8347ad3c 100644 --- a/lez/indexer/ffi/indexer_ffi.h +++ b/lez/indexer/ffi/indexer_ffi.h @@ -24,35 +24,6 @@ typedef enum FfiBedrockStatus { Finalized, } FfiBedrockStatus; -/** - * FFI-owned indexer. - * - * Has three fields behind `c_void` (so that cbindgen never needs to see their Rust layout): - * - An [`IndexerCore`] used to answer queries - * - The background task [`JoinHandle`] that drives ingestion (consuming the block stream so the - * store stays populated) - * - The [`Runtime`] they run on. It owns the underlying tokio runtime when we created it (and - * drops it on teardown) and merely borrows it when the caller supplied one. - */ -typedef struct IndexerServiceFFI { - void *core; - void *ingest_handle; - void *runtime; -} IndexerServiceFFI; - -/** - * Simple wrapper around a pointer to a value or an error. - * - * Pointer is not guaranteed. You should check the error field before - * dereferencing the pointer. - */ -typedef struct PointerResult_IndexerServiceFFI__OperationStatus { - struct IndexerServiceFFI *value; - enum OperationStatus error; -} PointerResult_IndexerServiceFFI__OperationStatus; - -typedef struct PointerResult_IndexerServiceFFI__OperationStatus InitializedIndexerServiceFFIResult; - typedef enum PointerKind_Tag { Owned, Borrowed, @@ -82,6 +53,34 @@ typedef struct Runtime { struct Pointer_Runtime inner; } Runtime; +/** + * FFI-owned indexer. + * + * - An [`IndexerCore`] used to answer queries + * - The background task [`JoinHandle`] that drives ingestion (consuming the block stream so the + * store stays populated) + * - The [`Runtime`] used to run async queries against the store (either owned or borrowed), + * already FFI-safe. + */ +typedef struct IndexerServiceFFI { + void *core; + void *ingest_handle; + struct Runtime runtime; +} IndexerServiceFFI; + +/** + * Simple wrapper around a pointer to a value or an error. + * + * Pointer is not guaranteed. You should check the error field before + * dereferencing the pointer. + */ +typedef struct PointerResult_IndexerServiceFFI__OperationStatus { + struct IndexerServiceFFI *value; + enum OperationStatus error; +} PointerResult_IndexerServiceFFI__OperationStatus; + +typedef struct PointerResult_IndexerServiceFFI__OperationStatus InitializedIndexerServiceFFIResult; + /** * Result of [`query_last_block`], returned **inline** (no heap allocation, so * there is no corresponding `free_*` to call). diff --git a/lez/indexer/ffi/src/indexer.rs b/lez/indexer/ffi/src/indexer.rs index 619373e6..0b6b874a 100644 --- a/lez/indexer/ffi/src/indexer.rs +++ b/lez/indexer/ffi/src/indexer.rs @@ -7,17 +7,16 @@ use crate::Runtime; /// FFI-owned indexer. /// -/// Has three fields behind `c_void` (so that cbindgen never needs to see their Rust layout): /// - An [`IndexerCore`] used to answer queries /// - The background task [`JoinHandle`] that drives ingestion (consuming the block stream so the /// store stays populated) -/// - The [`Runtime`] they run on. It owns the underlying tokio runtime when we created it (and -/// drops it on teardown) and merely borrows it when the caller supplied one. +/// - The [`Runtime`] used to run async queries against the store (either owned or borrowed), +/// already FFI-safe. #[repr(C)] pub struct IndexerServiceFFI { core: *mut c_void, ingest_handle: *mut c_void, - runtime: *mut c_void, + runtime: Runtime, } impl IndexerServiceFFI { @@ -26,7 +25,7 @@ impl IndexerServiceFFI { Self { core: Box::into_raw(Box::new(core)).cast::(), ingest_handle: Box::into_raw(Box::new(ingest_handle)).cast::(), - runtime: Box::into_raw(Box::new(runtime)).cast::(), + runtime, } } @@ -44,38 +43,24 @@ impl IndexerServiceFFI { /// Borrow the runtime to `block_on` an async store query. #[must_use] pub const fn runtime(&self) -> &Runtime { - unsafe { - self.runtime - .cast::() - .as_ref() - .expect("Runtime must be a non-null pointer") - } + &self.runtime } } -// Implement Drop to stop ingestion and free the boxed resources. impl Drop for IndexerServiceFFI { fn drop(&mut self) { - let Self { - core, - ingest_handle, - runtime, - } = self; - - if !ingest_handle.is_null() { - // Stop the background ingestion task before tearing down the core. - let handle = unsafe { Box::from_raw(ingest_handle.cast::>()) }; + if !self.ingest_handle.is_null() { + let handle = unsafe { Box::from_raw(self.ingest_handle.cast::>()) }; + // stop the background ingestion task before tearing down the core. handle.abort(); drop(handle); } - if !core.is_null() { - drop(unsafe { Box::from_raw(core.cast::()) }); - } - // Dropping the `Runtime` shuts down the tokio runtime only if we own it - // (a borrowed one is left for its external owner). Done last, and from - // the consumer thread, so it never drops from within a runtime worker. - if !runtime.is_null() { - drop(unsafe { Box::from_raw(runtime.cast::()) }); + if !self.core.is_null() { + drop(unsafe { Box::from_raw(self.core.cast::()) }); } + + // `runtime` field is dropped automatically on return here: + // - if runtime was owned, it is shutdown at this point + // - if it was borrowed, it continues to live within the external owner } } From 735235efa1dc18240258cb033f0694d77c00828e Mon Sep 17 00:00:00 2001 From: erhant Date: Tue, 23 Jun 2026 14:33:38 +0300 Subject: [PATCH 13/14] fix(indexer)!: address @Arjentix comments on `IndexerStatus` --- Cargo.lock | 1 + Cargo.toml | 1 + lez/indexer/core/Cargo.toml | 1 + lez/indexer/core/src/lib.rs | 53 ++++++++++---------- lez/indexer/core/src/status.rs | 88 +++++++++++++++++++++++++++++----- 5 files changed, 102 insertions(+), 42 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 318b47e6..3a4db53d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3799,6 +3799,7 @@ name = "indexer_core" version = "0.1.0" dependencies = [ "anyhow", + "arc-swap", "async-stream", "authenticated_transfer_core", "borsh", diff --git a/Cargo.toml b/Cargo.toml index 25f03774..4c64aaba 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -117,6 +117,7 @@ hex = "0.4.3" bytemuck = "1.24.0" bytesize = { version = "2.3.1", features = ["serde"] } humantime-serde = "1.1" +arc-swap = "1.7" humantime = "2.1" aes-gcm = "0.10.3" toml = "0.9.8" diff --git a/lez/indexer/core/Cargo.toml b/lez/indexer/core/Cargo.toml index c6cc5fc6..758acdd6 100644 --- a/lez/indexer/core/Cargo.toml +++ b/lez/indexer/core/Cargo.toml @@ -16,6 +16,7 @@ storage.workspace = true testnet_initial_state.workspace = true anyhow.workspace = true +arc-swap.workspace = true log.workspace = true serde.workspace = true humantime-serde.workspace = true diff --git a/lez/indexer/core/src/lib.rs b/lez/indexer/core/src/lib.rs index deb4d7af..0d595fc0 100644 --- a/lez/indexer/core/src/lib.rs +++ b/lez/indexer/core/src/lib.rs @@ -1,9 +1,7 @@ -use std::{ - path::Path, - sync::{Arc, Mutex}, -}; +use std::{path::Path, sync::Arc}; use anyhow::Result; +use arc_swap::ArcSwap; use common::block::Block; // ToDo: Remove after testnet use futures::StreamExt as _; @@ -16,7 +14,7 @@ use logos_blockchain_zone_sdk::{ use crate::{ block_store::IndexerStore, config::IndexerConfig, - status::{IndexerStatus, SyncState}, + status::{IndexerStatus, IndexerSyncStatus}, }; pub mod block_store; @@ -29,7 +27,7 @@ pub struct IndexerCore { pub config: IndexerConfig, pub store: IndexerStore, /// Live ingestion status; updated by the ingest stream, read by `status`. - pub status: Arc>, + pub status: Arc>, } impl IndexerCore { @@ -49,29 +47,27 @@ impl IndexerCore { zone_indexer: Arc::new(zone_indexer), config, store: IndexerStore::open_db(&home)?, - status: Arc::new(Mutex::new(IndexerStatus::starting())), + status: Arc::new(ArcSwap::from_pointee(IndexerSyncStatus::starting())), }) } /// Snapshot of the current ingestion status (sync state + indexed tip). /// - /// Combines the ingest loop's live state with the L2 tip read fresh from the + /// Combines the ingest loop's live status with the L2 tip read fresh from the /// store, so callers (FFI/RPC) can tell "catching up" from "failed". #[must_use] pub fn status(&self) -> IndexerStatus { - let mut snapshot = self - .status - .lock() - .expect("indexer status mutex poisoned") - .clone(); - snapshot.indexed_block_id = self.store.get_last_block_id().ok().flatten(); - snapshot + let sync = IndexerSyncStatus::clone(&self.status.load()); + let indexed_block_id = self.store.get_last_block_id().ok().flatten(); + IndexerStatus { + sync, + indexed_block_id, + } } - /// Apply a short, non-blocking update to the shared status. - fn set_status(&self, update: impl FnOnce(&mut IndexerStatus)) { - let mut status = self.status.lock().expect("indexer status mutex poisoned"); - update(&mut status); + /// Atomically publish a new ingestion status for readers of `status`. + fn set_status(&self, status: IndexerSyncStatus) { + self.status.store(Arc::new(status)); } pub fn subscribe_parse_block_stream(&self) -> impl futures::Stream> + '_ { @@ -97,10 +93,9 @@ impl IndexerCore { // `next_messages` reads L1 consensus info internally, so // this also covers an unreachable/misconfigured L1 node. error!("Failed to start zone-sdk next_messages stream: {err}"); - self.set_status(|s| { - s.state = SyncState::Error; - s.last_error = Some(format!("cannot reach L1 / read channel: {err}")); - }); + self.set_status(IndexerSyncStatus::error(format!( + "cannot reach L1 / read channel: {err}" + ))); tokio::time::sleep(poll_interval).await; continue; } @@ -115,10 +110,7 @@ impl IndexerCore { while let Some((msg, slot)) = stream.next().await { if !announced_syncing { - self.set_status(|s| { - s.state = SyncState::Syncing; - s.last_error = None; - }); + self.set_status(IndexerSyncStatus::syncing()); announced_syncing = true; } @@ -159,8 +151,11 @@ impl IndexerCore { yield Ok(block); } - // Stream drained: caught up to LIB as of this cycle. Sleep then poll again. - self.set_status(|s| s.state = SyncState::CaughtUp); + // Stream drained: caught up to LIB as of this cycle. Clears any + // prior error (e.g. a transient L1 disconnect that left no + // backlog, so the `Syncing` branch above never ran). Sleep then + // poll again. + self.set_status(IndexerSyncStatus::caught_up()); tokio::time::sleep(poll_interval).await; } } diff --git a/lez/indexer/core/src/status.rs b/lez/indexer/core/src/status.rs index a8a8b845..a7b91f8b 100644 --- a/lez/indexer/core/src/status.rs +++ b/lez/indexer/core/src/status.rs @@ -4,7 +4,7 @@ use serde::Serialize; /// "still catching up" apart from "something went wrong". #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] #[serde(rename_all = "snake_case")] -pub enum SyncState { +pub enum IndexerSyncState { /// Booted; no ingestion cycle has run yet. Starting, /// Streaming finalized messages toward the L1 frontier. @@ -15,27 +15,89 @@ pub enum SyncState { Error, } -/// Snapshot of the indexer's sync status. -/// -/// `state` is the coarse phase and `last_error` carries the reason when it is -/// `Error`. `indexed_block_id` is the L2 tip, filled from the store at read time -/// (so it is always `None` in the value held behind the status mutex — see -/// `IndexerCore::status`). +/// Live ingestion status owned by the ingest loop: the coarse `state` plus the +/// reason when it is `Error`. #[derive(Debug, Clone, Serialize)] #[serde(rename_all = "camelCase")] -pub struct IndexerStatus { - pub state: SyncState, - pub indexed_block_id: Option, +pub struct IndexerSyncStatus { + pub state: IndexerSyncState, pub last_error: Option, } -impl IndexerStatus { +impl IndexerSyncStatus { /// Initial status before any ingestion cycle has run. pub(crate) const fn starting() -> Self { Self { - state: SyncState::Starting, - indexed_block_id: None, + state: IndexerSyncState::Starting, last_error: None, } } + + /// Actively streaming finalized messages toward the L1 frontier. + pub(crate) const fn syncing() -> Self { + Self { + state: IndexerSyncState::Syncing, + last_error: None, + } + } + + /// Drained the stream up to LIB; idle until new blocks finalize. + pub(crate) const fn caught_up() -> Self { + Self { + state: IndexerSyncState::CaughtUp, + last_error: None, + } + } + + /// The last cycle failed; `reason` explains why. + pub(crate) const fn error(reason: String) -> Self { + Self { + state: IndexerSyncState::Error, + last_error: Some(reason), + } + } +} + +/// Full status snapshot returned to callers (FFI/RPC): the live [`SyncStatus`] +/// plus the L2 tip (`indexed_block_id`) read fresh from the store at query time. +/// +/// The tip is tracked by the store, not the ingest loop, so it lives here on the +/// returned snapshot rather than inside the shared [`SyncStatus`]. +#[derive(Debug, Clone, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct IndexerStatus { + #[serde(flatten)] + pub sync: IndexerSyncStatus, + pub indexed_block_id: Option, +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn indexer_status_serializes_to_flat_object() { + let status = IndexerStatus { + sync: IndexerSyncStatus::error("boom".to_owned()), + indexed_block_id: Some(7), + }; + let value = serde_json::to_value(&status).expect("serialize"); + assert_eq!( + value, + serde_json::json!({ + "state": "error", + "lastError": "boom", + "indexedBlockId": 7, + }) + ); + } + + #[test] + fn caught_up_clears_error() { + let value = serde_json::to_value(IndexerSyncStatus::caught_up()).expect("serialize"); + assert_eq!( + value, + serde_json::json!({ "state": "caught_up", "lastError": null }) + ); + } } From c7336c321658bede70daee9cd213ae32e2f75622 Mon Sep 17 00:00:00 2001 From: erhant Date: Tue, 23 Jun 2026 14:46:46 +0300 Subject: [PATCH 14/14] fix(indexer): tiny comment link fix [skip ci] --- lez/indexer/core/src/status.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/lez/indexer/core/src/status.rs b/lez/indexer/core/src/status.rs index a7b91f8b..1193e124 100644 --- a/lez/indexer/core/src/status.rs +++ b/lez/indexer/core/src/status.rs @@ -58,11 +58,11 @@ impl IndexerSyncStatus { } } -/// Full status snapshot returned to callers (FFI/RPC): the live [`SyncStatus`] +/// Full status snapshot returned to callers (FFI/RPC): the live [`IndexerSyncStatus`] /// plus the L2 tip (`indexed_block_id`) read fresh from the store at query time. /// /// The tip is tracked by the store, not the ingest loop, so it lives here on the -/// returned snapshot rather than inside the shared [`SyncStatus`]. +/// returned snapshot rather than inside the shared [`IndexerSyncStatus`]. #[derive(Debug, Clone, Serialize)] #[serde(rename_all = "camelCase")] pub struct IndexerStatus {