chore: enumify head change type

CHANGELOG.md

@@ -59,6 +59,8 @@
 
 - [#7480](https://github.com/ChainSafe/forest/pull/7480): `forest-tool api serve --index-backfill-epochs N` now backfills N epochs below the chain head. Previously the chain head consumed one of the requested epochs, so only N-1 were indexed.
 
+- [#7442](https://github.com/ChainSafe/forest/issues/7442): Fixed `ChainNotify` bug where the first notification was skipped; the behavior is now consistent with Lotus.
+
 ## Forest v0.35.0 "Shravan"
 
 Non-mandatory release for all node operators. It includes some fixes and improvements, notably around state-related RPC. Note that this release contains breaking changes, so please read the changelog carefully before upgrading.

src/chain/store/chain_store.rs

@@ -20,6 +20,7 @@ use crate::shim::{
 };
 use crate::state_manager::ExecutedTipset;
 use crate::utils::db::{BlockstoreExt, CborStoreExt};
+use crate::utils::publisher::Publisher;
 use crate::{
     blocks::{CachingBlockHeader, Tipset, TipsetKey, TxMeta},
     db::{DbImpl, EthMappingsStoreExt as _, HeaviestTipsetKeyProvider},
@@ -40,10 +41,6 @@ use std::{
     num::NonZeroUsize,
     sync::atomic::{self, AtomicI64},
 };
-use tokio::sync::broadcast;
-
-// A cap on the size of the future_sink
-const SINK_CAP: usize = 200;
 
 // Assume a tipset has 5 blocks on average, we cache 1-day-worth of validated blocks. (5 * 2 * 60 * 24 = 14400)
 const VALIDATED_BLOCKS_CACHE_SIZE: NonZeroUsize = nonzero!(14400usize);
@@ -82,7 +79,7 @@ pub type HeadChanges = PathChanges<Tipset>;
 /// to allow a consistent `ChainStore` to be shared across tasks.
 pub struct ChainStore {
     /// Publisher for head change events
-    head_changes_tx: broadcast::Sender<HeadChanges>,
+    head_changes: Publisher<HeadChanges>,
 
     /// Heaviest tipset cache
     heaviest_tipset: Arc<ArcSwap<Tipset>>,
@@ -118,7 +115,7 @@ pub struct ChainStore {
 impl ShallowClone for ChainStore {
     fn shallow_clone(&self) -> Self {
         Self {
-            head_changes_tx: self.head_changes_tx.clone(),
+            head_changes: self.head_changes.clone(),
             heaviest_tipset: self.heaviest_tipset.shallow_clone(),
             f3_finalized_tipset: self.f3_finalized_tipset.shallow_clone(),
             ec_calculator_finalized_epoch: self.ec_calculator_finalized_epoch.shallow_clone(),
@@ -142,7 +139,6 @@ impl ChainStore {
         let db = db.into();
         let genesis = genesis.into();
         anyhow::ensure!(genesis.epoch() == 0, "genesis tipset must be at epoch 0");
-        let (publisher, _) = broadcast::channel(SINK_CAP);
         let head = if let Some(head_tsk) = db
             .heaviest_tipset_key()
             .context("failed to load head tipset key")?
@@ -166,7 +162,7 @@ impl ChainStore {
             }
         }));
         Ok(Self {
-            head_changes_tx: publisher,
+            head_changes: Publisher::default(),
             chain_index,
             tipset_tracker: TipsetTracker::new(db, chain_config.clone()),
             heaviest_tipset,
@@ -254,7 +250,7 @@ impl ChainStore {
         }
 
         let old_head = self.heaviest_tipset.swap(head.shallow_clone().into());
-        if crate::utils::broadcast::has_subscribers(&self.head_changes_tx) {
+        if old_head.key() != head.key() && self.head_changes.has_subscribers() {
             let changes = match crate::rpc::chain::chain_get_path(self, old_head.key(), head.key())
             {
                 Ok(changes) => changes,
@@ -270,8 +266,8 @@ impl ChainStore {
                     }
                 }
             };
-            if self.head_changes_tx.send(changes).is_err() {
-                debug!("did not publish changes, no active receivers");
+            if !changes.is_empty() {
+                self.head_changes.publish(changes);
             }
         }
 
@@ -344,9 +340,18 @@ impl ChainStore {
         self.heaviest_tipset.load().as_ref().shallow_clone()
     }
 
-    /// Subscribes head changes.
-    pub fn subscribe_head_changes(&self) -> broadcast::Receiver<HeadChanges> {
-        self.head_changes_tx.subscribe()
+    /// Subscribes to head changes with an unbounded, lossless queue. Use this for
+    /// consumers that must not miss any head change (e.g. the chain/message indexers).
+    pub fn subscribe_head_changes(&self) -> flume::Receiver<HeadChanges> {
+        self.head_changes.subscribe()
+    }
+
+    /// Subscribes to head changes with a bounded queue that drops events for this
+    /// subscriber alone once it falls `cap` behind. Use this for best-effort consumers
+    /// that must not be able to grow memory without bound, e.g. ones driven by an
+    /// untrusted RPC client's read rate.
+    pub fn subscribe_head_changes_bounded(&self, cap: usize) -> flume::Receiver<HeadChanges> {
+        self.head_changes.subscribe_bounded(cap)
     }
 
     /// Returns a borrowed key-value store instance.

src/daemon/db_util.rs

@@ -29,7 +29,6 @@ use std::{
     time,
 };
 use tokio::io::AsyncWriteExt;
-use tokio::sync::broadcast::error::TryRecvError;
 use tokio_util::sync::CancellationToken;
 use tracing::{debug, info, warn};
 use url::Url;
@@ -742,7 +741,7 @@ pub async fn run_backfill(
     let cancel = guard.cancellation_token();
 
     // Subscribe before the walk so applies/reverts that happen during it are observed.
-    let mut head_rx = state_manager.chain_store().subscribe_head_changes();
+    let head_rx = state_manager.chain_store().subscribe_head_changes();
 
     // Optionally clamp the start below finality to avoid indexing revert-prone near-head tipsets.
     let start_ts = if options.allow_near_head {
@@ -824,32 +823,16 @@ pub async fn run_backfill(
     // Re-index tipsets applied during the walk so the canonical mapping wins.
     if !report.cancelled {
         let mut extra: Vec<(SignedMessage, u64)> = vec![];
-        loop {
-            match head_rx.try_recv() {
-                Ok(changes) => {
-                    for ts in changes.applies {
-                        if ts.epoch() >= lowest_epoch && ts.epoch() <= start_ts.epoch() {
-                            tracing::debug!(
-                                "re-indexing tipset @{} applied during backfill",
-                                ts.epoch()
-                            );
-                            if let Err(e) =
-                                process_ts(&ts, state_manager, &mut extra, options.allow_recompute)
-                                    .await
-                            {
-                                tracing::warn!(
-                                    "failed to re-index applied tipset @{}: {e:#}",
-                                    ts.epoch()
-                                );
-                            }
-                        }
+        for changes in head_rx.try_iter() {
+            for ts in changes.applies {
+                if ts.epoch() >= lowest_epoch && ts.epoch() <= start_ts.epoch() {
+                    tracing::debug!("re-indexing tipset @{} applied during backfill", ts.epoch());
+                    if let Err(e) =
+                        process_ts(&ts, state_manager, &mut extra, options.allow_recompute).await
+                    {
+                        tracing::warn!("failed to re-index applied tipset @{}: {e:#}", ts.epoch());
                     }
                 }
-                Err(TryRecvError::Empty) | Err(TryRecvError::Closed) => break,
-                Err(TryRecvError::Lagged(n)) => {
-                    tracing::warn!("backfill head-change listener lagged: skipped {n} events");
-                    continue;
-                }
             }
         }
         if !extra.is_empty() {

src/daemon/mod.rs

@@ -743,30 +743,23 @@ fn maybe_start_indexer_service(
         && !opts.stateless
         && !ctx.state_manager.chain_config().is_devnet()
     {
-        let mut head_changes_rx = ctx.state_manager.chain_store().subscribe_head_changes();
+        let head_changes_rx = ctx.state_manager.chain_store().subscribe_head_changes();
         let chain_store = ctx.state_manager.chain_store().shallow_clone();
         services.spawn(async move {
             tracing::info!("Starting indexer service");
 
             // Continuously listen for head changes
-            loop {
-                match head_changes_rx.recv().await {
-                    Ok(changes) => {
-                        for ts in changes.applies {
-                            tracing::debug!("Indexing tipset {}", ts.key());
-                            let delegated_messages = chain_store
-                                .headers_delegated_messages(ts.block_headers().iter())?;
-                            // Head indexing writes the newest tipset, so use the blind-write
-                            // fast path (no read-before-write timestamp comparison).
-                            chain_store.process_signed_messages(&delegated_messages, false)?;
-                        }
-                    }
-                    Err(RecvError::Lagged(n)) => {
-                        warn!("indexer service lagged: skipping {n} events")
-                    }
-                    Err(RecvError::Closed) => break Ok(()),
+            while let Ok(changes) = head_changes_rx.recv_async().await {
+                for ts in changes.applies {
+                    tracing::debug!("Indexing tipset {}", ts.key());
+                    let delegated_messages =
+                        chain_store.headers_delegated_messages(ts.block_headers().iter())?;
+                    // Head indexing writes the newest tipset, so use the blind-write
+                    // fast path (no read-before-write timestamp comparison).
+                    chain_store.process_signed_messages(&delegated_messages, false)?;
                 }
             }
+            Ok(())
         });
 
         // Run the collector only if chain indexer is enabled

src/message_pool/msgpool/msg_pool.rs

@@ -43,7 +43,7 @@ use nonzero_ext::nonzero;
 use parking_lot::RwLock as SyncRwLock;
 use std::num::NonZeroUsize;
 use std::time::Duration;
-use tokio::{sync::broadcast::error::RecvError, task::JoinSet, time::interval};
+use tokio::{task::JoinSet, time::interval};
 use tracing::warn;
 
 /// Maximum size of a serialized message in bytes. Anti-DoS measure to keep
@@ -541,23 +541,15 @@ where
         // Reacts to new HeadChanges
         {
             let mp = mp.shallow_clone();
-            let mut head_changes_rx = mp.api.subscribe_head_changes();
+            let head_changes_rx = mp.api.subscribe_head_changes();
             services.spawn(async move {
-                loop {
-                    match head_changes_rx.recv().await {
-                        Ok(HeadChanges { reverts, applies }) => {
-                            if let Err(e) = mp.apply_head_change(reverts, applies).await {
-                                tracing::warn!("Error changing head: {e}");
-                            }
-                        }
-                        Err(RecvError::Lagged(e)) => {
-                            warn!("Head change subscriber lagged: skipping {e} events");
-                        }
-                        Err(RecvError::Closed) => {
-                            break Ok(());
-                        }
+                while let Ok(HeadChanges { reverts, applies }) = head_changes_rx.recv_async().await
+                {
+                    if let Err(e) = mp.apply_head_change(reverts, applies).await {
+                        tracing::warn!("Error changing head: {e}");
                     }
                 }
+                Ok(())
             });
         }
 

src/message_pool/msgpool/provider.rs

@@ -18,15 +18,14 @@ 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
 /// required for the message pool.
 #[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<HeadChanges>;
+    fn subscribe_head_changes(&self) -> flume::Receiver<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
@@ -65,7 +64,7 @@ pub trait Provider {
 }
 
 impl Provider for ChainStore {
-    fn subscribe_head_changes(&self) -> broadcast::Receiver<HeadChanges> {
+    fn subscribe_head_changes(&self) -> flume::Receiver<HeadChanges> {
         self.subscribe_head_changes()
     }
 

src/message_pool/msgpool/test_provider.rs

@@ -14,17 +14,17 @@ use crate::cid_collections::CidHashMap;
 use crate::message::{ChainMessage, MessageRead as _, SignedMessage};
 use crate::message_pool::{Error, provider::Provider};
 use crate::shim::{address::Address, econ::TokenAmount, message::Message, state_tree::ActorState};
+use crate::utils::publisher::Publisher;
 use ahash::HashMap;
 use cid::Cid;
 use num::BigInt;
 use parking_lot::Mutex;
-use tokio::sync::broadcast;
 
 /// Structure used for creating a provider when writing tests involving message
 /// pool
 pub struct TestApi {
     pub inner: Mutex<TestApiInner>,
-    pub head_changes_tx: broadcast::Sender<HeadChanges>,
+    pub head_changes: Publisher<HeadChanges>,
 }
 
 #[derive(Default)]
@@ -42,27 +42,25 @@ pub struct TestApiInner {
 impl Default for TestApi {
     /// Create a new `TestApi`
     fn default() -> Self {
-        let (head_changes_tx, _) = broadcast::channel(1);
         TestApi {
             inner: Mutex::new(TestApiInner {
                 max_actor_pending_messages: 20000,
                 ..TestApiInner::default()
             }),
-            head_changes_tx,
+            head_changes: Publisher::default(),
         }
     }
 }
 
 impl TestApi {
     /// Constructor for a `TestApi` with custom number of max pending messages
     pub fn with_max_actor_pending_messages(max_actor_pending_messages: u64) -> Self {
-        let (publisher, _) = broadcast::channel(1);
         TestApi {
             inner: Mutex::new(TestApiInner {
                 max_actor_pending_messages,
                 ..TestApiInner::default()
             }),
-            head_changes_tx: publisher,
+            head_changes: Publisher::default(),
         }
     }
 
@@ -83,12 +81,10 @@ impl TestApi {
 
     /// Set the heaviest tipset for `TestApi`
     pub fn set_heaviest_tipset(&self, ts: Tipset) {
-        self.head_changes_tx
-            .send(HeadChanges {
-                applies: vec![ts],
-                reverts: vec![],
-            })
-            .unwrap();
+        self.head_changes.publish(HeadChanges {
+            applies: vec![ts],
+            reverts: vec![],
+        });
     }
 
     pub fn next_block(&self) -> CachingBlockHeader {
@@ -140,8 +136,8 @@ impl TestApiInner {
 }
 
 impl Provider for TestApi {
-    fn subscribe_head_changes(&self) -> broadcast::Receiver<HeadChanges> {
-        self.head_changes_tx.subscribe()
+    fn subscribe_head_changes(&self) -> flume::Receiver<HeadChanges> {
+        self.head_changes.subscribe()
     }
 
     fn get_heaviest_tipset(&self) -> Tipset {