From 688cf11de427490e59bdb33260a3b91cb60c3ab0 Mon Sep 17 00:00:00 2001 From: Bengineer Date: Sat, 4 Apr 2026 10:48:42 +0100 Subject: [PATCH 1/2] feat(pathfinder): add pathfinding example and update dependencies --- Cargo.lock | 5 +---- Cargo.toml | 5 +++++ crates/pathfinder/Cargo.toml | 1 - crates/pathfinder/src/fetcher.rs | 36 +++++++++++++++++++++++--------- examples/pathfinder/main.rs | 23 ++++++++++++++++++++ 5 files changed, 55 insertions(+), 15 deletions(-) create mode 100644 examples/pathfinder/main.rs diff --git a/Cargo.lock b/Cargo.lock index 5ce80744..28ea17b6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1031,7 +1031,6 @@ dependencies = [ [[package]] name = "dojo-introspect" version = "0.1.0" -source = "git+https://github.com/dojoengine/dojo-introspect?rev=aadc3c9#aadc3c980706596a4a083413813a0a3ab01fded7" dependencies = [ "async-trait", "introspect-rust-macros", @@ -2012,7 +2011,6 @@ dependencies = [ [[package]] name = "introspect-events" version = "0.1.2" -source = "git+https://github.com/cartridge-gg/introspect?rev=34e93c1#34e93c10c867c53c622cce03abb6431c9dae0ef5" dependencies = [ "cainome-cairo-serde", "introspect-types", @@ -2024,7 +2022,6 @@ dependencies = [ [[package]] name = "introspect-rust-macros" version = "0.1.0" -source = "git+https://github.com/cartridge-gg/introspect?rev=34e93c1#34e93c10c867c53c622cce03abb6431c9dae0ef5" dependencies = [ "paste", "proc-macro2", @@ -2036,7 +2033,6 @@ dependencies = [ [[package]] name = "introspect-types" version = "0.1.2" -source = "git+https://github.com/cartridge-gg/introspect?rev=34e93c1#34e93c10c867c53c622cce03abb6431c9dae0ef5" dependencies = [ "blake3", "convert_case", @@ -4874,6 +4870,7 @@ dependencies = [ "torii-introspect", "torii-introspect-postgres-sink", "torii-log-sink", + "torii-pathfinder", "torii-sql-sink", "torii-test-utils", "tower 0.5.3", diff --git a/Cargo.toml b/Cargo.toml index 7043f6e0..4c14aeb6 100644 --- a/Cargo.toml +++ b/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 diff --git a/crates/pathfinder/Cargo.toml b/crates/pathfinder/Cargo.toml index 92bad707..35a1306b 100644 --- a/crates/pathfinder/Cargo.toml +++ b/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"] diff --git a/crates/pathfinder/src/fetcher.rs b/crates/pathfinder/src/fetcher.rs index 8f64f919..2866aa0a 100644 --- a/crates/pathfinder/src/fetcher.rs +++ b/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,20 @@ 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) => { + if ctx.number == block.block_number { + break ctx; + } else if ctx.number > block.block_number { + return Err(PFError::block_context_missing(block.block_number)); + } else { + contexts.push(ctx.into()); + } + } + None => return Err(PFError::block_context_missing(block.block_number)), + } + }; for transaction in block.transactions { let transaction_hash = tx_hashes .next() diff --git a/examples/pathfinder/main.rs b/examples/pathfinder/main.rs new file mode 100644 index 00000000..57bf5370 --- /dev/null +++ b/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; + } +} From 958377f31110cfd96c1a2fd54c8eff47d0e25c33 Mon Sep 17 00:00:00 2001 From: Bengineer Date: Sat, 4 Apr 2026 11:02:57 +0100 Subject: [PATCH 2/2] refactor(fetcher): simplify block context retrieval logic in EventFetcher implementation --- crates/pathfinder/src/fetcher.rs | 14 ++++++-------- 1 file changed, 6 insertions(+), 8 deletions(-) diff --git a/crates/pathfinder/src/fetcher.rs b/crates/pathfinder/src/fetcher.rs index 2866aa0a..e3aebabf 100644 --- a/crates/pathfinder/src/fetcher.rs +++ b/crates/pathfinder/src/fetcher.rs @@ -154,15 +154,13 @@ impl EventFetcher for Connection { let block = BlockEvents::try_from(row)?; let ctx = loop { match ctx_iter.next() { - Some(ctx) => { - if ctx.number == block.block_number { - break ctx; - } else if ctx.number > block.block_number { - return Err(PFError::block_context_missing(block.block_number)); - } else { - contexts.push(ctx.into()); + 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)), } };