From 3888ede959a908e9b0c19169506762eb819cef35 Mon Sep 17 00:00:00 2001 From: hanabi1224 Date: Thu, 12 Mar 2026 20:29:48 +0800 Subject: [PATCH 01/10] fix: broadcast HeadChange::Revert(Tipset) in set_heaviest_tipset --- src/chain/store/chain_store.rs | 38 ++++++++++++++++++++-------- src/daemon/mod.rs | 14 +++++----- src/message_pool/msgpool/msg_pool.rs | 17 ++++++------- src/rpc/methods/chain.rs | 33 ++++++++++++------------ src/state_manager/mod.rs | 3 ++- 5 files changed, 60 insertions(+), 45 deletions(-) diff --git a/src/chain/store/chain_store.rs b/src/chain/store/chain_store.rs index fb18955ed0f..0a732758d31 100644 --- a/src/chain/store/chain_store.rs +++ b/src/chain/store/chain_store.rs @@ -6,7 +6,6 @@ use super::{ index::{ChainIndex, ResolveNullTipset}, tipset_tracker::TipsetTracker, }; -use crate::db::{EthMappingsStore, EthMappingsStoreExt}; use crate::interpreter::{BlockMessages, VMTrace}; use crate::libp2p_bitswap::{BitswapStoreRead, BitswapStoreReadWrite}; use crate::message::{ChainMessage, Message as MessageTrait, SignedMessage}; @@ -23,6 +22,10 @@ use crate::{ blocks::{CachingBlockHeader, Tipset, TipsetKey, TxMeta}, db::HeaviestTipsetKeyProvider, }; +use crate::{ + db::{EthMappingsStore, EthMappingsStoreExt}, + rpc::chain::PathChange, +}; use crate::{fil_cns, utils::cache::SizeTrackingLruCache}; use ahash::{HashMap, HashMapExt, HashSet}; use anyhow::Context as _; @@ -47,10 +50,7 @@ pub type ChainEpochDelta = ChainEpoch; /// `Enum` for `pubsub` channel that defines message type variant and data /// contained in message type. -#[derive(Clone, Debug)] -pub enum HeadChange { - Apply(Tipset), -} +pub type HeadChange = PathChange; /// Stores chain data such as heaviest tipset and cached tipset info at each /// epoch. This structure is thread-safe, and all caches are wrapped in a mutex @@ -142,14 +142,30 @@ where } /// Sets heaviest tipset - pub fn set_heaviest_tipset(&self, ts: Tipset) -> Result<(), Error> { + pub fn set_heaviest_tipset(&self, head: Tipset) -> Result<(), Error> { self.heaviest_tipset_key_provider - .set_heaviest_tipset_key(ts.key())?; - *self.heaviest_tipset_cache.write() = Some(ts.clone()); - ts.key().save(self.blockstore())?; - if self.publisher.send(HeadChange::Apply(ts)).is_err() { - debug!("did not publish head change, no active receivers"); + .set_heaviest_tipset_key(head.key())?; + let old_head = (*self.heaviest_tipset_cache.write()).replace(head.clone()); + head.key().save(self.blockstore())?; + if let Some(old_head) = old_head { + match crate::rpc::chain::chain_get_path(self, old_head.key(), head.key()) { + Ok(changes) => { + for change in changes { + if self.publisher.send(change).is_err() { + debug!("did not publish change, no active receivers"); + } + } + } + Err(e) => { + warn!("failed to get chain path changes: {e}") + } + } + } else { + if self.publisher.send(HeadChange::Apply(head)).is_err() { + debug!("did not publish head change, no active receivers"); + } } + Ok(()) } diff --git a/src/daemon/mod.rs b/src/daemon/mod.rs index 765519caba4..a25efa1caf3 100644 --- a/src/daemon/mod.rs +++ b/src/daemon/mod.rs @@ -505,14 +505,12 @@ fn maybe_start_indexer_service( // Continuously listen for head changes loop { - let HeadChange::Apply(ts) = receiver.recv().await?; - - tracing::debug!("Indexing tipset {}", ts.key()); - - let delegated_messages = - chain_store.headers_delegated_messages(ts.block_headers().iter())?; - - chain_store.process_signed_messages(&delegated_messages)?; + if let HeadChange::Apply(ts) = receiver.recv().await? { + tracing::debug!("Indexing tipset {}", ts.key()); + let delegated_messages = + chain_store.headers_delegated_messages(ts.block_headers().iter())?; + chain_store.process_signed_messages(&delegated_messages)?; + } } }); diff --git a/src/message_pool/msgpool/msg_pool.rs b/src/message_pool/msgpool/msg_pool.rs index 6333df24deb..a20801af248 100644 --- a/src/message_pool/msgpool/msg_pool.rs +++ b/src/message_pool/msgpool/msg_pool.rs @@ -551,18 +551,17 @@ where let pending = mp.pending.clone(); let republished = mp.republished.clone(); - let cur_tipset = mp.cur_tipset.clone(); + let current_ts = mp.cur_tipset.clone(); let repub_trigger = mp.repub_trigger.clone(); // Reacts to new HeadChanges services.spawn(async move { loop { match subscriber.recv().await { - Ok(ts) => { - let (cur, rev, app) = match ts { - HeadChange::Apply(tipset) => { - (cur_tipset.clone(), Vec::new(), vec![tipset]) - } + Ok(change) => { + let (reverts, applies) = match change { + HeadChange::Apply(ts) => (vec![], vec![ts]), + HeadChange::Revert(ts) => (vec![ts], vec![]), }; head_change( api.as_ref(), @@ -570,9 +569,9 @@ where repub_trigger.clone(), republished.as_ref(), pending.as_ref(), - cur.as_ref(), - rev, - app, + ¤t_ts, + reverts, + applies, ) .await .context("Error changing head")?; diff --git a/src/rpc/methods/chain.rs b/src/rpc/methods/chain.rs index 136c386cecd..6cd9145b0f7 100644 --- a/src/rpc/methods/chain.rs +++ b/src/rpc/methods/chain.rs @@ -88,22 +88,20 @@ pub(crate) fn new_heads( let handle = tokio::spawn(async move { while let Ok(v) = subscriber.recv().await { - let headers = match v { - HeadChange::Apply(ts) => { - // Convert the tipset to an Ethereum block with full transaction info - // Note: In Filecoin's Eth RPC, a tipset maps to a single Ethereum block - match EthBlock::from_filecoin_tipset(data.clone(), ts, TxInfo::Full).await { - Ok(block) => ApiHeaders(block), - Err(e) => { - tracing::error!("Failed to convert tipset to eth block: {}", e); - continue; + if let HeadChange::Apply(ts) = v { + // Convert the tipset to an Ethereum block with full transaction info + // Note: In Filecoin's Eth RPC, a tipset maps to a single Ethereum block + match EthBlock::from_filecoin_tipset(data.clone(), ts, TxInfo::Full).await { + Ok(block) => { + if let Err(e) = sender.send(ApiHeaders(block)) { + tracing::error!("Failed to send headers: {}", e); + break; } } + Err(e) => { + tracing::error!("Failed to convert tipset to eth block: {}", e); + } } - }; - if let Err(e) = sender.send(headers) { - tracing::error!("Failed to send headers: {}", e); - break; } } }); @@ -149,6 +147,7 @@ pub(crate) fn logs( } } } + HeadChange::Revert(_) => {} } } }); @@ -850,7 +849,7 @@ impl RpcMethod<2> for ChainGetPath { (from, to): Self::Params, _: &http::Extensions, ) -> Result { - impl_chain_get_path(ctx.chain_store(), &from, &to).map_err(Into::into) + chain_get_path(ctx.chain_store(), &from, &to).map_err(Into::into) } } @@ -871,7 +870,7 @@ impl RpcMethod<2> for ChainGetPath { /// ``` /// /// Exposes errors from the [`Blockstore`], and returns an error if there is no common ancestor. -fn impl_chain_get_path( +pub fn chain_get_path( chain_store: &ChainStore, from: &TipsetKey, to: &TipsetKey, @@ -903,6 +902,7 @@ fn impl_chain_get_path( to_apply = next; } } + Ok(all_reverts .into_iter() .map(PathChange::Revert) @@ -1332,6 +1332,7 @@ pub(crate) fn chain_notify( while let Ok(v) = subscriber.recv().await { let (change, tipset) = match v { HeadChange::Apply(ts) => ("apply".into(), ts), + HeadChange::Revert(ts) => ("revert".into(), ts), }; if sender.send(vec![ApiHeadChange { change, tipset }]).is_err() { @@ -1791,7 +1792,7 @@ mod tests { } let actual = - impl_chain_get_path(store, from.make_tipset().key(), to.make_tipset().key()).unwrap(); + chain_get_path(store, from.make_tipset().key(), to.make_tipset().key()).unwrap(); let expected = expected .into_iter() .map(|change| match change { diff --git a/src/state_manager/mod.rs b/src/state_manager/mod.rs index c042351222c..053ef3dfe26 100644 --- a/src/state_manager/mod.rs +++ b/src/state_manager/mod.rs @@ -1210,7 +1210,7 @@ where let mut subscriber_poll = tokio::task::spawn(async move { loop { match subscriber.recv().await { - Ok(subscriber) => match subscriber { + Ok(head_change) => match head_change { HeadChange::Apply(tipset) => { if candidate_tipset .as_ref() @@ -1237,6 +1237,7 @@ where candidate_receipt = Some(receipt) } } + HeadChange::Revert(_) => {} }, Err(RecvError::Lagged(i)) => { warn!( From 4c5900807c0bef88877c80585f6894f5bc23aec9 Mon Sep 17 00:00:00 2001 From: hanabi1224 Date: Thu, 12 Mar 2026 21:00:27 +0800 Subject: [PATCH 02/10] resolve AI comments --- src/chain/store/chain_store.rs | 51 +++++++++++++++++----------------- 1 file changed, 25 insertions(+), 26 deletions(-) diff --git a/src/chain/store/chain_store.rs b/src/chain/store/chain_store.rs index 0a732758d31..2e936397c85 100644 --- a/src/chain/store/chain_store.rs +++ b/src/chain/store/chain_store.rs @@ -66,7 +66,7 @@ pub struct ChainStore { heaviest_tipset_key_provider: Arc, /// Heaviest tipset cache - heaviest_tipset_cache: Arc>>, + heaviest_tipset_cache: Arc>, /// Used as a cache for tipset `lookbacks`. chain_index: Arc>>, @@ -124,14 +124,20 @@ where let (publisher, _) = broadcast::channel(SINK_CAP); let chain_index = Arc::new(ChainIndex::new(Arc::clone(&db))); let validated_blocks = Mutex::new(HashSet::default()); - + let head = if let Ok(head_tsk) = heaviest_tipset_key_provider.heaviest_tipset_key() + && let Ok(head) = chain_index.load_required_tipset(&head_tsk) + { + head + } else { + Tipset::from(&genesis_block_header) + }; let cs = Self { publisher, chain_index, tipset_tracker: TipsetTracker::new(Arc::clone(&db), chain_config.clone()), db, heaviest_tipset_key_provider, - heaviest_tipset_cache: Default::default(), + heaviest_tipset_cache: Arc::new(RwLock::new(head)), genesis_block_header, validated_blocks, eth_mappings, @@ -145,24 +151,26 @@ where pub fn set_heaviest_tipset(&self, head: Tipset) -> Result<(), Error> { self.heaviest_tipset_key_provider .set_heaviest_tipset_key(head.key())?; - let old_head = (*self.heaviest_tipset_cache.write()).replace(head.clone()); + let old_head = std::mem::replace(&mut *self.heaviest_tipset_cache.write(), head.clone()); head.key().save(self.blockstore())?; - if let Some(old_head) = old_head { - match crate::rpc::chain::chain_get_path(self, old_head.key(), head.key()) { - Ok(changes) => { - for change in changes { - if self.publisher.send(change).is_err() { - debug!("did not publish change, no active receivers"); + + match crate::rpc::chain::chain_get_path(self, old_head.key(), head.key()) { + Ok(changes) => { + for change in changes { + let change_text = match &change { + HeadChange::Apply(ts) => format!("apply@{}: {}", ts.epoch(), ts.key()), + HeadChange::Revert(ts) => { + format!("revert@{}: {}", ts.epoch(), ts.key()) } + }; + tracing::info!("head change: {change_text}"); + if self.publisher.send(change).is_err() { + debug!("did not publish change, no active receivers"); } } - Err(e) => { - warn!("failed to get chain path changes: {e}") - } } - } else { - if self.publisher.send(HeadChange::Apply(head)).is_err() { - debug!("did not publish head change, no active receivers"); + Err(e) => { + warn!("failed to get chain path changes: {e}") } } @@ -216,16 +224,7 @@ where /// Returns the currently tracked heaviest tipset. pub fn heaviest_tipset(&self) -> Tipset { - if let Some(ts) = &*self.heaviest_tipset_cache.read() { - return ts.clone(); - } - let tsk = self - .heaviest_tipset_key_provider - .heaviest_tipset_key() - .unwrap_or_else(|_| TipsetKey::from(nunny::vec![*self.genesis_block_header.cid()])); - self.chain_index - .load_required_tipset(&tsk) - .expect("failed to load heaviest tipset") + self.heaviest_tipset_cache.read().clone() } /// Returns the genesis tipset. From b26e0c2e88a96dff37b7c6a2d55a33fb965329be Mon Sep 17 00:00:00 2001 From: hanabi1224 Date: Thu, 12 Mar 2026 21:34:04 +0800 Subject: [PATCH 03/10] resolve comments --- src/chain/store/chain_store.rs | 28 +++--- src/daemon/mod.rs | 3 +- src/message_pool/msgpool/msg_pool.rs | 8 +- src/message_pool/msgpool/provider.rs | 8 +- src/message_pool/msgpool/test_provider.rs | 13 ++- src/rpc/methods/chain.rs | 106 +++++++++++++--------- src/state_manager/mod.rs | 9 +- 7 files changed, 95 insertions(+), 80 deletions(-) diff --git a/src/chain/store/chain_store.rs b/src/chain/store/chain_store.rs index 2e936397c85..7b8d94eaef1 100644 --- a/src/chain/store/chain_store.rs +++ b/src/chain/store/chain_store.rs @@ -6,7 +6,6 @@ use super::{ index::{ChainIndex, ResolveNullTipset}, tipset_tracker::TipsetTracker, }; -use crate::interpreter::{BlockMessages, VMTrace}; use crate::libp2p_bitswap::{BitswapStoreRead, BitswapStoreReadWrite}; use crate::message::{ChainMessage, Message as MessageTrait, SignedMessage}; use crate::networks::{ChainConfig, Height}; @@ -27,6 +26,10 @@ use crate::{ rpc::chain::PathChange, }; use crate::{fil_cns, utils::cache::SizeTrackingLruCache}; +use crate::{ + interpreter::{BlockMessages, VMTrace}, + rpc::chain::PathChanges, +}; use ahash::{HashMap, HashMapExt, HashSet}; use anyhow::Context as _; use cid::Cid; @@ -52,12 +55,14 @@ pub type ChainEpochDelta = ChainEpoch; /// contained in message type. pub type HeadChange = PathChange; +pub type HeadChanges = PathChanges; + /// Stores chain data such as heaviest tipset and cached tipset info at each /// 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 - publisher: Publisher, + publisher: Publisher, /// key-value `datastore`. db: Arc, @@ -125,7 +130,9 @@ where let chain_index = Arc::new(ChainIndex::new(Arc::clone(&db))); let validated_blocks = Mutex::new(HashSet::default()); let head = if let Ok(head_tsk) = heaviest_tipset_key_provider.heaviest_tipset_key() - && let Ok(head) = chain_index.load_required_tipset(&head_tsk) + && let Some(head) = chain_index + .load_tipset(&head_tsk) + .context("failed to load head tipset")? { head } else { @@ -156,17 +163,8 @@ where match crate::rpc::chain::chain_get_path(self, old_head.key(), head.key()) { Ok(changes) => { - for change in changes { - let change_text = match &change { - HeadChange::Apply(ts) => format!("apply@{}: {}", ts.epoch(), ts.key()), - HeadChange::Revert(ts) => { - format!("revert@{}: {}", ts.epoch(), ts.key()) - } - }; - tracing::info!("head change: {change_text}"); - if self.publisher.send(change).is_err() { - debug!("did not publish change, no active receivers"); - } + if self.publisher.send(changes).is_err() { + debug!("did not publish changes, no active receivers"); } } Err(e) => { @@ -233,7 +231,7 @@ where } /// Returns a reference to the publisher of head changes. - pub fn publisher(&self) -> &Publisher { + pub fn publisher(&self) -> &Publisher { &self.publisher } diff --git a/src/daemon/mod.rs b/src/daemon/mod.rs index a25efa1caf3..339441bcc20 100644 --- a/src/daemon/mod.rs +++ b/src/daemon/mod.rs @@ -7,7 +7,6 @@ pub mod db_util; pub mod main; use crate::blocks::Tipset; -use crate::chain::HeadChange; use crate::chain::index::ResolveNullTipset; use crate::chain_sync::network_context::SyncNetworkContext; use crate::chain_sync::{ChainFollower, SyncStatus}; @@ -505,7 +504,7 @@ fn maybe_start_indexer_service( // Continuously listen for head changes loop { - if let HeadChange::Apply(ts) = receiver.recv().await? { + for ts in receiver.recv().await?.applies { tracing::debug!("Indexing tipset {}", ts.key()); let delegated_messages = chain_store.headers_delegated_messages(ts.block_headers().iter())?; diff --git a/src/message_pool/msgpool/msg_pool.rs b/src/message_pool/msgpool/msg_pool.rs index a20801af248..6371fc16c34 100644 --- a/src/message_pool/msgpool/msg_pool.rs +++ b/src/message_pool/msgpool/msg_pool.rs @@ -9,7 +9,7 @@ use std::{num::NonZeroUsize, sync::Arc, time::Duration}; use crate::blocks::{CachingBlockHeader, Tipset}; -use crate::chain::{HeadChange, MINIMUM_BASE_FEE}; +use crate::chain::{HeadChanges, MINIMUM_BASE_FEE}; #[cfg(test)] use crate::db::SettingsStore; use crate::eth::is_valid_eth_tx_for_sending; @@ -558,11 +558,7 @@ where services.spawn(async move { loop { match subscriber.recv().await { - Ok(change) => { - let (reverts, applies) = match change { - HeadChange::Apply(ts) => (vec![], vec![ts]), - HeadChange::Revert(ts) => (vec![ts], vec![]), - }; + Ok(HeadChanges { reverts, applies }) => { head_change( api.as_ref(), bls_sig_cache.as_ref(), diff --git a/src/message_pool/msgpool/provider.rs b/src/message_pool/msgpool/provider.rs index a685810344d..8180b71d19e 100644 --- a/src/message_pool/msgpool/provider.rs +++ b/src/message_pool/msgpool/provider.rs @@ -4,7 +4,7 @@ use std::sync::Arc; use crate::blocks::{CachingBlockHeader, Tipset, TipsetKey}; -use crate::chain::HeadChange; +use crate::chain::HeadChanges; use crate::message::{ChainMessage, SignedMessage}; use crate::message_pool::msg_pool::{ MAX_ACTOR_PENDING_MESSAGES, MAX_UNTRUSTED_ACTOR_PENDING_MESSAGES, @@ -31,7 +31,7 @@ use crate::message_pool::errors::Error; #[async_trait] pub trait Provider { /// Update `Mpool`'s `cur_tipset` whenever there is a change to the provider - fn subscribe_head_changes(&self) -> Subscriber; + fn subscribe_head_changes(&self) -> Subscriber; /// Get the heaviest Tipset in the provider fn get_heaviest_tipset(&self) -> Tipset; /// Add a message to the `MpoolProvider`, return either Cid or Error @@ -64,7 +64,7 @@ pub trait Provider { /// `mpool` RPC. #[derive(derive_more::Constructor)] pub struct MpoolRpcProvider { - subscriber: Publisher, + subscriber: Publisher, sm: Arc>, } @@ -73,7 +73,7 @@ impl Provider for MpoolRpcProvider where DB: Blockstore + Sync + Send + 'static, { - fn subscribe_head_changes(&self) -> Subscriber { + fn subscribe_head_changes(&self) -> Subscriber { self.subscriber.subscribe() } diff --git a/src/message_pool/msgpool/test_provider.rs b/src/message_pool/msgpool/test_provider.rs index c00f7be60b3..0e18816deab 100644 --- a/src/message_pool/msgpool/test_provider.rs +++ b/src/message_pool/msgpool/test_provider.rs @@ -8,7 +8,7 @@ use std::convert::TryFrom; use crate::blocks::{ CachingBlockHeader, ElectionProof, RawBlockHeader, Ticket, Tipset, TipsetKey, VRFProof, }; -use crate::chain::HeadChange; +use crate::chain::HeadChanges; use crate::cid_collections::CidHashMap; use crate::message::{ChainMessage, Message as MessageTrait, SignedMessage}; use crate::message_pool::{Error, provider::Provider}; @@ -25,7 +25,7 @@ use tokio::sync::broadcast::{Receiver as Subscriber, Sender as Publisher}; /// pool pub struct TestApi { pub inner: Mutex, - pub publisher: Publisher, + pub publisher: Publisher, } #[derive(Default)] @@ -81,7 +81,12 @@ impl TestApi { /// Set the heaviest tipset for `TestApi` pub fn set_heaviest_tipset(&self, ts: Tipset) { - self.publisher.send(HeadChange::Apply(ts)).unwrap(); + self.publisher + .send(HeadChanges { + applies: vec![ts], + reverts: vec![], + }) + .unwrap(); } pub fn next_block(&self) -> CachingBlockHeader { @@ -119,7 +124,7 @@ impl TestApiInner { #[async_trait] impl Provider for TestApi { - fn subscribe_head_changes(&self) -> Subscriber { + fn subscribe_head_changes(&self) -> Subscriber { self.publisher.subscribe() } diff --git a/src/rpc/methods/chain.rs b/src/rpc/methods/chain.rs index 6cd9145b0f7..61152c4fc8a 100644 --- a/src/rpc/methods/chain.rs +++ b/src/rpc/methods/chain.rs @@ -87,8 +87,8 @@ pub(crate) fn new_heads( let mut subscriber = data.chain_store().publisher().subscribe(); let handle = tokio::spawn(async move { - while let Ok(v) = subscriber.recv().await { - if let HeadChange::Apply(ts) = v { + while let Ok(changes) = subscriber.recv().await { + for ts in changes.applies { // Convert the tipset to an Ethereum block with full transaction info // Note: In Filecoin's Eth RPC, a tipset maps to a single Ethereum block match EthBlock::from_filecoin_tipset(data.clone(), ts, TxInfo::Full).await { @@ -126,28 +126,21 @@ pub(crate) fn logs( let ctx = ctx.clone(); let handle = tokio::spawn(async move { - while let Ok(v) = subscriber.recv().await { - match v { - HeadChange::Apply(ts) => { - match eth_logs_with_filter(&ctx, &ts, filter.clone(), None).await { - Ok(logs) => { - if !logs.is_empty() - && let Err(e) = sender.send(logs) - { - tracing::error!( - "Failed to send logs for tipset {}: {}", - ts.key(), - e - ); - break; - } - } - Err(e) => { - tracing::error!("Failed to fetch logs for tipset {}: {}", ts.key(), e); + while let Ok(changes) = subscriber.recv().await { + for ts in changes.applies { + match eth_logs_with_filter(&ctx, &ts, filter.clone(), None).await { + Ok(logs) => { + if !logs.is_empty() + && let Err(e) = sender.send(logs) + { + tracing::error!("Failed to send logs for tipset {}: {}", ts.key(), e); + break; } } + Err(e) => { + tracing::error!("Failed to fetch logs for tipset {}: {}", ts.key(), e); + } } - HeadChange::Revert(_) => {} } } }); @@ -849,11 +842,11 @@ impl RpcMethod<2> for ChainGetPath { (from, to): Self::Params, _: &http::Extensions, ) -> Result { - chain_get_path(ctx.chain_store(), &from, &to).map_err(Into::into) + Ok(chain_get_path(ctx.chain_store(), &from, &to)?.into_change_vec()) } } -/// Find the path between two tipsets, as a series of [`PathChange`]s. +/// Find the path between two tipsets, as [`PathChanges`]. /// /// ```text /// 0 - A - B - C - D @@ -874,7 +867,7 @@ pub fn chain_get_path( chain_store: &ChainStore, from: &TipsetKey, to: &TipsetKey, -) -> anyhow::Result> { +) -> anyhow::Result { let mut to_revert = chain_store .load_required_tipset_or_heaviest(from) .context("couldn't load `from`")?; @@ -882,8 +875,8 @@ pub fn chain_get_path( .load_required_tipset_or_heaviest(to) .context("couldn't load `to`")?; - let mut all_reverts = vec![]; - let mut all_applies = vec![]; + let mut reverts = vec![]; + let mut applies = vec![]; // This loop is guaranteed to terminate if the blockstore contain no cycles. // This is currently computationally infeasible. @@ -892,22 +885,18 @@ pub fn chain_get_path( let next = chain_store .load_required_tipset_or_heaviest(to_revert.parents()) .context("couldn't load ancestor of `from`")?; - all_reverts.push(to_revert); + reverts.push(to_revert); to_revert = next; } else { let next = chain_store .load_required_tipset_or_heaviest(to_apply.parents()) .context("couldn't load ancestor of `to`")?; - all_applies.push(to_apply); + applies.push(to_apply); to_apply = next; } } - - Ok(all_reverts - .into_iter() - .map(PathChange::Revert) - .chain(all_applies.into_iter().rev().map(PathChange::Apply)) - .collect()) + applies.reverse(); + Ok(PathChanges { reverts, applies }) } /// Get tipset at epoch. Pick younger tipset if epoch points to a @@ -1329,14 +1318,15 @@ pub(crate) fn chain_notify( // Skip first message let _ = subscriber.recv().await; - while let Ok(v) = subscriber.recv().await { - let (change, tipset) = match v { - HeadChange::Apply(ts) => ("apply".into(), ts), - HeadChange::Revert(ts) => ("revert".into(), ts), - }; - - if sender.send(vec![ApiHeadChange { change, tipset }]).is_err() { - break; + while let Ok(changes) = subscriber.recv().await { + for change in changes.into_change_vec() { + let (change, tipset) = match change { + HeadChange::Apply(ts) => ("apply".into(), ts), + HeadChange::Revert(ts) => ("revert".into(), ts), + }; + if sender.send(vec![ApiHeadChange { change, tipset }]).is_err() { + break; + } } } }); @@ -1564,6 +1554,33 @@ impl HasLotusJson for PathChange { } } +#[derive(Debug)] +pub struct PathChanges { + pub reverts: Vec, + pub applies: Vec, +} + +impl Clone for PathChanges { + fn clone(&self) -> Self { + let Self { reverts, applies } = self; + Self { + reverts: reverts.clone(), + applies: applies.clone(), + } + } +} + +impl PathChanges { + pub fn into_change_vec(self) -> Vec> { + let Self { reverts, applies } = self; + reverts + .into_iter() + .map(PathChange::Revert) + .chain(applies.into_iter().map(PathChange::Apply)) + .collect() + } +} + #[cfg(test)] impl quickcheck::Arbitrary for PathChange where @@ -1791,8 +1808,9 @@ mod tests { ) } - let actual = - chain_get_path(store, from.make_tipset().key(), to.make_tipset().key()).unwrap(); + let actual = chain_get_path(store, from.make_tipset().key(), to.make_tipset().key()) + .unwrap() + .into_change_vec(); let expected = expected .into_iter() .map(|change| match change { diff --git a/src/state_manager/mod.rs b/src/state_manager/mod.rs index 053ef3dfe26..0333cfaf31a 100644 --- a/src/state_manager/mod.rs +++ b/src/state_manager/mod.rs @@ -16,7 +16,7 @@ use self::utils::structured; use crate::beacon::{BeaconEntry, BeaconSchedule}; use crate::blocks::{Tipset, TipsetKey}; use crate::chain::{ - ChainStore, HeadChange, + ChainStore, index::{ChainIndex, ResolveNullTipset}, }; use crate::interpreter::{ @@ -1210,8 +1210,8 @@ where let mut subscriber_poll = tokio::task::spawn(async move { loop { match subscriber.recv().await { - Ok(head_change) => match head_change { - HeadChange::Apply(tipset) => { + Ok(head_changes) => { + for tipset in head_changes.applies { if candidate_tipset .as_ref() .map(|s| tipset.epoch() >= s.epoch() + confidence) @@ -1237,8 +1237,7 @@ where candidate_receipt = Some(receipt) } } - HeadChange::Revert(_) => {} - }, + } Err(RecvError::Lagged(i)) => { warn!( "wait for message head change subscriber lagged, skipped {} events", From a7581831bc544aa86a5375977138de7eefcd5259 Mon Sep 17 00:00:00 2001 From: hanabi1224 Date: Thu, 12 Mar 2026 21:44:17 +0800 Subject: [PATCH 04/10] resolve comments --- src/chain/store/chain_store.rs | 6 ++++-- src/db/car/many.rs | 12 ++++++------ src/db/gc/snapshot.rs | 2 +- src/db/memory.rs | 5 ++--- src/db/mod.rs | 4 ++-- src/db/parity_db.rs | 5 ++--- .../subcommands/api_cmd/generate_test_snapshot.rs | 2 +- 7 files changed, 18 insertions(+), 18 deletions(-) diff --git a/src/chain/store/chain_store.rs b/src/chain/store/chain_store.rs index 7b8d94eaef1..1218e23e168 100644 --- a/src/chain/store/chain_store.rs +++ b/src/chain/store/chain_store.rs @@ -129,7 +129,9 @@ where let (publisher, _) = broadcast::channel(SINK_CAP); let chain_index = Arc::new(ChainIndex::new(Arc::clone(&db))); let validated_blocks = Mutex::new(HashSet::default()); - let head = if let Ok(head_tsk) = heaviest_tipset_key_provider.heaviest_tipset_key() + let head = if let Some(head_tsk) = heaviest_tipset_key_provider + .heaviest_tipset_key() + .context("failed to load head tipset key")? && let Some(head) = chain_index .load_tipset(&head_tsk) .context("failed to load head tipset")? @@ -156,10 +158,10 @@ where /// Sets heaviest tipset pub fn set_heaviest_tipset(&self, head: Tipset) -> Result<(), Error> { + head.key().save(self.blockstore())?; self.heaviest_tipset_key_provider .set_heaviest_tipset_key(head.key())?; let old_head = std::mem::replace(&mut *self.heaviest_tipset_cache.write(), head.clone()); - head.key().save(self.blockstore())?; match crate::rpc::chain::chain_get_path(self, old_head.key(), head.key()) { Ok(changes) => { diff --git a/src/db/car/many.rs b/src/db/car/many.rs index 6b53c3ba674..28184d03beb 100644 --- a/src/db/car/many.rs +++ b/src/db/car/many.rs @@ -130,12 +130,12 @@ impl ManyCar { Ok(()) } - pub fn heaviest_tipset_key(&self) -> anyhow::Result { - self.read_only + pub fn heaviest_tipset_key(&self) -> anyhow::Result> { + Ok(self + .read_only .read() .peek() - .map(|w| AnyCar::heaviest_tipset_key(&w.car)) - .context("ManyCar store doesn't have a heaviest tipset key") + .map(|w| AnyCar::heaviest_tipset_key(&w.car))) } pub fn heaviest_tipset(&self) -> anyhow::Result { @@ -252,9 +252,9 @@ impl EthMappingsStore for ManyCar { } impl super::super::HeaviestTipsetKeyProvider for ManyCar { - fn heaviest_tipset_key(&self) -> anyhow::Result { + fn heaviest_tipset_key(&self) -> anyhow::Result> { match SettingsStoreExt::read_obj::(self, crate::db::setting_keys::HEAD_KEY)? { - Some(tsk) => Ok(tsk), + Some(tsk) => Ok(Some(tsk)), None => self.heaviest_tipset_key(), } } diff --git a/src/db/gc/snapshot.rs b/src/db/gc/snapshot.rs index 6417214988d..60c53e266b6 100644 --- a/src/db/gc/snapshot.rs +++ b/src/db/gc/snapshot.rs @@ -282,7 +282,7 @@ where tracing::warn!("{e}"); } - *self.memory_db_head_key.write() = db.heaviest_tipset_key().ok(); + *self.memory_db_head_key.write() = db.heaviest_tipset_key()?; db.unsubscribe_write_ops(); match joinset.join_next().await { Some(Ok(map)) => { diff --git a/src/db/memory.rs b/src/db/memory.rs index 5419d06f2ff..42a4fa70f2c 100644 --- a/src/db/memory.rs +++ b/src/db/memory.rs @@ -154,9 +154,8 @@ impl BitswapStoreReadWrite for MemoryDB { } impl super::HeaviestTipsetKeyProvider for MemoryDB { - fn heaviest_tipset_key(&self) -> anyhow::Result { - SettingsStoreExt::read_obj::(self, crate::db::setting_keys::HEAD_KEY)? - .context("head key not found") + fn heaviest_tipset_key(&self) -> anyhow::Result> { + SettingsStoreExt::read_obj::(self, crate::db::setting_keys::HEAD_KEY) } fn set_heaviest_tipset_key(&self, tsk: &TipsetKey) -> anyhow::Result<()> { diff --git a/src/db/mod.rs b/src/db/mod.rs index 48cd6d9d4ff..4dec56e9712 100644 --- a/src/db/mod.rs +++ b/src/db/mod.rs @@ -237,14 +237,14 @@ impl PersistentStore for &Arc { pub trait HeaviestTipsetKeyProvider { /// Returns the currently tracked heaviest tipset. - fn heaviest_tipset_key(&self) -> anyhow::Result; + fn heaviest_tipset_key(&self) -> anyhow::Result>; /// Sets heaviest tipset. fn set_heaviest_tipset_key(&self, tsk: &TipsetKey) -> anyhow::Result<()>; } impl HeaviestTipsetKeyProvider for Arc { - fn heaviest_tipset_key(&self) -> anyhow::Result { + fn heaviest_tipset_key(&self) -> anyhow::Result> { self.as_ref().heaviest_tipset_key() } diff --git a/src/db/parity_db.rs b/src/db/parity_db.rs index 8c6970c1ef1..1d80e78482a 100644 --- a/src/db/parity_db.rs +++ b/src/db/parity_db.rs @@ -173,9 +173,8 @@ impl SettingsStore for ParityDb { } impl super::HeaviestTipsetKeyProvider for ParityDb { - fn heaviest_tipset_key(&self) -> anyhow::Result { - super::SettingsStoreExt::read_obj::(self, super::setting_keys::HEAD_KEY)? - .context("head key not found") + fn heaviest_tipset_key(&self) -> anyhow::Result> { + super::SettingsStoreExt::read_obj::(self, super::setting_keys::HEAD_KEY) } fn set_heaviest_tipset_key(&self, tsk: &TipsetKey) -> anyhow::Result<()> { diff --git a/src/tool/subcommands/api_cmd/generate_test_snapshot.rs b/src/tool/subcommands/api_cmd/generate_test_snapshot.rs index 6cedaef505e..9b99378cd76 100644 --- a/src/tool/subcommands/api_cmd/generate_test_snapshot.rs +++ b/src/tool/subcommands/api_cmd/generate_test_snapshot.rs @@ -223,7 +223,7 @@ where } impl HeaviestTipsetKeyProvider for ReadOpsTrackingStore { - fn heaviest_tipset_key(&self) -> anyhow::Result { + fn heaviest_tipset_key(&self) -> anyhow::Result> { self.inner.heaviest_tipset_key() } From b7c0d1984af57e6de920878d1faddfd0b61b4bf7 Mon Sep 17 00:00:00 2001 From: hanabi1224 Date: Thu, 12 Mar 2026 21:53:13 +0800 Subject: [PATCH 05/10] fix --- src/rpc/methods/chain.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/rpc/methods/chain.rs b/src/rpc/methods/chain.rs index 61152c4fc8a..13e23024376 100644 --- a/src/rpc/methods/chain.rs +++ b/src/rpc/methods/chain.rs @@ -95,7 +95,7 @@ pub(crate) fn new_heads( Ok(block) => { if let Err(e) = sender.send(ApiHeaders(block)) { tracing::error!("Failed to send headers: {}", e); - break; + return; } } Err(e) => { From 20df24cfe77d0cc000128bdd765a4aed979c6886 Mon Sep 17 00:00:00 2001 From: hanabi1224 Date: Thu, 12 Mar 2026 22:06:46 +0800 Subject: [PATCH 06/10] fix --- src/rpc/methods/chain.rs | 30 ++++++++++++++----- .../api_cmd/generate_test_snapshot.rs | 5 +++- 2 files changed, 26 insertions(+), 9 deletions(-) diff --git a/src/rpc/methods/chain.rs b/src/rpc/methods/chain.rs index 13e23024376..a7cd68fcc93 100644 --- a/src/rpc/methods/chain.rs +++ b/src/rpc/methods/chain.rs @@ -1319,14 +1319,13 @@ pub(crate) fn chain_notify( let _ = subscriber.recv().await; while let Ok(changes) = subscriber.recv().await { - for change in changes.into_change_vec() { - let (change, tipset) = match change { - HeadChange::Apply(ts) => ("apply".into(), ts), - HeadChange::Revert(ts) => ("revert".into(), ts), - }; - if sender.send(vec![ApiHeadChange { change, tipset }]).is_err() { - break; - } + let api_changes = changes + .into_change_vec() + .into_iter() + .map(From::from) + .collect(); + if sender.send(api_changes).is_err() { + break; } } }); @@ -1497,6 +1496,21 @@ pub struct ApiHeadChange { } lotus_json_with_self!(ApiHeadChange); +impl From for ApiHeadChange { + fn from(change: HeadChange) -> Self { + match change { + HeadChange::Apply(tipset) => Self { + change: "apply".into(), + tipset, + }, + HeadChange::Revert(tipset) => Self { + change: "revert".into(), + tipset, + }, + } + } +} + #[derive(PartialEq, Debug, Serialize, Deserialize, Clone, JsonSchema)] #[serde(tag = "Type", content = "Val", rename_all = "snake_case")] pub enum PathChange { diff --git a/src/tool/subcommands/api_cmd/generate_test_snapshot.rs b/src/tool/subcommands/api_cmd/generate_test_snapshot.rs index 9b99378cd76..c7a39a99b3e 100644 --- a/src/tool/subcommands/api_cmd/generate_test_snapshot.rs +++ b/src/tool/subcommands/api_cmd/generate_test_snapshot.rs @@ -194,7 +194,10 @@ where SettingsStoreExt::write_obj( &self.tracker, crate::db::setting_keys::HEAD_KEY, - &self.inner.heaviest_tipset_key()?, + &self + .inner + .heaviest_tipset_key()? + .context("heaviest tipset key not found")?, )?; } From d753d31652a8bf7aed4a47a53079db68a14f75c3 Mon Sep 17 00:00:00 2001 From: hanabi1224 Date: Thu, 12 Mar 2026 22:17:46 +0800 Subject: [PATCH 07/10] fix --- src/rpc/methods/chain.rs | 1 - src/state_manager/mod.rs | 9 +++++++++ 2 files changed, 9 insertions(+), 1 deletion(-) diff --git a/src/rpc/methods/chain.rs b/src/rpc/methods/chain.rs index a7cd68fcc93..42525daae44 100644 --- a/src/rpc/methods/chain.rs +++ b/src/rpc/methods/chain.rs @@ -1317,7 +1317,6 @@ pub(crate) fn chain_notify( tokio::spawn(async move { // Skip first message let _ = subscriber.recv().await; - while let Ok(changes) = subscriber.recv().await { let api_changes = changes .into_change_vec() diff --git a/src/state_manager/mod.rs b/src/state_manager/mod.rs index 0333cfaf31a..7f2c5179fdd 100644 --- a/src/state_manager/mod.rs +++ b/src/state_manager/mod.rs @@ -1211,6 +1211,15 @@ where loop { match subscriber.recv().await { Ok(head_changes) => { + for tipset in head_changes.reverts { + if candidate_tipset + .as_ref() + .is_some_and(|candidate| candidate.key() == tipset.key()) + { + candidate_tipset = None; + candidate_receipt = None; + } + } for tipset in head_changes.applies { if candidate_tipset .as_ref() From c6b0c0ca92ec1c8d1e5a6e9b1c0e6483a0911b43 Mon Sep 17 00:00:00 2001 From: hanabi1224 Date: Thu, 12 Mar 2026 22:30:57 +0800 Subject: [PATCH 08/10] fix --- src/message_pool/msgpool/msg_pool.rs | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/src/message_pool/msgpool/msg_pool.rs b/src/message_pool/msgpool/msg_pool.rs index 6371fc16c34..26e9736aad9 100644 --- a/src/message_pool/msgpool/msg_pool.rs +++ b/src/message_pool/msgpool/msg_pool.rs @@ -559,7 +559,7 @@ where loop { match subscriber.recv().await { Ok(HeadChanges { reverts, applies }) => { - head_change( + if let Err(e) = head_change( api.as_ref(), bls_sig_cache.as_ref(), repub_trigger.clone(), @@ -570,10 +570,12 @@ where applies, ) .await - .context("Error changing head")?; + { + tracing::warn!("Error changing head: {e}"); + } } Err(RecvError::Lagged(e)) => { - warn!("Head change subscriber lagged: skipping {} events", e); + warn!("Head change subscriber lagged: skipping {e} events"); } Err(RecvError::Closed) => { break Ok(()); From 843febee88a2f30bd8449601131c9725d120f066 Mon Sep 17 00:00:00 2001 From: hanabi1224 Date: Thu, 12 Mar 2026 22:51:44 +0800 Subject: [PATCH 09/10] fix --- src/message_pool/msgpool/mod.rs | 21 +++++++++++++++++---- 1 file changed, 17 insertions(+), 4 deletions(-) diff --git a/src/message_pool/msgpool/mod.rs b/src/message_pool/msgpool/mod.rs index 720d39c3615..6fe48064d8b 100644 --- a/src/message_pool/msgpool/mod.rs +++ b/src/message_pool/msgpool/mod.rs @@ -226,15 +226,25 @@ where let mut repub = false; let mut rmsgs: HashMap> = HashMap::new(); for ts in revert { - let pts = api.load_tipset(ts.parents())?; + let Ok(pts) = api.load_tipset(ts.parents()) else { + tracing::error!("error loading reverted tipset parent"); + continue; + }; *cur_tipset.write() = pts; let mut msgs: Vec = Vec::new(); for block in ts.block_headers() { - let (umsg, smsgs) = api.messages_for_block(block)?; + let Ok((umsg, smsgs)) = api.messages_for_block(block) else { + tracing::error!("error retrieving messages for reverted block"); + continue; + }; msgs.extend(smsgs); for msg in umsg { - let smsg = recover_sig(bls_sig_cache, msg)?; + let msg_cid = msg.cid(); + let Ok(smsg) = recover_sig(bls_sig_cache, msg) else { + tracing::debug!("could not recover signature for bls message {}", msg_cid); + continue; + }; msgs.push(smsg) } } @@ -246,7 +256,10 @@ where for ts in apply { for b in ts.block_headers() { - let (msgs, smsgs) = api.messages_for_block(b)?; + let Ok((msgs, smsgs)) = api.messages_for_block(b) else { + tracing::error!("error retrieving messages for block"); + continue; + }; for msg in smsgs { remove_from_selected_msgs( From 279d588a8913bc2f827c0565600e043b4e8e5e8b Mon Sep 17 00:00:00 2001 From: hanabi1224 Date: Fri, 13 Mar 2026 07:13:46 +0800 Subject: [PATCH 10/10] checks for chain_get_path --- src/chain/store/chain_store.rs | 22 +++++++++++++++------- src/rpc/methods/chain.rs | 8 ++++++++ 2 files changed, 23 insertions(+), 7 deletions(-) diff --git a/src/chain/store/chain_store.rs b/src/chain/store/chain_store.rs index 1218e23e168..d2815890a6c 100644 --- a/src/chain/store/chain_store.rs +++ b/src/chain/store/chain_store.rs @@ -163,15 +163,23 @@ where .set_heaviest_tipset_key(head.key())?; let old_head = std::mem::replace(&mut *self.heaviest_tipset_cache.write(), head.clone()); - match crate::rpc::chain::chain_get_path(self, old_head.key(), head.key()) { - Ok(changes) => { - if self.publisher.send(changes).is_err() { - debug!("did not publish changes, no active receivers"); - } - } + let changes = match crate::rpc::chain::chain_get_path(self, old_head.key(), head.key()) { + Ok(changes) => changes, Err(e) => { - warn!("failed to get chain path changes: {e}") + // Do not warn when the old head is genesis + if old_head.epoch() > 0 { + warn!("failed to get chain path changes: {e}"); + } + // Fallback to single apply + PathChanges { + applies: vec![head], + reverts: vec![], + } } + }; + + if self.publisher.send(changes).is_err() { + debug!("did not publish changes, no active receivers"); } Ok(()) diff --git a/src/rpc/methods/chain.rs b/src/rpc/methods/chain.rs index 42525daae44..34318f09f9d 100644 --- a/src/rpc/methods/chain.rs +++ b/src/rpc/methods/chain.rs @@ -868,6 +868,7 @@ pub fn chain_get_path( from: &TipsetKey, to: &TipsetKey, ) -> anyhow::Result { + let finality = chain_store.chain_config().policy.chain_finality; let mut to_revert = chain_store .load_required_tipset_or_heaviest(from) .context("couldn't load `from`")?; @@ -875,6 +876,13 @@ pub fn chain_get_path( .load_required_tipset_or_heaviest(to) .context("couldn't load `to`")?; + anyhow::ensure!( + (to_apply.epoch() - to_revert.epoch()).abs() <= finality, + "the gap between the new head ({}) and the old head ({}) is larger than chain finality ({finality})", + to_apply.epoch(), + to_revert.epoch() + ); + let mut reverts = vec![]; let mut applies = vec![];