From e4de3e2c158806163f9a7ab12fade321bfe60908 Mon Sep 17 00:00:00 2001 From: hanabi1224 Date: Fri, 7 Aug 2026 14:40:03 +0800 Subject: [PATCH 01/12] fix: disable overflow in head change broadcast channel --- Cargo.lock | 13 +++++++++++ Cargo.toml | 1 + src/chain/store/chain_store.rs | 28 +++++++++++++++-------- src/daemon/db_util.rs | 9 ++++---- src/daemon/mod.rs | 7 +++--- src/message_pool/msgpool/msg_pool.rs | 9 ++++---- src/message_pool/msgpool/provider.rs | 5 ++-- src/message_pool/msgpool/test_provider.rs | 19 +++++++-------- src/rpc/methods/chain.rs | 7 +++--- src/rpc/methods/eth/pubsub.rs | 4 ++-- src/state_manager/message_search.rs | 12 ++++------ 11 files changed, 66 insertions(+), 48 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 874d455ac4ee..73699ab275a5 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 0cbe27fd6252..1f3cfa3684de 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 b68ac96d37aa..1ad9c0b8ad28 100644 --- a/src/chain/store/chain_store.rs +++ b/src/chain/store/chain_store.rs @@ -40,10 +40,9 @@ 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; +// 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); @@ -82,7 +81,10 @@ 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_tx: async_broadcast::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>, @@ -119,6 +121,7 @@ impl ShallowClone for ChainStore { fn shallow_clone(&self) -> Self { Self { head_changes_tx: self.head_changes_tx.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 +145,10 @@ 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 let head = if let Some(head_tsk) = db .heaviest_tipset_key() .context("failed to load head tipset key")? @@ -166,7 +172,8 @@ impl ChainStore { } })); Ok(Self { - head_changes_tx: publisher, + head_changes_tx, + head_changes_rx_inactive: head_changes_rx.deactivate().into(), chain_index, tipset_tracker: TipsetTracker::new(db, chain_config.clone()), heaviest_tipset, @@ -254,7 +261,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 if there are active subscribers. + if self.head_changes_tx.receiver_count() > 0 { let changes = match crate::rpc::chain::chain_get_path(self, old_head.key(), head.key()) { Ok(changes) => changes, @@ -270,7 +278,7 @@ impl ChainStore { } } }; - if self.head_changes_tx.send(changes).is_err() { + if self.head_changes_tx.broadcast_blocking(changes).is_err() { debug!("did not publish changes, no active receivers"); } } @@ -345,8 +353,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_tx.new_receiver() } /// Returns a borrowed key-value store instance. diff --git a/src/daemon/db_util.rs b/src/daemon/db_util.rs index c32bf3b8cc2c..e369f7e25039 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; @@ -829,9 +828,11 @@ pub async fn run_backfill( } } } - Err(TryRecvError::Empty) | Err(TryRecvError::Closed) => break, - Err(TryRecvError::Lagged(n)) => { - tracing::warn!("backfill head-change listener lagged: skipped {n} events"); + Err(async_broadcast::TryRecvError::Empty) + | Err(async_broadcast::TryRecvError::Closed) => break, + Err(async_broadcast::TryRecvError::Overflowed(n)) => { + // This is unexpected as overflow is disabled + error!("unexpected broadcast overflow {n}"); continue; } } diff --git a/src/daemon/mod.rs b/src/daemon/mod.rs index 3b284e93e555..488ba4c1f75c 100644 --- a/src/daemon/mod.rs +++ b/src/daemon/mod.rs @@ -764,10 +764,11 @@ fn maybe_start_indexer_service( chain_store.process_signed_messages(&delegated_messages, false)?; } } - Err(RecvError::Lagged(n)) => { - warn!("indexer service lagged: skipping {n} events") + Err(async_broadcast::RecvError::Overflowed(n)) => { + // This is unexpected as overflow is disabled + error!("unexpected broadcast overflow {n}"); } - Err(RecvError::Closed) => break Ok(()), + Err(async_broadcast::RecvError::Closed) => break Ok(()), } } }); diff --git a/src/message_pool/msgpool/msg_pool.rs b/src/message_pool/msgpool/msg_pool.rs index f1c43bd43366..762a3a36eeb8 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 @@ -550,10 +550,11 @@ where tracing::warn!("Error changing head: {e}"); } } - Err(RecvError::Lagged(e)) => { - warn!("Head change subscriber lagged: skipping {e} events"); + Err(async_broadcast::RecvError::Overflowed(n)) => { + // This is unexpected as overflow is disabled + error!("unexpected broadcast overflow {n}"); } - Err(RecvError::Closed) => { + Err(async_broadcast::RecvError::Closed) => { break Ok(()); } } diff --git a/src/message_pool/msgpool/provider.rs b/src/message_pool/msgpool/provider.rs index 396ec7dcb1c4..b01ecb002a6b 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 2e9b22767033..f696277de7a2 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,18 @@ 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, } #[derive(Default)] @@ -42,7 +39,7 @@ 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, _) = async_broadcast::broadcast(100); TestApi { inner: Mutex::new(TestApiInner { max_actor_pending_messages: 20000, @@ -56,7 +53,7 @@ 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 (publisher, _) = async_broadcast::broadcast(100); TestApi { inner: Mutex::new(TestApiInner { max_actor_pending_messages, @@ -84,7 +81,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 +137,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 6472547dc706..373062333bf9 100644 --- a/src/rpc/methods/chain.rs +++ b/src/rpc/methods/chain.rs @@ -1760,12 +1760,13 @@ pub(crate) fn chain_notify( break; } } - Err(tokio::sync::broadcast::error::RecvError::Closed) => { + Err(async_broadcast::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"); + Err(async_broadcast::RecvError::Overflowed(n)) => { + // This is unexpected as overflow is disabled + error!("unexpected broadcast overflow {n}"); } } } diff --git a/src/rpc/methods/eth/pubsub.rs b/src/rpc/methods/eth/pubsub.rs index d7cdc29f3470..f562d33cf451 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 d55151eb557d..2b71dd22d521 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; @@ -377,13 +375,11 @@ impl StateManager { } } } - Err(RecvError::Lagged(i)) => { - warn!( - "wait for message head change subscriber lagged, skipped {} events", - i - ); + Err(async_broadcast::RecvError::Overflowed(n)) => { + // This is unexpected as overflow is disabled + error!("unexpected broadcast overflow {n}"); } - Err(RecvError::Closed) => break, + Err(async_broadcast::RecvError::Closed) => break, } } Err(Error::other("cancelled")) From c8d8c7022f701bc001f432428d1547683df4ecd9 Mon Sep 17 00:00:00 2001 From: hanabi1224 Date: Fri, 7 Aug 2026 15:32:59 +0800 Subject: [PATCH 02/12] tests --- src/chain/store/chain_store.rs | 60 ++++++++++++++++++++++++++++++++-- src/rpc/methods/chain.rs | 11 +++++++ 2 files changed, 68 insertions(+), 3 deletions(-) diff --git a/src/chain/store/chain_store.rs b/src/chain/store/chain_store.rs index 1ad9c0b8ad28..b28b6ab637e7 100644 --- a/src/chain/store/chain_store.rs +++ b/src/chain/store/chain_store.rs @@ -278,7 +278,11 @@ impl ChainStore { } } }; - if self.head_changes_tx.broadcast_blocking(changes).is_err() { + // Do not publish empty change and check active receivers again + if !changes.is_empty() + && self.head_changes_tx.receiver_count() > 0 + && self.head_changes_tx.broadcast_blocking(changes).is_err() + { debug!("did not publish changes, no active receivers"); } } @@ -851,9 +855,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() { @@ -968,4 +977,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/rpc/methods/chain.rs b/src/rpc/methods/chain.rs index 373062333bf9..4cb1550e9bec 100644 --- a/src/rpc/methods/chain.rs +++ b/src/rpc/methods/chain.rs @@ -1139,6 +1139,13 @@ pub fn chain_get_path( from: &TipsetKey, to: &TipsetKey, ) -> anyhow::Result { + if from == to { + return Ok(PathChanges { + reverts: vec![], + applies: vec![], + }); + } + let finality = chain_store.chain_config().policy.chain_finality; let mut to_revert = chain_store .load_required_tipset_or_heaviest(from) @@ -2051,6 +2058,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 From caee8f70557bd0df8231f3334d4c9e8cbbc32594 Mon Sep 17 00:00:00 2001 From: hanabi1224 Date: Fri, 7 Aug 2026 15:53:20 +0800 Subject: [PATCH 03/12] simplify code with Stream API --- src/daemon/db_util.rs | 38 +++------ src/daemon/mod.rs | 27 +++---- src/lib.rs | 2 +- src/message_pool/msgpool/msg_pool.rs | 18 +---- src/rpc/methods/chain.rs | 31 +++----- src/state_manager/message_search.rs | 115 ++++++++++++--------------- 6 files changed, 86 insertions(+), 145 deletions(-) diff --git a/src/daemon/db_util.rs b/src/daemon/db_util.rs index e369f7e25039..bdf4eec1d808 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}; @@ -807,34 +807,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() - ); - } - } + while let Some(changes) = head_rx.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(async_broadcast::TryRecvError::Empty) - | Err(async_broadcast::TryRecvError::Closed) => break, - Err(async_broadcast::TryRecvError::Overflowed(n)) => { - // This is unexpected as overflow is disabled - error!("unexpected broadcast overflow {n}"); - continue; - } } } if !extra.is_empty() { diff --git a/src/daemon/mod.rs b/src/daemon/mod.rs index 488ba4c1f75c..d933dae41fe8 100644 --- a/src/daemon/mod.rs +++ b/src/daemon/mod.rs @@ -750,27 +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(async_broadcast::RecvError::Overflowed(n)) => { - // This is unexpected as overflow is disabled - error!("unexpected broadcast overflow {n}"); - } - Err(async_broadcast::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 3dc91c5a8244..0b3224254e7d 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 762a3a36eeb8..424d107d7f77 100644 --- a/src/message_pool/msgpool/msg_pool.rs +++ b/src/message_pool/msgpool/msg_pool.rs @@ -543,22 +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(async_broadcast::RecvError::Overflowed(n)) => { - // This is unexpected as overflow is disabled - error!("unexpected broadcast overflow {n}"); - } - Err(async_broadcast::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/rpc/methods/chain.rs b/src/rpc/methods/chain.rs index 4cb1550e9bec..dda48a0dd48f 100644 --- a/src/rpc/methods/chain.rs +++ b/src/rpc/methods/chain.rs @@ -1754,29 +1754,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(async_broadcast::RecvError::Closed) => { - tracing::info!("head changes channel closed"); - break; - } - Err(async_broadcast::RecvError::Overflowed(n)) => { - // This is unexpected as overflow is disabled - error!("unexpected broadcast overflow {n}"); - } + 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 } diff --git a/src/state_manager/message_search.rs b/src/state_manager/message_search.rs index 2b71dd22d521..5e836a41c846 100644 --- a/src/state_manager/message_search.rs +++ b/src/state_manager/message_search.rs @@ -315,71 +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(async_broadcast::RecvError::Overflowed(n)) => { - // This is unexpected as overflow is disabled - error!("unexpected broadcast overflow {n}"); + + // 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(async_broadcast::RecvError::Closed) => break, } } Err(Error::other("cancelled")) From 92502ce16a201310bbf0373e923f2e4ee17d81b7 Mon Sep 17 00:00:00 2001 From: hanabi1224 Date: Fri, 7 Aug 2026 15:59:48 +0800 Subject: [PATCH 04/12] fix --- src/chain/store/chain_store.rs | 4 ++-- src/rpc/methods/chain.rs | 7 ------- 2 files changed, 2 insertions(+), 9 deletions(-) diff --git a/src/chain/store/chain_store.rs b/src/chain/store/chain_store.rs index b28b6ab637e7..1d75bbfb6b58 100644 --- a/src/chain/store/chain_store.rs +++ b/src/chain/store/chain_store.rs @@ -261,8 +261,8 @@ impl ChainStore { } let old_head = self.heaviest_tipset.swap(head.shallow_clone().into()); - // Only publish head changes if there are active subscribers. - if self.head_changes_tx.receiver_count() > 0 { + // Only publish head changes when there are active subscribers and head is changed. + if self.head_changes_tx.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, diff --git a/src/rpc/methods/chain.rs b/src/rpc/methods/chain.rs index dda48a0dd48f..7128cd037325 100644 --- a/src/rpc/methods/chain.rs +++ b/src/rpc/methods/chain.rs @@ -1139,13 +1139,6 @@ pub fn chain_get_path( from: &TipsetKey, to: &TipsetKey, ) -> anyhow::Result { - if from == to { - return Ok(PathChanges { - reverts: vec![], - applies: vec![], - }); - } - let finality = chain_store.chain_config().policy.chain_finality; let mut to_revert = chain_store .load_required_tipset_or_heaviest(from) From cd20d45b72792579435aa2ab37656ac309c3282b Mon Sep 17 00:00:00 2001 From: hanabi1224 Date: Fri, 7 Aug 2026 16:08:59 +0800 Subject: [PATCH 05/12] fix ut --- src/message_pool/msgpool/test_provider.rs | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/src/message_pool/msgpool/test_provider.rs b/src/message_pool/msgpool/test_provider.rs index f696277de7a2..b82fd1187523 100644 --- a/src/message_pool/msgpool/test_provider.rs +++ b/src/message_pool/msgpool/test_provider.rs @@ -22,6 +22,7 @@ use std::convert::TryFrom; pub struct TestApi { pub inner: Mutex, pub head_changes_tx: async_broadcast::Sender, + head_changes_rx_inactive: async_broadcast::InactiveReceiver, } #[derive(Default)] @@ -39,13 +40,14 @@ pub struct TestApiInner { impl Default for TestApi { /// Create a new `TestApi` fn default() -> Self { - let (head_changes_tx, _) = async_broadcast::broadcast(100); + 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(), } } } @@ -53,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, _) = async_broadcast::broadcast(100); + 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(), } } From 5bf32ae6010cc7c5aebf13230b5a4fe5b3a613e4 Mon Sep 17 00:00:00 2001 From: hanabi1224 Date: Fri, 7 Aug 2026 16:12:48 +0800 Subject: [PATCH 06/12] fix --- src/chain/store/chain_store.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/src/chain/store/chain_store.rs b/src/chain/store/chain_store.rs index 1d75bbfb6b58..80bae6f9b13c 100644 --- a/src/chain/store/chain_store.rs +++ b/src/chain/store/chain_store.rs @@ -281,9 +281,12 @@ impl ChainStore { // Do not publish empty change and check active receivers again if !changes.is_empty() && self.head_changes_tx.receiver_count() > 0 + // 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.broadcast_blocking(changes).is_err() { - debug!("did not publish changes, no active receivers"); + debug!("no active receivers"); } } From 04f5a10bf5f87d7d9192e5238349a90ede3c002a Mon Sep 17 00:00:00 2001 From: hanabi1224 Date: Fri, 7 Aug 2026 16:45:53 +0800 Subject: [PATCH 07/12] unbounded back pressure channel --- src/chain/store/chain_store.rs | 40 +++++++++++++++++++++++++--------- src/daemon/db_util.rs | 4 ++-- 2 files changed, 32 insertions(+), 12 deletions(-) diff --git a/src/chain/store/chain_store.rs b/src/chain/store/chain_store.rs index 80bae6f9b13c..a218903d9e48 100644 --- a/src/chain/store/chain_store.rs +++ b/src/chain/store/chain_store.rs @@ -39,6 +39,7 @@ use serde::{Serialize, de::DeserializeOwned}; use std::{ num::NonZeroUsize, sync::atomic::{self, AtomicI64}, + time::Duration, }; // Capacity of the head change broadcast channel @@ -80,8 +81,8 @@ 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: async_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>, @@ -120,7 +121,7 @@ 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(), @@ -149,6 +150,24 @@ impl ChainStore { 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) + tokio::spawn(async move { + 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." + ); + } + } + }); let head = if let Some(head_tsk) = db .heaviest_tipset_key() .context("failed to load head tipset key")? @@ -172,7 +191,7 @@ impl ChainStore { } })); Ok(Self { - head_changes_tx, + head_changes_tx_bridge, head_changes_rx_inactive: head_changes_rx.deactivate().into(), chain_index, tipset_tracker: TipsetTracker::new(db, chain_config.clone()), @@ -262,7 +281,7 @@ impl ChainStore { let old_head = self.heaviest_tipset.swap(head.shallow_clone().into()); // Only publish head changes when there are active subscribers and head is changed. - if self.head_changes_tx.receiver_count() > 0 && old_head.key() != head.key() { + 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, @@ -280,11 +299,12 @@ impl ChainStore { }; // Do not publish empty change and check active receivers again if !changes.is_empty() - && self.head_changes_tx.receiver_count() > 0 - // head change is only published after tipset validation and the 30s block delay + && self.head_changes_rx_inactive.receiver_count() > 0 + // Use an unbounded bridge channel to avoid blocking `set_heaviest_tipset`. + // 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.broadcast_blocking(changes).is_err() + // 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"); } @@ -361,7 +381,7 @@ impl ChainStore { /// Subscribes head changes. pub fn subscribe_head_changes(&self) -> async_broadcast::Receiver { - self.head_changes_tx.new_receiver() + self.head_changes_rx_inactive.activate_cloned() } /// Returns a borrowed key-value store instance. diff --git a/src/daemon/db_util.rs b/src/daemon/db_util.rs index bdf4eec1d808..2495d26d59cf 100644 --- a/src/daemon/db_util.rs +++ b/src/daemon/db_util.rs @@ -725,7 +725,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 mut head_changes_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 { @@ -807,7 +807,7 @@ 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![]; - while let Some(changes) = head_rx.next().await { + while let Some(changes) = head_changes_rx.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()); From 4ca98ae1ccd643f424813341ca3dbee5ee914654 Mon Sep 17 00:00:00 2001 From: hanabi1224 Date: Fri, 7 Aug 2026 16:49:22 +0800 Subject: [PATCH 08/12] fix --- src/chain/store/chain_store.rs | 1 + src/message_pool/msgpool/test_provider.rs | 6 +++--- 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/src/chain/store/chain_store.rs b/src/chain/store/chain_store.rs index a218903d9e48..6099b193376b 100644 --- a/src/chain/store/chain_store.rs +++ b/src/chain/store/chain_store.rs @@ -156,6 +156,7 @@ impl ChainStore { let (head_changes_tx_bridge, head_changes_rx_bridge) = flume::unbounded(); // Warn if the broadcast channel is blocked (timed out after 1 second) 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)) diff --git a/src/message_pool/msgpool/test_provider.rs b/src/message_pool/msgpool/test_provider.rs index b82fd1187523..14c43260ac87 100644 --- a/src/message_pool/msgpool/test_provider.rs +++ b/src/message_pool/msgpool/test_provider.rs @@ -22,7 +22,7 @@ use std::convert::TryFrom; pub struct TestApi { pub inner: Mutex, pub head_changes_tx: async_broadcast::Sender, - head_changes_rx_inactive: async_broadcast::InactiveReceiver, + _head_changes_rx_inactive: async_broadcast::InactiveReceiver, } #[derive(Default)] @@ -47,7 +47,7 @@ impl Default for TestApi { ..TestApiInner::default() }), head_changes_tx, - head_changes_rx_inactive: head_changes_rx.deactivate(), + _head_changes_rx_inactive: head_changes_rx.deactivate(), } } } @@ -62,7 +62,7 @@ impl TestApi { ..TestApiInner::default() }), head_changes_tx, - head_changes_rx_inactive: head_changes_rx.deactivate(), + _head_changes_rx_inactive: head_changes_rx.deactivate(), } } From 8a5f3ea660cd2c6cb345f7b916bc8d77b529a7d8 Mon Sep 17 00:00:00 2001 From: hanabi1224 Date: Fri, 7 Aug 2026 16:50:58 +0800 Subject: [PATCH 09/12] comment --- src/chain/store/chain_store.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/src/chain/store/chain_store.rs b/src/chain/store/chain_store.rs index 6099b193376b..f75527f43b8a 100644 --- a/src/chain/store/chain_store.rs +++ b/src/chain/store/chain_store.rs @@ -302,6 +302,7 @@ impl ChainStore { 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. From 3fd859450eb3ddd5a5fe313b13bdd46bcf3b02b7 Mon Sep 17 00:00:00 2001 From: hanabi1224 Date: Fri, 7 Aug 2026 17:16:37 +0800 Subject: [PATCH 10/12] fix tests --- src/chain/store/chain_store.rs | 34 ++++++++++++++++++++++------------ 1 file changed, 22 insertions(+), 12 deletions(-) diff --git a/src/chain/store/chain_store.rs b/src/chain/store/chain_store.rs index f75527f43b8a..e4bdf57e8366 100644 --- a/src/chain/store/chain_store.rs +++ b/src/chain/store/chain_store.rs @@ -155,20 +155,30 @@ impl ChainStore { // 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) - 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." - ); + 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")? From 45aac95637e2a44e5968a3dd4c1d4eb7b8b595eb Mon Sep 17 00:00:00 2001 From: hanabi1224 Date: Fri, 7 Aug 2026 19:30:47 +0800 Subject: [PATCH 11/12] fix --- src/daemon/db_util.rs | 33 +++++++++++++++++++++++++-------- 1 file changed, 25 insertions(+), 8 deletions(-) diff --git a/src/daemon/db_util.rs b/src/daemon/db_util.rs index 2495d26d59cf..08c87f2d4185 100644 --- a/src/daemon/db_util.rs +++ b/src/daemon/db_util.rs @@ -807,16 +807,33 @@ 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![]; - while let Some(changes) = head_changes_rx.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()); + loop { + match head_changes_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() + ); + } + } } } + Err(async_broadcast::TryRecvError::Empty) + | Err(async_broadcast::TryRecvError::Closed) => break, + Err(async_broadcast::TryRecvError::Overflowed(n)) => { + error!("unexpected head change overflow during backfill: {n} changes lost"); + continue; + } } } if !extra.is_empty() { From b64bb793d86c07d9ece1daa99810d2c76fecb041 Mon Sep 17 00:00:00 2001 From: hanabi1224 Date: Fri, 7 Aug 2026 20:09:21 +0800 Subject: [PATCH 12/12] consume head_changes_rx eagerly in index backfill --- src/daemon/db_util.rs | 51 +++++++++++++++++++++---------------------- 1 file changed, 25 insertions(+), 26 deletions(-) diff --git a/src/daemon/db_util.rs b/src/daemon/db_util.rs index 08c87f2d4185..83fe00361c00 100644 --- a/src/daemon/db_util.rs +++ b/src/daemon/db_util.rs @@ -29,7 +29,7 @@ use std::{ time, }; use tokio::io::AsyncWriteExt; -use tokio_util::sync::CancellationToken; +use tokio_util::{sync::CancellationToken, task::AbortOnDropHandle}; use tracing::{debug, info, warn}; use url::Url; use walkdir::WalkDir; @@ -725,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. + // 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 { @@ -806,34 +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_changes_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(async_broadcast::TryRecvError::Empty) - | Err(async_broadcast::TryRecvError::Closed) => break, - Err(async_broadcast::TryRecvError::Overflowed(n)) => { - error!("unexpected head change overflow during backfill: {n} changes lost"); - continue; - } } } if !extra.is_empty() {