feat(storage): add Bloom filter based event lookups

CHANGELOG.md

@@ -13,6 +13,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - `starknet_getEvents` incorrectly evaluates empty sub-lists in key filters for pending events.
 
+### Changed
+
+- `starknet_getEvents` implementation is now using a much simpler implementation that no longer relies on SQLite queries. In general this leads to more consistent query times and a roughly 20% smaller database.
+  - The migration step involves computing Bloom filters for all blocks and dropping database tables no longer needed. This takes more than one hour for a mainnet database.
+  - The new `storage.event-bloom-filter-cache-size`, `rpc.get-events-max-blocks-to-scan` and `rpc.get-events-max-bloom-filters-to-load` arguments control some aspects of the algorithm.
+
 ## [0.10.3] - 2024-01-04
 
 ### Added

Cargo.toml

@@ -73,6 +73,9 @@ serde_with = "3.0.0"
 sha3 = "0.10"
 # This one needs to match the version used by blockifier
 starknet_api = "=0.6.0"
+test-log = { version = "0.2.12", default-features = false, features = [
+    "trace",
+] }
 thiserror = "1.0.48"
 tokio = "1.29.1"
 tracing = "0.1.37"

crates/crypto/src/algebra/field/felt.rs

@@ -95,6 +95,11 @@ impl Felt {
         &self.0
     }
 
+    /// Big-endian mutable representation of this [Felt].
+    pub fn as_mut_be_bytes(&mut self) -> &mut [u8; 32] {
+        &mut self.0
+    }
+
     /// Convenience function which extends [Felt::from_be_bytes] to work with slices.
     pub const fn from_be_slice(bytes: &[u8]) -> Result<Self, OverflowError> {
         if bytes.len() > 32 {

crates/executor/src/execution_state.rs

@@ -50,9 +50,9 @@ impl<'tx> ExecutionState<'tx> {
         if self.execute_on_parent_state && self.header.number.get() >= 10 {
             let block_number_whose_hash_becomes_available =
                 pathfinder_common::BlockNumber::new_or_panic(self.header.number.get() - 10);
-            let (_, block_hash) = self
+            let block_hash = self
                 .transaction
-                .block_id(block_number_whose_hash_becomes_available.into())?
+                .block_hash(block_number_whose_hash_becomes_available.into())?
                 .context("Getting historical block hash")?;
 
             tracing::trace!(%block_number_whose_hash_becomes_available, %block_hash, "Setting historical block hash");

crates/gateway-client/Cargo.toml

@@ -39,9 +39,7 @@ lazy_static = { workspace = true }
 pathfinder-crypto = { path = "../crypto" }
 pretty_assertions_sorted = { workspace = true }
 starknet-gateway-test-fixtures = { path = "../gateway-test-fixtures" }
-test-log = { version = "0.2.12", default-features = false, features = [
-    "trace",
-] }
+test-log = { workspace = true }
 tracing-subscriber = { workspace = true }
 
 [[test]]

crates/p2p/Cargo.toml

@@ -55,8 +55,6 @@ fake = { workspace = true }
 hex = { workspace = true }
 rand = { workspace = true }
 rstest = { workspace = true }
-test-log = { version = "0.2.12", default-features = false, features = [
-    "trace",
-] }
+test-log = { workspace = true }
 tokio = { version = "1.32.0", features = ["test-util"] }
 tracing-subscriber = { version = "0.3.17", features = ["env-filter"] }

crates/pathfinder/examples/feeder_gateway.rs

@@ -53,6 +53,7 @@ async fn serve() -> anyhow::Result<()> {
     let storage = pathfinder_storage::Storage::migrate(
         database_path.into(),
         pathfinder_storage::JournalMode::WAL,
+        1,
     )?
     .create_pool(NonZeroU32::new(10).unwrap())
     .unwrap();
@@ -295,10 +296,9 @@ fn get_chain(tx: &pathfinder_storage::Transaction<'_>) -> anyhow::Result<Chain>
     };
 
     let genesis_hash = tx
-        .block_id(BlockNumber::GENESIS.into())
+        .block_hash(BlockNumber::GENESIS.into())
         .unwrap()
-        .context("Getting genesis hash")?
-        .1;
+        .context("Getting genesis hash")?;
 
     let chain = match genesis_hash {
         MAINNET_GENESIS_HASH => Chain::Mainnet,

crates/pathfinder/examples/migrate_db.rs

@@ -20,7 +20,7 @@ fn main() {
     let size_before = std::fs::metadata(&path).expect("Path does not exist").len() as i64;
 
     let started_at = std::time::Instant::now();
-    pathfinder_storage::Storage::migrate(path.clone(), pathfinder_storage::JournalMode::WAL)
+    pathfinder_storage::Storage::migrate(path.clone(), pathfinder_storage::JournalMode::WAL, 1)
         .unwrap();
     let migrated_at = std::time::Instant::now();
 

crates/pathfinder/examples/re_execute.rs

@@ -29,7 +29,7 @@ fn main() -> anyhow::Result<()> {
     let n_cpus = std::thread::available_parallelism().unwrap().get();
 
     let database_path = std::env::args().nth(1).unwrap();
-    let storage = Storage::migrate(database_path.into(), JournalMode::WAL)?
+    let storage = Storage::migrate(database_path.into(), JournalMode::WAL, 1)?
         .create_pool(NonZeroU32::new(n_cpus as u32 * 2).unwrap())?;
     let mut db = storage
         .connection()

crates/pathfinder/examples/verify_block_hashes.rs

@@ -24,7 +24,7 @@ fn main() -> anyhow::Result<()> {
     };
 
     let database_path = std::env::args().nth(2).unwrap();
-    let storage = Storage::migrate(database_path.into(), JournalMode::WAL)?
+    let storage = Storage::migrate(database_path.into(), JournalMode::WAL, 1)?
         .create_pool(NonZeroU32::new(1).unwrap())
         .unwrap();
     let mut db = storage

crates/pathfinder/examples/verify_transaction_hashes.rs

@@ -29,7 +29,7 @@ fn main() -> anyhow::Result<()> {
 
     println!("Migrating database...");
 
-    let storage = Storage::migrate(database_path.into(), JournalMode::WAL)?
+    let storage = Storage::migrate(database_path.into(), JournalMode::WAL, 1)?
         .create_pool(NonZeroU32::new(1).unwrap())
         .unwrap();
     let mut db = storage

crates/pathfinder/src/bin/pathfinder/config.rs

@@ -205,6 +205,34 @@ This should only be enabled for debugging purposes as it adds substantial proces
         env = "PATHFINDER_GATEWAY_API_KEY"
     )]
     gateway_api_key: Option<String>,
+
+    #[arg(
+        long = "storage.event-bloom-filter-cache-size",
+        long_help = "The number of blocks whose event bloom filters are cached in memory. \
+            This cache speeds up event related RPC queries at the cost of using extra memory. \
+            Each cached filter takes 2 KiB of memory.",
+        env = "PATHFINDER_STORAGE_BLOOM_FILTER_CACHE_SIZE",
+        default_value = "524288"
+    )]
+    event_bloom_filter_cache_size: std::num::NonZeroUsize,
+
+    #[arg(
+        long = "rpc.get-events-max-blocks-to-scan",
+        long_help = "The number of blocks to scan for events when querying for events. \
+            This limit is used to prevent queries from taking too long.",
+        env = "PATHFINDER_RPC_GET_EVENTS_MAX_BLOCKS_TO_SCAN",
+        default_value = "500"
+    )]
+    get_events_max_blocks_to_scan: std::num::NonZeroUsize,
+
+    #[arg(
+        long = "rpc.get-events-max-bloom-filters-to-load",
+        long_help = "The number of Bloom filters to load for events when querying for events. \
+            This limit is used to prevent queries from taking too long.",
+        env = "PATHFINDER_RPC_GET_EVENTS_MAX_BLOOM_FILTERS_TO_LOAD",
+        default_value = "100000"
+    )]
+    get_events_max_bloom_filters_to_load: std::num::NonZeroUsize,
 }
 
 #[derive(clap::ValueEnum, Debug, Clone, Copy, PartialEq)]
@@ -461,6 +489,9 @@ pub struct Config {
     pub is_sync_enabled: bool,
     pub is_rpc_enabled: bool,
     pub gateway_api_key: Option<String>,
+    pub event_bloom_filter_cache_size: NonZeroUsize,
+    pub get_events_max_blocks_to_scan: NonZeroUsize,
+    pub get_events_max_bloom_filters_to_load: NonZeroUsize,
 }
 
 pub struct Ethereum {
@@ -631,6 +662,9 @@ impl Config {
             is_sync_enabled: cli.is_sync_enabled,
             is_rpc_enabled: cli.is_rpc_enabled,
             gateway_api_key: cli.gateway_api_key,
+            event_bloom_filter_cache_size: cli.event_bloom_filter_cache_size,
+            get_events_max_blocks_to_scan: cli.get_events_max_blocks_to_scan,
+            get_events_max_bloom_filters_to_load: cli.get_events_max_bloom_filters_to_load,
         }
     }
 }

crates/pathfinder/src/bin/pathfinder/main.rs

@@ -105,8 +105,13 @@ async fn async_main() -> anyhow::Result<()> {
     verify_networks(pathfinder_context.network, ethereum.chain)?;
 
     // Setup and verify database
-    let storage_manager =
-        Storage::migrate(pathfinder_context.database.clone(), config.sqlite_wal).unwrap();
+
+    let storage_manager = Storage::migrate(
+        pathfinder_context.database.clone(),
+        config.sqlite_wal,
+        config.event_bloom_filter_cache_size.get(),
+    )
+    .unwrap();
     let sync_storage = storage_manager
         // 5 is enough for normal sync operations, and then `available_parallelism` for
         // the rayon thread pool workers to use.
@@ -165,14 +170,20 @@ Hint: This is usually caused by exceeding the file descriptor limit of your syst
 
     let (tx_pending, rx_pending) = tokio::sync::watch::channel(Default::default());
 
+    let rpc_config = pathfinder_rpc::context::RpcConfig {
+        batch_concurrency_limit: config.rpc_batch_concurrency_limit,
+        get_events_max_blocks_to_scan: config.get_events_max_blocks_to_scan,
+        get_events_max_bloom_filters_to_load: config.get_events_max_bloom_filters_to_load,
+    };
+
     let context = pathfinder_rpc::context::RpcContext::new(
         rpc_storage,
         execution_storage,
         sync_state.clone(),
         pathfinder_context.network_id,
         pathfinder_context.gateway.clone(),
         rx_pending,
-        config.rpc_batch_concurrency_limit,
+        rpc_config,
     );
 
     let context = if config.websocket.enabled {

crates/pathfinder/src/p2p_network/sync_handlers.rs

@@ -251,7 +251,7 @@ fn get_transactions_for_block(
     block_number: BlockNumber,
     responses: &mut Vec<TransactionsResponse>,
 ) -> anyhow::Result<bool> {
-    let Some((_, block_hash)) = tx.block_id(block_number.into())? else {
+    let Some(block_hash) = tx.block_hash(block_number.into())? else {
         return Ok(false);
     };
 
@@ -286,7 +286,7 @@ fn get_receipts_for_block(
     block_number: BlockNumber,
     responses: &mut Vec<ReceiptsResponse>,
 ) -> anyhow::Result<bool> {
-    let Some((_, block_hash)) = tx.block_id(block_number.into())? else {
+    let Some(block_hash) = tx.block_hash(block_number.into())? else {
         return Ok(false);
     };
 
@@ -318,7 +318,7 @@ fn get_events_for_block(
     block_number: BlockNumber,
     responses: &mut Vec<EventsResponse>,
 ) -> anyhow::Result<bool> {
-    let Some((_, block_hash)) = tx.block_id(block_number.into())? else {
+    let Some(block_hash) = tx.block_hash(block_number.into())? else {
         return Ok(false);
     };
 

crates/pathfinder/src/state/sync.rs

@@ -671,9 +671,8 @@ async fn l1_update(
             .context("Insert update")?;
 
         let l2_hash = transaction
-            .block_id(update.block_number.into())
-            .context("Fetching block hash")?
-            .map(|(_, hash)| hash);
+            .block_hash(update.block_number.into())
+            .context("Fetching block hash")?;
 
         if let Some(l2_hash) = l2_hash {
             if l2_hash == update.block_hash {

crates/rpc/Cargo.toml

@@ -66,8 +66,6 @@ pathfinder-crypto = { path = "../crypto" }
 pretty_assertions_sorted = { workspace = true }
 rstest = { workspace = true }
 tempfile = "3.6"
-test-log = { version = "0.2.12", default-features = false, features = [
-    "trace",
-] }
+test-log = { workspace = true }
 tokio-tungstenite = "0.20"
 tracing-subscriber = { workspace = true }

crates/rpc/src/context.rs

@@ -11,6 +11,13 @@ use std::sync::Arc;
 type SequencerClient = starknet_gateway_client::Client;
 use tokio::sync::watch as tokio_watch;
 
+#[derive(Clone)]
+pub struct RpcConfig {
+    pub batch_concurrency_limit: NonZeroUsize,
+    pub get_events_max_blocks_to_scan: NonZeroUsize,
+    pub get_events_max_bloom_filters_to_load: NonZeroUsize,
+}
+
 #[derive(Clone)]
 pub struct RpcContext {
     pub storage: Storage,
@@ -21,7 +28,7 @@ pub struct RpcContext {
     pub eth_gas_price: gas_price::Cached,
     pub sequencer: SequencerClient,
     pub websocket: Option<WebsocketContext>,
-    pub batch_concurrency_limit: NonZeroUsize,
+    pub config: RpcConfig,
 }
 
 impl RpcContext {
@@ -32,7 +39,7 @@ impl RpcContext {
         chain_id: ChainId,
         sequencer: SequencerClient,
         pending_data: tokio_watch::Receiver<PendingData>,
-        batch_concurrency_limit: NonZeroUsize,
+        config: RpcConfig,
     ) -> Self {
         let pending_data = PendingWatcher::new(pending_data);
         Self {
@@ -44,7 +51,7 @@ impl RpcContext {
             eth_gas_price: gas_price::Cached::new(sequencer.clone()),
             sequencer,
             websocket: None,
-            batch_concurrency_limit,
+            config,
         }
     }
 
@@ -73,14 +80,20 @@ impl RpcContext {
         let sync_state = Arc::new(SyncState::default());
         let (_, rx) = tokio_watch::channel(Default::default());
 
+        let config = RpcConfig {
+            batch_concurrency_limit: NonZeroUsize::new(8).unwrap(),
+            get_events_max_blocks_to_scan: NonZeroUsize::new(1000).unwrap(),
+            get_events_max_bloom_filters_to_load: NonZeroUsize::new(1000).unwrap(),
+        };
+
         Self::new(
             storage.clone(),
             storage,
             sync_state,
             chain_id,
             sequencer.disable_retry_for_tests(),
             rx,
-            NonZeroUsize::new(8).unwrap(),
+            config,
         )
     }