feat: Provider::watch_blocks_from and Provider::watch_canonical_blocks_from

[0xForerunner]

• Adds watch_blocks_from and watch_canonical_blocks_from provider APIs for historical backfill + live tailing using polling.
• Streams yield request futures so callers can control concurrency via buffering and reuse transport retry layers.
• Includes unit tests for progression, recovery after errors, buffering, provider drop semantics, fixed block tags, and cursor-advance guard behavior.

We can now get a historical, concurrent, and reorg aware block stream with live tailing just like this:

    provider
        .watch_blocks_from(start_block)
        .poll_interval(Duration::from_millis(500))
        .hashes()
        .block_tag(BlockNumberOrTag::Latest)
        .canonical()
        .max_reorg_depth(16)
        .rpc_concurrency(rpc_concurrency)
        .into_stream()
        // Concurrently process canonical events.
        .for_each_concurrent(process_concurrency, |result| async move {
            match result {
                Ok(CanonicalEvent::Added(block)) => {
                    let number = block.header().number();
                    let txs = block.transactions().len();
                    println!("processed block {number} ({txs} txs)");
                }
                Ok(CanonicalEvent::Removed(block)) => {
                    let number = block.header().number();
                    println!("reorged block {number}");
                }
                Err(err) => eprintln!("failed to fetch block: {err}"),
            }
        })
        .await;

PR Checklist

• Added Tests
• Added Documentation
• Breaking changes

[0xForerunner]

I've updated this PR to remove watch_logs_from, because I think without reorg handling it's not useful. Sticking with watch_blocks_from and watch_canonical_blocks_from.

[0xForerunner]

@DaniPopes @mattsse Would love a review on this when you've got a minute!

[mattsse]

this feature makes sense and I wanted something like this in the past.

because I believe we eventually want some additional logic on this, I really want this to be an actual stream type because maitnaining ~100 of stream! code is very difficult

[mattsse]

I want named types for this, because eventually we always need extra logic, and this isnt really easy to maintain

[0xForerunner]

Do you want a dedicated type here for the Stream, the future, or both?

[mattsse]

I'd much prefer if we have a dedicated stream impl for this

[0xForerunner]

Works for me! Are you okay with futures::stream::unfold, or do you prefer a direct impl Stream

[0xForerunner]

Maybe you could let me know what you'd like the function signature to look like for WatchBlocksFrom::into_stream, so I can make sure I make the right types here.

[0xForerunner]

@mattsse I've switched to using manual impl Stream state machines. I am still using boxed Futures. If you would like me to switch over to named and manually impled futures lemme know! We can just name the one Future that's in the api, or if you want to get rid of all the boxed futures I can implement a state machine for each of these.

[0xForerunner]

@mattsse I've also created concrete types for the futures here, and gotten rid of the Boxing. Feel free to revert the last two commits if you preferred it before. Should be good to go for another review, Thanks!

[0xForerunner]

@mattsse @klkvr bumping!

[0xForerunner]

@mattsse bumping again!