feat: Run Scanner in a Separate Tokio Task #138

0xNeshi · 2025-10-23T11:49:51Z

Towards #73
Closes #120
Closes #118

avoids forcing lib users to manually wrap scanners in tokio tasks
better error bubbling DevEx
fixes flaky tests <- used to fail due to race conditions; with this PR, scanner logic becomes more deterministic

Additional changes:

remove redundant code
add additional macro helpers for stream assertions
test refactors <- use macro helpers

…mations

LeoPatOZ

Happy to merge (once conflicts are resolved ofc)

0xNeshi · 2025-10-27T12:09:44Z

src/block_range_scanner.rs

-    Unsubscribe {
-        response: oneshot::Sender<Result<(), ScannerError>>,
-    },
-    Shutdown {
-        response: oneshot::Sender<Result<(), ScannerError>>,
-    },


redundant, we'll add back if and when necessary

0xNeshi · 2025-10-27T12:31:42Z

src/block_range_scanner.rs


        info!(start_block = start_block_num, end_block = end_block_num, "Syncing historical data");

-        let sender = self.subscriber.take().ok_or_else(|| ScannerError::ServiceShutdown)?;


No more need to manually drop the sender (with .take() call), as this is done implicitly with sender being a locally scoped parameter.

0xNeshi · 2025-10-27T12:32:21Z

src/block_range_scanner.rs

-        let live_subscription_task = tokio::spawn(async move {
+        tokio::spawn(async move {


No need to manually stop live task, as stream_historical_blocks no longer returns errors

0xNeshi · 2025-10-27T12:36:27Z

src/block_range_scanner.rs

            Command::StreamLive { sender, block_confirmations, response } => {
-                self.ensure_no_subscriber()?;
                info!("Starting live stream");
-                self.subscriber = Some(sender);
-                let result = self.handle_live(block_confirmations).await;
+                let result = self.handle_live(block_confirmations, sender).await;
                let _ = response.send(result);


It is redundant to:

ensure no subscribers exist

set option field self.subscriber

immediately after .take() it to ensure the subscriber channel is closed after scanner is done.

pass "taken" subscriber into tokio tasks

Much more straightforward to pass the subscriber directly.

0xNeshi · 2025-10-27T12:38:08Z

src/block_range_scanner.rs

-        assert_next!(stream, None);
+        assert_closed!(stream);


intention is clearer

0xNeshi · 2025-10-27T12:38:46Z

src/block_range_scanner.rs

    }

    #[tokio::test]
-    async fn mixed_ranges_are_handled_correctly() -> anyhow::Result<()> {


unrealistic case, and anyway handled by other related tests

0xNeshi · 2025-10-27T12:39:15Z

src/block_range_scanner.rs

-    }
-
-    #[tokio::test]
-    async fn cutoff_at_zero_handles_all_ranges() -> anyhow::Result<()> {


Implicitly handled by buffered_messages_after_cutoff_are_all_passed

0xNeshi · 2025-10-27T12:39:36Z

src/error.rs

-    #[error("Only one subscriber allowed at a time")]
-    MultipleSubscribers,


Can no longer occur

0xNeshi · 2025-10-27T12:40:11Z

src/event_scanner/modes/common.rs

+    tokio::spawn(async move {
+        while let Some(message) = stream.next().await {
+            if let Err(err) = range_tx.send(message) {
+                error!(error = %err, "No receivers, stopping broadcast");
+                break;
+            }
        }
-    }
+    });


Block range broadcasting is no longer blocking

0xNeshi · 2025-10-27T12:41:16Z

tests/historic_mode/basic.rs

    let mut stream = setup.stream.take(expected_event_count);

-    tokio::spawn(async move { scanner.start().await });
+    scanner.start().await?;


Since scanner starts processes in other tokio tasks, initialization errors can now easily be returned immediately and there's no more need for this tokio spawn ritual

LeoPatOZ · 2025-10-28T07:47:17Z

src/block_range_scanner.rs

-                    .hash();
+                tip_hash = match provider.get_block_by_number(from.into()).await {
+                    Ok(block) => {
+                        block.expect("Chain should have the same height post-reorg").header().hash()


should we avoid string errors and have like unwrap_or / unwrap_or_else ScannerError ?

Will change to unwrap_or_else to be able to reference from value.

The reason we do not return ScannerError and instead panic is because this is an invariant that must always hold true. I.e. one of our core assumptions is that a block number with value from will exist pre- and post-reorg. If this doesn't happen, our assumptions are terribly wrong and the scanner crashes, indicating we need to fix our assumptions (and the scanner).

LeoPatOZ · 2025-10-28T08:04:21Z

tests/historic_to_live/basic.rs

    let mut stream = setup.stream;

-    tokio::spawn(async move { scanner.start().await });
+    contract.increase().send().await?.watch().await?;


why the unravel ?

it's clearer and shorter (less LoC)

LeoPatOZ · 2025-10-28T08:05:59Z

tests/historic_mode/basic.rs

Goes for all the test should we not use the new assert_next and increase macro for all the tests? Currently it seems inconsistent

assert_next: yes! didn't bother doing that in this PR, as it's too much work; only made the changes for: unreliable tests that I was debugging + new tests + tests where it was convenient to update. We'll make these updates either incrementally, or when we have nothing better to do (e.g. waiting on reviews, blocked otherwise etc.)

increase macro: in my other PR feat: Add 'Get Latest X Events and Switch to Live Streaming' Method #133 , I introduced a common fn increase, which we'll be able to use instead of the locally defined macros with the same name. The reason for a regular function instead of the macro is that fn is sufficient and is easier to debug

LeoPatOZ

nice very clean!

0xNeshi added 6 commits October 23, 2025 12:02

feat: run sync operations in separate threads

fa8122f

feat: run start ops in separate thread

c3771ce

test: refactor replays_historical_then_switches_to_live

95e56b5

feat: rewind runs in separate thread

f203e03

test: refactor live_mode_processes_all_blocks_respecting_block_confir…

d2c4440

…mations

fix: should stream range when end block == cutoff + test refactors

31f2a6a

0xNeshi self-assigned this Oct 23, 2025

0xNeshi requested review from LeoPatOZ and pepebndc as code owners October 23, 2025 11:49

LeoPatOZ previously approved these changes Oct 27, 2025

View reviewed changes

chore: merge with main

2dc61bd

0xNeshi dismissed LeoPatOZ’s stale review via 2dc61bd October 27, 2025 11:42

0xNeshi added 5 commits October 27, 2025 12:51

ref: clippy + typos

597a531

ref: remove unsubscribe & shutdown

ff66cda

ref: turn try_send helper into ext. try_stream method on Sender

4cd1303

test: fix examples

71a9113

test: start scanner in the same thread

ed29d6a

0xNeshi commented Oct 27, 2025

View reviewed changes

0xNeshi added 6 commits October 27, 2025 13:18

ref: rename sender -> subscriber

3c5ba0b

feat: remove multiple subscribers

743632d

ref: share TryStream

c42e49d

ref: rename EventListener's sender -> subscriber

0f2a296

ref: revert import order

b385423

ref: rename command handler's sender -> subscriber

2dc6aeb

0xNeshi commented Oct 27, 2025

View reviewed changes

0xNeshi added 3 commits October 27, 2025 13:33

ref: remove redundant sync info log

2e75d05

ref: rename subscriber -> sender

c68a58f

ref: clippy

6ae9712

0xNeshi commented Oct 27, 2025

View reviewed changes

0xNeshi requested a review from LeoPatOZ October 27, 2025 12:41

Merge branch 'main' into test-fixes

2a8c030

LeoPatOZ reviewed Oct 28, 2025

View reviewed changes

feat: update error message

6ef8685

0xNeshi requested a review from LeoPatOZ October 28, 2025 10:07

LeoPatOZ approved these changes Oct 28, 2025

View reviewed changes

0xNeshi merged commit 7a3ae85 into main Oct 28, 2025
13 checks passed

0xNeshi deleted the test-fixes branch October 28, 2025 12:02

github-actions bot locked and limited conversation to collaborators Oct 28, 2025


		info!(start_block = start_block_num, end_block = end_block_num, "Syncing historical data");

		let sender = self.subscriber.take().ok_or_else(\|\| ScannerError::ServiceShutdown)?;

		let live_subscription_task = tokio::spawn(async move {
		tokio::spawn(async move {

		#[error("Only one subscriber allowed at a time")]
		MultipleSubscribers,

feat: Run Scanner in a Separate Tokio Task #138

feat: Run Scanner in a Separate Tokio Task #138

Uh oh!

Conversation

0xNeshi commented Oct 23, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

LeoPatOZ left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

0xNeshi Oct 27, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

LeoPatOZ Oct 28, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

0xNeshi Oct 28, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

0xNeshi Oct 28, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

LeoPatOZ left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

0xNeshi commented Oct 23, 2025 •

edited

Loading

0xNeshi Oct 27, 2025 •

edited

Loading

LeoPatOZ Oct 28, 2025 •

edited

Loading

0xNeshi Oct 28, 2025 •

edited

Loading

0xNeshi Oct 28, 2025 •

edited

Loading