fix: broadcast HeadChange::Revert(Tipset) in set_heaviest_tipset

src/chain/store/chain_store.rs

@@ -6,8 +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};
 use crate::networks::{ChainConfig, Height};
@@ -23,7 +21,15 @@ 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 crate::{
+    interpreter::{BlockMessages, VMTrace},
+    rpc::chain::PathChanges,
+};
 use ahash::{HashMap, HashMapExt, HashSet};
 use anyhow::Context as _;
 use cid::Cid;
@@ -47,17 +53,16 @@ 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<Tipset>;
+
+pub type HeadChanges = PathChanges<Tipset>;
 
 /// 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<DB> {
     /// Publisher for head change events
-    publisher: Publisher<HeadChange>,
+    publisher: Publisher<HeadChanges>,
 
     /// key-value `datastore`.
     db: Arc<DB>,
@@ -66,7 +71,7 @@ pub struct ChainStore<DB> {
     heaviest_tipset_key_provider: Arc<dyn HeaviestTipsetKeyProvider + Sync + Send>,
 
     /// Heaviest tipset cache
-    heaviest_tipset_cache: Arc<RwLock<Option<Tipset>>>,
+    heaviest_tipset_cache: Arc<RwLock<Tipset>>,
 
     /// Used as a cache for tipset `lookbacks`.
     chain_index: Arc<ChainIndex<Arc<DB>>>,
@@ -124,14 +129,24 @@ 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 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")?
+        {
+            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,
@@ -142,14 +157,23 @@ where
     }
 
     /// Sets heaviest tipset
-    pub fn set_heaviest_tipset(&self, ts: Tipset) -> Result<(), Error> {
+    pub fn set_heaviest_tipset(&self, head: Tipset) -> Result<(), Error> {
+        head.key().save(self.blockstore())?;
         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 = 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");
+                }
+            }
+            Err(e) => {
+                warn!("failed to get chain path changes: {e}")
+            }
         }
+
         Ok(())
     }
 
@@ -200,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.
@@ -218,7 +233,7 @@ where
     }
 
     /// Returns a reference to the publisher of head changes.
-    pub fn publisher(&self) -> &Publisher<HeadChange> {
+    pub fn publisher(&self) -> &Publisher<HeadChanges> {
         &self.publisher
     }
 

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,14 +504,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)?;
+                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())?;
+                    chain_store.process_signed_messages(&delegated_messages)?;
+                }
             }
         });
 

src/db/car/many.rs

@@ -130,12 +130,12 @@ impl<WriterT> ManyCar<WriterT> {
         Ok(())
     }
 
-    pub fn heaviest_tipset_key(&self) -> anyhow::Result<TipsetKey> {
-        self.read_only
+    pub fn heaviest_tipset_key(&self) -> anyhow::Result<Option<TipsetKey>> {
+        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<Tipset> {
@@ -252,9 +252,9 @@ impl<WriterT: EthMappingsStore> EthMappingsStore for ManyCar<WriterT> {
 }
 
 impl<T: Blockstore + SettingsStore> super::super::HeaviestTipsetKeyProvider for ManyCar<T> {
-    fn heaviest_tipset_key(&self) -> anyhow::Result<TipsetKey> {
+    fn heaviest_tipset_key(&self) -> anyhow::Result<Option<TipsetKey>> {
         match SettingsStoreExt::read_obj::<TipsetKey>(self, crate::db::setting_keys::HEAD_KEY)? {
-            Some(tsk) => Ok(tsk),
+            Some(tsk) => Ok(Some(tsk)),
             None => self.heaviest_tipset_key(),
         }
     }

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)) => {

src/db/memory.rs

@@ -154,9 +154,8 @@ impl BitswapStoreReadWrite for MemoryDB {
 }
 
 impl super::HeaviestTipsetKeyProvider for MemoryDB {
-    fn heaviest_tipset_key(&self) -> anyhow::Result<TipsetKey> {
-        SettingsStoreExt::read_obj::<TipsetKey>(self, crate::db::setting_keys::HEAD_KEY)?
-            .context("head key not found")
+    fn heaviest_tipset_key(&self) -> anyhow::Result<Option<TipsetKey>> {
+        SettingsStoreExt::read_obj::<TipsetKey>(self, crate::db::setting_keys::HEAD_KEY)
     }
 
     fn set_heaviest_tipset_key(&self, tsk: &TipsetKey) -> anyhow::Result<()> {

src/db/mod.rs

@@ -237,14 +237,14 @@ impl<T: PersistentStore> PersistentStore for &Arc<T> {
 
 pub trait HeaviestTipsetKeyProvider {
     /// Returns the currently tracked heaviest tipset.
-    fn heaviest_tipset_key(&self) -> anyhow::Result<TipsetKey>;
+    fn heaviest_tipset_key(&self) -> anyhow::Result<Option<TipsetKey>>;
 
     /// Sets heaviest tipset.
     fn set_heaviest_tipset_key(&self, tsk: &TipsetKey) -> anyhow::Result<()>;
 }
 
 impl<T: HeaviestTipsetKeyProvider> HeaviestTipsetKeyProvider for Arc<T> {
-    fn heaviest_tipset_key(&self) -> anyhow::Result<TipsetKey> {
+    fn heaviest_tipset_key(&self) -> anyhow::Result<Option<TipsetKey>> {
         self.as_ref().heaviest_tipset_key()
     }
 

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<TipsetKey> {
-        super::SettingsStoreExt::read_obj::<TipsetKey>(self, super::setting_keys::HEAD_KEY)?
-            .context("head key not found")
+    fn heaviest_tipset_key(&self) -> anyhow::Result<Option<TipsetKey>> {
+        super::SettingsStoreExt::read_obj::<TipsetKey>(self, super::setting_keys::HEAD_KEY)
     }
 
     fn set_heaviest_tipset_key(&self, tsk: &TipsetKey) -> anyhow::Result<()> {

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;
@@ -551,28 +551,23 @@ 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(HeadChanges { reverts, applies }) => {
                         head_change(
                             api.as_ref(),
                             bls_sig_cache.as_ref(),
                             repub_trigger.clone(),
                             republished.as_ref(),
                             pending.as_ref(),
-                            cur.as_ref(),
-                            rev,
-                            app,
+                            &current_ts,
+                            reverts,
+                            applies,
                         )
                         .await
                         .context("Error changing head")?;

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<HeadChange>;
+    fn subscribe_head_changes(&self) -> Subscriber<HeadChanges>;
     /// 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<DB> {
-    subscriber: Publisher<HeadChange>,
+    subscriber: Publisher<HeadChanges>,
     sm: Arc<StateManager<DB>>,
 }
 
@@ -73,7 +73,7 @@ impl<DB> Provider for MpoolRpcProvider<DB>
 where
     DB: Blockstore + Sync + Send + 'static,
 {
-    fn subscribe_head_changes(&self) -> Subscriber<HeadChange> {
+    fn subscribe_head_changes(&self) -> Subscriber<HeadChanges> {
         self.subscriber.subscribe()
     }
 

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<TestApiInner>,
-    pub publisher: Publisher<HeadChange>,
+    pub publisher: Publisher<HeadChanges>,
 }
 
 #[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<HeadChange> {
+    fn subscribe_head_changes(&self) -> Subscriber<HeadChanges> {
         self.publisher.subscribe()
     }