feat: subscribe to finalized and safe headers #9402
Conversation
|
|
||
| /// A type that allows to register finalized block related event subscriptions | ||
| /// and get notified when a new finalized block header is available. | ||
| pub trait FinalizedBlockHeaderSubscription: Send + Sync { |
There was a problem hiding this comment.
let's combine safe and finalized into one trait
maybe we can come up with a name that isn't also tied to headers, essentially these are derived from forkchoice, so perhaps Forkchoice(Block)Subscriptions
| /// and get notified when a new finalized block header is available. | ||
| pub trait FinalizedBlockHeaderSubscription: Send + Sync { | ||
| /// Get notified when a new finalized block header is available. | ||
| fn subscribe_to_finalized_header(&self) -> broadcast::Receiver<SealedHeader>; |
There was a problem hiding this comment.
I'd like to roll a standalone wrapper type for this, because perhaps we want to put some more features into this,
I know we didn't do this for CanonStateNotifications but we should have
greged93
force-pushed
the
feat/finalize-safe-subscription
branch
from
7534153
to
f13dbbc
Compare
| @@ -139,3 +139,44 @@ impl CanonStateNotification { | |||
| receipts | |||
| } | |||
| } | |||
|
|
|||
| /// Wrapper around a broadcast receiver that receives fork choice notifications. | |||
| #[derive(Debug)] | |||
There was a problem hiding this comment.
can we please derive derive_more::Deref and derive_more::DerefMut for wrapper types
| Some(Err(err)) => { | ||
| debug!(%err, "finalized header notification stream lagging behind"); | ||
| continue | ||
| } |
There was a problem hiding this comment.
| Some(Err(err)) => { | |
| debug!(%err, "finalized header notification stream lagging behind"); | |
| continue | |
| } | |
| Some(Err(err)) => { | |
| debug!(target: "provider::fcu", | |
| %err, | |
| "finalized header notification stream lagging behind" | |
| ); | |
| cx.waker().wake_by_ref(); | |
| return Poll::Pending | |
| } |
if we don't return the error here
There was a problem hiding this comment.
should I also modify CanonStateNotificationStream?
There was a problem hiding this comment.
pushed without for now, if needed I can make a follow up PR
There was a problem hiding this comment.
yeah, please modify CanonStateNotificationStream in a new PR, you can ref this comment in the PR description
greged93
force-pushed
the
feat/finalize-safe-subscription
branch
from
951b89a
to
278a19f
Compare
| return match ready!(self.as_mut().project().st.poll_next(cx)) { | ||
| Some(Ok(notification)) => Poll::Ready(Some(notification)), | ||
| Some(Err(err)) => { | ||
| debug!(target: "provider::fcu", | ||
| %err, | ||
| "finalized header notification stream lagging behind" | ||
| ); | ||
|
|
||
| cx.waker().wake_by_ref(); | ||
| Poll::Pending | ||
| } | ||
| None => Poll::Ready(None), | ||
| }; |
There was a problem hiding this comment.
this should be in a loop then we don't need the manual waker call
There was a problem hiding this comment.
made this update from a comment from @emhane, afaiu it's better to remove the loop, call the waker and yield to the executor no? can't you otherwise end up with the stream taking up excessive resources if it keeps erroring?
There was a problem hiding this comment.
exactly, it can lead to starvation (for no need)
There was a problem hiding this comment.
remove the loop, call the waker and yield to the executor no?
this is effectively a loop with extra steps
| return match ready!(self.as_mut().project().st.poll_next(cx)) { | ||
| Some(Ok(notification)) => Poll::Ready(Some(notification)), | ||
| Some(Err(err)) => { | ||
| debug!(target: "provider::fcu", | ||
| %err, | ||
| "finalized header notification stream lagging behind" | ||
| ); | ||
|
|
||
| cx.waker().wake_by_ref(); | ||
| Poll::Pending | ||
| } | ||
| None => Poll::Ready(None), | ||
| }; |
There was a problem hiding this comment.
exactly, it can lead to starvation (for no need)
| @@ -139,3 +139,47 @@ impl CanonStateNotification { | |||
| receipts | |||
| } | |||
| } | |||
|
|
|||
| /// Wrapper around a broadcast receiver that receives fork choice notifications. | |||
| #[derive(Debug, derive_more::Deref, derive_more::DerefMut)] | |||
There was a problem hiding this comment.
Deref and DerefMut can be brought into scope in the prelude
Resolves #9269.