From ae3a4b36b78c5b15546b676211fc7df4296b9fdc Mon Sep 17 00:00:00 2001 From: Hubert Bugaj Date: Thu, 13 Aug 2026 17:55:23 +0200 Subject: [PATCH 1/2] fix: head change publisher --- CHANGELOG.md | 2 + src/chain/store/chain_store.rs | 33 ++-- src/daemon/db_util.rs | 35 ++--- src/daemon/mod.rs | 27 ++-- src/message_pool/msgpool/msg_pool.rs | 22 +-- src/message_pool/msgpool/provider.rs | 5 +- src/message_pool/msgpool/test_provider.rs | 24 ++- src/rpc/methods/chain.rs | 148 ++++++++++++++---- src/rpc/methods/eth/pubsub.rs | 22 ++- src/state_manager/message_search.rs | 180 ++++++++++++++-------- src/utils/mod.rs | 1 + src/utils/publisher.rs | 128 +++++++++++++++ 12 files changed, 435 insertions(+), 192 deletions(-) create mode 100644 src/utils/publisher.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 806cff6aed1e..cd781b1b05fe 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -59,6 +59,8 @@ - [#7480](https://github.com/ChainSafe/forest/pull/7480): `forest-tool api serve --index-backfill-epochs N` now backfills N epochs below the chain head. Previously the chain head consumed one of the requested epochs, so only N-1 were indexed. +- [#7442](https://github.com/ChainSafe/forest/issues/7442): Fixed `ChainNotify` bug where the first notification was skipped; the behavior is now consistent with Lotus. + ## Forest v0.35.0 "Shravan" Non-mandatory release for all node operators. It includes some fixes and improvements, notably around state-related RPC. Note that this release contains breaking changes, so please read the changelog carefully before upgrading. diff --git a/src/chain/store/chain_store.rs b/src/chain/store/chain_store.rs index 98294e1234bc..b87a8a015abd 100644 --- a/src/chain/store/chain_store.rs +++ b/src/chain/store/chain_store.rs @@ -20,6 +20,7 @@ use crate::shim::{ }; use crate::state_manager::ExecutedTipset; use crate::utils::db::{BlockstoreExt, CborStoreExt}; +use crate::utils::publisher::Publisher; use crate::{ blocks::{CachingBlockHeader, Tipset, TipsetKey, TxMeta}, db::{DbImpl, EthMappingsStoreExt as _, HeaviestTipsetKeyProvider}, @@ -40,10 +41,6 @@ use std::{ num::NonZeroUsize, sync::atomic::{self, AtomicI64}, }; -use tokio::sync::broadcast; - -// A cap on the size of the future_sink -const SINK_CAP: usize = 200; // 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); @@ -82,7 +79,7 @@ pub type HeadChanges = PathChanges; /// to allow a consistent `ChainStore` to be shared across tasks. pub struct ChainStore { /// Publisher for head change events - head_changes_tx: broadcast::Sender, + head_changes: Publisher, /// Heaviest tipset cache heaviest_tipset: Arc>, @@ -118,7 +115,7 @@ pub struct ChainStore { impl ShallowClone for ChainStore { fn shallow_clone(&self) -> Self { Self { - head_changes_tx: self.head_changes_tx.clone(), + head_changes: self.head_changes.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 +139,6 @@ 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 head = if let Some(head_tsk) = db .heaviest_tipset_key() .context("failed to load head tipset key")? @@ -166,7 +162,7 @@ impl ChainStore { } })); Ok(Self { - head_changes_tx: publisher, + head_changes: Publisher::default(), chain_index, tipset_tracker: TipsetTracker::new(db, chain_config.clone()), heaviest_tipset, @@ -254,7 +250,7 @@ impl ChainStore { } let old_head = self.heaviest_tipset.swap(head.shallow_clone().into()); - if crate::utils::broadcast::has_subscribers(&self.head_changes_tx) { + if old_head.key() != head.key() && self.head_changes.has_subscribers() { let changes = match crate::rpc::chain::chain_get_path(self, old_head.key(), head.key()) { Ok(changes) => changes, @@ -270,8 +266,8 @@ impl ChainStore { } } }; - if self.head_changes_tx.send(changes).is_err() { - debug!("did not publish changes, no active receivers"); + if !changes.is_empty() { + self.head_changes.publish(changes); } } @@ -344,9 +340,18 @@ impl ChainStore { self.heaviest_tipset.load().as_ref().shallow_clone() } - /// Subscribes head changes. - pub fn subscribe_head_changes(&self) -> broadcast::Receiver { - self.head_changes_tx.subscribe() + /// Subscribes to head changes with an unbounded, lossless queue. Use this for + /// consumers that must not miss any head change (e.g. the chain/message indexers). + pub fn subscribe_head_changes(&self) -> flume::Receiver { + self.head_changes.subscribe() + } + + /// Subscribes to head changes with a bounded queue that drops events for this + /// subscriber alone once it falls `cap` behind. Use this for best-effort consumers + /// that must not be able to grow memory without bound, e.g. ones driven by an + /// untrusted RPC client's read rate. + pub fn subscribe_head_changes_bounded(&self, cap: usize) -> flume::Receiver { + self.head_changes.subscribe_bounded(cap) } /// Returns a borrowed key-value store instance. diff --git a/src/daemon/db_util.rs b/src/daemon/db_util.rs index 181ecf0a4f8d..887a2e15eaf3 100644 --- a/src/daemon/db_util.rs +++ b/src/daemon/db_util.rs @@ -29,7 +29,6 @@ use std::{ time, }; use tokio::io::AsyncWriteExt; -use tokio::sync::broadcast::error::TryRecvError; use tokio_util::sync::CancellationToken; use tracing::{debug, info, warn}; use url::Url; @@ -742,7 +741,7 @@ 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(); + let head_rx = state_manager.chain_store().subscribe_head_changes(); // Optionally clamp the start below finality to avoid indexing revert-prone near-head tipsets. let start_ts = if options.allow_near_head { @@ -824,32 +823,16 @@ pub async fn run_backfill( // Re-index tipsets applied during the walk so the canonical mapping wins. if !report.cancelled { 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() - ); - } - } + for changes in head_rx.try_iter() { + 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 1e879b1e3baa..154e40b82db9 100644 --- a/src/daemon/mod.rs +++ b/src/daemon/mod.rs @@ -743,30 +743,23 @@ fn maybe_start_indexer_service( && !opts.stateless && !ctx.state_manager.chain_config().is_devnet() { - let mut head_changes_rx = ctx.state_manager.chain_store().subscribe_head_changes(); + let head_changes_rx = ctx.state_manager.chain_store().subscribe_head_changes(); 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 Ok(changes) = head_changes_rx.recv_async().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/message_pool/msgpool/msg_pool.rs b/src/message_pool/msgpool/msg_pool.rs index f1c43bd43366..95c0697e0369 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 @@ -541,23 +541,15 @@ where // Reacts to new HeadChanges { let mp = mp.shallow_clone(); - let mut head_changes_rx = mp.api.subscribe_head_changes(); + let 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 Ok(HeadChanges { reverts, applies }) = head_changes_rx.recv_async().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 396ec7dcb1c4..65f20079cdeb 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) -> flume::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) -> flume::Receiver { self.subscribe_head_changes() } diff --git a/src/message_pool/msgpool/test_provider.rs b/src/message_pool/msgpool/test_provider.rs index 2e9b22767033..5c23bd55270d 100644 --- a/src/message_pool/msgpool/test_provider.rs +++ b/src/message_pool/msgpool/test_provider.rs @@ -14,17 +14,17 @@ use crate::cid_collections::CidHashMap; use crate::message::{ChainMessage, MessageRead as _, SignedMessage}; use crate::message_pool::{Error, provider::Provider}; use crate::shim::{address::Address, econ::TokenAmount, message::Message, state_tree::ActorState}; +use crate::utils::publisher::Publisher; use ahash::HashMap; use cid::Cid; use num::BigInt; use parking_lot::Mutex; -use tokio::sync::broadcast; /// 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: Publisher, } #[derive(Default)] @@ -42,13 +42,12 @@ pub struct TestApiInner { impl Default for TestApi { /// Create a new `TestApi` fn default() -> Self { - let (head_changes_tx, _) = broadcast::channel(1); TestApi { inner: Mutex::new(TestApiInner { max_actor_pending_messages: 20000, ..TestApiInner::default() }), - head_changes_tx, + head_changes: Publisher::default(), } } } @@ -56,13 +55,12 @@ 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); TestApi { inner: Mutex::new(TestApiInner { max_actor_pending_messages, ..TestApiInner::default() }), - head_changes_tx: publisher, + head_changes: Publisher::default(), } } @@ -83,12 +81,10 @@ impl TestApi { /// Set the heaviest tipset for `TestApi` pub fn set_heaviest_tipset(&self, ts: Tipset) { - self.head_changes_tx - .send(HeadChanges { - applies: vec![ts], - reverts: vec![], - }) - .unwrap(); + self.head_changes.publish(HeadChanges { + applies: vec![ts], + reverts: vec![], + }); } pub fn next_block(&self) -> CachingBlockHeader { @@ -140,8 +136,8 @@ impl TestApiInner { } impl Provider for TestApi { - fn subscribe_head_changes(&self) -> broadcast::Receiver { - self.head_changes_tx.subscribe() + fn subscribe_head_changes(&self) -> flume::Receiver { + self.head_changes.subscribe() } fn get_heaviest_tipset(&self) -> Tipset { diff --git a/src/rpc/methods/chain.rs b/src/rpc/methods/chain.rs index 963a3428fd4e..dd03bf009d12 100644 --- a/src/rpc/methods/chain.rs +++ b/src/rpc/methods/chain.rs @@ -1729,42 +1729,31 @@ pub(crate) fn chain_notify( _params: Params<'_>, data: &crate::rpc::RPCState, ) -> Subscriber> { + chain_notify_inner(data.chain_store()) +} + +fn chain_notify_inner(chain_store: &ChainStore) -> Subscriber> { let (sender, receiver) = broadcast::channel(HEAD_CHANNEL_CAPACITY); - // As soon as the channel is created, send the current tipset - let current = data.chain_store().heaviest_tipset(); - let (change, tipset) = ("current".into(), current); + // Subscribe before sampling the head, else a change landing in between is lost. + let head_changes_rx = chain_store.subscribe_head_changes(); + let current = chain_store.heaviest_tipset(); sender - .send(vec![ApiHeadChange { change, tipset }]) + .send(vec![ApiHeadChange { + change: "current".into(), + tipset: current, + }]) .expect("receiver is not dropped"); - let mut head_changes_rx = data.chain_store().subscribe_head_changes(); - 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 Ok(changes) = head_changes_rx.recv_async().await { + let api_changes = changes.into_change_vec().into_iter().map_into().collect(); + if sender.send(api_changes).is_err() { + tracing::info!("chain notify subscribers are all closed"); + break; } } + tracing::info!("head changes channel closed"); }); receiver } @@ -2046,6 +2035,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 @@ -2227,6 +2220,105 @@ mod tests { let _ = (a, c1); } + #[test] + fn head_changes_published_deduped_and_ordered() { + let cs = ChainStore::calibnet(); + let db = Chain4U::with_blockstore(cs.db_owned()); + chain4u! { + in db; + [_genesis = cs.genesis_block_header()] + -> [a] -> [b] -> [c, d] -> [e] + }; + + let rx = cs.subscribe_head_changes(); + + cs.set_heaviest_tipset(a.make_tipset()).unwrap(); + // Re-setting the same head must not publish a change. + cs.set_heaviest_tipset(a.make_tipset()).unwrap(); + cs.set_heaviest_tipset(b.make_tipset()).unwrap(); + cs.set_heaviest_tipset([c, d].make_tipset()).unwrap(); + cs.set_heaviest_tipset(e.make_tipset()).unwrap(); + + let drained = rx.try_iter().collect_vec(); + let applied = drained.iter().map(|c| c.applies.clone()).collect_vec(); + assert_eq!( + applied, + vec![ + vec![a.make_tipset()], + vec![b.make_tipset()], + vec![[c, d].make_tipset()], + vec![e.make_tipset()], + ] + ); + assert!(drained.iter().all(|c| c.reverts.is_empty())); + } + + #[test] + fn head_changes_publishes_reverts_on_reorg() { + let cs = ChainStore::calibnet(); + let db = Chain4U::with_blockstore(cs.db_owned()); + chain4u! { + in db; + [_genesis = cs.genesis_block_header()] + -> [a] -> [b1] + }; + chain4u! { + from [a] in db; + [b2] + }; + + let rx = cs.subscribe_head_changes(); + cs.set_heaviest_tipset(a.make_tipset()).unwrap(); + cs.set_heaviest_tipset(b1.make_tipset()).unwrap(); + cs.set_heaviest_tipset(b2.make_tipset()).unwrap(); // reorg b1 -> b2 + + let last = rx.try_iter().last().unwrap(); + assert_eq!(last.reverts, vec![b1.make_tipset()]); + assert_eq!(last.applies, vec![b2.make_tipset()]); + } + + #[tokio::test] + async fn chain_notify_delivers_every_apply_from_subscription() { + let cs = ChainStore::calibnet(); + let db = Chain4U::with_blockstore(cs.db_owned()); + chain4u! { + in db; + [_genesis = cs.genesis_block_header()] + -> [a] -> [b] -> [c] + }; + + let mut rx = chain_notify_inner(&cs); + + // First message is the current head. + let first = tokio::time::timeout(std::time::Duration::from_secs(1), rx.recv()) + .await + .unwrap() + .unwrap(); + assert_eq!(first.len(), 1); + assert_eq!(first[0].change, "current"); + + for ts in [&a, &b, &c] { + cs.set_heaviest_tipset(ts.make_tipset()).unwrap(); + } + + // Every applied tipset must be delivered, in order, with none dropped. + let mut applied = vec![]; + for _ in 0..3 { + match tokio::time::timeout(std::time::Duration::from_millis(500), rx.recv()).await { + Ok(Ok(msg)) => applied.extend( + msg.into_iter() + .filter(|c| c.change == "apply") + .map(|c| c.tipset), + ), + _ => break, + } + } + assert_eq!( + applied, + vec![a.make_tipset(), b.make_tipset(), c.make_tipset()] + ); + } + impl ChainStore { fn _load(genesis_car: &'static [u8], genesis_cid: Cid) -> Self { let db = Arc::new( diff --git a/src/rpc/methods/eth/pubsub.rs b/src/rpc/methods/eth/pubsub.rs index d7cdc29f3470..a8a221362bbd 100644 --- a/src/rpc/methods/eth/pubsub.rs +++ b/src/rpc/methods/eth/pubsub.rs @@ -76,6 +76,10 @@ use tokio::sync::broadcast; /// A cap on the number of in-flight per-tipset log batches in the shared logs feed. const LOGS_FEED_CAP: usize = 256; +/// A cap on head changes buffered per `newHeads` subscription. A client that reads slower +/// than blocks arrive drops the excess rather than growing memory without bound. +const NEW_HEADS_SUBSCRIBER_CAP: usize = 1000; + /// Sender half of the shared logs feed; see [`RPCState::eth_logs_feed`]. pub type LogsFeed = broadcast::Sender>>; @@ -108,11 +112,13 @@ impl EthPubSubApiServer for EthPubSub { } /// Stream of "message tipsets", the parent of each newly applied tipset. -/// Reverts are ignored; lagged events are dropped (and logged) by [`subscription_stream`]. +/// Reverts are ignored. fn head_message_tipsets(ctx: &Arc) -> impl Stream + Send + use<> { - let rx = ctx.chain_store().subscribe_head_changes(); + let rx = ctx + .chain_store() + .subscribe_head_changes_bounded(NEW_HEADS_SUBSCRIBER_CAP); let ctx = ctx.shallow_clone(); - subscription_stream(rx).flat_map(move |changes| { + rx.into_stream().flat_map(move |changes| { let ctx = ctx.shallow_clone(); let items: Vec<_> = changes .applies @@ -154,8 +160,8 @@ 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()); - while let Some(changes) = head_changes.next().await { + let head_changes = ctx.chain_store().subscribe_head_changes(); + while let Ok(changes) = head_changes.recv_async().await { // Collecting events is not free; skip the work entirely while no subscription is live. if feed.receiver_count() == 0 { continue; @@ -246,9 +252,9 @@ fn spawn_pending_transactions(sink: SubscriptionSink, ctx: Arc) { } /// Forward stream items to the subscription sink until the sink is closed, -/// the client disconnects, or the upstream stream ends. The stream is -/// expected to absorb upstream backpressure (e.g. `Lagged`) on its own; this -/// helper only cares about the sink side. +/// the client disconnects, or the upstream stream ends. The stream is expected to +/// absorb upstream backpressure on its own (by dropping events when the client reads +/// too slowly); this helper only cares about the sink side. async fn pipe_stream_to_sink(mut stream: S, sink: SubscriptionSink) where S: Stream + Unpin + Send, diff --git a/src/state_manager/message_search.rs b/src/state_manager/message_search.rs index 70f90422b39f..fc87bc05bdef 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; @@ -256,7 +254,7 @@ impl StateManager { // between sampling and subscribing cannot be missed. Otherwise a revert // of the sampled head could go unseen and a reverted receipt could be // released after `confidence` epochs. - let mut head_changes_rx = self.cs.subscribe_head_changes(); + let head_changes_rx = self.cs.subscribe_head_changes(); let current_ts = self.heaviest_tipset(); let maybe_message_receipt = self.tipset_executed_message(¤t_ts, &message, allow_replaced.unwrap_or(true))?; @@ -318,72 +316,60 @@ impl StateManager { 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)); - } - } - } + let Ok(head_changes) = head_changes_rx.recv_async().await else { + break; + }; + 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); } - Err(RecvError::Lagged(i)) => { - warn!( - "wait for message head change subscriber lagged, skipped {} events", - i - ); + + 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")) @@ -888,6 +874,66 @@ mod tests { assert!(receipt.exit_code().is_success()); } + /// A candidate seeded at the head must not be returned once a reorg reverts that tipset, + /// even after the new chain advances past the confidence window. + #[tokio::test] + async fn wait_for_message_reverted_candidate_is_not_returned() { + let db = Arc::new(MemoryDB::default()); + let message = message_with_nonce(5); + let msg_cid = db.put_cbor_default(&message).unwrap(); + + let root_before = state_root_with_sender_nonce(&db, 5); + let root_after = state_root_with_sender_nonce(&db, 6); + let messages = tx_meta(&db, msg_cid); + let receipts = receipts_root(&db); + let c4u = Chain4U::with_blockstore(db.clone()); + chain4u! { + in c4u; + [genesis = HeaderBuilder::new().with_timestamp(7777)] + -> [_e1 = HeaderBuilder::new().with_state_root(root_before)] + -> [_e2 = HeaderBuilder::new() + .with_state_root(root_before) + .with_messages(messages)] + -> exec @ [_e3 = HeaderBuilder::new() + .with_state_root(root_after) + .with_message_receipts(receipts)] + }; + // A competing fork from epoch 1 that never includes the message, advancing to epoch 5. + chain4u! { + from [_e1] in c4u; + [_f2 = HeaderBuilder::new().with_state_root(root_before)] + -> [_f3 = HeaderBuilder::new().with_state_root(root_before)] + -> [_f4 = HeaderBuilder::new().with_state_root(root_before)] + -> fork_head @ [_f5 = HeaderBuilder::new().with_state_root(root_before)] + }; + let state_manager = state_manager_with_head(db.clone(), genesis, exec); + + let token = CancellationToken::new(); + let mut wait = + Box::pin(state_manager.wait_for_message(msg_cid, 2, None, Some(true), &token)); + + // Seed the candidate at the head (epoch 3) and subscribe; confidence 2 is unmet. + assert!( + tokio::time::timeout(Duration::from_millis(100), &mut wait) + .await + .is_err() + ); + + // Reorg: exec is reverted onto a fork (epoch 5) that never executed the message. + state_manager + .chain_store() + .set_heaviest_tipset(fork_head.clone()) + .unwrap(); + + // The reverted candidate must not be released even though the fork is past confidence. + assert!( + tokio::time::timeout(Duration::from_millis(300), &mut wait) + .await + .is_err(), + "a reverted candidate must not be returned" + ); + } + #[rstest] #[case::zero_confidence(10, 10, 0, true)] #[case::exact_confidence(15, 10, 5, true)] diff --git a/src/utils/mod.rs b/src/utils/mod.rs index 035b52d8ca69..67a466d522f0 100644 --- a/src/utils/mod.rs +++ b/src/utils/mod.rs @@ -15,6 +15,7 @@ pub mod multihash; pub mod net; pub mod p2p; pub mod proofs_api; +pub mod publisher; pub mod rand; pub mod reqwest_resume; mod shallow_clone; diff --git a/src/utils/publisher.rs b/src/utils/publisher.rs new file mode 100644 index 000000000000..14dd11e1d382 --- /dev/null +++ b/src/utils/publisher.rs @@ -0,0 +1,128 @@ +// Copyright 2019-2026 ChainSafe Systems +// SPDX-License-Identifier: Apache-2.0, MIT + +use parking_lot::Mutex; +use std::sync::Arc; + +/// A non-blocking fan-out publisher. +/// +/// Each subscriber gets its own [`flume`] queue and cloning shares the subscriber +/// registry. [`Self::subscribe`] gives a subscriber an unbounded, lossless queue; +/// [`Self::subscribe_bounded`] gives a bounded queue that drops new events for that +/// subscriber alone once it is full (use it for best-effort consumers that must not be +/// able to grow memory without bound, e.g. ones fed by untrusted clients). Either way the +/// producer never blocks and a slow subscriber never stalls the others. +pub struct Publisher(Arc>>>); + +impl Clone for Publisher { + fn clone(&self) -> Self { + Self(self.0.clone()) + } +} + +impl Default for Publisher { + fn default() -> Self { + Self(Arc::new(Mutex::new(Vec::new()))) + } +} + +impl Publisher { + /// Registers a new subscriber with an unbounded, lossless queue and returns its receiver. + pub fn subscribe(&self) -> flume::Receiver { + let (tx, rx) = flume::unbounded(); + self.0.lock().push(tx); + rx + } + + /// Registers a new subscriber with a bounded queue of capacity `cap`. When the subscriber + /// falls `cap` events behind, the newest events are dropped for it alone (it keeps the + /// oldest `cap`; the producer and other subscribers are unaffected). + pub fn subscribe_bounded(&self, cap: usize) -> flume::Receiver { + let (tx, rx) = flume::bounded(cap); + self.0.lock().push(tx); + rx + } + + /// Delivers `msg` to every subscriber. Never blocks: for a bounded subscriber that is + /// full the event is dropped for that subscriber; a subscriber whose receiver is gone + /// is pruned. + pub fn publish(&self, msg: T) { + self.0.lock().retain(|tx| { + !matches!( + tx.try_send(msg.clone()), + Err(flume::TrySendError::Disconnected(_)) + ) + }); + } + + /// Cheap check for whether any subscriber is registered. Does not prune, so it may + /// briefly report `true` after the last receiver is gone (until the next [`Self::publish`] + /// prunes it), but never reports `false` while a live subscriber exists. + pub fn has_subscribers(&self) -> bool { + !self.0.lock().is_empty() + } +} + +#[cfg(test)] +mod tests { + use super::*; + use itertools::Itertools as _; + + #[test] + fn publisher_is_lossless_under_lag() { + let publisher = Publisher::default(); + let rx1 = publisher.subscribe(); + let rx2 = publisher.subscribe(); + + // Far more than any bounded channel would hold; nothing is drained meanwhile. + const N: u32 = 10_000; + for i in 0..N { + publisher.publish(i); + } + + for rx in [&rx1, &rx2] { + for expected in 0..N { + assert_eq!(rx.recv().unwrap(), expected); + } + assert!(rx.try_recv().is_err()); + } + } + + #[test] + fn publisher_prunes_dropped_subscribers() { + let publisher = Publisher::::default(); + let rx_live = publisher.subscribe(); + let rx_dead = publisher.subscribe(); + assert!(publisher.has_subscribers()); + + drop(rx_dead); + // Publishing prunes the dead sender while still delivering to the live one. + publisher.publish(7); + assert!(publisher.has_subscribers()); + assert_eq!(rx_live.recv().unwrap(), 7); + + drop(rx_live); + publisher.publish(8); + assert!(!publisher.has_subscribers()); + } + + #[test] + fn publisher_bounded_subscriber_drops_without_blocking_others() { + let publisher = Publisher::default(); + let unbounded = publisher.subscribe(); + let bounded = publisher.subscribe_bounded(2); + + // Publishing well past the bound must not block and must not affect the unbounded sub. + for i in 0..10 { + publisher.publish(i); + } + + // Bounded subscriber kept only up to its capacity; the excess was dropped for it alone. + let bounded_items = bounded.try_iter().collect_vec(); + assert_eq!(bounded_items, vec![0, 1]); + + // Unbounded subscriber still received everything, in order. + let unbounded_items = unbounded.try_iter().collect_vec(); + assert_eq!(unbounded_items, (0..10).collect_vec()); + } +} From 3d0cf21d405ff9e7a64e9812b22a118762fd83fe Mon Sep 17 00:00:00 2001 From: Hubert Bugaj Date: Fri, 14 Aug 2026 13:00:11 +0200 Subject: [PATCH 2/2] chore: enumify head change type --- src/rpc/methods/chain.rs | 21 +++++++++++++++------ 1 file changed, 15 insertions(+), 6 deletions(-) diff --git a/src/rpc/methods/chain.rs b/src/rpc/methods/chain.rs index dd03bf009d12..9d33aa8cee36 100644 --- a/src/rpc/methods/chain.rs +++ b/src/rpc/methods/chain.rs @@ -1740,7 +1740,7 @@ fn chain_notify_inner(chain_store: &ChainStore) -> Subscriber let current = chain_store.heaviest_tipset(); sender .send(vec![ApiHeadChange { - change: "current".into(), + change: HeadChangeType::Current, tipset: current, }]) .expect("receiver is not dropped"); @@ -1910,11 +1910,20 @@ pub struct ChainExportParams { } lotus_json_with_self!(ChainExportParams); +/// The kind of head change delivered by `Filecoin.ChainNotify`. +#[derive(PartialEq, Eq, Debug, Clone, Copy, Serialize, Deserialize, JsonSchema)] +#[serde(rename_all = "lowercase")] +pub enum HeadChangeType { + Current, + Apply, + Revert, +} + #[derive(PartialEq, Debug, Serialize, Deserialize, Clone, JsonSchema)] #[serde(rename_all = "PascalCase")] pub struct ApiHeadChange { #[serde(rename = "Type")] - pub change: String, + pub change: HeadChangeType, #[serde(rename = "Val", with = "crate::lotus_json")] #[schemars(with = "LotusJson")] pub tipset: Tipset, @@ -1925,11 +1934,11 @@ impl From for ApiHeadChange { fn from(change: HeadChange) -> Self { match change { HeadChange::Apply(tipset) => Self { - change: "apply".into(), + change: HeadChangeType::Apply, tipset, }, HeadChange::Revert(tipset) => Self { - change: "revert".into(), + change: HeadChangeType::Revert, tipset, }, } @@ -2295,7 +2304,7 @@ mod tests { .unwrap() .unwrap(); assert_eq!(first.len(), 1); - assert_eq!(first[0].change, "current"); + assert_eq!(first[0].change, HeadChangeType::Current); for ts in [&a, &b, &c] { cs.set_heaviest_tipset(ts.make_tipset()).unwrap(); @@ -2307,7 +2316,7 @@ mod tests { match tokio::time::timeout(std::time::Duration::from_millis(500), rx.recv()).await { Ok(Ok(msg)) => applied.extend( msg.into_iter() - .filter(|c| c.change == "apply") + .filter(|c| c.change == HeadChangeType::Apply) .map(|c| c.tipset), ), _ => break,