diff --git a/CHANGELOG.md b/CHANGELOG.md index 8b849414be89..0119c8638c89 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -33,6 +33,8 @@ ### Added +- [#6720](https://github.com/ChainSafe/forest/pull/6720): Added SQL index and `forest-cli index validate-backfill`(Compatible with `lotus index validate-backfill`) for backfilling the SQL index. + - [#7426](https://github.com/ChainSafe/forest/pull/7426): Added `forest-tool db import-tipset-lookup` for importing a tipset lookup snapshot. - [#7408](https://github.com/ChainSafe/forest/pull/7408): Impl `Forest.IndexBackfill`, `Forest.IndexBackfillStatus` and `Forest.IndexBackfillCancel` RPC methods and the `forest-cli index backfill` command to back-fill the chain index (Ethereum mappings, events, block blooms) through the running daemon, so the node no longer needs to be stopped. An interrupted or cancelled run can be resumed with `--resume`. diff --git a/Cargo.toml b/Cargo.toml index 0cbe27fd6252..7db9418f0e69 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -211,6 +211,7 @@ smallvec = "1" smart-default = "0.7" sonic-rs = "0.5" spire_enum = "1" +sqlx = { version = "0.9", default-features = false, features = ["sqlite", "runtime-tokio", "macros"] } stacker = "0.1" static_assertions = "1" statrs = "0.19" @@ -241,7 +242,6 @@ zstd = "0.13" # optional dependencies console-subscriber = { version = "0.5", features = ["parking_lot"], optional = true } libtest-mimic = "0.8" -sqlx = { version = "0.9", default-features = false, features = ["sqlite", "runtime-tokio", "macros"], optional = true } tikv-jemallocator = { version = "0.7", optional = true } tracing-loki = { version = "0.2", default-features = false, features = ["compat-0-2-1", "rustls"], optional = true } @@ -348,7 +348,7 @@ debug = "line-tables-only" # These should be refactored (probably removed) in #2984 [features] -default = ["jemalloc", "tracing-loki", "sqlite"] +default = ["jemalloc", "tracing-loki"] test = [] # default feature set for unit tests slim = ["rustalloc"] cargo-test = [] # group of tests that is recommended to run with `cargo test` instead of `nextest` @@ -360,7 +360,6 @@ jemalloc-profiling = [ "tikv-jemallocator?/profiling", "tikv-jemallocator?/unprefixed_malloc_on_supported_platforms", ] # enable jemalloc profiling features, use with `release-symbols` build profile for best results -sqlite = ["dep:sqlx"] # Allocator. Use at most one of these. rustalloc = [] diff --git a/docs/docs/users/reference/env_variables.md b/docs/docs/users/reference/env_variables.md index c99c51f95d05..9df9df6460af 100644 --- a/docs/docs/users/reference/env_variables.md +++ b/docs/docs/users/reference/env_variables.md @@ -51,6 +51,7 @@ process. | `FOREST_DRAND_INCENTINET_CONFIG` | string | empty | refer to Drand config format section | Override `DRAND_INCENTINET` config | | `FOREST_TRACE_FILTER_MAX_RESULT` | positive integer | 500 | 1000 | Sets the maximum results returned per request by `trace_filter` | | `FOREST_CHAIN_INDEXER_ENABLED` | 1, true, 0, or false | true | 0 | Whether or not to index the chain to support the Ethereum RPC API. Enabled by default; set to a falsy value (`0`/`false`) to opt out | +| `FOREST_CHAIN_INDEXER_GC_RETENTION_EPOCHS` | non-negative integer | 2880 | 2880, or 0 for disabled | Chain index data retention epochs | | `FOREST_MESSAGES_IN_TIPSET_CACHE_SIZE` | positive integer | 8192 | 42 | The size of an internal cache of tipsets to messages | | `FOREST_STATE_MIGRATION_DB_WRITE_BUFFER` | non-negative integer | 10000 | 100000 | The size of db write buffer for state migration (`~10MB` RAM per `10k` buffer) | | `FOREST_SNAPSHOT_GC_INTERVAL_EPOCHS` | non-negative integer | 20160 | 8000 | The interval in epochs for scheduling snapshot GC | @@ -79,6 +80,7 @@ process. | `FOREST_ETH_RPC_COMPUTE_BLOOM_ON_MISS` | 1 or true | false | 1 | Allows `eth` block RPC methods to compute (and store) the block `logsBloom` when it is not already stored, otherwise such blocks report an all-ones bloom | | `FOREST_RPC_METRICS_DISABLED` | 1 or true | false | 1 | Disable per-method JSON-RPC metrics only, leaving the metrics endpoint and all other metrics (cache, sync, database, ...) intact. To turn off metrics entirely, disable the endpoint instead with `--no-metrics` | | `FOREST_STATE_COMPUTE_MAX_RANGE` | positive integer | 2000 | 100 | The maximum `nEpochs` value `Forest.StateCompute` accepts | +| `FOREST_RPC_SQL_RANGE_THRESHOLD` | positive integer | 10000 | 1000 | The RPC tipset range threshold above which SQL index is prefered over in-memory event collection | ### `FOREST_F3_SIDECAR_FFI_BUILD_OPT_OUT` diff --git a/scripts/tests/api_compare/docker-compose.yml b/scripts/tests/api_compare/docker-compose.yml index 64a513db08ce..9a820f9da3fc 100644 --- a/scripts/tests/api_compare/docker-compose.yml +++ b/scripts/tests/api_compare/docker-compose.yml @@ -42,6 +42,7 @@ services: - FIL_PROOFS_PARAMETER_CACHE=${FIL_PROOFS_PARAMETER_CACHE} - FULLNODE_API_INFO=/dns/forest/tcp/${FOREST_RPC_PORT}/http - FOREST_CHAIN_INDEXER_ENABLED=1 + - FOREST_CHAIN_INDEXER_GC_RETENTION_EPOCHS=2880 entrypoint: ["/bin/bash", "-c"] user: 0:0 command: @@ -91,6 +92,9 @@ services: # Backfill the NOSQL index db forest-cli index backfill --from $$SNAPSHOT_EPOCH --n-tipsets 200 --recompute --allow-near-head + + # backfill the SQL index db + forest-cli index validate-backfill --from $$SNAPSHOT_EPOCH --to $(($$SNAPSHOT_EPOCH - 200)) forest-wallet-import: depends_on: forest: diff --git a/scripts/tests/api_compare/filter-list-gateway b/scripts/tests/api_compare/filter-list-gateway index 82ffc606c92c..4c68a0556b47 100644 --- a/scripts/tests/api_compare/filter-list-gateway +++ b/scripts/tests/api_compare/filter-list-gateway @@ -8,6 +8,7 @@ !Filecoin.ChainSetHead !Filecoin.ChainStatObj !Filecoin.ChainTipSetWeight +!Filecoin.ChainValidateIndex !Filecoin.EthGetBlockReceiptsLimited !Filecoin.EthGetTransactionByHashLimited !Filecoin.F3 diff --git a/scripts/tests/api_compare/filter-list-offline b/scripts/tests/api_compare/filter-list-offline index 6546cb5d6b72..eb77cf888636 100644 --- a/scripts/tests/api_compare/filter-list-offline +++ b/scripts/tests/api_compare/filter-list-offline @@ -1,5 +1,7 @@ # This list contains potentially broken methods (or tests) that are ignored. # They should be considered bugged, and not used until the root cause is resolved. +# Indexer disabled +!Filecoin.ChainValidateIndex !Filecoin.EthSyncing !eth_syncing !Filecoin.NetAddrsListen diff --git a/scripts/tests/harness.sh b/scripts/tests/harness.sh index b9639b6310a5..56f6d53a229a 100644 --- a/scripts/tests/harness.sh +++ b/scripts/tests/harness.sh @@ -4,6 +4,7 @@ # executed directly. export FOREST_CHAIN_INDEXER_ENABLED="1" +export FOREST_CHAIN_INDEXER_GC_RETENTION_EPOCHS="2880" # Path to the `forest` daemon binary. Note this is intentionally NOT named # `FOREST_PATH`: that environment variable is honored by Forest itself to diff --git a/src/beacon/drand.rs b/src/beacon/drand.rs index 61ebddbca410..9761c90891ec 100644 --- a/src/beacon/drand.rs +++ b/src/beacon/drand.rs @@ -23,7 +23,6 @@ use crate::utils::net::global_http_client; use ambassador::{Delegate, delegatable_trait}; use backon::{ExponentialBuilder, Retryable}; use bls_signatures::Serialize as _; -use nonzero_ext::nonzero; use serde::{Deserialize as SerdeDeserialize, Serialize as SerdeSerialize}; use tracing::debug; use url::Url; diff --git a/src/chain/store/indexer.rs b/src/chain/store/indexer.rs new file mode 100644 index 000000000000..7e181318a9d4 --- /dev/null +++ b/src/chain/store/indexer.rs @@ -0,0 +1,890 @@ +// Copyright 2019-2026 ChainSafe Systems +// SPDX-License-Identifier: Apache-2.0, MIT + +mod ddls; +pub use ddls::{DDLS, PreparedStatements}; +mod events; +pub use events::IndexerEventFilter; +#[cfg(test)] +mod tests; + +use crate::{ + blocks::Tipset, + chain::{HeadChanges, index::ResolveNullTipset}, + message::SignedMessage, + prelude::*, + rpc::{ + chain::types::ChainIndexValidation, + eth::{eth_tx_from_signed_eth_message, types::EthHash}, + }, + shim::{ + ActorID, + address::Address, + clock::{ChainEpoch, EPOCHS_IN_DAY}, + }, + state_manager::{ExecutedMessage, StateManager}, + utils::sqlite, +}; +use ahash::HashMap; +use sqlx::Row as _; +use std::{ + ops::DerefMut as _, + time::{Duration, Instant}, +}; + +struct IndexedTipsetData { + pub indexed_messages_count: u64, + pub indexed_events_count: u64, + pub indexed_event_entries_count: u64, +} + +#[derive(Debug, smart_default::SmartDefault)] +pub struct SqliteIndexerOptions { + pub gc_retention_epochs: Option, +} + +impl SqliteIndexerOptions { + pub fn validate(&self) -> anyhow::Result<()> { + if let Some(gc_retention_epochs) = self.gc_retention_epochs { + anyhow::ensure!( + gc_retention_epochs >= EPOCHS_IN_DAY, + "gc retention epochs must be unset or no less than {EPOCHS_IN_DAY}" + ); + } + Ok(()) + } + + pub fn with_gc_retention_epochs(mut self, gc_retention_epochs: Option) -> Self { + self.gc_retention_epochs = gc_retention_epochs; + self + } +} + +pub struct SqliteIndexer { + lock: tokio::sync::Mutex<()>, + options: SqliteIndexerOptions, + sm: StateManager, + db: sqlx::SqlitePool, + stmts: PreparedStatements, +} + +impl SqliteIndexer { + pub async fn new( + db: sqlx::SqlitePool, + sm: StateManager, + options: SqliteIndexerOptions, + ) -> anyhow::Result { + options.validate()?; + sqlite::init_db( + &db, + "chain index", + DDLS.iter().cloned().map(sqlx::query), + vec![], + ) + .await?; + let stmts = PreparedStatements::default(); + Ok(Self { + lock: Default::default(), + options, + sm, + db, + stmts, + }) + } + + /// Acquires write lock. Note that `WAL` (Write-Ahead Logging) mode is enabled to allow + /// concurrent reads to occur while a single write transaction is active + pub async fn acquire_write_lock(&self) -> tokio::sync::MutexGuard<'_, ()> { + self.lock.lock().await + } + + pub fn db(&self) -> &sqlx::SqlitePool { + &self.db + } + + pub fn options(&self) -> &SqliteIndexerOptions { + &self.options + } + + fn actor_to_delegated_address(&self, actor_id: ActorID, ts: &Tipset) -> Address { + let id_addr = Address::new_id(actor_id); + match self.sm.get_required_actor(&id_addr, *ts.parent_state()) { + Ok(actor) => actor + .delegated_address + .map(Address::from) + .unwrap_or(id_addr), + Err(_) => id_addr, + } + } + + pub async fn index_loop( + &self, + mut head_changes_rx: tokio::sync::broadcast::Receiver, + ) -> ! { + loop { + if let Ok(HeadChanges { reverts, applies }) = head_changes_rx.recv().await { + let _lock = self.acquire_write_lock().await; + for ts in reverts { + if let Err(e) = self.revert_tipset(&ts).await { + tracing::warn!( + "failed to revert new head@{}({}): {e}", + ts.epoch(), + ts.key() + ); + } + } + for ts in applies { + if let Err(e) = self.index_tipset(&ts).await { + tracing::warn!( + "failed to index new head@{}({}): {e}", + ts.epoch(), + ts.key() + ); + } + } + } + } + } + + pub async fn gc_loop(&self) -> ! { + let mut ticker = tokio::time::interval(Duration::from_hours(4)); + loop { + ticker.tick().await; + self.gc().await; + } + } + + async fn gc(&self) { + let Some(gc_retention_epochs) = self.options.gc_retention_epochs else { + return; + }; + let _lock = self.acquire_write_lock().await; + tracing::info!("starting index gc"); + let head = self.sm.chain_store().heaviest_tipset(); + let removal_epoch = head.epoch() - gc_retention_epochs - 10; // 10 is for some grace period + if removal_epoch <= 0 { + tracing::info!("no tipsets to gc"); + return; + } + + tracing::info!( + "gc'ing all (reverted and non-reverted) tipsets before epoch {removal_epoch}" + ); + match sqlx::query(self.stmts.remove_tipsets_before_height) + .bind(removal_epoch) + .execute(&self.db) + .await + { + Ok(r) => { + tracing::info!( + "gc'd {} entries before epoch {removal_epoch}", + r.rows_affected() + ); + } + Err(e) => { + tracing::error!( + "failed to remove reverted tipsets before height {removal_epoch}: {e}" + ); + return; + } + } + + // ------------------------------------------------------------------------------------------------- + // Also GC eth hashes + + // Convert `gc_retention_epochs` to number of days + let gc_retention_days = gc_retention_epochs / EPOCHS_IN_DAY; + if gc_retention_days < 1 { + tracing::info!("skipping gc of eth hashes as retention days is less than 1"); + return; + } + + tracing::info!("gc'ing eth hashes older than {gc_retention_days} days"); + match sqlx::query(self.stmts.remove_eth_hashes_older_than) + .bind(format!("-{gc_retention_days} day")) + .execute(&self.db) + .await + { + Ok(r) => { + tracing::info!( + "gc'd {} eth hashes older than {gc_retention_days} days", + r.rows_affected() + ); + } + Err(e) => { + tracing::error!("failed to gc eth hashes older than {gc_retention_days} days: {e}"); + } + } + } + + pub async fn populate_after_snapshot_import(&self) -> anyhow::Result<()> { + let _lock = self.acquire_write_lock().await; + let start = Instant::now(); + let head = self.sm.chain_store().heaviest_tipset(); + tracing::info!( + "starting to populate chainindex at head epoch {}", + head.epoch() + ); + let mut tx = self.db.begin().await?; + let mut total_indexed = 0; + for ts in head.chain(self.sm.db()) { + if let Err(e) = self + .index_tipset_and_parent_events_with_tx(&mut tx, &ts) + .await + { + tracing::info!( + "stopping chainindex population at epoch {}: {e}", + ts.epoch() + ); + break; + } + total_indexed += 1; + } + tx.commit().await?; + tracing::info!( + "successfully populated chain index with {total_indexed} tipsets, took {}", + humantime::format_duration(start.elapsed()) + ); + Ok(()) + } + + pub async fn validate_index( + &self, + epoch: ChainEpoch, + backfill: bool, + ) -> anyhow::Result { + let _lock = if backfill { + Some(self.acquire_write_lock().await) + } else { + None + }; + let head = self.sm.chain_store().heaviest_tipset(); + anyhow::ensure!( + epoch < head.epoch(), + "cannot validate index at epoch {epoch}, can only validate at an epoch less than chain head epoch {}", + head.epoch() + ); + let ts = self + .sm + .chain_index() + .load_required_tipset_by_height(epoch, head, ResolveNullTipset::TakeOlder) + .await?; + let is_index_empty: bool = sqlx::query(self.stmts.is_index_empty) + .fetch_one(&self.db) + .await? + .get(0); + + // Canonical chain has a null round at the epoch -> return if index is empty otherwise validate that index also + // has a null round at this epoch i.e. it does not have anything indexed at all for this epoch + if ts.epoch() != epoch { + if is_index_empty { + return Ok(ChainIndexValidation { + height: epoch, + is_null_round: true, + ..Default::default() + }); + } + // validate the db has a hole here and error if not, we don't attempt to repair because something must be very wrong for this to fail + return self.validate_is_null_round(epoch).await; + } + + // if the index is empty -> short-circuit and simply backfill if applicable + if is_index_empty { + check_backfill_required(epoch, backfill)?; + return self.backfill_missing_tipset(&ts).await; + } + + // see if the tipset at this epoch is already indexed or if we need to backfill + if let Some((reverted_count, non_reverted_count)) = + self.get_tipset_counts_at_height(epoch).await? + { + if reverted_count == 0 && non_reverted_count == 0 { + check_backfill_required(epoch, backfill)?; + return self.backfill_missing_tipset(&ts).await; + } else if reverted_count > 0 && non_reverted_count == 0 { + anyhow::bail!("index corruption: height {epoch} only has reverted tipsets"); + } else if non_reverted_count > 1 { + anyhow::bail!("index corruption: height {epoch} has multiple non-reverted tipsets"); + } + } else { + check_backfill_required(epoch, backfill)?; + return self.backfill_missing_tipset(&ts).await; + } + + // fetch the non-reverted tipset at this epoch + let indexed_tsk_cid_bytes: Vec = + sqlx::query(self.stmts.get_non_reverted_tipset_at_height) + .bind(epoch) + .fetch_one(&self.db) + .await? + .get(0); + let indexed_tsk_cid = Cid::read_bytes(indexed_tsk_cid_bytes.as_slice())?; + let expected_tsk_cid = ts.key().cid()?; + anyhow::ensure!( + indexed_tsk_cid == expected_tsk_cid, + "index corruption: indexed tipset at height {epoch} has key {indexed_tsk_cid}, but canonical chain has {expected_tsk_cid}", + ); + let ( + IndexedTipsetData { + indexed_messages_count, + indexed_events_count, + indexed_event_entries_count, + }, + backfilled, + ) = match self.get_and_verify_indexed_data(&ts).await { + Ok(r) => (r, false), + Err(_) if backfill => { + self.backfill_missing_tipset(&ts).await?; + (self.get_and_verify_indexed_data(&ts).await?, true) + } + Err(e) => anyhow::bail!( + "failed to validate indexed data for tipset at height {epoch}(backfill disabled): {e:#?}" + ), + }; + Ok(ChainIndexValidation { + tip_set_key: ts.key().clone().into(), + height: ts.epoch(), + backfilled, + indexed_messages_count, + indexed_events_count, + indexed_event_entries_count, + is_null_round: false, + }) + } + + async fn validate_is_null_round( + &self, + epoch: ChainEpoch, + ) -> anyhow::Result { + // make sure we do not have tipset(reverted or non-reverted) indexed at this epoch + let is_null_round: bool = sqlx::query(self.stmts.has_null_round_at_height) + .bind(epoch) + .fetch_one(&self.db) + .await? + .get(0); + anyhow::ensure!( + is_null_round, + "index corruption: height {epoch} should be a null round but is not" + ); + Ok(ChainIndexValidation { + height: epoch, + is_null_round: true, + ..Default::default() + }) + } + + async fn backfill_missing_tipset(&self, ts: &Tipset) -> anyhow::Result { + let execution_ts = self.get_next_tipset(ts).await?; + let mut tx = self.db.begin().await?; + self.index_tipset_and_parent_events_with_tx(&mut tx, &execution_ts) + .await?; + tx.commit().await?; + let IndexedTipsetData { + indexed_messages_count, + indexed_events_count, + indexed_event_entries_count, + } = self.get_indexed_tipset_data(ts).await?; + Ok(ChainIndexValidation { + tip_set_key: ts.key().clone().into(), + height: ts.epoch(), + backfilled: true, + indexed_messages_count, + indexed_events_count, + indexed_event_entries_count, + is_null_round: false, + }) + } + + async fn get_next_tipset(&self, ts: &Tipset) -> anyhow::Result { + let child = self + .sm + .chain_index() + .load_required_tipset_by_height( + ts.epoch() + 1, + self.sm.chain_store().heaviest_tipset(), + ResolveNullTipset::TakeNewer, + ) + .await?; + anyhow::ensure!( + child.parents() == ts.key(), + "chain forked at height {}; please retry your request; err: chain forked", + ts.epoch() + ); + Ok(child) + } + + async fn get_and_verify_indexed_data(&self, ts: &Tipset) -> anyhow::Result { + let indexed_tipset_data = self.get_indexed_tipset_data(ts).await?; + self.verify_indexed_data(ts, &indexed_tipset_data).await?; + Ok(indexed_tipset_data) + } + + /// verifies that the indexed data for a tipset is correct by comparing the number of messages and events + /// in the chain store to the number of messages and events indexed. + async fn verify_indexed_data( + &self, + ts: &Tipset, + indexed_tipset_data: &IndexedTipsetData, + ) -> anyhow::Result<()> { + let tsk_cid = ts.key().cid()?; + let tsk_cid_bytes = tsk_cid.to_bytes(); + let execution_ts = self.get_next_tipset(ts).await?; + + // given that `ts` is on the canonical chain and `execution_ts` is the next tipset in the chain + // `ts` can not have reverted events + let has_reverted_events_in_tipset: bool = + sqlx::query(self.stmts.has_reverted_events_in_tipset) + .bind(&tsk_cid_bytes) + .fetch_one(&self.db) + .await? + .get(0); + anyhow::ensure!( + !has_reverted_events_in_tipset, + "index corruption: reverted events found for an executed tipset {tsk_cid} at height {}", + ts.epoch() + ); + let executed_messages = Arc::unwrap_or_clone( + self.sm + .load_executed_tipset_with_receipt(ts, &execution_ts) + .await? + .executed_messages, + ); + anyhow::ensure!( + executed_messages.len() as u64 == indexed_tipset_data.indexed_messages_count, + "message count mismatch for height {}: chainstore has {}, index has {}", + ts.epoch(), + executed_messages.len(), + indexed_tipset_data.indexed_messages_count + ); + let mut events_count = 0; + let mut event_entries_count = 0; + for ExecutedMessage { events, .. } in &executed_messages { + if let Some(events) = events { + events_count += events.len(); + for event in events { + event_entries_count += event.entries().len(); + } + } + } + anyhow::ensure!( + events_count as u64 == indexed_tipset_data.indexed_events_count, + "event count mismatch for height {}: chainstore has {events_count}, index has {}", + ts.epoch(), + indexed_tipset_data.indexed_events_count + ); + anyhow::ensure!( + event_entries_count as u64 == indexed_tipset_data.indexed_event_entries_count, + "event entries count mismatch for height {}: chainstore has {event_entries_count}, index has {}", + ts.epoch(), + indexed_tipset_data.indexed_event_entries_count + ); + + // compare the events AMT root between the indexed events and the events in the chain state + for ExecutedMessage { + message, receipt, .. + } in executed_messages + { + let msg_cid = message.cid(); + let indexed_events_root = self.amt_root_for_events(&tsk_cid, &msg_cid).await?; + match (indexed_events_root, receipt.events_root()) { + (Some(a), Some(b)) => { + anyhow::ensure!( + a == b, + "index corruption: events AMT root mismatch for message {msg_cid} at height {}. Index root: {a}, Receipt root: {b}", + ts.epoch() + ); + } + (None, None) => continue, + (Some(_), None) => { + anyhow::bail!( + "index corruption: events found in index for message {msg_cid} at height {}, but message receipt has no events root", + ts.epoch() + ) + } + (None, Some(b)) => { + anyhow::bail!( + "index corruption: no events found in index for message {msg_cid} at height {}, but message receipt has events root {b}", + ts.epoch() + ) + } + } + } + + Ok(()) + } + + async fn get_indexed_tipset_data(&self, ts: &Tipset) -> anyhow::Result { + let tsk_cid_bytes = ts.key().cid()?.to_bytes(); + let indexed_messages_count = sqlx::query(self.stmts.get_non_reverted_tipset_message_count) + .bind(&tsk_cid_bytes) + .fetch_one(&self.db) + .await? + .get(0); + let indexed_events_count = sqlx::query(self.stmts.get_non_reverted_tipset_event_count) + .bind(&tsk_cid_bytes) + .fetch_one(&self.db) + .await? + .get(0); + let indexed_event_entries_count = + sqlx::query(self.stmts.get_non_reverted_tipset_event_entries_count) + .bind(&tsk_cid_bytes) + .fetch_one(&self.db) + .await? + .get(0); + Ok(IndexedTipsetData { + indexed_messages_count, + indexed_events_count, + indexed_event_entries_count, + }) + } + + async fn get_tipset_counts_at_height( + &self, + epoch: ChainEpoch, + ) -> anyhow::Result> { + let row = sqlx::query(self.stmts.count_tipsets_at_height) + .bind(epoch) + .fetch_optional(&self.db) + .await?; + Ok(row.map(|r| (r.get(0), r.get(1)))) + } + + async fn amt_root_for_events( + &self, + tsk_cid: &Cid, + msg_cid: &Cid, + ) -> anyhow::Result> { + use crate::state_manager::EVENTS_AMT_BITWIDTH; + use fil_actors_shared::fvm_ipld_amt::Amt; + use fvm_ipld_blockstore::MemoryBlockstore; + use fvm_shared4::event::{ActorEvent, Entry, Flags, StampedEvent}; + + let mut tx = self.db.begin().await?; + let rows = sqlx::query(self.stmts.get_event_id_and_emitter_id) + .bind(tsk_cid.to_bytes()) + .bind(msg_cid.to_bytes()) + .fetch_all(tx.deref_mut()) + .await?; + if rows.is_empty() { + Ok(None) + } else { + let mut events = Amt::new_with_bit_width(MemoryBlockstore::new(), EVENTS_AMT_BITWIDTH); + for (i, row) in rows.iter().enumerate() { + let event_id: i64 = row.get(0); + let actor_id: i64 = row.get(1); + let mut event = StampedEvent { + emitter: actor_id as _, + event: ActorEvent { entries: vec![] }, + }; + let rows2 = sqlx::query(self.stmts.get_event_entries) + .bind(event_id) + .fetch_all(tx.deref_mut()) + .await?; + for row2 in rows2 { + let flags: Vec = row2.get(0); + let flags = *flags.first().with_context(|| { + format!("index corruption: empty flags blob for event {event_id}") + })?; + let key: String = row2.get(1); + let codec: i64 = row2.get(2); + let value: Vec = row2.get(3); + event.event.entries.push(Entry { + flags: Flags::from_bits_retain(u64::from(flags)), + key, + codec: codec as _, + value, + }); + } + events.set(i as _, event)?; + } + + Ok(Some(events.flush()?)) + } + } + + async fn revert_tipset(&self, ts: &Tipset) -> anyhow::Result<()> { + tracing::debug!("reverting tipset@{}[{}]", ts.epoch(), ts.key().terse()); + let tsk_cid_bytes = ts.key().cid()?.to_bytes(); + // Because of deferred execution in Filecoin, events at tipset T are reverted when a tipset T+1 is reverted. + // However, the tipet `T` itself is not reverted. + let pts = Tipset::load_required(self.sm.db(), ts.parents())?; + let events_tsk_cid_bytes = pts.key().cid()?.to_bytes(); + let mut tx = self.db.begin().await?; + sqlx::query(self.stmts.update_tipset_to_reverted) + .bind(&tsk_cid_bytes) + .execute(tx.deref_mut()) + .await?; + // events are indexed against the message inclusion tipset, not the message execution tipset. + // So we need to revert the events for the message inclusion tipset + sqlx::query(self.stmts.update_events_to_reverted) + .bind(&events_tsk_cid_bytes) + .execute(tx.deref_mut()) + .await?; + tx.commit().await?; + Ok(()) + } + + async fn index_tipset(&self, ts: &Tipset) -> anyhow::Result<()> { + tracing::debug!("indexing tipset@{}[{}]", ts.epoch(), ts.key().terse()); + let mut tx = self.db.begin().await?; + self.index_tipset_and_parent_events_with_tx(&mut tx, ts) + .await?; + tx.commit().await?; + Ok(()) + } + + async fn index_tipset_with_tx<'a>( + &self, + tx: &mut sqlx::SqliteTransaction<'a>, + ts: &Tipset, + ) -> anyhow::Result<()> { + let tsk_cid_bytes = ts.key().cid()?.to_bytes(); + if self + .restore_tipset_if_exists_with_tx(tx, &tsk_cid_bytes) + .await? + { + Ok(()) + } else { + let msgs = self + .sm + .chain_store() + .messages_for_tipset(ts) + .map_err(|e| anyhow::anyhow!("failed to get messages for tipset: {e}"))?; + if msgs.is_empty() { + // If there are no messages, just insert the tipset and return + sqlx::query(self.stmts.insert_tipset_message) + .bind(&tsk_cid_bytes) + .bind(ts.epoch()) + .bind(0) + .bind(None::<&[u8]>) + .bind(-1) + .execute(tx.deref_mut()) + .await + .map_err(|e| anyhow::anyhow!("failed to insert empty tipset: {e}"))?; + } else { + for (i, msg) in msgs.iter().enumerate() { + sqlx::query(self.stmts.insert_tipset_message) + .bind(&tsk_cid_bytes) + .bind(ts.epoch()) + .bind(0) + .bind(msg.cid().to_bytes()) + .bind(i as i64) + .execute(tx.deref_mut()) + .await + .map_err(|e| anyhow::anyhow!("failed to insert tipset message: {e}"))?; + } + + for block in ts.block_headers() { + let (_, smsgs) = crate::chain::block_messages(self.sm.db(), block) + .map_err(|e| anyhow::anyhow!("failed to get messages for block: {e}"))?; + for smsg in smsgs.into_iter().filter(SignedMessage::is_delegated) { + self.index_signed_message_with_tx(tx, &smsg) + .await + .map_err(|e| anyhow::anyhow!("failed to index eth tx hash: {e}"))?; + } + } + } + Ok(()) + } + } + + async fn index_tipset_and_parent_events_with_tx<'a>( + &self, + tx: &mut sqlx::SqliteTransaction<'a>, + ts: &Tipset, + ) -> anyhow::Result<()> { + self.index_tipset_with_tx(tx, ts) + .await + .map_err(|e| anyhow::anyhow!("failed to index tipset: {e}"))?; + if ts.epoch() == 0 { + // Skip parent if ts is genesis + return Ok(()); + } + let pts = Tipset::load_required(self.sm.db(), ts.parents())?; + // Index the parent tipset if it doesn't exist yet. + // This is necessary to properly index events produced by executing + // messages included in the parent tipset by the current tipset (deferred execution). + self.index_tipset_with_tx(tx, &pts) + .await + .map_err(|e| anyhow::anyhow!("failed to index parent tipset: {e}"))?; + // Now Index events + self.index_events_with_tx(tx, &pts, ts) + .await + .map_err(|e| anyhow::anyhow!("failed to index events: {e}")) + } + + async fn index_events_with_tx<'a>( + &self, + tx: &mut sqlx::SqliteTransaction<'a>, + msg_ts: &Tipset, + execution_ts: &Tipset, + ) -> anyhow::Result<()> { + // check if we have an event indexed for any message in the `msg_ts` tipset -> if so, there's nothig to do here + // this makes event inserts idempotent + let msg_tsk_cid_bytes = msg_ts.key().cid()?.to_bytes(); + + // if we've already indexed events for this tipset, mark them as unreverted and return + let rows_affected = sqlx::query(self.stmts.update_events_to_non_reverted) + .bind(&msg_tsk_cid_bytes) + .execute(tx.deref_mut()) + .await + .map_err(|e| anyhow::anyhow!("failed to unrevert events for tipset: {e}"))? + .rows_affected(); + if rows_affected > 0 { + tracing::debug!( + "unreverted {rows_affected} events for tipset {}", + msg_ts.key() + ); + return Ok(()); + } + let executed_messages = Arc::unwrap_or_clone( + self.sm + .load_executed_tipset_with_receipt(msg_ts, execution_ts) + .await + .context("failed to load executed message")? + .executed_messages, + ); + // let executed_messages = self + // .load_executed_messages(msg_ts.shallow_clone(), execution_ts.shallow_clone()) + // .await + // .map_err(|e| anyhow::anyhow!("failed to load executed message: {e}"))?; + let mut event_count = 0; + let mut address_lookups = HashMap::default(); + for ExecutedMessage { + message, events, .. + } in executed_messages + { + let msg_cid_bytes = message.cid().to_bytes(); + + // read message id for this message cid and tipset key cid + let message_id: i64 = sqlx::query(self.stmts.get_msg_id_for_msg_cid_and_tipset) + .bind(&msg_tsk_cid_bytes) + .bind(&msg_cid_bytes) + .fetch_optional(tx.deref_mut()) + .await? + .with_context(|| { + format!( + "message id not found for message cid {} and tipset key {}", + message.cid(), + msg_ts.key() + ) + })? + .get(0); + + // Insert events for this message + if let Some(events) = events { + for event in events { + let emitter = event.emitter(); + let addr = if let Some(addr) = address_lookups.get(&emitter) { + *addr + } else { + let addr = self.actor_to_delegated_address(emitter, execution_ts); + address_lookups.insert(emitter, addr); + addr + }; + + let robust_addr_bytes = if addr.is_delegated() { + addr.to_bytes() + } else { + vec![] + }; + + // Insert event into events table + let event_id = sqlx::query(self.stmts.insert_event) + .bind(message_id) + .bind(event_count) + .bind(emitter as i64) + .bind(robust_addr_bytes) + .bind(0) + .execute(tx.deref_mut()) + .await? + .last_insert_rowid(); + + for entry in event.entries() { + let (flags, key, codec, value) = entry.into_parts(); + sqlx::query(self.stmts.insert_event_entry) + .bind(event_id) + .bind(is_indexed_flag(flags)) + .bind([flags as u8].as_slice()) + .bind(key) + .bind(codec as i64) + .bind(&value) + .execute(tx.deref_mut()) + .await?; + } + + event_count += 1; + } + } + } + Ok(()) + } + + async fn restore_tipset_if_exists_with_tx<'a>( + &self, + tx: &mut sqlx::SqliteTransaction<'a>, + tsk_cid_bytes: &[u8], + ) -> anyhow::Result { + match sqlx::query(self.stmts.has_tipset) + .bind(tsk_cid_bytes) + .fetch_one(tx.deref_mut()) + .await + .map(|r| r.get(0)) + { + Ok(exists) => { + if exists { + sqlx::query(self.stmts.update_tipset_to_non_reverted) + .bind(tsk_cid_bytes) + .execute(tx.deref_mut()) + .await + .map_err(|e| anyhow::anyhow!("failed to restore tipset: {e}"))?; + } + Ok(exists) + } + Err(e) => anyhow::bail!("failed to check if tipset exists: {e}"), + } + } + + async fn index_signed_message_with_tx<'a>( + &self, + tx: &mut sqlx::SqliteTransaction<'a>, + smsg: &SignedMessage, + ) -> anyhow::Result<()> { + let (_, eth_tx) = eth_tx_from_signed_eth_message(smsg, self.sm.chain_config().eth_chain_id) + .map_err(|e| anyhow::anyhow!("failed to convert filecoin message to eth tx: {e}"))?; + let tx_hash = EthHash( + eth_tx + .eth_hash() + .map_err(|e| anyhow::anyhow!("failed to hash transaction: {e}"))?, + ); + self.index_eth_tx_hash_with_tx(tx, tx_hash, smsg.cid()) + .await + } + + async fn index_eth_tx_hash_with_tx<'a>( + &self, + tx: &mut sqlx::SqliteTransaction<'a>, + tx_hash: EthHash, + msg_cid: Cid, + ) -> anyhow::Result<()> { + _ = sqlx::query(self.stmts.insert_eth_tx_hash) + .bind(tx_hash.to_string()) + .bind(msg_cid.to_string()) + .execute(tx.deref_mut()) + .await?; + Ok(()) + } +} + +fn is_indexed_flag(flag: u64) -> bool { + use crate::shim::fvm_shared_latest::event::Flags; + flag & (Flags::FLAG_INDEXED_KEY.bits() | Flags::FLAG_INDEXED_VALUE.bits()) > 0 +} + +fn check_backfill_required(epoch: ChainEpoch, backfill: bool) -> anyhow::Result<()> { + anyhow::ensure!( + backfill, + "missing tipset at height {epoch} in the chain index, set backfill flag to true to fix" + ); + Ok(()) +} diff --git a/src/chain/store/indexer/ddls.rs b/src/chain/store/indexer/ddls.rs new file mode 100644 index 000000000000..704fe5750b77 --- /dev/null +++ b/src/chain/store/indexer/ddls.rs @@ -0,0 +1,126 @@ +// Copyright 2019-2026 ChainSafe Systems +// SPDX-License-Identifier: Apache-2.0, MIT + +pub static DDLS: [&str; 10] = [ + r#"CREATE TABLE IF NOT EXISTS tipset_message ( + id INTEGER PRIMARY KEY, + tipset_key_cid BLOB NOT NULL, + height INTEGER NOT NULL, + reverted INTEGER NOT NULL, + message_cid BLOB, + message_index INTEGER, + UNIQUE (tipset_key_cid, message_cid) + )"#, + r#"CREATE TABLE IF NOT EXISTS eth_tx_hash ( + tx_hash TEXT PRIMARY KEY, + message_cid BLOB NOT NULL, + inserted_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + )"#, + r#"CREATE TABLE IF NOT EXISTS event ( + id INTEGER PRIMARY KEY, + message_id INTEGER NOT NULL, + event_index INTEGER NOT NULL, + emitter_id INTEGER NOT NULL, + emitter_addr BLOB, + reverted INTEGER NOT NULL, + FOREIGN KEY (message_id) REFERENCES tipset_message(id) ON DELETE CASCADE, + UNIQUE (message_id, event_index) + )"#, + r#"CREATE TABLE IF NOT EXISTS event_entry ( + event_id INTEGER NOT NULL, + indexed INTEGER NOT NULL, + flags BLOB NOT NULL, + key TEXT NOT NULL, + codec INTEGER, + value BLOB NOT NULL, + FOREIGN KEY (event_id) REFERENCES event(id) ON DELETE CASCADE + )"#, + "CREATE INDEX IF NOT EXISTS insertion_time_index ON eth_tx_hash (inserted_at)", + "CREATE INDEX IF NOT EXISTS idx_message_cid ON tipset_message (message_cid)", + "CREATE INDEX IF NOT EXISTS idx_tipset_key_cid ON tipset_message (tipset_key_cid)", + "CREATE INDEX IF NOT EXISTS idx_event_message_id ON event (message_id)", + "CREATE INDEX IF NOT EXISTS idx_height ON tipset_message (height)", + "CREATE INDEX IF NOT EXISTS event_entry_event_id ON event_entry(event_id)", +]; + +pub struct PreparedStatements { + pub has_tipset: &'static str, + pub is_index_empty: &'static str, + pub has_null_round_at_height: &'static str, + pub get_non_reverted_tipset_at_height: &'static str, + pub count_tipsets_at_height: &'static str, + pub get_non_reverted_tipset_message_count: &'static str, + pub get_non_reverted_tipset_event_count: &'static str, + pub has_reverted_events_in_tipset: &'static str, + pub get_non_reverted_tipset_event_entries_count: &'static str, + pub insert_eth_tx_hash: &'static str, + pub insert_tipset_message: &'static str, + pub update_tipset_to_non_reverted: &'static str, + pub update_tipset_to_reverted: &'static str, + pub update_events_to_non_reverted: &'static str, + pub update_events_to_reverted: &'static str, + pub get_msg_id_for_msg_cid_and_tipset: &'static str, + pub insert_event: &'static str, + pub insert_event_entry: &'static str, + pub remove_tipsets_before_height: &'static str, + pub remove_eth_hashes_older_than: &'static str, + pub get_event_entries: &'static str, + pub get_event_id_and_emitter_id: &'static str, +} + +impl Default for PreparedStatements { + fn default() -> Self { + let has_tipset = "SELECT EXISTS(SELECT 1 FROM tipset_message WHERE tipset_key_cid = ?)"; + let is_index_empty = "SELECT NOT EXISTS(SELECT 1 FROM tipset_message LIMIT 1)"; + let has_null_round_at_height = + "SELECT NOT EXISTS(SELECT 1 FROM tipset_message WHERE height = ?)"; + let get_non_reverted_tipset_at_height = + "SELECT tipset_key_cid FROM tipset_message WHERE height = ? AND reverted = 0 LIMIT 1"; + let count_tipsets_at_height = "SELECT COUNT(CASE WHEN reverted = 1 THEN 1 END) AS reverted_count, COUNT(CASE WHEN reverted = 0 THEN 1 END) AS non_reverted_count FROM (SELECT tipset_key_cid, MAX(reverted) AS reverted FROM tipset_message WHERE height = ? GROUP BY tipset_key_cid) AS unique_tipsets"; + let get_non_reverted_tipset_message_count = "SELECT COUNT(*) FROM tipset_message WHERE tipset_key_cid = ? AND reverted = 0 AND message_cid IS NOT NULL"; + let get_non_reverted_tipset_event_count = "SELECT COUNT(*) FROM event WHERE reverted = 0 AND message_id IN (SELECT id FROM tipset_message WHERE tipset_key_cid = ? AND reverted = 0)"; + let has_reverted_events_in_tipset = "SELECT EXISTS(SELECT 1 FROM event WHERE reverted = 1 AND message_id IN (SELECT id FROM tipset_message WHERE tipset_key_cid = ?))"; + let get_non_reverted_tipset_event_entries_count = "SELECT COUNT(ee.event_id) AS entry_count FROM event_entry ee JOIN event e ON ee.event_id = e.id JOIN tipset_message tm ON e.message_id = tm.id WHERE tm.tipset_key_cid = ? AND tm.reverted = 0"; + let insert_eth_tx_hash = "INSERT INTO eth_tx_hash (tx_hash, message_cid) VALUES (?, ?) ON CONFLICT (tx_hash) DO UPDATE SET inserted_at = CURRENT_TIMESTAMP"; + let insert_tipset_message = "INSERT INTO tipset_message (tipset_key_cid, height, reverted, message_cid, message_index) VALUES (?, ?, ?, ?, ?) ON CONFLICT (tipset_key_cid, message_cid) DO UPDATE SET reverted = 0"; + let update_tipset_to_non_reverted = + "UPDATE tipset_message SET reverted = 0 WHERE tipset_key_cid = ?"; + let update_tipset_to_reverted = + "UPDATE tipset_message SET reverted = 1 WHERE tipset_key_cid = ?"; + let update_events_to_non_reverted = "UPDATE event SET reverted = 0 WHERE message_id IN (SELECT id FROM tipset_message WHERE tipset_key_cid = ?)"; + let update_events_to_reverted = "UPDATE event SET reverted = 1 WHERE message_id IN (SELECT id FROM tipset_message WHERE tipset_key_cid = ?)"; + let get_msg_id_for_msg_cid_and_tipset = "SELECT id FROM tipset_message WHERE tipset_key_cid = ? AND message_cid = ? AND reverted = 0"; + let insert_event = "INSERT INTO event (message_id, event_index, emitter_id, emitter_addr, reverted) VALUES (?, ?, ?, ?, ?)"; + let insert_event_entry = "INSERT INTO event_entry (event_id, indexed, flags, key, codec, value) VALUES (?, ?, ?, ?, ?, ?)"; + let remove_tipsets_before_height = "DELETE FROM tipset_message WHERE height < ?"; + let remove_eth_hashes_older_than = + "DELETE FROM eth_tx_hash WHERE inserted_at < datetime('now', ?)"; + let get_event_entries = "SELECT flags, key, codec, value FROM event_entry WHERE event_id=? ORDER BY _rowid_ ASC"; + let get_event_id_and_emitter_id = "SELECT e.id, e.emitter_id FROM event e JOIN tipset_message tm ON e.message_id = tm.id WHERE tm.tipset_key_cid = ? AND tm.message_cid = ? ORDER BY e.event_index ASC"; + + Self { + has_tipset, + is_index_empty, + has_null_round_at_height, + get_non_reverted_tipset_at_height, + count_tipsets_at_height, + get_non_reverted_tipset_message_count, + get_non_reverted_tipset_event_count, + has_reverted_events_in_tipset, + get_non_reverted_tipset_event_entries_count, + insert_eth_tx_hash, + insert_tipset_message, + update_tipset_to_non_reverted, + update_tipset_to_reverted, + update_events_to_non_reverted, + update_events_to_reverted, + get_msg_id_for_msg_cid_and_tipset, + insert_event, + insert_event_entry, + remove_tipsets_before_height, + remove_eth_hashes_older_than, + get_event_entries, + get_event_id_and_emitter_id, + } + } +} diff --git a/src/chain/store/indexer/events.rs b/src/chain/store/indexer/events.rs new file mode 100644 index 000000000000..7da34a11147f --- /dev/null +++ b/src/chain/store/indexer/events.rs @@ -0,0 +1,285 @@ +// Copyright 2019-2026 ChainSafe Systems +// SPDX-License-Identifier: Apache-2.0, MIT + +use super::*; +use crate::blocks::TipsetKey; +use crate::prelude::*; +use crate::rpc::eth::filter::{SkipEvent, ensure_filter_cap}; +use crate::rpc::eth::{ + CollectedEvent, + filter::{ActorEventBlock, ParsedFilter, ParsedFilterTipsets}, +}; +use crate::rpc::types::EventEntry; +use crate::shim::address::Protocol; +use ahash::HashSet; +use futures::StreamExt as _; +use sqlx::{Arguments as _, FromRow}; +use std::borrow::Cow; + +#[derive(Debug, sqlx::FromRow)] +struct EventRow { + id: i64, + height: i64, + tipset_key_cid: Vec, + #[sqlx(try_from = "i64")] + emitter_id: u64, + emitter_addr: Vec, + #[sqlx(try_from = "i64")] + event_index: u64, + message_cid: Vec, + #[sqlx(try_from = "i64")] + message_index: u64, + reverted: bool, + flags: Vec, + key: String, + #[sqlx(try_from = "i64")] + codec: u64, + value: Vec, +} + +impl SqliteIndexer { + pub async fn get_events_for_filter( + &self, + filter: IndexerEventFilter, + max_filter_results: usize, + skip_event: SkipEvent, + ) -> anyhow::Result> { + let bs = self.sm.db(); + let mut qb = filter.to_query_builder()?; + let query = qb.build(); + let mut results = query.fetch(self.db()); + let mut current_id = -1; + let mut last_height = -1; + let mut tipsets_seen = 0; + let mut collected_events = vec![]; + let mut ce = None; + while let Some(row) = results.next().await { + let event = EventRow::from_row(&row?).inspect_err(|e| { + tracing::warn!("{e}"); + })?; + // The query returns all entries for all matching events; create a new CollectedEvent each time we see a new id. + if event.id != current_id { + if let Some(ce) = ce.take() { + collected_events.push(ce); + } + + if event.height != last_height { + tipsets_seen += 1; + last_height = event.height; + } + + ensure_filter_cap(max_filter_results, tipsets_seen, collected_events.len() + 1)?; + + current_id = event.id; + let tsk_cid = Cid::read_bytes(event.tipset_key_cid.as_slice())?; + let emitter_addr = if !event.emitter_addr.is_empty() { + Address::from_bytes(event.emitter_addr.as_slice())? + } else if matches!(skip_event, SkipEvent::OnUnresolvedAddress) { + continue; + } else { + Address::new_id(event.emitter_id) + }; + ce = Some(CollectedEvent { + event_idx: event.event_index, + reverted: event.reverted, + height: event.height, + msg_idx: event.message_index, + msg_cid: Cid::read_bytes(event.message_cid.as_slice())?, + tipset_key: TipsetKey::load(bs, &tsk_cid)?, + emitter_addr, + entries: vec![], + }); + } + + if let Some(ce) = ce.as_mut() { + ce.entries.push(EventEntry { + flags: event + .flags + .first() + .copied() + .context("failed to get flags")? + .into(), + key: event.key, + codec: event.codec, + value: event.value.into(), + }); + } + } + if let Some(ce) = ce.take() { + collected_events.push(ce); + } + Ok(collected_events) + } +} + +#[derive(Debug, Clone, Default)] +pub struct IndexerEventFilter { + pub min_height: ChainEpoch, + pub max_height: ChainEpoch, + pub tipset_cid: Option, + pub msg_cid: Option, + pub addresses: Vec
, + pub keys: HashMap>, +} + +impl IndexerEventFilter { + pub fn to_query_builder(&self) -> anyhow::Result> { + let arg_err = |e| anyhow::anyhow!("failed to push argument: {e}"); + + let mut clauses: Vec> = vec![]; + let mut joins = vec![]; + let mut args = sqlx::sqlite::SqliteArguments::default(); + if let Some(ts_cid) = self.tipset_cid { + clauses.push("tm.tipset_key_cid=?".into()); + args.add(ts_cid.to_bytes()).map_err(arg_err)?; + } else { + if self.min_height >= 0 && self.min_height == self.max_height { + clauses.push("tm.height=?".into()); + args.add(self.min_height).map_err(arg_err)?; + } else if self.min_height >= 0 && self.max_height >= 0 { + anyhow::ensure!( + self.max_height >= self.min_height, + "max_height should not be less that min_height" + ); + clauses.push("tm.height BETWEEN ? AND ?".into()); + args.add(self.min_height).map_err(arg_err)?; + args.add(self.max_height).map_err(arg_err)?; + } else if self.min_height >= 0 { + clauses.push("tm.height >= ?".into()); + args.add(self.min_height).map_err(arg_err)?; + } else if self.max_height >= 0 { + clauses.push("tm.height <= ?".into()); + args.add(self.max_height).map_err(arg_err)?; + } else { + anyhow::bail!("filter must specify either a tipset or a height range"); + } + // unless asking for a specific tipset, we never want to see reverted historical events + clauses.push("e.reverted=?".into()); + args.add(false).map_err(arg_err)?; + } + + if let Some(msg_cid) = self.msg_cid { + clauses.push("tm.message_cid=?".into()); + args.add(msg_cid.to_bytes()).map_err(arg_err)?; + } + + if !self.addresses.is_empty() { + let mut id_addresses = HashSet::default(); + let mut delegated_addresses = HashSet::default(); + for addr in self.addresses.iter() { + match addr.protocol() { + Protocol::ID => { + id_addresses.insert(addr.id()?); + } + Protocol::Delegated => { + delegated_addresses.insert(addr.to_bytes()); + } + p => { + anyhow::bail!( + "can only query events by ID or Delegated addresses; but request has {p} address" + ); + } + } + } + + if !id_addresses.is_empty() { + let placeholders = id_addresses.iter().map(|_| "?").join(","); + clauses.push(format!("e.emitter_id IN ({placeholders})").into()); + for id in id_addresses { + args.add(id as i64).map_err(arg_err)?; + } + } + + if !delegated_addresses.is_empty() { + let placeholders = delegated_addresses.iter().map(|_| "?").join(","); + clauses.push(format!("e.emitter_addr IN ({placeholders})").into()); + for addr in delegated_addresses { + args.add(addr).map_err(arg_err)?; + } + } + } + + // join + if !self.keys.is_empty() { + let mut idx = 0; + for (key, vals) in self.keys.iter() { + if !vals.is_empty() { + idx += 1; + let alias = format!("ee{idx}"); + joins.push(format!("event_entry {alias} ON e.id={alias}.event_id")); + clauses.push(format!("{alias}.indexed=1 AND {alias}.key=?").into()); + args.add(key).map_err(arg_err)?; + let mut subclauses = Vec::with_capacity(vals.len()); + for val in vals { + subclauses.push(format!("({alias}.codec=? AND {alias}.value=?)")); + args.add(val.codec as i64).map_err(arg_err)?; + args.add(&val.value).map_err(arg_err)?; + } + clauses.push(format!("({})", subclauses.join(" OR ")).into()); + } + } + } + + let mut qb = sqlx::QueryBuilder::with_arguments( + "SELECT + e.id, + tm.height, + tm.tipset_key_cid, + e.emitter_id, + e.emitter_addr, + e.event_index, + tm.message_cid, + tm.message_index, + e.reverted, + ee.flags, + ee.key, + ee.codec, + ee.value + FROM event e + JOIN tipset_message tm ON e.message_id = tm.id + JOIN event_entry ee ON e.id = ee.event_id", + args, + ); + + // join + if !joins.is_empty() { + qb.push(format!(", {}", joins.join(", "))); + } + + // where + if !clauses.is_empty() { + qb.push(format!(" WHERE {}", clauses.join(" AND "))); + } + + // order: retain insertion order of event_entry rows + qb.push(" ORDER BY tm.height ASC, tm.message_index ASC, e.event_index ASC, ee._rowid_ ASC"); + Ok(qb) + } +} + +impl TryFrom for IndexerEventFilter { + type Error = anyhow::Error; + + fn try_from( + ParsedFilter { + tipsets, + addresses, + keys, + msg_cid, + }: ParsedFilter, + ) -> Result { + let (min_height, max_height, tipset_cid) = match tipsets { + ParsedFilterTipsets::Hash(h) => (-1, -1, Some(h.to_cid())), + ParsedFilterTipsets::Range(r) => (*r.start(), *r.end(), None), + ParsedFilterTipsets::Key(k) => (-1, -1, Some(k.cid()?)), + }; + Ok(Self { + min_height, + max_height, + tipset_cid, + msg_cid, + addresses, + keys, + }) + } +} diff --git a/src/chain/store/indexer/tests.rs b/src/chain/store/indexer/tests.rs new file mode 100644 index 000000000000..c0b61e664e1a --- /dev/null +++ b/src/chain/store/indexer/tests.rs @@ -0,0 +1,38 @@ +// Copyright 2019-2026 ChainSafe Systems +// SPDX-License-Identifier: Apache-2.0, MIT + +use super::*; +use crate::{ + blocks::{Chain4U, HeaderBuilder, chain4u}, + chain::ChainStore, + db::MemoryDB, + networks::ChainConfig, +}; + +#[tokio::test(flavor = "multi_thread")] +async fn test_indexer_new() { + let c4u = Chain4U::new(); + chain4u! { + in c4u; + t0 @ [_b0 = HeaderBuilder::new().with_timestamp(100)] + }; + + let bs = Arc::new(MemoryDB::default()); + let cs = ChainStore::new( + bs, + Arc::new(ChainConfig::devnet()), + t0.min_ticket_block().clone(), + ) + .unwrap(); + let sm = StateManager::new(cs).unwrap(); + let temp_db_path = tempfile::Builder::new() + .suffix(".sqlite3") + .tempfile_in(std::env::temp_dir()) + .unwrap(); + let db = crate::utils::sqlite::open_file(temp_db_path.path()) + .await + .unwrap(); + SqliteIndexer::new(db, sm, Default::default()) + .await + .unwrap(); +} diff --git a/src/chain/store/mod.rs b/src/chain/store/mod.rs index 0aba76672816..dc6665a6aa05 100644 --- a/src/chain/store/mod.rs +++ b/src/chain/store/mod.rs @@ -5,6 +5,7 @@ pub mod base_fee; mod chain_store; mod errors; pub mod index; +pub mod indexer; mod tipset_tracker; mod weighted_quick_select; diff --git a/src/cli/subcommands/index_cmd.rs b/src/cli/subcommands/index_cmd.rs index 9826d355013d..fe5411a1f4fd 100644 --- a/src/cli/subcommands/index_cmd.rs +++ b/src/cli/subcommands/index_cmd.rs @@ -2,16 +2,20 @@ // SPDX-License-Identifier: Apache-2.0, MIT use crate::ipld::ChainExportState; -use crate::rpc::chain::{ - ApiIndexBackfillStatus, IndexBackfill, IndexBackfillCancel, IndexBackfillParams, - IndexBackfillStatus, +use crate::prelude::*; +use crate::rpc::{ + self, + chain::{ + ApiIndexBackfillStatus, ChainHead, ChainValidateIndex, IndexBackfill, IndexBackfillCancel, + IndexBackfillParams, IndexBackfillStatus, + }, + prelude::*, }; -use crate::rpc::{self, prelude::*}; -use crate::shim::clock::ChainEpoch; use clap::Subcommand; use indicatif::{ProgressBar, ProgressStyle}; -use std::time::Duration; +use std::time::{Duration, Instant}; +/// Manage the chain index #[derive(Debug, Subcommand)] pub enum IndexCommands { /// Backfill the chain index (Ethereum mappings, events, block blooms) using the running node. @@ -54,6 +58,20 @@ pub enum IndexCommands { }, /// Cancel the in-progress index backfill. BackfillCancel {}, + /// validates the SQL chain index entries for each epoch in descending order in the specified range, checking for missing or + /// inconsistent entries (i.e. the indexed data does not match the actual chain state). If '--backfill' is enabled + /// (which it is by default), it will attempt to backfill any missing entries using the `ChainValidateIndex` API. + ValidateBackfill { + /// specifies the starting tipset epoch for validation (inclusive) + #[arg(long, required = true)] + from: ChainEpoch, + /// specifies the ending tipset epoch for validation (inclusive) + #[arg(long, required = true)] + to: ChainEpoch, + /// determines whether to backfill missing index entries during validation + #[arg(long, default_missing_value = "true", default_value = "true")] + backfill: Option, + }, } impl IndexCommands { @@ -107,6 +125,9 @@ impl IndexCommands { } Ok(()) } + Self::ValidateBackfill { from, to, backfill } => { + validate_backfill(&client, from, to, backfill.unwrap_or_default()).await + } } } } @@ -150,3 +171,62 @@ async fn wait_for_backfill(client: &rpc::Client) -> anyhow::Result<()> { } Ok(()) } + +async fn validate_backfill( + client: &rpc::Client, + from: ChainEpoch, + to: ChainEpoch, + backfill: bool, +) -> anyhow::Result<()> { + anyhow::ensure!( + from > 0, + "invalid from epoch: {from}, must be greater than 0" + ); + anyhow::ensure!(to > 0, "invalid to epoch: {to}, must be greater than 0"); + anyhow::ensure!( + to <= from, + "to epoch ({to}) must be less than or equal to from epoch ({from})" + ); + let head = ChainHead::call(client, ()) + .await + .context("failed to get chain head for index validation")?; + anyhow::ensure!( + from < head.epoch(), + "from epoch ({from}) must be less than chain head ({})", + head.epoch() + ); + let start = Instant::now(); + tracing::info!( + "starting chainindex validation; from epoch: {from}; to epoch: {to}; backfill: {backfill};" + ); + let mut backfills = 0; + let mut null_rounds = 0; + let mut validations = 0; + let mut failures = 0usize; + for epoch in (to..=from).rev() { + match ChainValidateIndex::call(client, (epoch, backfill)).await { + Ok(r) => { + if r.backfilled { + backfills += 1; + } else if r.is_null_round { + null_rounds += 1; + } else { + validations += 1; + } + } + Err(e) => { + failures += 1; + tracing::warn!("Failed to validate index at epoch {epoch}: {e}"); + } + } + } + anyhow::ensure!( + failures == 0, + "index validation failed with {failures} errors" + ); + tracing::info!( + "done with {backfills} backfills, {null_rounds} null rounds, {validations} validations, took {}", + humantime::format_duration(start.elapsed()) + ); + Ok(()) +} diff --git a/src/cli/subcommands/mod.rs b/src/cli/subcommands/mod.rs index 3151507164b9..092ec86172d1 100644 --- a/src/cli/subcommands/mod.rs +++ b/src/cli/subcommands/mod.rs @@ -101,7 +101,7 @@ pub enum Subcommand { #[command(subcommand)] Healthcheck(HealthcheckCommand), - /// Manages Filecoin Fast Finality (F3) interactions + /// Manage Filecoin Fast Finality (F3) interactions #[command(subcommand)] F3(F3Commands), diff --git a/src/cli_shared/cli/config.rs b/src/cli_shared/cli/config.rs index da9d716ee30b..70975371668d 100644 --- a/src/cli_shared/cli/config.rs +++ b/src/cli_shared/cli/config.rs @@ -2,17 +2,16 @@ // SPDX-License-Identifier: Apache-2.0, MIT use super::client::Client; +use crate::chain::indexer::SqliteIndexerOptions; use crate::db::db_engine::DbConfig; +use crate::def_is_env_set_and_truthy; use crate::libp2p::Libp2pConfig; use crate::shim::clock::ChainEpoch; use crate::shim::econ::TokenAmount; -use crate::utils::misc::env::is_env_set_and_truthy; use crate::{chain_sync::SyncConfig, networks::NetworkChain}; use serde::{Deserialize, Serialize}; use std::path::PathBuf; -const FOREST_CHAIN_INDEXER_ENABLED: &str = "FOREST_CHAIN_INDEXER_ENABLED"; - /// Structure that defines daemon configuration when process is detached #[derive(Deserialize, Serialize, PartialEq, Eq, Debug, Clone)] #[cfg_attr(test, derive(derive_quickcheck_arbitrary::Arbitrary))] @@ -103,9 +102,34 @@ pub struct ChainIndexerConfig { impl Default for ChainIndexerConfig { fn default() -> Self { + def_is_env_set_and_truthy!(indexer_enabled, "FOREST_CHAIN_INDEXER_ENABLED"); + static GC_RETENTION_EPOCHS: std::sync::LazyLock> = + std::sync::LazyLock::new(|| { + std::env::var("FOREST_CHAIN_INDEXER_GC_RETENTION_EPOCHS") + .ok() + .and_then(|val| val.parse::().ok()) + .and_then(|val| { + if val == 0 { + // Zero means disabled + None + } else { + if let Err(e) = SqliteIndexerOptions::default() + .with_gc_retention_epochs(Some(val.into())) + .validate() + { + tracing::error!( + "Invalid FOREST_CHAIN_INDEXER_GC_RETENTION_EPOCHS value: {e}" + ); + None + } else { + Some(val) + } + } + }) + }); Self { - enable_indexer: is_env_set_and_truthy(FOREST_CHAIN_INDEXER_ENABLED).unwrap_or(true), - gc_retention_epochs: None, + enable_indexer: indexer_enabled().unwrap_or(true), + gc_retention_epochs: *GC_RETENTION_EPOCHS, } } } diff --git a/src/daemon/context.rs b/src/daemon/context.rs index f81d3284920a..dbf70bb0ae80 100644 --- a/src/daemon/context.rs +++ b/src/daemon/context.rs @@ -3,15 +3,16 @@ use crate::auth::{ADMIN, create_token, generate_priv_key}; use crate::chain::ChainStore; +use crate::chain::indexer::{SqliteIndexer, SqliteIndexerOptions}; use crate::cli_shared::chain_path; use crate::cli_shared::cli::CliOpts; use crate::daemon::asyncify; use crate::daemon::bundle::load_actor_bundles; use crate::daemon::db_util::load_all_forest_cars_with_cleanup; -use crate::db::CAR_DB_DIR_NAME; use crate::db::car::ManyCar; use crate::db::db_engine::db_root; use crate::db::parity_db::{GarbageCollectableParityDb, ParityDb}; +use crate::db::{CAR_DB_DIR_NAME, INDEX_DB_DIR_NAME, INDEX_DB_FILE_NAME}; use crate::genesis::read_genesis_header; use crate::libp2p::{Keypair, PeerId}; use crate::networks::ChainConfig; @@ -28,7 +29,7 @@ use dialoguer::console::Term; use fvm_shared4::address::Network; use parking_lot::RwLock; use std::cell::RefCell; -use std::path::PathBuf; +use std::path::{Path, PathBuf}; use std::sync::Arc; use tracing::{info, warn}; @@ -40,6 +41,7 @@ pub struct AppContext { pub db: DbType, pub db_meta_data: DbMetadata, pub state_manager: StateManager, + pub chain_indexer: Option>, pub keystore: Arc>, pub admin_jwt: String, pub snapshot_progress_tracker: SnapshotProgressTracker, @@ -55,6 +57,24 @@ impl AppContext { let state_manager = create_state_manager(cfg, &db, &chain_cfg).await?; let (keystore, admin_jwt) = load_or_create_keystore_and_configure_jwt(opts, cfg).await?; let snapshot_progress_tracker = SnapshotProgressTracker::default(); + let chain_indexer = if cfg.chain_indexer.enable_indexer + && !opts.stateless + && !state_manager.chain_config().is_devnet() + { + let options = SqliteIndexerOptions::default() + .with_gc_retention_epochs(cfg.chain_indexer.gc_retention_epochs.map(From::from)); + options.validate()?; + Some(Arc::new( + SqliteIndexer::new( + crate::utils::sqlite::open_file(db_meta_data.index_db_path()).await?, + state_manager.shallow_clone(), + options, + ) + .await?, + )) + } else { + None + }; let temp_dir = chain_path(cfg).join("tmp"); std::fs::create_dir_all(&temp_dir).context("Failed to create temporary directory")?; Ok(Self { @@ -63,6 +83,7 @@ impl AppContext { db, db_meta_data, state_manager, + chain_indexer, keystore, admin_jwt, snapshot_progress_tracker, @@ -189,15 +210,20 @@ fn maybe_migrate_db(config: &Config) { pub(crate) struct DbMetadata { db_root_dir: PathBuf, forest_car_db_dir: PathBuf, + index_db_path: PathBuf, } impl DbMetadata { - pub(crate) fn get_root_dir(&self) -> PathBuf { - self.db_root_dir.clone() + pub(crate) fn root_dir(&self) -> &Path { + &self.db_root_dir } - pub(crate) fn get_forest_car_db_dir(&self) -> PathBuf { - self.forest_car_db_dir.clone() + pub(crate) fn forest_car_db_dir(&self) -> &Path { + &self.forest_car_db_dir + } + + pub(crate) fn index_db_path(&self) -> &Path { + &self.index_db_path } } @@ -206,6 +232,7 @@ impl DbMetadata { /// - load parity-db /// - load CAR database /// - load actor bundles +/// - setup index db folder and file async fn setup_db(opts: &CliOpts, config: &Config) -> anyhow::Result<(DbType, DbMetadata)> { maybe_migrate_db(config); let chain_data_path = chain_path(config); @@ -217,6 +244,11 @@ async fn setup_db(opts: &CliOpts, config: &Config) -> anyhow::Result<(DbType, Db let db = Arc::new(ManyCar::new(db_writer.clone())); let forest_car_db_dir = db_root_dir.join(CAR_DB_DIR_NAME); load_all_forest_cars_with_cleanup(&db, &forest_car_db_dir)?; + let index_db_dir = db_root_dir.join(INDEX_DB_DIR_NAME); + if !index_db_dir.is_dir() { + std::fs::create_dir_all(&index_db_dir)?; + } + let index_db_path = index_db_dir.join(INDEX_DB_FILE_NAME); if config.client.load_actors && !opts.stateless { load_actor_bundles(&db, config.chain()).await?; } @@ -225,6 +257,7 @@ async fn setup_db(opts: &CliOpts, config: &Config) -> anyhow::Result<(DbType, Db DbMetadata { db_root_dir, forest_car_db_dir, + index_db_path, }, )) } diff --git a/src/daemon/mod.rs b/src/daemon/mod.rs index 3b284e93e555..e7d87887dcea 100644 --- a/src/daemon/mod.rs +++ b/src/daemon/mod.rs @@ -134,7 +134,7 @@ async fn maybe_import_snapshot( chain_config, ctx.state_manager.chain_store().heaviest_tipset().epoch(), opts.auto_download_snapshot, - &ctx.db_meta_data.get_root_dir(), + ctx.db_meta_data.root_dir(), ) .await?; } @@ -146,7 +146,7 @@ async fn maybe_import_snapshot( { let (car_db_path, ts) = import_chain_as_forest_car( path, - &ctx.db_meta_data.get_forest_car_db_dir(), + ctx.db_meta_data.forest_car_db_dir(), config.client.import_mode, config.client.rpc_v1_endpoint()?, &crate::f3::get_f3_root(config), @@ -164,6 +164,12 @@ async fn maybe_import_snapshot( "Loaded car DB at {} and set current head to epoch {ts_epoch}", car_db_path.display(), ); + // populate chain index if enabled + if let Some(chain_indexer) = &ctx.chain_indexer + && let Err(e) = chain_indexer.populate_after_snapshot_import().await + { + tracing::warn!("failed to populate chain index from snapshot: {e}"); + } } // If the snapshot progress state is not completed, @@ -190,7 +196,6 @@ async fn maybe_import_snapshot( }) .await??; } - Ok(()) } @@ -599,6 +604,7 @@ fn maybe_start_rpc_service( } services.spawn({ let state_manager = ctx.state_manager.shallow_clone(); + let chain_indexer = ctx.chain_indexer.clone(); let bad_blocks = chain_follower.bad_blocks.shallow_clone(); let sync_status = chain_follower.sync_status.shallow_clone(); let sync_network_context = chain_follower.network.shallow_clone(); @@ -619,6 +625,7 @@ fn maybe_start_rpc_service( state_manager, keystore, mpool, + chain_indexer, bad_blocks, sync_status, eth_event_handler, @@ -746,6 +753,7 @@ fn maybe_start_indexer_service( && !opts.stateless && !ctx.state_manager.chain_config().is_devnet() { + // Nosql indexer let mut head_changes_rx = ctx.state_manager.chain_store().subscribe_head_changes(); let chain_store = ctx.state_manager.chain_store().shallow_clone(); services.spawn(async move { @@ -786,6 +794,28 @@ fn maybe_start_indexer_service( collector.run().await }); } + + // New SQLITE indexer + { + if let Some(indexer) = &ctx.chain_indexer { + services.spawn({ + let head_changes_rx = ctx.state_manager.chain_store().subscribe_head_changes(); + let indexer = indexer.clone(); + async move { indexer.index_loop(head_changes_rx).await } + }); + services.spawn({ + let indexer = indexer.clone(); + async move { + if indexer.options().gc_retention_epochs.is_none() { + tracing::info!("gc retention epochs is not set, skipping gc"); + Ok(()) + } else { + indexer.gc_loop().await; + } + } + }); + } + } } } diff --git a/src/db/mod.rs b/src/db/mod.rs index 90c0f466423a..c16f9060a535 100644 --- a/src/db/mod.rs +++ b/src/db/mod.rs @@ -30,6 +30,8 @@ use serde::Serialize; use serde::de::DeserializeOwned; pub const CAR_DB_DIR_NAME: &str = "car_db"; +pub const INDEX_DB_DIR_NAME: &str = "index_db"; +pub const INDEX_DB_FILE_NAME: &str = "chainindex.db"; pub mod setting_keys { /// Key used to store the heaviest tipset in the settings store. This is expected to be a [`crate::blocks::TipsetKey`]s diff --git a/src/lib.rs b/src/lib.rs index 3dc91c5a8244..28d2842ec64b 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -86,6 +86,7 @@ mod prelude { pub use cid::Cid; pub use futures::FutureExt as _; pub use itertools::Itertools as _; + pub use nonzero_ext::nonzero; pub use std::{ops::Deref as _, sync::Arc}; pub use tracing::{debug, error, info, trace, warn}; } diff --git a/src/message_pool/msgpool/test_provider.rs b/src/message_pool/msgpool/test_provider.rs index 2e9b22767033..6dac3cce9630 100644 --- a/src/message_pool/msgpool/test_provider.rs +++ b/src/message_pool/msgpool/test_provider.rs @@ -56,13 +56,13 @@ impl Default for TestApi { impl TestApi { /// Constructor for a `TestApi` with custom number of max pending messages pub fn with_max_actor_pending_messages(max_actor_pending_messages: u64) -> Self { - let (publisher, _) = broadcast::channel(1); + let (head_changes_tx, _) = broadcast::channel(1); TestApi { inner: Mutex::new(TestApiInner { max_actor_pending_messages, ..TestApiInner::default() }), - head_changes_tx: publisher, + head_changes_tx, } } diff --git a/src/rpc/methods/chain.rs b/src/rpc/methods/chain.rs index 6472547dc706..ee73de5197e4 100644 --- a/src/rpc/methods/chain.rs +++ b/src/rpc/methods/chain.rs @@ -92,6 +92,31 @@ impl RpcMethod<0> for ChainGetFinalizedTipset { } } +pub enum ChainValidateIndex {} +impl RpcMethod<2> for ChainValidateIndex { + const NAME: &'static str = "Filecoin.ChainValidateIndex"; + const PARAM_NAMES: [&'static str; 2] = ["epoch", "backfill"]; + const API_PATHS: BitFlags = ApiPaths::all(); + const PERMISSION: Permission = Permission::Write; + const DESCRIPTION: &'static str = + "Validates the chain index at the given epoch, optionally backfill when missing"; + + type Params = (ChainEpoch, bool); + type Ok = ChainIndexValidation; + + async fn handle( + ctx: Ctx, + (epoch, backfill): Self::Params, + _: &http::Extensions, + ) -> Result { + if let Some(ci) = &ctx.chain_indexer { + Ok(ci.validate_index(epoch, backfill).await?) + } else { + Err(anyhow::anyhow!("chain indexer is disabled").into()) + } + } +} + pub enum ChainGetMessage {} impl RpcMethod<1> for ChainGetMessage { const NAME: &'static str = "Filecoin.ChainGetMessage"; diff --git a/src/rpc/methods/chain/types.rs b/src/rpc/methods/chain/types.rs index eb0f2b038de0..fafb3a1d7c35 100644 --- a/src/rpc/methods/chain/types.rs +++ b/src/rpc/methods/chain/types.rs @@ -11,6 +11,28 @@ pub struct ObjStat { } lotus_json_with_self!(ObjStat); +#[derive(Serialize, Deserialize, JsonSchema, Clone, Debug, Eq, PartialEq, Default)] +#[serde(rename_all = "PascalCase")] +pub struct ChainIndexValidation { + /// the key of the canonical tipset for this epoch + #[serde(with = "crate::lotus_json")] + #[schemars(with = "LotusJson")] + pub tip_set_key: ApiTipsetKey, + /// the epoch height at which the validation is performed. + pub height: ChainEpoch, + /// the number of indexed messages for the canonical tipset at this epoch + pub indexed_messages_count: u64, + /// the number of indexed events for the canonical tipset at this epoch + pub indexed_events_count: u64, + /// the number of indexed event entries for the canonical tipset at this epoch + pub indexed_event_entries_count: u64, + /// whether missing data was successfully backfilled into the index during validation + pub backfilled: bool, + /// if the epoch corresponds to a null round and therefore does not have any indexed messages or events + pub is_null_round: bool, +} +lotus_json_with_self!(ChainIndexValidation); + /// Describes how the node is currently determining finality, /// combining probabilistic EC finality (based on observed chain health) with /// F3 fast finality when available. diff --git a/src/rpc/methods/eth.rs b/src/rpc/methods/eth.rs index f7a302e054c0..60248ee8b72d 100644 --- a/src/rpc/methods/eth.rs +++ b/src/rpc/methods/eth.rs @@ -25,7 +25,9 @@ use self::trace::types::*; use self::types::*; use super::gas; use crate::blocks::{Tipset, TipsetKey}; -use crate::chain::{ChainStore, compute_base_fee, index::ResolveNullTipset}; +use crate::chain::{ + ChainStore, compute_base_fee, index::ResolveNullTipset, indexer::IndexerEventFilter, +}; use crate::chain_sync::NodeSyncStatus; use crate::cid_collections::CidHashSet; use crate::db::DbImpl; @@ -3201,7 +3203,7 @@ pub struct CollectedEvent { pub(crate) reverted: bool, pub(crate) height: ChainEpoch, pub(crate) tipset_key: TipsetKey, - msg_idx: u64, + pub(crate) msg_idx: u64, pub(crate) msg_cid: Cid, } @@ -3328,9 +3330,9 @@ fn eth_filter_logs_from_events( events: &[CollectedEvent], ) -> anyhow::Result> { let chain_id = ctx.state_manager.chain_config().eth_chain_id; - let mut tx_hash_by_msg: HashMap = HashMap::new(); - let mut block_hash_by_tipset: HashMap = HashMap::new(); - let mut eth_addr_by_emitter: HashMap = HashMap::new(); + let mut tx_hash_by_msg: HashMap = HashMap::default(); + let mut block_hash_by_tipset: HashMap = HashMap::default(); + let mut eth_addr_by_emitter: HashMap = HashMap::default(); let mut logs = Vec::with_capacity(events.len()); for event in events { @@ -3420,23 +3422,42 @@ impl RpcMethod<1> for EthGetLogs { (eth_filter,): Self::Params, _: &http::Extensions, ) -> Result { - let pf = Arc::new( - ctx.eth_event_handler - .parse_eth_filter_spec(&ctx, ð_filter) - .map_err(|e| { - if e.downcast_ref::().is_some_and(|eth_err| { - matches!(eth_err, EthErrors::BlockRangeExceeded { .. }) - }) { - return e; - } - e.context("failed to parse events for filter") - })?, - ); - let events = ctx + let pf = ctx .eth_event_handler - .get_events_for_parsed_filter(&ctx, &pf, SkipEvent::OnUnresolvedAddress) - .await - .context("failed to get events for filter")?; + .parse_eth_filter_spec(&ctx, ð_filter) + .map_err(|e| { + if e.downcast_ref::() + .is_some_and(|eth_err| matches!(eth_err, EthErrors::BlockRangeExceeded { .. })) + { + return e; + } + e.context("failed to parse events for filter") + })?; + let events = if pf + .tipsets + .is_large_range_for_sql(ctx.chain_store().heaviest_tipset().epoch()) + && let Some(chain_indexer) = ctx.chain_indexer() + { + // we can't return events for the heaviest tipset as the transactions in that tipset will be executed + // in the next non-null tipset (because of Filecoin's "deferred execution" model) + if let ParsedFilterTipsets::Range(range) = &pf.tipsets + && *range.end() >= ctx.chain_store().heaviest_tipset().epoch() + { + return Err(EthErrors::EventsNotYetAvailable.into()); + } + chain_indexer + .get_events_for_filter( + IndexerEventFilter::try_from(pf)?, + ctx.eth_event_handler.max_filter_results, + SkipEvent::OnUnresolvedAddress, + ) + .await + } else { + ctx.eth_event_handler + .get_events_for_parsed_filter(&ctx, &pf.into(), SkipEvent::OnUnresolvedAddress) + .await + } + .context("failed to get events for filter")?; Ok(eth_filter_result_from_events(&ctx, &events)?) } } diff --git a/src/rpc/methods/eth/filter/mod.rs b/src/rpc/methods/eth/filter/mod.rs index f780e99bca5d..3bdf729f4407 100644 --- a/src/rpc/methods/eth/filter/mod.rs +++ b/src/rpc/methods/eth/filter/mod.rs @@ -24,8 +24,7 @@ use super::BlockNumberOrHash; use super::CollectedEvent; use super::Predefined; use super::get_tipset_from_hash; -use crate::blocks::Tipset; -use crate::blocks::TipsetKey; +use crate::blocks::{Tipset, TipsetKey}; use crate::chain::index::ResolveNullTipset; use crate::cli_shared::cli::EventsConfig; use crate::eth::EthChainId; @@ -50,8 +49,10 @@ use anyhow::{Error, anyhow, bail, ensure}; use futures::{TryStreamExt as _, stream::FuturesOrdered}; use fvm_ipld_encoding::IPLD_RAW; use serde::*; +use std::num::NonZeroU32; use std::ops::RangeInclusive; use std::sync::Arc; +use std::sync::LazyLock; use std::time::Duration; use store::*; use tokio_util::task::AbortOnDropHandle; @@ -105,7 +106,7 @@ pub trait FilterManager { /// (`tipsets_contributing <= 1`) always pass — the natural unit is the tipset. /// Once two or more tipsets have contributed events, returns an error if the /// running total exceeds `max_filter_results`. -fn ensure_filter_cap( +pub fn ensure_filter_cap( max_filter_results: usize, tipsets_contributing: usize, total_events: usize, @@ -743,8 +744,8 @@ fn parse_eth_topics( #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] pub(crate) struct ActorEventBlock { - codec: u64, - value: Vec, + pub codec: u64, + pub value: Vec, } fn keys_to_keys_with_codec( @@ -774,7 +775,37 @@ pub enum ParsedFilterTipsets { Key(TipsetKey), } -#[derive(Debug)] +impl ParsedFilterTipsets { + /// Returns true if the filter's tipset range is no less than the threshold for prefering SQL query. + /// head epoch is used to determine the effective end of the range when the end is negative. + pub fn is_large_range_for_sql(&self, head_epoch: ChainEpoch) -> bool { + static LARGE_RANGE_THRESHOLD: LazyLock = LazyLock::new(|| { + std::env::var("FOREST_RPC_SQL_RANGE_THRESHOLD") + .ok() + .and_then(|i| { + i.parse().ok().inspect(|i| { + tracing::info!("RPC SQL range threshold set to {i} from `FOREST_RPC_SQL_RANGE_THRESHOLD`"); + }) + }) + // The current default value disables SQL by default as the maximum allowed range is 2880. + .unwrap_or(nonzero!(10000_u32)) + }); + + if let ParsedFilterTipsets::Range(range) = self { + let range_end = match *range.end() { + // use pending_epoch(head_epoch - 1) as head when end is negative, because the heaviest tipset doesn't have events yet + end if end < 0 => (head_epoch - 1).max(0), + end => end, + }; + let range_size = range_end - *range.start(); + range_size >= i64::from(LARGE_RANGE_THRESHOLD.get()) + } else { + false + } + } +} + +#[derive(Debug, Clone)] pub struct ParsedFilter { pub(crate) tipsets: ParsedFilterTipsets, pub(crate) addresses: Vec
, @@ -898,6 +929,7 @@ mod tests { use base64::{Engine, prelude::BASE64_STANDARD}; use fvm_ipld_encoding::DAG_CBOR; use fvm_shared4::event::Flags; + use rstest::rstest; use std::str::FromStr; #[test] @@ -1745,4 +1777,32 @@ mod tests { assert_eq!(event.emitter_addr, Address::new_id(1234)); } } + + #[rstest] + #[case(0..=0, 20000, false)] + #[case(0..=1, 20000, false)] + #[case(0..=9999, 20000, false)] + #[case(0..=10000, 20000, true)] + #[case(0..=10001, 20000, true)] + #[allow(clippy::reversed_empty_ranges)] + #[case(0..=(-1), 20000, true)] + #[allow(clippy::reversed_empty_ranges)] + #[case(0..=(-1), 100, false)] + fn test_is_large_range_for_sql_range( + #[case] range: RangeInclusive, + #[case] head_epoch: ChainEpoch, + #[case] expected: bool, + ) { + let filter = ParsedFilterTipsets::Range(range); + assert_eq!(filter.is_large_range_for_sql(head_epoch), expected); + } + + #[test] + fn test_is_large_range_for_sql_no_range() { + let filter = ParsedFilterTipsets::Hash(EthHash::default()); + assert!(!filter.is_large_range_for_sql(1000)); + + let filter = ParsedFilterTipsets::Key(nunny::vec![Cid::default()].into()); + assert!(!filter.is_large_range_for_sql(1000)); + } } diff --git a/src/rpc/methods/sync.rs b/src/rpc/methods/sync.rs index 8d62cb31ded8..3a4177d08f1f 100644 --- a/src/rpc/methods/sync.rs +++ b/src/rpc/methods/sync.rs @@ -235,6 +235,7 @@ mod tests { state_manager, keystore: Arc::new(RwLock::new(KeyStore::new(KeyStoreConfig::Memory).unwrap())), mpool, + chain_indexer: Default::default(), bad_blocks: Some(Default::default()), sync_status: Default::default(), eth_event_handler: Arc::new(EthEventHandler::new()), diff --git a/src/rpc/mod.rs b/src/rpc/mod.rs index 554ae0251983..5070b8696920 100644 --- a/src/rpc/mod.rs +++ b/src/rpc/mod.rs @@ -102,6 +102,7 @@ macro_rules! for_each_rpc_method { $callback!($crate::rpc::chain::IndexBackfillStatus); $callback!($crate::rpc::chain::IndexBackfillCancel); $callback!($crate::rpc::chain::ChainGetTipsetByParentState); + $callback!($crate::rpc::chain::ChainValidateIndex); // common vertical $callback!($crate::rpc::common::Session); @@ -501,6 +502,7 @@ pub struct RPCState { pub keystore: Arc>, pub state_manager: crate::state_manager::StateManager, pub mpool: crate::message_pool::MessagePool, + pub chain_indexer: Option>, pub bad_blocks: Option, pub sync_status: crate::chain_sync::SyncStatus, pub eth_event_handler: Arc, @@ -532,6 +534,10 @@ impl RPCState { self.state_manager.chain_config() } + pub fn chain_indexer(&self) -> Option<&Arc> { + self.chain_indexer.as_ref() + } + pub fn genesis_info(&self) -> &Arc { self.state_manager.genesis_info() } diff --git a/src/rpc/snapshots/forest__rpc__tests__rpc__v0.snap b/src/rpc/snapshots/forest__rpc__tests__rpc__v0.snap index a38f1fb5172e..1818c3148a20 100644 --- a/src/rpc/snapshots/forest__rpc__tests__rpc__v0.snap +++ b/src/rpc/snapshots/forest__rpc__tests__rpc__v0.snap @@ -481,6 +481,24 @@ methods: - $ref: "#/components/schemas/Tipset" - type: "null" paramStructure: by-position + - name: Filecoin.ChainValidateIndex + description: "Validates the chain index at the given epoch, optionally backfill when missing" + params: + - name: epoch + required: true + schema: + type: integer + format: int64 + - name: backfill + required: true + schema: + type: boolean + result: + name: Filecoin.ChainValidateIndex.Result + required: true + schema: + $ref: "#/components/schemas/ChainIndexValidation" + paramStructure: by-position - name: Filecoin.Session description: Returns a UUID that uniquely identifies this node for the current session. params: [] @@ -5446,6 +5464,45 @@ components: - Succeeded - Cancelled - Failed + ChainIndexValidation: + type: object + properties: + Backfilled: + description: whether missing data was successfully backfilled into the index during validation + type: boolean + Height: + description: the epoch height at which the validation is performed. + type: integer + format: int64 + IndexedEventEntriesCount: + description: the number of indexed event entries for the canonical tipset at this epoch + type: integer + format: uint64 + minimum: 0 + IndexedEventsCount: + description: the number of indexed events for the canonical tipset at this epoch + type: integer + format: uint64 + minimum: 0 + IndexedMessagesCount: + description: the number of indexed messages for the canonical tipset at this epoch + type: integer + format: uint64 + minimum: 0 + IsNullRound: + description: if the epoch corresponds to a null round and therefore does not have any indexed messages or events + type: boolean + TipSetKey: + description: the key of the canonical tipset for this epoch + $ref: "#/components/schemas/Nullable_Array_of_Cid" + required: + - TipSetKey + - Height + - IndexedMessagesCount + - IndexedEventsCount + - IndexedEventEntriesCount + - Backfilled + - IsNullRound ChangedType: description: Represents a changed value with before and after states. type: object diff --git a/src/rpc/snapshots/forest__rpc__tests__rpc__v1.snap b/src/rpc/snapshots/forest__rpc__tests__rpc__v1.snap index da0fa25318c4..34327616d00f 100644 --- a/src/rpc/snapshots/forest__rpc__tests__rpc__v1.snap +++ b/src/rpc/snapshots/forest__rpc__tests__rpc__v1.snap @@ -476,6 +476,24 @@ methods: - $ref: "#/components/schemas/Tipset" - type: "null" paramStructure: by-position + - name: Filecoin.ChainValidateIndex + description: "Validates the chain index at the given epoch, optionally backfill when missing" + params: + - name: epoch + required: true + schema: + type: integer + format: int64 + - name: backfill + required: true + schema: + type: boolean + result: + name: Filecoin.ChainValidateIndex.Result + required: true + schema: + $ref: "#/components/schemas/ChainIndexValidation" + paramStructure: by-position - name: Filecoin.Session description: Returns a UUID that uniquely identifies this node for the current session. params: [] @@ -5515,6 +5533,45 @@ components: - Succeeded - Cancelled - Failed + ChainIndexValidation: + type: object + properties: + Backfilled: + description: whether missing data was successfully backfilled into the index during validation + type: boolean + Height: + description: the epoch height at which the validation is performed. + type: integer + format: int64 + IndexedEventEntriesCount: + description: the number of indexed event entries for the canonical tipset at this epoch + type: integer + format: uint64 + minimum: 0 + IndexedEventsCount: + description: the number of indexed events for the canonical tipset at this epoch + type: integer + format: uint64 + minimum: 0 + IndexedMessagesCount: + description: the number of indexed messages for the canonical tipset at this epoch + type: integer + format: uint64 + minimum: 0 + IsNullRound: + description: if the epoch corresponds to a null round and therefore does not have any indexed messages or events + type: boolean + TipSetKey: + description: the key of the canonical tipset for this epoch + $ref: "#/components/schemas/Nullable_Array_of_Cid" + required: + - TipSetKey + - Height + - IndexedMessagesCount + - IndexedEventsCount + - IndexedEventEntriesCount + - Backfilled + - IsNullRound ChangedType: description: Represents a changed value with before and after states. type: object diff --git a/src/shim/address.rs b/src/shim/address.rs index c380700c63a7..8db2e37b4e26 100644 --- a/src/shim/address.rs +++ b/src/shim/address.rs @@ -177,6 +177,10 @@ impl Address { self.0.protocol() } + pub fn is_delegated(&self) -> bool { + self.protocol() == Protocol::Delegated + } + pub fn into_payload(self) -> Payload { self.0.into_payload() } diff --git a/src/shim/mod.rs b/src/shim/mod.rs index 50b75411b4d4..23aabc107fdf 100644 --- a/src/shim/mod.rs +++ b/src/shim/mod.rs @@ -39,3 +39,5 @@ pub mod fvm_latest { } pub type MethodNum = fvm_shared_latest::MethodNum; + +pub type ActorID = fvm_shared_latest::ActorID; diff --git a/src/tool/offline_server/server.rs b/src/tool/offline_server/server.rs index ad41405a5ef7..9c576e5c68f4 100644 --- a/src/tool/offline_server/server.rs +++ b/src/tool/offline_server/server.rs @@ -114,6 +114,7 @@ pub async fn offline_rpc_state( state_manager, keystore: Arc::new(RwLock::new(keystore)), mpool: message_pool, + chain_indexer: Default::default(), bad_blocks: Default::default(), sync_status: Arc::new(ArcSwap::from_pointee(SyncStatusReport::init())), eth_event_handler, diff --git a/src/tool/subcommands/api_cmd/api_compare_tests.rs b/src/tool/subcommands/api_cmd/api_compare_tests.rs index 3bedbee19308..31e3d2ebc686 100644 --- a/src/tool/subcommands/api_cmd/api_compare_tests.rs +++ b/src/tool/subcommands/api_cmd/api_compare_tests.rs @@ -623,6 +623,8 @@ fn chain_tests_with_tipset( },))?, offline, ), + RpcTest::basic(ChainValidateIndex::request((tipset.epoch(), true))?), + RpcTest::basic(ChainValidateIndex::request((tipset.epoch(), false))?), ]); } diff --git a/src/tool/subcommands/api_cmd/generate_test_snapshot.rs b/src/tool/subcommands/api_cmd/generate_test_snapshot.rs index a423edcf6da5..4abb1723625a 100644 --- a/src/tool/subcommands/api_cmd/generate_test_snapshot.rs +++ b/src/tool/subcommands/api_cmd/generate_test_snapshot.rs @@ -157,6 +157,7 @@ async fn ctx( state_manager, keystore: Arc::new(RwLock::new(KeyStore::new(KeyStoreConfig::Memory)?)), mpool: message_pool, + chain_indexer: Default::default(), bad_blocks: Default::default(), sync_status: Arc::new(ArcSwap::from_pointee(SyncStatusReport::init())), eth_event_handler, diff --git a/src/tool/subcommands/api_cmd/test_snapshot.rs b/src/tool/subcommands/api_cmd/test_snapshot.rs index bb9ac76150fc..18d4ccd62f7e 100644 --- a/src/tool/subcommands/api_cmd/test_snapshot.rs +++ b/src/tool/subcommands/api_cmd/test_snapshot.rs @@ -259,6 +259,7 @@ async fn ctx( state_manager, keystore: Arc::new(RwLock::new(KeyStore::new(KeyStoreConfig::Memory)?)), mpool: message_pool, + chain_indexer: Default::default(), bad_blocks: Default::default(), sync_status: Arc::new(ArcSwap::from_pointee(SyncStatusReport::init())), eth_event_handler, diff --git a/src/tool/subcommands/api_cmd/test_snapshots_ignored.txt b/src/tool/subcommands/api_cmd/test_snapshots_ignored.txt index 212f3f880a60..8305c0f2bca4 100644 --- a/src/tool/subcommands/api_cmd/test_snapshots_ignored.txt +++ b/src/tool/subcommands/api_cmd/test_snapshots_ignored.txt @@ -15,6 +15,7 @@ Filecoin.ChainExport Filecoin.ChainGetEvents Filecoin.ChainGetFinalizedTipset Filecoin.ChainSetHead +Filecoin.ChainValidateIndex Filecoin.EthEstimateGas Filecoin.EthGetFilterChanges Filecoin.EthGetFilterLogs diff --git a/src/utils/misc/env.rs b/src/utils/misc/env.rs index df194139f704..9009bc9a8d4a 100644 --- a/src/utils/misc/env.rs +++ b/src/utils/misc/env.rs @@ -56,6 +56,24 @@ macro_rules! def_is_env_truthy { }; } +#[macro_export] +macro_rules! def_is_env_set_and_truthy { + ($fn_name:ident, $env: expr) => { + #[inline] + pub fn $fn_name() -> Option { + cfg_if::cfg_if! { + if #[cfg(test)] { + $crate::utils::misc::env::is_env_set_and_truthy($env) + } else{ + static ENV: std::sync::LazyLock> = + std::sync::LazyLock::new(|| $crate::utils::misc::env::is_env_set_and_truthy($env)); + *ENV + } + } + } + }; +} + #[cfg(test)] mod tests { use super::*; diff --git a/src/utils/mod.rs b/src/utils/mod.rs index 15f578eab789..8c6678c5bec6 100644 --- a/src/utils/mod.rs +++ b/src/utils/mod.rs @@ -19,7 +19,6 @@ pub mod rand; pub mod reqwest_resume; mod shallow_clone; pub use shallow_clone::ShallowClone; -#[cfg(feature = "sqlite")] pub mod sqlite; pub mod stats; pub mod stream; diff --git a/src/utils/sqlite/mod.rs b/src/utils/sqlite/mod.rs index 2dfe7375ce10..9ab75d4907b9 100644 --- a/src/utils/sqlite/mod.rs +++ b/src/utils/sqlite/mod.rs @@ -34,16 +34,6 @@ pub async fn open_file(file: &Path) -> anyhow::Result { Ok(open(options).await?) } -/// Opens for creates an in-memory database -pub async fn open_memory() -> sqlx::Result { - open( - SqliteConnectOptions::new() - .in_memory(true) - .shared_cache(true), - ) - .await -} - /// Opens a database at the given path. If the database does not exist, it will be created. pub async fn open(options: SqliteConnectOptions) -> sqlx::Result { let options = options @@ -54,7 +44,8 @@ pub async fn open(options: SqliteConnectOptions) -> sqlx::Result { .journal_mode(SqliteJournalMode::Wal) .pragma("journal_size_limit", "0") // always reset journal and wal files .foreign_keys(true) - .read_only(false); + .read_only(false) + .create_if_missing(true); SqlitePool::connect_with(options).await } @@ -89,13 +80,15 @@ pub async fn init_db<'q>( tx.commit().await }; - if sqlx::query("SELECT name FROM sqlite_master WHERE type='table' AND name='_meta';") + if sqlx::query("SELECT name FROM sqlite_master WHERE type='table' AND name='_meta'") .fetch_optional(db) .await .with_context(|| format!("error looking for {name} database _meta table"))? .is_none() { - init(db, schema_version).await?; + init(db, schema_version).await.map_err(|e| { + anyhow::anyhow!("failed to initialize db version {schema_version}: {e}") + })?; } let found_version: u64 = sqlx::query_scalar("SELECT max(version) FROM _meta")