1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194
//! Canonical chain state notification trait and types.
use auto_impl::auto_impl;
use derive_more::{Deref, DerefMut};
use reth_execution_types::{BlockReceipts, Chain};
use reth_primitives::{SealedBlockWithSenders, SealedHeader};
use std::{
pin::Pin,
sync::Arc,
task::{ready, Context, Poll},
};
use tokio::sync::{broadcast, watch};
use tokio_stream::{
wrappers::{BroadcastStream, WatchStream},
Stream,
};
use tracing::debug;
/// Type alias for a receiver that receives [`CanonStateNotification`]
pub type CanonStateNotifications = broadcast::Receiver<CanonStateNotification>;
/// Type alias for a sender that sends [`CanonStateNotification`]
pub type CanonStateNotificationSender = broadcast::Sender<CanonStateNotification>;
/// A type that allows to register chain related event subscriptions.
#[auto_impl(&, Arc)]
pub trait CanonStateSubscriptions: Send + Sync {
/// Get notified when a new canonical chain was imported.
///
/// A canonical chain be one or more blocks, a reorg or a revert.
fn subscribe_to_canonical_state(&self) -> CanonStateNotifications;
/// Convenience method to get a stream of [`CanonStateNotification`].
fn canonical_state_stream(&self) -> CanonStateNotificationStream {
CanonStateNotificationStream {
st: BroadcastStream::new(self.subscribe_to_canonical_state()),
}
}
}
/// A Stream of [`CanonStateNotification`].
#[derive(Debug)]
#[pin_project::pin_project]
pub struct CanonStateNotificationStream {
#[pin]
st: BroadcastStream<CanonStateNotification>,
}
impl Stream for CanonStateNotificationStream {
type Item = CanonStateNotification;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
loop {
return match ready!(self.as_mut().project().st.poll_next(cx)) {
Some(Ok(notification)) => Poll::Ready(Some(notification)),
Some(Err(err)) => {
debug!(%err, "canonical state notification stream lagging behind");
continue
}
None => Poll::Ready(None),
}
}
}
}
/// A notification that is sent when a new block is imported, or an old block is reverted.
///
/// The notification contains at least one [`Chain`] with the imported segment. If some blocks were
/// reverted (e.g. during a reorg), the old chain is also returned.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum CanonStateNotification {
/// The canonical chain was extended.
Commit {
/// The newly added chain segment.
new: Arc<Chain>,
},
/// A chain segment was reverted or reorged.
///
/// - In the case of a reorg, the reverted blocks are present in `old`, and the new blocks are
/// present in `new`.
/// - In the case of a revert, the reverted blocks are present in `old`, and `new` is an empty
/// chain segment.
Reorg {
/// The chain segment that was reverted.
old: Arc<Chain>,
/// The chain segment that was added on top of the canonical chain, minus the reverted
/// blocks.
///
/// In the case of a revert, not a reorg, this chain segment is empty.
new: Arc<Chain>,
},
}
impl CanonStateNotification {
/// Get the chain segment that was reverted, if any.
pub fn reverted(&self) -> Option<Arc<Chain>> {
match self {
Self::Commit { .. } => None,
Self::Reorg { old, .. } => Some(old.clone()),
}
}
/// Get the newly imported chain segment, if any.
pub fn committed(&self) -> Arc<Chain> {
match self {
Self::Commit { new } | Self::Reorg { new, .. } => new.clone(),
}
}
/// Get the new tip of the chain.
///
/// Returns the new tip for [`Self::Reorg`] and [`Self::Commit`] variants which commit at least
/// 1 new block.
pub fn tip(&self) -> &SealedBlockWithSenders {
match self {
Self::Commit { new } | Self::Reorg { new, .. } => new.tip(),
}
}
/// Get receipts in the reverted and newly imported chain segments with their corresponding
/// block numbers and transaction hashes.
///
/// The boolean in the tuple (2nd element) denotes whether the receipt was from the reverted
/// chain segment.
pub fn block_receipts(&self) -> Vec<(BlockReceipts, bool)> {
let mut receipts = Vec::new();
// get old receipts
if let Some(old) = self.reverted() {
receipts
.extend(old.receipts_with_attachment().into_iter().map(|receipt| (receipt, true)));
}
// get new receipts
receipts.extend(
self.committed().receipts_with_attachment().into_iter().map(|receipt| (receipt, false)),
);
receipts
}
}
/// Wrapper around a broadcast receiver that receives fork choice notifications.
#[derive(Debug, Deref, DerefMut)]
pub struct ForkChoiceNotifications(pub watch::Receiver<Option<SealedHeader>>);
/// A trait that allows to register to fork choice related events
/// and get notified when a new fork choice is available.
pub trait ForkChoiceSubscriptions: Send + Sync {
/// Get notified when a new safe block of the chain is selected.
fn subscribe_safe_block(&self) -> ForkChoiceNotifications;
/// Get notified when a new finalized block of the chain is selected.
fn subscribe_finalized_block(&self) -> ForkChoiceNotifications;
/// Convenience method to get a stream of the new safe blocks of the chain.
fn safe_block_stream(&self) -> ForkChoiceStream<SealedHeader> {
ForkChoiceStream::<SealedHeader> { st: WatchStream::new(self.subscribe_safe_block().0) }
}
/// Convenience method to get a stream of the new finalized blocks of the chain.
fn finalized_block_stream(&self) -> ForkChoiceStream<SealedHeader> {
ForkChoiceStream::<SealedHeader> {
st: WatchStream::new(self.subscribe_finalized_block().0),
}
}
}
/// A stream for fork choice watch channels (pending, safe or finalized watchers)
#[derive(Debug)]
#[pin_project::pin_project]
pub struct ForkChoiceStream<T> {
#[pin]
st: WatchStream<Option<T>>,
}
impl<T: Clone + Sync + Send + 'static> ForkChoiceStream<T> {
/// Creates a new `ForkChoiceStream`
pub fn new(rx: watch::Receiver<Option<T>>) -> Self {
Self { st: WatchStream::new(rx) }
}
}
impl<T: Clone + Sync + Send + 'static> Stream for ForkChoiceStream<T> {
type Item = T;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
loop {
match ready!(self.as_mut().project().st.poll_next(cx)) {
Some(Some(notification)) => return Poll::Ready(Some(notification)),
Some(None) => continue,
None => return Poll::Ready(None),
}
}
}
}