diff --git a/Cargo.toml b/Cargo.toml index 411feb006402..d6667e26958c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -212,6 +212,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.18" @@ -242,7 +243,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 } @@ -345,7 +345,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` @@ -357,7 +357,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/scripts/tests/api_compare/docker-compose.yml b/scripts/tests/api_compare/docker-compose.yml index f4cca6295a71..c69c537ddde5 100644 --- a/scripts/tests/api_compare/docker-compose.yml +++ b/scripts/tests/api_compare/docker-compose.yml @@ -58,7 +58,7 @@ services: --halt-after-import SNAPSHOT_EPOCH="$(ls /data/*.car.zst | tail -n 1 | grep -Eo '[0-9]+' | tail -n 1)" - # backfill the index db + # backfill the index db (nosql) forest-tool index backfill --from $$SNAPSHOT_EPOCH --n-tipsets 200 --chain ${CHAIN} forest --chain ${CHAIN} --encrypt-keystore false --no-gc \ @@ -90,6 +90,10 @@ services: export FULLNODE_API_INFO="$(cat /data/forest-token):/dns/forest/tcp/${FOREST_RPC_PORT}/http" echo "Waiting till Forest is ready" forest-cli healthcheck ready --healthcheck-port ${FOREST_HEALTHZ_RPC_PORT} --wait + + # backfill the index db (sql) + SNAPSHOT_EPOCH="$(ls /data/*.car.zst | tail -n 1 | grep -Eo '[0-9]+' | tail -n 1)" + 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 36053caf02be..c74ac96e63f1 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/src/chain/store/indexer.rs b/src/chain/store/indexer.rs new file mode 100644 index 000000000000..c4aac9895b7a --- /dev/null +++ b/src/chain/store/indexer.rs @@ -0,0 +1,929 @@ +// Copyright 2019-2026 ChainSafe Systems +// SPDX-License-Identifier: Apache-2.0, MIT + +mod ddls; +mod events; +#[cfg(test)] +mod tests; + +use ahash::HashMap; +pub use ddls::{DDLS, PreparedStatements}; +pub use events::IndexerEventFilter; +use sqlx::Row as _; + +use crate::{ + blocks::Tipset, + chain::{ChainStore, HeadChanges, index::ResolveNullTipset}, + message::{ChainMessage, SignedMessage}, + prelude::*, + rpc::{ + chain::types::ChainIndexValidation, + eth::{eth_tx_from_signed_eth_message, types::EthHash}, + }, + shim::{ + ActorID, + address::Address, + clock::{ChainEpoch, EPOCHS_IN_DAY}, + executor::{Receipt, StampedEvent}, + }, + utils::sqlite, +}; +use std::{ + ops::DerefMut as _, + sync::Arc, + time::{Duration, Instant}, +}; + +type ActorToDelegatedAddressFunc = + Arc anyhow::Result
+ Send + Sync + 'static>; + +type RecomputeTipsetStateFunc = Arc anyhow::Result<()> + Send + Sync + 'static>; + +type ExecutedMessage = (ChainMessage, Receipt, Option>); + +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: ChainEpoch, +} + +impl SqliteIndexerOptions { + fn validate(&self) -> anyhow::Result<()> { + anyhow::ensure!( + self.gc_retention_epochs == 0 || self.gc_retention_epochs >= EPOCHS_IN_DAY, + "gc retention epochs must be 0 or greater than {EPOCHS_IN_DAY}" + ); + Ok(()) + } + + pub fn with_gc_retention_epochs(mut self, gc_retention_epochs: ChainEpoch) -> Self { + self.gc_retention_epochs = gc_retention_epochs; + self + } +} + +pub struct SqliteIndexer { + lock: tokio::sync::Mutex<()>, + options: SqliteIndexerOptions, + cs: ChainStore, + db: sqlx::SqlitePool, + stmts: PreparedStatements, + actor_to_delegated_address_func: Option, + recompute_tipset_state_func: Option, +} + +impl SqliteIndexer { + pub async fn new( + db: sqlx::SqlitePool, + cs: ChainStore, + 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, + cs, + db, + stmts, + actor_to_delegated_address_func: None, + recompute_tipset_state_func: None, + }) + } + + /// 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 aquire_write_lock(&self) -> tokio::sync::MutexGuard<'_, ()> { + self.lock.lock().await + } + + pub fn db(&self) -> &sqlx::SqlitePool { + &self.db + } + + pub fn with_actor_to_delegated_address_func(mut self, f: ActorToDelegatedAddressFunc) -> Self { + self.actor_to_delegated_address_func = Some(f); + self + } + + pub fn with_recompute_tipset_state_func(mut self, f: RecomputeTipsetStateFunc) -> Self { + self.recompute_tipset_state_func = Some(f); + self + } + + pub async fn index_loop( + &self, + mut head_changes_rx: tokio::sync::broadcast::Receiver, + ) -> anyhow::Result<()> { + loop { + let HeadChanges { reverts, applies } = head_changes_rx.recv().await?; + let _lock = self.aquire_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) { + if self.options.gc_retention_epochs <= 0 { + tracing::info!("gc retention epochs is not set, skipping gc"); + return; + } + + let mut ticker = tokio::time::interval(Duration::from_hours(4)); + loop { + ticker.tick().await; + self.gc().await; + } + } + + async fn gc(&self) { + let _lock = self.aquire_write_lock().await; + tracing::info!("starting index gc"); + let head = self.cs.heaviest_tipset(); + let removal_epoch = head.epoch() - self.options.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) + .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 = self.options.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) + .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(&self) -> anyhow::Result<()> { + let _lock = self.aquire_write_lock().await; + let start = Instant::now(); + let head = self.cs.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.cs.db()) { + if let Err(e) = self.index_tipset_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.aquire_write_lock().await) + } else { + None + }; + let head = self.cs.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 + .cs + .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, + ) = if let Ok(r) = self.get_and_verify_indexed_data(&ts).await { + (r, false) + } else { + self.backfill_missing_tipset(&ts).await?; + (self.get_and_verify_indexed_data(&ts).await?, true) + }; + 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 + .cs + .chain_index() + .load_required_tipset_by_height( + ts.epoch() + 1, + self.cs.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 = self.load_executed_messages(ts, &execution_ts)?; + 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 (_, _, 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 (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); + if let Some(&flags) = flags.first() { + 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()?)) + } + } + + fn load_executed_messages( + &self, + msg_ts: &Tipset, + receipt_ts: &Tipset, + ) -> anyhow::Result> { + let recompute_tipset_state_func = self + .recompute_tipset_state_func + .as_ref() + .context("recompute_tipset_state_func not set")?; + let msgs = self.cs.messages_for_tipset(msg_ts)?; + if msgs.is_empty() { + return Ok(vec![]); + } + let mut recomputed = false; + let recompute = || { + let tsk_cid = receipt_ts.key().cid()?; + tracing::warn!( + "failed to load receipts for tipset {tsk_cid} (epoch {}); recomputing tipset state", + receipt_ts.epoch() + ); + recompute_tipset_state_func(msg_ts.clone())?; + tracing::warn!( + "successfully recomputed tipset state and loaded events for tipset {tsk_cid} (epoch {})", + receipt_ts.epoch() + ); + anyhow::Ok(()) + }; + let receipts = + match Receipt::get_receipts(self.cs.db(), *receipt_ts.parent_message_receipts()) { + Ok(receipts) => receipts, + Err(_) => { + recompute()?; + recomputed = true; + Receipt::get_receipts(self.cs.db(), *receipt_ts.parent_message_receipts())? + } + }; + anyhow::ensure!( + msgs.len() == receipts.len(), + "mismatching message and receipt counts ({} msgs, {} rcts)", + msgs.len(), + receipts.len() + ); + let mut executed = Vec::with_capacity(msgs.len()); + for (message, receipt) in msgs.iter().zip(receipts) { + let events = if let Some(events_root) = receipt.events_root() { + Some(match StampedEvent::get_events(self.cs.db(), &events_root) { + Ok(events) => events, + Err(e) if recomputed => return Err(e), + Err(_) => { + recompute()?; + recomputed = true; + StampedEvent::get_events(self.cs.db(), &events_root)? + } + }) + } else { + None + }; + executed.push((message.clone(), receipt, events)); + } + Ok(executed) + } + + 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.cs.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 + .cs + .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.cs.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.cs.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<()> { + let actor_to_delegated_address_func = self + .actor_to_delegated_address_func + .as_ref() + .context("indexer can not index events without an address resolver")?; + // 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 = self + .load_executed_messages(msg_ts, execution_ts) + .map_err(|e| anyhow::anyhow!("failed to load executed message: {e}"))?; + let mut event_count = 0; + let mut address_lookups = HashMap::default(); + for (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 = actor_to_delegated_address_func(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.cs.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..64000cb7e95e --- /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 height >= ?)"; + 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..192c04c2722b --- /dev/null +++ b/src/chain/store/indexer/events.rs @@ -0,0 +1,279 @@ +// 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::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 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, + ) -> anyhow::Result> { + let bs = self.cs.db(); + let mut qb = filter.to_query_builder()?; + let query = qb.build(); + let results = query.fetch_all(self.db()).await?; + let mut current_id = -1; + let mut last_height = -1; + let mut tipsets_seen = 0; + let mut collected_events = vec![]; + let mut ce = None; + for row in results { + 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::new_id(event.emitter_id) + } else { + Address::from_bytes(event.emitter_addr.as_slice())? + }; + 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..b76a1896df80 --- /dev/null +++ b/src/chain/store/indexer/tests.rs @@ -0,0 +1,36 @@ +// Copyright 2019-2026 ChainSafe Systems +// SPDX-License-Identifier: Apache-2.0, MIT + +use super::*; +use crate::{ + blocks::{Chain4U, chain4u}, + db::MemoryDB, + networks::ChainConfig, +}; + +#[tokio::test(flavor = "multi_thread")] +async fn test_indexer_new() { + let c4u = Chain4U::new(); + chain4u! { + in c4u; + t0 @ [_b0] + }; + + let bs = Arc::new(MemoryDB::default()); + let cs = ChainStore::new( + bs, + Arc::new(ChainConfig::devnet()), + t0.min_ticket_block().clone(), + ) + .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, cs, 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 new file mode 100644 index 000000000000..73bb586f62c9 --- /dev/null +++ b/src/cli/subcommands/index_cmd.rs @@ -0,0 +1,92 @@ +// Copyright 2019-2026 ChainSafe Systems +// SPDX-License-Identifier: Apache-2.0, MIT + +use crate::{ + rpc::{ + self, RpcMethodExt as _, + chain::{ChainHead, ChainValidateIndex}, + }, + shim::clock::ChainEpoch, +}; +use clap::Subcommand; +use std::time::Instant; + +/// Manage the chain index +#[derive(Debug, Subcommand)] +pub enum IndexCommands { + /// validates the 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 { + pub async fn run(self, client: rpc::Client) -> anyhow::Result<()> { + match self { + Self::ValidateBackfill { from, to, backfill } => { + validate_backfill(&client, from, to, backfill.unwrap_or_default()).await + } + } + } +} + +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?; + 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; + 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) => { + tracing::warn!("Failed to validate index at epoch {epoch}: {e}"); + } + } + } + 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 29546f14b625..48dd6a2dbe2d 100644 --- a/src/cli/subcommands/mod.rs +++ b/src/cli/subcommands/mod.rs @@ -11,6 +11,7 @@ mod chain_cmd; mod config_cmd; mod f3_cmd; mod healthcheck_cmd; +mod index_cmd; mod info_cmd; mod mpool_cmd; mod net_cmd; @@ -22,9 +23,10 @@ mod wait_api_cmd; pub(super) use self::{ auth_cmd::AuthCommands, chain_cmd::ChainCommands, config_cmd::ConfigCommands, - f3_cmd::F3Commands, healthcheck_cmd::HealthcheckCommand, mpool_cmd::MpoolCommands, - net_cmd::NetCommands, shutdown_cmd::ShutdownCommand, snapshot_cmd::SnapshotCommands, - state_cmd::StateCommands, sync_cmd::SyncCommands, wait_api_cmd::WaitApiCommand, + f3_cmd::F3Commands, healthcheck_cmd::HealthcheckCommand, index_cmd::IndexCommands, + mpool_cmd::MpoolCommands, net_cmd::NetCommands, shutdown_cmd::ShutdownCommand, + snapshot_cmd::SnapshotCommands, state_cmd::StateCommands, sync_cmd::SyncCommands, + wait_api_cmd::WaitApiCommand, }; use crate::cli::subcommands::info_cmd::InfoCommand; pub(crate) use crate::cli_shared::cli::Config; @@ -95,12 +97,15 @@ pub enum Subcommand { #[command(subcommand)] Healthcheck(HealthcheckCommand), - /// Manages Filecoin Fast Finality (F3) interactions + /// Manage Filecoin Fast Finality (F3) interactions #[command(subcommand)] F3(F3Commands), /// Wait for lotus API to come online WaitApi(WaitApiCommand), + + #[command(subcommand)] + Index(IndexCommands), } impl Subcommand { diff --git a/src/daemon/context.rs b/src/daemon/context.rs index f81d3284920a..d2a33806e3bc 100644 --- a/src/daemon/context.rs +++ b/src/daemon/context.rs @@ -3,22 +3,24 @@ 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::interpreter::VMTrace; use crate::libp2p::{Keypair, PeerId}; use crate::networks::ChainConfig; use crate::prelude::*; use crate::rpc::sync::SnapshotProgressTracker; -use crate::shim::address::CurrentNetwork; -use crate::state_manager::StateManager; +use crate::shim::address::{Address, CurrentNetwork}; +use crate::state_manager::{NO_CALLBACK, StateManager}; use crate::{ Config, ENCRYPTED_KEYSTORE_NAME, FOREST_KEYSTORE_PHRASE_ENV, JWT_IDENTIFIER, KeyStore, KeyStoreConfig, @@ -28,7 +30,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 +42,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 +58,46 @@ 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 { + Some(Arc::new( + SqliteIndexer::new( + crate::utils::sqlite::open_file(db_meta_data.index_db_path()).await?, + state_manager.chain_store().shallow_clone(), + SqliteIndexerOptions::default().with_gc_retention_epochs(i64::from( + cfg.chain_indexer.gc_retention_epochs.unwrap_or_default(), + )), + ) + .await? + .with_actor_to_delegated_address_func(Arc::new({ + let state_manager = state_manager.shallow_clone(); + move |actor_id, ts| { + let id_addr = Address::new_id(actor_id); + Ok( + match state_manager.get_required_actor(&id_addr, *ts.parent_state()) { + Ok(actor) => actor + .delegated_address + .map(Address::from) + .unwrap_or(id_addr), + Err(_) => id_addr, + }, + ) + } + })) + .with_recompute_tipset_state_func(Arc::new({ + let state_manager = state_manager.shallow_clone(); + move |ts| { + state_manager.compute_tipset_state_blocking( + ts, + NO_CALLBACK, + VMTrace::NotTraced, + )?; + Ok(()) + } + })), + )) + } 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 +106,7 @@ impl AppContext { db, db_meta_data, state_manager, + chain_indexer, keystore, admin_jwt, snapshot_progress_tracker, @@ -189,15 +233,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 forest_car_db_dir(&self) -> &Path { + &self.forest_car_db_dir } - pub(crate) fn get_forest_car_db_dir(&self) -> PathBuf { - self.forest_car_db_dir.clone() + pub(crate) fn index_db_path(&self) -> &Path { + &self.index_db_path } } @@ -206,6 +255,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 +267,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 +280,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 cea07105d515..cc06001ddaf3 100644 --- a/src/daemon/mod.rs +++ b/src/daemon/mod.rs @@ -135,7 +135,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?; } @@ -147,7 +147,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), @@ -165,6 +165,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().await + { + tracing::warn!("failed to populate chain index from snapshot: {e}"); + } } // If the snapshot progress state is not completed, @@ -191,7 +197,6 @@ async fn maybe_import_snapshot( }) .await??; } - Ok(()) } @@ -571,6 +576,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(); @@ -591,6 +597,7 @@ fn maybe_start_rpc_service( state_manager, keystore, mpool, + chain_indexer, bad_blocks, sync_status, eth_event_handler, @@ -718,43 +725,69 @@ fn maybe_start_indexer_service( && !opts.stateless && !ctx.state_manager.chain_config().is_devnet() { - 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 { - tracing::info!("Starting indexer service"); + // Old 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 { + tracing::info!("Starting indexer service"); - // Continuously listen for head changes - loop { - match head_changes_rx.recv().await { - Ok(changes) => { - for ts in changes.applies { - tracing::debug!("Indexing tipset {}", ts.key()); - let delegated_messages = chain_store - .headers_delegated_messages(ts.block_headers().iter())?; - chain_store.process_signed_messages(&delegated_messages)?; + // Continuously listen for head changes + loop { + match head_changes_rx.recv().await { + Ok(changes) => { + for ts in changes.applies { + tracing::debug!("Indexing tipset {}", ts.key()); + let delegated_messages = chain_store + .headers_delegated_messages(ts.block_headers().iter())?; + chain_store.process_signed_messages(&delegated_messages)?; + } } + Err(RecvError::Lagged(n)) => { + warn!("indexer service lagged: skipping {n} events") + } + Err(RecvError::Closed) => break Ok(()), } - Err(RecvError::Lagged(n)) => { - warn!("indexer service lagged: skipping {n} events") - } - Err(RecvError::Closed) => break Ok(()), } + }); + + // Run the collector only if chain indexer is enabled + if let Some(retention_epochs) = config.chain_indexer.gc_retention_epochs { + let chain_store = ctx.state_manager.chain_store().shallow_clone(); + let chain_config = ctx.state_manager.chain_config().clone(); + services.spawn(async move { + tracing::info!("Starting collector for eth_mappings"); + let mut collector = EthMappingCollector::new( + chain_store.db_owned(), + chain_config.eth_chain_id, + retention_epochs.into(), + ); + collector.run().await + }); } - }); + } - // Run the collector only if chain indexer is enabled - if let Some(retention_epochs) = config.chain_indexer.gc_retention_epochs { - let chain_store = ctx.state_manager.chain_store().shallow_clone(); - let chain_config = ctx.state_manager.chain_config().clone(); - services.spawn(async move { - tracing::info!("Starting collector for eth_mappings"); - let mut collector = EthMappingCollector::new( - chain_store.db_owned(), - chain_config.eth_chain_id, - retention_epochs.into(), - ); - 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 { + if let Err(e) = indexer.index_loop(head_changes_rx).await { + tracing::warn!("indexer stopped unexpectedly: {e}"); + } + Ok(()) + } + }); + services.spawn({ + let indexer = indexer.clone(); + async move { + indexer.gc_loop().await; + Ok(()) + } + }); + } } } } diff --git a/src/db/mod.rs b/src/db/mod.rs index bcaf31628049..575d74e75dce 100644 --- a/src/db/mod.rs +++ b/src/db/mod.rs @@ -29,6 +29,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/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 f4087a574a38..ea0f0d3df51b 100644 --- a/src/rpc/methods/chain.rs +++ b/src/rpc/methods/chain.rs @@ -87,6 +87,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 0598c3b86289..bf5cf24bb296 100644 --- a/src/rpc/methods/eth.rs +++ b/src/rpc/methods/eth.rs @@ -19,7 +19,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; @@ -3193,7 +3195,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, } @@ -3320,9 +3322,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 { @@ -3411,23 +3413,30 @@ 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 let Some(chain_indexer) = ctx.chain_indexer() { + chain_indexer + .get_events_for_filter( + IndexerEventFilter::try_from(pf)?, + ctx.eth_event_handler.max_filter_results, + ) + .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 23dc2a3fb1f1..d04b24396ffb 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; @@ -99,7 +98,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, @@ -701,8 +700,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( @@ -732,7 +731,7 @@ pub enum ParsedFilterTipsets { Key(TipsetKey), } -#[derive(Debug)] +#[derive(Debug, Clone)] pub struct ParsedFilter { pub(crate) tipsets: ParsedFilterTipsets, pub(crate) addresses: Vec
, diff --git a/src/rpc/methods/sync.rs b/src/rpc/methods/sync.rs index 35c761cb1001..777c8f31ccbd 100644 --- a/src/rpc/methods/sync.rs +++ b/src/rpc/methods/sync.rs @@ -234,6 +234,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 152e6c00ba67..6690f6e0b982 100644 --- a/src/rpc/mod.rs +++ b/src/rpc/mod.rs @@ -99,6 +99,7 @@ macro_rules! for_each_rpc_method { $callback!($crate::rpc::chain::ForestChainExportStatus); $callback!($crate::rpc::chain::ForestChainExportCancel); $callback!($crate::rpc::chain::ChainGetTipsetByParentState); + $callback!($crate::rpc::chain::ChainValidateIndex); // common vertical $callback!($crate::rpc::common::Session); @@ -498,6 +499,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, @@ -529,6 +531,10 @@ impl RPCState { self.state_manager.chain_config() } + pub fn chain_indexer(&self) -> Option<&Arc> { + self.chain_indexer.as_ref() + } + pub fn db(&self) -> &DbImpl { self.state_manager.db() } diff --git a/src/shim/address.rs b/src/shim/address.rs index 18799c70f2d4..b062b029c2f4 100644 --- a/src/shim/address.rs +++ b/src/shim/address.rs @@ -176,6 +176,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 c3ab92bec16e..b6dd17c5ac5c 100644 --- a/src/tool/offline_server/server.rs +++ b/src/tool/offline_server/server.rs @@ -108,6 +108,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 c49fe06488a0..b4eeddebcf1a 100644 --- a/src/tool/subcommands/api_cmd/api_compare_tests.rs +++ b/src/tool/subcommands/api_cmd/api_compare_tests.rs @@ -532,6 +532,7 @@ fn chain_tests_with_tipset( tipset: &Tipset, ) -> anyhow::Result> { let mut tests = vec![ + RpcTest::identity(ChainValidateIndex::request((tipset.epoch(), true))?), RpcTest::identity(ChainGetTipSetByHeight::request(( tipset.epoch(), Default::default(), diff --git a/src/tool/subcommands/api_cmd/generate_test_snapshot.rs b/src/tool/subcommands/api_cmd/generate_test_snapshot.rs index 91cc562df79f..9badb12b5d26 100644 --- a/src/tool/subcommands/api_cmd/generate_test_snapshot.rs +++ b/src/tool/subcommands/api_cmd/generate_test_snapshot.rs @@ -149,6 +149,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 0edf43b4aede..2111892b60d9 100644 --- a/src/tool/subcommands/api_cmd/test_snapshot.rs +++ b/src/tool/subcommands/api_cmd/test_snapshot.rs @@ -211,6 +211,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/utils/mod.rs b/src/utils/mod.rs index 5c0dce2a7f82..1c5fa9aead9b 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")