Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat(ethexe): impl headers stream resubscription for observer service #4481

Merged
merged 2 commits into from
Jan 28, 2025
Merged

feat(ethexe): impl headers stream resubscription for observer service #4481

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
780be90
impl it!
breathx Jan 28, 2025
7d2306d
to-do, to-do, to-do, todooooo
breathx Jan 28, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
95 changes: 55 additions & 40 deletions ethexe/observer/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,7 @@ use ethexe_signer::Address;
use futures::{
future::BoxFuture,
stream::{FusedStream, FuturesUnordered},
Stream, StreamExt,
FutureExt, Stream, StreamExt,
};
use gprimitives::{CodeId, H256};
use parity_scale_codec::{Decode, Encode};
Expand All @@ -56,7 +56,7 @@ pub use observer::*;
pub use query::*;

type BlobDownloadFuture = BoxFuture<'static, Result<(CodeId, Vec<u8>)>>;
type BlocksStream = dyn Stream<Item = (H256, BlockHeader, Vec<BlockEvent>)> + Send;
type BlockFuture = BoxFuture<'static, Result<(H256, BlockHeader, Vec<BlockEvent>)>>;

#[derive(Clone, Debug)]
pub struct EthereumConfig {
Expand All @@ -81,19 +81,44 @@ pub struct ObserverService {

last_block_number: u32,

stream: Pin<Box<BlocksStream>>,
headers_stream: SubscriptionStream<Header>,
block_future: Option<BlockFuture>,

codes_futures: FuturesUnordered<BlobDownloadFuture>,
}

impl Stream for ObserverService {
type Item = Result<ObserverEvent>;

fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
if let Poll::Ready(Some((hash, header, events))) = self.stream.poll_next_unpin(cx) {
let event = Ok(self.handle_stream_next(hash, header, events));
if self.block_future.is_none() {
if let Poll::Ready(res) = self.headers_stream.poll_next_unpin(cx) {
if let Some(header) = res {
self.block_future = Some(Box::pin(Self::get_block(
header,
self.provider.clone(),
self.router,
)));
} else {
// TODO: test resubscribe works.
log::warn!("Alloy headers stream ended, resubscribing");
self.headers_stream = self.subscription.resubscribe().into_stream();
cx.waker().wake_by_ref();
return Poll::Pending;
}
}
}

return Poll::Ready(Some(event));
};
if let Some(fut) = self.block_future.as_mut() {
if let Poll::Ready(res) = fut.poll_unpin(cx) {
let event =
res.map(|(hash, header, events)| self.handle_stream_next(hash, header, events));

self.block_future = None;

return Poll::Ready(Some(event));
}
}

if let Poll::Ready(Some(res)) = self.codes_futures.poll_next_unpin(cx) {
let event = res.map(|(code_id, code)| ObserverEvent::Blob { code_id, code });
Expand Down Expand Up @@ -136,21 +161,16 @@ impl ObserverService {
.await
.context("failed to subscribe blocks")?;

let blocks_stream = subscription.resubscribe().into_stream();

let stream = Box::pin(Self::events_all(
blocks_stream,
provider.clone(),
config.router_address,
));
let headers_stream = subscription.resubscribe().into_stream();

Ok(Self {
blobs,
provider,
subscription,
router: config.router_address,
last_block_number: 0,
stream,
headers_stream,
block_future: None,
codes_futures: FuturesUnordered::new(),
})
}
Expand All @@ -175,29 +195,25 @@ impl ObserverService {
)));
}

fn events_all(
mut stream: SubscriptionStream<Header>,
async fn get_block(
header: Header,
provider: Provider,
router: Address,
) -> impl Stream<Item = (H256, BlockHeader, Vec<BlockEvent>)> {
async_stream::stream! {
while let Some(header) = stream.next().await {
let hash = (*header.hash).into();
let parent_hash = (*header.parent_hash).into();
let block_number = header.number as u32;
let block_timestamp = header.timestamp;

let header = BlockHeader {
height: block_number,
timestamp: block_timestamp,
parent_hash,
};

let events = read_block_events(hash, &provider, router.0.into()).await.unwrap();

yield (hash, header, events);
}
}
) -> Result<(H256, BlockHeader, Vec<BlockEvent>)> {
let hash = (*header.hash).into();
let parent_hash = (*header.parent_hash).into();
let block_number = header.number as u32;
let block_timestamp = header.timestamp;

let header = BlockHeader {
height: block_number,
timestamp: block_timestamp,
parent_hash,
};

read_block_events(hash, &provider, router.0.into())
.await
.map(|events| (hash, header, events))
}

fn handle_stream_next(
Expand Down Expand Up @@ -231,17 +247,16 @@ impl ObserverService {
impl Clone for ObserverService {
fn clone(&self) -> Self {
let subscription = self.subscription.resubscribe();
let stream = subscription.resubscribe().into_stream();

let stream = Self::events_all(stream, self.provider.clone(), self.router);
let headers_stream = subscription.resubscribe().into_stream();

Self {
blobs: self.blobs.clone(),
provider: self.provider.clone(),
subscription,
router: self.router,
last_block_number: self.last_block_number,
stream: Box::pin(stream),
headers_stream,
block_future: None,
codes_futures: FuturesUnordered::new(),
}
}
Expand Down
Loading