|Fixed empty block causing issue

Cargo.toml

@@ -185,6 +185,10 @@ path = "examples/introspect/simple.rs"
 name = "introspect_restart"
 path = "examples/introspect/restart.rs"
 
+[[example]]
+name = "pathfinder"
+path = "examples/pathfinder/main.rs"
+
 [dependencies]
 anyhow.workspace = true
 async-trait.workspace = true
@@ -245,6 +249,7 @@ torii-introspect.workspace = true
 torii-common.workspace = true
 torii-introspect-postgres-sink.workspace = true
 torii-test-utils.workspace = true
+torii-pathfinder.workspace = true
 
 
 # Example dependencies

crates/pathfinder/Cargo.toml

@@ -23,5 +23,4 @@ anyhow = { workspace = true, optional = true }
 workspace = true
 
 [features]
-default = ["etl"]
 etl = ["torii", "async-trait", "anyhow", "torii-starknet/starknet"]

crates/pathfinder/src/fetcher.rs

@@ -104,10 +104,18 @@ impl EventFetcher for Connection {
 
         for row in self.get_block_events_rows(from_block, to_block)? {
             let block: BlockEvents = row.try_into()?;
-            let block_hash = hash_rows
-                .next()
-                .map(|r| r.hash)
-                .ok_or_else(|| PFError::block_hash_missing(block.block_number))?;
+            let block_hash = loop {
+                match hash_rows.next() {
+                    Some(hash_row) => {
+                        if hash_row.number == block.block_number {
+                            break hash_row.hash;
+                        } else if hash_row.number > block.block_number {
+                            return Err(PFError::block_hash_missing(block.block_number));
+                        }
+                    }
+                    None => return Err(PFError::block_hash_missing(block.block_number)),
+                }
+            };
             for transaction in block.transactions {
                 let transaction_hash = tx_hashes
                     .next()
@@ -144,12 +152,18 @@ impl EventFetcher for Connection {
 
         for row in self.get_block_events_rows(from_block, to_block)? {
             let block = BlockEvents::try_from(row)?;
-            let ctx = ctx_iter
-                .next()
-                .ok_or_else(|| PFError::block_context_missing(block.block_number))?;
-            if ctx.number != block.block_number {
-                return Err(PFError::block_context_missing(block.block_number));
-            }
+            let ctx = loop {
+                match ctx_iter.next() {
+                    Some(ctx) => match ctx.number.cmp(&block.block_number) {
+                        std::cmp::Ordering::Equal => break ctx,
+                        std::cmp::Ordering::Less => contexts.push(ctx.into()),
+                        std::cmp::Ordering::Greater => {
+                            return Err(PFError::block_context_missing(block.block_number))
+                        }
+                    },
+                    None => return Err(PFError::block_context_missing(block.block_number)),
+                }
+            };
             for transaction in block.transactions {
                 let transaction_hash = tx_hashes
                     .next()

examples/pathfinder/main.rs

@@ -0,0 +1,23 @@
+use torii_pathfinder::{connect, EventFetcher};
+
+const DB_PATH: &str = "/mnt/store/mainnet.sqlite";
+
+const BATCH_SIZE: u64 = 10000;
+
+fn main() {
+    let conn = connect(DB_PATH).unwrap();
+    let mut current_block = 6000000;
+    for _ in 0..100 {
+        let (blocks, events) = conn
+            .get_emitted_events_with_context(current_block, current_block + BATCH_SIZE - 1)
+            .expect("failed to fetch events with context");
+        println!(
+            "Fetched {} blocks and {} events for blocks {} to {}",
+            blocks.len(),
+            events.len(),
+            current_block,
+            current_block + BATCH_SIZE - 1
+        );
+        current_block += BATCH_SIZE;
+    }
+}