refactor(withdrawer): read from block subscription stream and get events on each block

[noot]

Summary

refactor the bridge withdrawer to read only from one block subscription stream instead of a block stream + event streams. then, when a new block is received, the events for that block are batched and sent to the submitter.

Background

this fixes race conditions between the block/event streams and ensures proper batching; ie. that one rollup block = one sequencer tx.

Changes

• refactor the bridge withdrawer to read only from one block subscription stream
• on startup, gets the events in each block from the last block submitted+1 up until the current block, and batches them by block and sends to submitter
• on receiving a new block, batches the events and sends them to the submitter

Testing

unit tests

[Fraser999]

Seems like in almost all places we're using Arc<Provider<Ws>> we can get away with just &Provider<Ws>. Looks like the only place we need the provider wrapped in an Arc is in IAstriaWithdrawer::new, and we can just create one inline there.

[noot]

unfortunately it needs to be in an Arc due to the loop here:

    loop {
        select! {
            () = shutdown_token.cancelled() => {
                info!("block watcher shutting down");
                return Ok(());
            }
            block = block_rx.next() => {
                if let Some(block) = block {
                    get_and_send_events_at_block(
                        provider.clone(),
                        contract_address,
                        block,
                        &converter,
                        &submitter_handle,
                    )
                    .await
                    .wrap_err("failed to get and send events at block")?;
                }
            }
        }
    }

have to clone it otherwise it gets moved in the previous loop iteration

[Fraser999]

You should be able to change get_and_send_events_at_block to take provider: &Provider<Ws> too though (and all the other places taking an Arc)?

[Fraser999]

Approved, but I still think wrapping Provider in an Arc is not required (except in just the single place indicated in the comment). Not a huge deal though - just a micro-optimization to avoid using atomic ref-counting :)