diff --git a/Cargo.lock b/Cargo.lock index 874d455ac4e..73699ab275a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -337,6 +337,18 @@ dependencies = [ "wait-timeout", ] +[[package]] +name = "async-broadcast" +version = "0.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "435a87a52755b8f27fcf321ac4f04b2802e337c8c4872923137471ec39c37532" +dependencies = [ + "event-listener", + "event-listener-strategy", + "futures-core", + "pin-project-lite", +] + [[package]] name = "async-channel" version = "2.5.0" @@ -3231,6 +3243,7 @@ dependencies = [ "arc-swap", "argon2", "assert_cmd", + "async-broadcast", "async-compression", "async-fs", "async-trait", diff --git a/Cargo.toml b/Cargo.toml index 0cbe27fd625..1f3cfa3684d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -37,6 +37,7 @@ anes = "0.2" anyhow = { workspace = true } arc-swap = "1" argon2 = "0.5" +async-broadcast = "0.7" async-compression = { version = "0.4", features = ["tokio", "zstd"] } async-fs = "2" async-trait = "0.1" diff --git a/src/chain/store/chain_store.rs b/src/chain/store/chain_store.rs index b68ac96d37a..e4bdf57e836 100644 --- a/src/chain/store/chain_store.rs +++ b/src/chain/store/chain_store.rs @@ -39,11 +39,11 @@ use serde::{Serialize, de::DeserializeOwned}; use std::{ num::NonZeroUsize, sync::atomic::{self, AtomicI64}, + time::Duration, }; -use tokio::sync::broadcast; -// A cap on the size of the future_sink -const SINK_CAP: usize = 200; +// Capacity of the head change broadcast channel +const HEAD_CHANGE_BROADCAST_CHANNEL_CAP: usize = 1000; // Assume a tipset has 5 blocks on average, we cache 1-day-worth of validated blocks. (5 * 2 * 60 * 24 = 14400) const VALIDATED_BLOCKS_CACHE_SIZE: NonZeroUsize = nonzero!(14400usize); @@ -81,8 +81,11 @@ pub type HeadChanges = PathChanges; /// epoch. This structure is thread-safe, and all caches are wrapped in a mutex /// to allow a consistent `ChainStore` to be shared across tasks. pub struct ChainStore { - /// Publisher for head change events - head_changes_tx: broadcast::Sender, + /// The non-blocking bridge sender for head change events. This is used to send head changes to the `head_changes_tx` channel. + head_changes_tx_bridge: flume::Sender, + + /// Inactive receiver for head change events. This is used to keep the channel alive even if there are no active subscribers. + head_changes_rx_inactive: Arc>, /// Heaviest tipset cache heaviest_tipset: Arc>, @@ -118,7 +121,8 @@ pub struct ChainStore { impl ShallowClone for ChainStore { fn shallow_clone(&self) -> Self { Self { - head_changes_tx: self.head_changes_tx.clone(), + head_changes_tx_bridge: self.head_changes_tx_bridge.clone(), + head_changes_rx_inactive: self.head_changes_rx_inactive.shallow_clone(), heaviest_tipset: self.heaviest_tipset.shallow_clone(), f3_finalized_tipset: self.f3_finalized_tipset.shallow_clone(), ec_calculator_finalized_epoch: self.ec_calculator_finalized_epoch.shallow_clone(), @@ -142,7 +146,39 @@ impl ChainStore { let db = db.into(); let genesis = genesis.into(); anyhow::ensure!(genesis.epoch() == 0, "genesis tipset must be at epoch 0"); - let (publisher, _) = broadcast::channel(SINK_CAP); + let (mut head_changes_tx, head_changes_rx) = + async_broadcast::broadcast(HEAD_CHANGE_BROADCAST_CHANNEL_CAP); + head_changes_tx.set_await_active(false); + head_changes_tx.set_overflow(false); // Disable overflow to not drop head changes + // Bridge the flume channel to the async_broadcast channel in a background task, + // it's unbounded to take the back pressure and not block `set_heaviest_head` + // in case `head_changes_tx` is unexpectedly full and blocked. + let (head_changes_tx_bridge, head_changes_rx_bridge) = flume::unbounded(); + // Warn if the broadcast channel is blocked (timed out after 1 second) + if tokio::runtime::Handle::try_current().is_ok() { + tokio::spawn(async move { + // The loop breaks when the flume channel is closed, which happens when the `ChainStore` is dropped. + while let Ok(m) = head_changes_rx_bridge.recv_async().await { + const TIMEOUT: Duration = Duration::from_secs(1); + if tokio::time::timeout(TIMEOUT, head_changes_tx.broadcast_direct(m)) + .await + .is_err() + { + error!( + "Head change broadcast channel is full. This indicates some consumers are not processing head changes fast enough." + ); + } + } + }); + } else { + cfg_if::cfg_if! { + if #[cfg(test)] { + warn!("ChainStore::new() is called outside of a Tokio runtime, head change broadcast channel is not working in this test"); + } else { + anyhow::bail!("ChainStore::new() must be called from within a Tokio runtime"); + } + } + } let head = if let Some(head_tsk) = db .heaviest_tipset_key() .context("failed to load head tipset key")? @@ -166,7 +202,8 @@ impl ChainStore { } })); Ok(Self { - head_changes_tx: publisher, + head_changes_tx_bridge, + head_changes_rx_inactive: head_changes_rx.deactivate().into(), chain_index, tipset_tracker: TipsetTracker::new(db, chain_config.clone()), heaviest_tipset, @@ -254,7 +291,8 @@ impl ChainStore { } let old_head = self.heaviest_tipset.swap(head.shallow_clone().into()); - if crate::utils::broadcast::has_subscribers(&self.head_changes_tx) { + // Only publish head changes when there are active subscribers and head is changed. + if self.head_changes_rx_inactive.receiver_count() > 0 && old_head.key() != head.key() { let changes = match crate::rpc::chain::chain_get_path(self, old_head.key(), head.key()) { Ok(changes) => changes, @@ -270,8 +308,17 @@ impl ChainStore { } } }; - if self.head_changes_tx.send(changes).is_err() { - debug!("did not publish changes, no active receivers"); + // Do not publish empty change and check active receivers again + if !changes.is_empty() + && self.head_changes_rx_inactive.receiver_count() > 0 + // Use an unbounded bridge channel to avoid blocking `set_heaviest_tipset`. + // Consider refactoring `set_heaviest_tipset` to be async and move the timeout logic here instead. + // Note: head change is only published after tipset validation and the 30s block delay + // should be sufficient for any consumer to catch up. If this blocks, the consumer logic + // needs to be fixed, e.g. spawning a non-blocking task the process the head changes. + && self.head_changes_tx_bridge.send(changes).is_err() + { + debug!("no active receivers"); } } @@ -345,8 +392,8 @@ impl ChainStore { } /// Subscribes head changes. - pub fn subscribe_head_changes(&self) -> broadcast::Receiver { - self.head_changes_tx.subscribe() + pub fn subscribe_head_changes(&self) -> async_broadcast::Receiver { + self.head_changes_rx_inactive.activate_cloned() } /// Returns a borrowed key-value store instance. @@ -843,9 +890,14 @@ pub fn get_parent_receipt( #[cfg(test)] mod tests { use super::*; - use crate::utils::multihash::prelude::*; - use crate::{blocks::RawBlockHeader, shim::address::Address}; + use crate::{ + blocks::{Chain4U, RawBlockHeader, chain4u}, + shim::address::Address, + utils::multihash::prelude::*, + }; use fvm_ipld_encoding::DAG_CBOR; + use std::time::Duration; + use tokio_util::task::AbortOnDropHandle; #[test] fn genesis_test() { @@ -960,4 +1012,49 @@ mod tests { ); assert!(inserter_executed.load(std::sync::atomic::Ordering::Relaxed)); } + + #[tokio::test] + async fn test_head_changes() { + let c4u = Chain4U::new(); + chain4u! { + in c4u; + t0 @ [genesis] + -> t1 @ [_b1_0] + -> t2 @ [_b2_0, _b2_1] + -> t3 @ [_b3_0] + -> t4 @ [_b4_1] + }; + + let db = DbImpl::from(Arc::new(crate::db::MemoryDB::default())); + let chain_config = Arc::new(ChainConfig::default()); + let cs = ChainStore::new(db, chain_config, genesis).unwrap(); + let mut rx = cs.subscribe_head_changes(); + + let handle = AbortOnDropHandle::new(tokio::spawn({ + let tipsets = vec![ + // This duplicate head should not be published + t0.shallow_clone(), + t1.shallow_clone(), + t2.shallow_clone(), + t3.shallow_clone(), + t4.shallow_clone(), + ]; + async move { + for ts in tipsets { + cs.set_heaviest_tipset(ts).unwrap(); + tokio::time::sleep(Duration::from_millis(100)).await; + } + } + })); + + // t0 is set as head in `ChainStore::new`, so the first published change is t1. + for ts in [&t1, &t2, &t3, &t4] { + let changes = rx.recv().await.unwrap(); + assert_eq!(changes.applies, vec![ts.shallow_clone()]); + } + + rx.try_recv().unwrap_err(); // no more messages + + handle.await.unwrap(); + } } diff --git a/src/daemon/db_util.rs b/src/daemon/db_util.rs index c32bf3b8cc2..83fe00361c0 100644 --- a/src/daemon/db_util.rs +++ b/src/daemon/db_util.rs @@ -17,8 +17,8 @@ use crate::state_manager::StateManager; use crate::utils::db::car_stream::CarStream; use crate::utils::io::EitherMmapOrRandomAccessFile; use crate::utils::net::{DownloadFileOption, download_to}; -use anyhow::{Context, bail}; -use futures::TryStreamExt; +use anyhow::bail; +use futures::TryStreamExt as _; use serde::{Deserialize, Serialize}; use std::sync::LazyLock; use std::sync::atomic::{AtomicI64, Ordering}; @@ -29,8 +29,7 @@ use std::{ time, }; use tokio::io::AsyncWriteExt; -use tokio::sync::broadcast::error::TryRecvError; -use tokio_util::sync::CancellationToken; +use tokio_util::{sync::CancellationToken, task::AbortOnDropHandle}; use tracing::{debug, info, warn}; use url::Url; use walkdir::WalkDir; @@ -726,7 +725,18 @@ pub async fn run_backfill( let cancel = guard.cancellation_token(); // Subscribe before the walk so applies/reverts that happen during it are observed. - let mut head_rx = state_manager.chain_store().subscribe_head_changes(); + // We should consume the channel ASAP to avoid blocking the sending part or dropping + // tipsets when overflow is enabled. + let mut head_changes_rx = state_manager.chain_store().subscribe_head_changes(); + let (tx, rx) = flume::unbounded(); + let bridge_handle = AbortOnDropHandle::new(tokio::spawn(async move { + while let Some(changes) = head_changes_rx.next().await { + if let Err(e) = tx.send_async(changes).await { + error!("failed to send head changes to backfill: {e}"); + break; + } + } + })); // Optionally clamp the start below finality to avoid indexing revert-prone near-head tipsets. let start_ts = if options.allow_near_head { @@ -807,33 +817,22 @@ pub async fn run_backfill( // Re-index tipsets applied during the walk so the canonical mapping wins. if !report.cancelled { + // drop the handle and the underlying channels + drop(bridge_handle); let mut extra: Vec<(SignedMessage, u64)> = vec![]; - loop { - match head_rx.try_recv() { - Ok(changes) => { - for ts in changes.applies { - if ts.epoch() >= lowest_epoch && ts.epoch() <= start_ts.epoch() { - tracing::debug!( - "re-indexing tipset @{} applied during backfill", - ts.epoch() - ); - if let Err(e) = - process_ts(&ts, state_manager, &mut extra, options.allow_recompute) - .await - { - tracing::warn!( - "failed to re-index applied tipset @{}: {e:#}", - ts.epoch() - ); - } - } + // Not using `rx.drain` to make sure tx is dropped and the channel is closed. + // Otherwise the loop below will block forever and timeout the CI tests. + let mut rx_stream = rx.into_stream(); + while let Some(changes) = rx_stream.next().await { + for ts in changes.applies { + if ts.epoch() >= lowest_epoch && ts.epoch() <= start_ts.epoch() { + tracing::debug!("re-indexing tipset @{} applied during backfill", ts.epoch()); + if let Err(e) = + process_ts(&ts, state_manager, &mut extra, options.allow_recompute).await + { + tracing::warn!("failed to re-index applied tipset @{}: {e:#}", ts.epoch()); } } - Err(TryRecvError::Empty) | Err(TryRecvError::Closed) => break, - Err(TryRecvError::Lagged(n)) => { - tracing::warn!("backfill head-change listener lagged: skipped {n} events"); - continue; - } } } if !extra.is_empty() { diff --git a/src/daemon/mod.rs b/src/daemon/mod.rs index 3b284e93e55..d933dae41fe 100644 --- a/src/daemon/mod.rs +++ b/src/daemon/mod.rs @@ -750,26 +750,18 @@ fn maybe_start_indexer_service( let chain_store = ctx.state_manager.chain_store().shallow_clone(); services.spawn(async move { tracing::info!("Starting indexer service"); - // Continuously listen for head changes - loop { - match head_changes_rx.recv().await { - Ok(changes) => { - for ts in changes.applies { - tracing::debug!("Indexing tipset {}", ts.key()); - let delegated_messages = chain_store - .headers_delegated_messages(ts.block_headers().iter())?; - // Head indexing writes the newest tipset, so use the blind-write - // fast path (no read-before-write timestamp comparison). - chain_store.process_signed_messages(&delegated_messages, false)?; - } - } - Err(RecvError::Lagged(n)) => { - warn!("indexer service lagged: skipping {n} events") - } - Err(RecvError::Closed) => break Ok(()), + while let Some(changes) = head_changes_rx.next().await { + for ts in changes.applies { + tracing::debug!("Indexing tipset {}", ts.key()); + let delegated_messages = + chain_store.headers_delegated_messages(ts.block_headers().iter())?; + // Head indexing writes the newest tipset, so use the blind-write + // fast path (no read-before-write timestamp comparison). + chain_store.process_signed_messages(&delegated_messages, false)?; } } + Ok(()) }); // Run the collector only if chain indexer is enabled diff --git a/src/lib.rs b/src/lib.rs index 3dc91c5a824..0b3224254e7 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -84,7 +84,7 @@ mod prelude { pub use ahash::{HashMapExt as _, HashSetExt as _}; pub use anyhow::Context as _; pub use cid::Cid; - pub use futures::FutureExt as _; + pub use futures::{FutureExt as _, StreamExt as _}; pub use itertools::Itertools as _; pub use std::{ops::Deref as _, sync::Arc}; pub use tracing::{debug, error, info, trace, warn}; diff --git a/src/message_pool/msgpool/msg_pool.rs b/src/message_pool/msgpool/msg_pool.rs index f1c43bd4336..424d107d7f7 100644 --- a/src/message_pool/msgpool/msg_pool.rs +++ b/src/message_pool/msgpool/msg_pool.rs @@ -43,7 +43,7 @@ use nonzero_ext::nonzero; use parking_lot::RwLock as SyncRwLock; use std::num::NonZeroUsize; use std::time::Duration; -use tokio::{sync::broadcast::error::RecvError, task::JoinSet, time::interval}; +use tokio::{task::JoinSet, time::interval}; use tracing::warn; /// Maximum size of a serialized message in bytes. Anti-DoS measure to keep @@ -543,21 +543,12 @@ where let mp = mp.shallow_clone(); let mut head_changes_rx = mp.api.subscribe_head_changes(); services.spawn(async move { - loop { - match head_changes_rx.recv().await { - Ok(HeadChanges { reverts, applies }) => { - if let Err(e) = mp.apply_head_change(reverts, applies).await { - tracing::warn!("Error changing head: {e}"); - } - } - Err(RecvError::Lagged(e)) => { - warn!("Head change subscriber lagged: skipping {e} events"); - } - Err(RecvError::Closed) => { - break Ok(()); - } + while let Some(HeadChanges { reverts, applies }) = head_changes_rx.next().await { + if let Err(e) = mp.apply_head_change(reverts, applies).await { + tracing::warn!("Error changing head: {e}"); } } + Ok(()) }); } diff --git a/src/message_pool/msgpool/provider.rs b/src/message_pool/msgpool/provider.rs index 396ec7dcb1c..b01ecb002a6 100644 --- a/src/message_pool/msgpool/provider.rs +++ b/src/message_pool/msgpool/provider.rs @@ -18,7 +18,6 @@ use crate::shim::{ }; use crate::utils::db::CborStoreExt; use auto_impl::auto_impl; -use tokio::sync::broadcast; /// Provider Trait. This trait will be used by the message pool to interact with /// some medium in order to do the operations that are listed below that are @@ -26,7 +25,7 @@ use tokio::sync::broadcast; #[auto_impl(Arc)] pub trait Provider { /// Update `Mpool`'s `cur_tipset` whenever there is a change to the provider - fn subscribe_head_changes(&self) -> broadcast::Receiver; + fn subscribe_head_changes(&self) -> async_broadcast::Receiver; /// Get the heaviest Tipset in the provider fn get_heaviest_tipset(&self) -> Tipset; /// Add a message to the `MpoolProvider`, return either Cid or Error @@ -65,7 +64,7 @@ pub trait Provider { } impl Provider for ChainStore { - fn subscribe_head_changes(&self) -> broadcast::Receiver { + fn subscribe_head_changes(&self) -> async_broadcast::Receiver { self.subscribe_head_changes() } diff --git a/src/message_pool/msgpool/test_provider.rs b/src/message_pool/msgpool/test_provider.rs index 2e9b2276703..14c43260ac8 100644 --- a/src/message_pool/msgpool/test_provider.rs +++ b/src/message_pool/msgpool/test_provider.rs @@ -3,9 +3,6 @@ //! Contains mock implementations for testing internal `MessagePool` APIs -use std::convert::TryFrom; -use std::sync::Arc; - use crate::blocks::{ CachingBlockHeader, ElectionProof, RawBlockHeader, Ticket, Tipset, TipsetKey, VRFProof, }; @@ -13,18 +10,19 @@ use crate::chain::HeadChanges; use crate::cid_collections::CidHashMap; use crate::message::{ChainMessage, MessageRead as _, SignedMessage}; use crate::message_pool::{Error, provider::Provider}; +use crate::prelude::*; use crate::shim::{address::Address, econ::TokenAmount, message::Message, state_tree::ActorState}; use ahash::HashMap; -use cid::Cid; use num::BigInt; use parking_lot::Mutex; -use tokio::sync::broadcast; +use std::convert::TryFrom; /// Structure used for creating a provider when writing tests involving message /// pool pub struct TestApi { pub inner: Mutex, - pub head_changes_tx: broadcast::Sender, + pub head_changes_tx: async_broadcast::Sender, + _head_changes_rx_inactive: async_broadcast::InactiveReceiver, } #[derive(Default)] @@ -42,13 +40,14 @@ pub struct TestApiInner { impl Default for TestApi { /// Create a new `TestApi` fn default() -> Self { - let (head_changes_tx, _) = broadcast::channel(1); + let (head_changes_tx, head_changes_rx) = async_broadcast::broadcast(100); TestApi { inner: Mutex::new(TestApiInner { max_actor_pending_messages: 20000, ..TestApiInner::default() }), head_changes_tx, + _head_changes_rx_inactive: head_changes_rx.deactivate(), } } } @@ -56,13 +55,14 @@ impl Default for TestApi { impl TestApi { /// Constructor for a `TestApi` with custom number of max pending messages pub fn with_max_actor_pending_messages(max_actor_pending_messages: u64) -> Self { - let (publisher, _) = broadcast::channel(1); + let (head_changes_tx, head_changes_rx) = async_broadcast::broadcast(100); TestApi { inner: Mutex::new(TestApiInner { max_actor_pending_messages, ..TestApiInner::default() }), - head_changes_tx: publisher, + head_changes_tx, + _head_changes_rx_inactive: head_changes_rx.deactivate(), } } @@ -84,7 +84,7 @@ impl TestApi { /// Set the heaviest tipset for `TestApi` pub fn set_heaviest_tipset(&self, ts: Tipset) { self.head_changes_tx - .send(HeadChanges { + .broadcast_blocking(HeadChanges { applies: vec![ts], reverts: vec![], }) @@ -140,8 +140,8 @@ impl TestApiInner { } impl Provider for TestApi { - fn subscribe_head_changes(&self) -> broadcast::Receiver { - self.head_changes_tx.subscribe() + fn subscribe_head_changes(&self) -> async_broadcast::Receiver { + self.head_changes_tx.new_receiver() } fn get_heaviest_tipset(&self) -> Tipset { diff --git a/src/rpc/methods/chain.rs b/src/rpc/methods/chain.rs index 6472547dc70..7128cd03732 100644 --- a/src/rpc/methods/chain.rs +++ b/src/rpc/methods/chain.rs @@ -1747,28 +1747,18 @@ pub(crate) fn chain_notify( tokio::spawn(async move { // Skip first message let _ = head_changes_rx.recv().await; - loop { - match head_changes_rx.recv().await { - Ok(changes) => { - let api_changes = changes - .into_change_vec() - .into_iter() - .map(From::from) - .collect(); - if sender.send(api_changes).is_err() { - tracing::info!("chain notify subscribers are all closed"); - break; - } - } - Err(tokio::sync::broadcast::error::RecvError::Closed) => { - tracing::info!("head changes channel closed"); - break; - } - Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => { - tracing::warn!("head changes channel lagged by {n} messages"); - } + while let Some(changes) = head_changes_rx.next().await { + let api_changes = changes + .into_change_vec() + .into_iter() + .map(From::from) + .collect(); + if sender.send(api_changes).is_err() { + tracing::info!("chain notify subscribers are all closed"); + break; } } + tracing::info!("head changes channel closed"); }); receiver } @@ -2050,6 +2040,10 @@ impl Clone for PathChanges { } impl PathChanges { + pub fn is_empty(&self) -> bool { + self.reverts.is_empty() && self.applies.is_empty() + } + pub fn into_change_vec(self) -> Vec> { let Self { reverts, applies } = self; reverts diff --git a/src/rpc/methods/eth/pubsub.rs b/src/rpc/methods/eth/pubsub.rs index d7cdc29f347..f562d33cf45 100644 --- a/src/rpc/methods/eth/pubsub.rs +++ b/src/rpc/methods/eth/pubsub.rs @@ -112,7 +112,7 @@ impl EthPubSubApiServer for EthPubSub { fn head_message_tipsets(ctx: &Arc) -> impl Stream + Send + use<> { let rx = ctx.chain_store().subscribe_head_changes(); let ctx = ctx.shallow_clone(); - subscription_stream(rx).flat_map(move |changes| { + rx.flat_map(move |changes| { let ctx = ctx.shallow_clone(); let items: Vec<_> = changes .applies @@ -154,7 +154,7 @@ fn spawn_new_heads(sink: SubscriptionSink, ctx: Arc) { /// Drives the shared logs feed for every chain head change, collects the Ethereum logs of the affected tipsets async fn run_logs_feed(ctx: Arc, feed: LogsFeed) { - let mut head_changes = subscription_stream(ctx.chain_store().subscribe_head_changes()); + let mut head_changes = ctx.chain_store().subscribe_head_changes(); while let Some(changes) = head_changes.next().await { // Collecting events is not free; skip the work entirely while no subscription is live. if feed.receiver_count() == 0 { diff --git a/src/state_manager/message_search.rs b/src/state_manager/message_search.rs index d55151eb557..5e836a41c84 100644 --- a/src/state_manager/message_search.rs +++ b/src/state_manager/message_search.rs @@ -8,9 +8,7 @@ use ahash::HashSet; use parking_lot::RwLock; use std::sync::OnceLock; use std::time::Duration; -use tokio::sync::broadcast::error::RecvError; use tokio_util::sync::CancellationToken; -use tracing::warn; /// Maximum allowed message confidence. const MAX_MESSAGE_CONFIDENCE: ChainEpoch = crate::shim::policy::policy_constants::CHAIN_FINALITY; @@ -317,73 +315,60 @@ impl StateManager { let sm = self.shallow_clone(); async move { let mut candidate: Option<(Tipset, Receipt)> = initial_candidate; - while !cancellation_token.is_cancelled() { - match head_changes_rx.recv().await { - Ok(head_changes) => { - for reverted_ts in head_changes.reverts { - reverted.write().insert(reverted_ts.key().clone()); - - if candidate - .as_ref() - .is_some_and(|(ts, _)| ts.key() == reverted_ts.key()) - { - candidate = None; - } - } - for applied_ts in head_changes.applies { - reverted.write().remove(applied_ts.key()); - - // Return if `search_back_candidate` meets confidence requirement - if let Some((candidate_ts, candidate_receipt)) = - search_back_candidate.get() - && confidence_reached( - applied_ts.epoch(), - candidate_ts.epoch(), - confidence, - ) - && !reverted.read().contains(candidate_ts.key()) - { - return Ok(( - candidate_ts.shallow_clone(), - candidate_receipt.clone(), - )); - } - - // Return if the candidate meets confidence requirement - if let Some((candidate_ts, _)) = &candidate - && confidence_reached( - applied_ts.epoch(), - candidate_ts.epoch(), - confidence, - ) - && let Some(candidate) = candidate - { - return Ok(candidate); - } - - let maybe_receipt = sm.tipset_executed_message( - &applied_ts, - &message, - allow_replaced.unwrap_or(true), - )?; - if let Some(receipt) = maybe_receipt { - if confidence == 0 { - // Return if there's no confidence requirement - return Ok((applied_ts, receipt)); - } else { - // Otherwise set it as candidate - candidate = Some((applied_ts, receipt)); - } - } - } + while !cancellation_token.is_cancelled() + && let Some(head_changes) = head_changes_rx.next().await + { + for reverted_ts in head_changes.reverts { + reverted.write().insert(reverted_ts.key().clone()); + + if candidate + .as_ref() + .is_some_and(|(ts, _)| ts.key() == reverted_ts.key()) + { + candidate = None; + } + } + for applied_ts in head_changes.applies { + reverted.write().remove(applied_ts.key()); + + // Return if `search_back_candidate` meets confidence requirement + if let Some((candidate_ts, candidate_receipt)) = search_back_candidate.get() + && confidence_reached( + applied_ts.epoch(), + candidate_ts.epoch(), + confidence, + ) + && !reverted.read().contains(candidate_ts.key()) + { + return Ok((candidate_ts.shallow_clone(), candidate_receipt.clone())); } - Err(RecvError::Lagged(i)) => { - warn!( - "wait for message head change subscriber lagged, skipped {} events", - i - ); + + // Return if the candidate meets confidence requirement + if let Some((candidate_ts, _)) = &candidate + && confidence_reached( + applied_ts.epoch(), + candidate_ts.epoch(), + confidence, + ) + && let Some(candidate) = candidate + { + return Ok(candidate); + } + + let maybe_receipt = sm.tipset_executed_message( + &applied_ts, + &message, + allow_replaced.unwrap_or(true), + )?; + if let Some(receipt) = maybe_receipt { + if confidence == 0 { + // Return if there's no confidence requirement + return Ok((applied_ts, receipt)); + } else { + // Otherwise set it as candidate + candidate = Some((applied_ts, receipt)); + } } - Err(RecvError::Closed) => break, } } Err(Error::other("cancelled"))