fix: broadcast head changes only when subscribed

src/chain/store/chain_store.rs

@@ -163,23 +163,25 @@ where
             .set_heaviest_tipset_key(head.key())?;
         let old_head = std::mem::replace(&mut *self.heaviest_tipset_cache.write(), head.clone());
 
-        let changes = match crate::rpc::chain::chain_get_path(self, old_head.key(), head.key()) {
-            Ok(changes) => changes,
-            Err(e) => {
-                // Do not warn when the old head is genesis
-                if old_head.epoch() > 0 {
-                    error!("failed to get chain path changes: {e}");
-                }
-                // Fallback to single apply
-                PathChanges {
-                    applies: vec![head],
-                    reverts: vec![],
+        if crate::utils::broadcast::has_subscribers(&self.head_changes_tx) {
+            let changes = match crate::rpc::chain::chain_get_path(self, old_head.key(), head.key())
+            {
+                Ok(changes) => changes,
+                Err(e) => {
+                    // Do not warn when the old head is genesis
+                    if old_head.epoch() > 0 {
+                        error!("failed to get chain path changes: {e}");
+                    }
+                    // Fallback to single apply
+                    PathChanges {
+                        applies: vec![head],
+                        reverts: vec![],
+                    }
                 }
+            };
+            if self.head_changes_tx.send(changes).is_err() {
+                debug!("did not publish changes, no active receivers");
             }
-        };
-
-        if self.head_changes_tx.send(changes).is_err() {
-            debug!("did not publish changes, no active receivers");
         }
 
         Ok(())

src/db/parity_db.rs

@@ -6,10 +6,9 @@ use crate::blocks::TipsetKey;
 use crate::db::{DBStatistics, parity_db_config::ParityDbConfig};
 use crate::libp2p_bitswap::{BitswapStoreRead, BitswapStoreReadWrite};
 use crate::rpc::eth::types::EthHash;
-use crate::utils::multihash::prelude::*;
+use crate::utils::{broadcast::has_subscribers, multihash::prelude::*};
 use anyhow::{Context as _, anyhow};
 use cid::Cid;
-use futures::FutureExt;
 use fvm_ipld_blockstore::Blockstore;
 use fvm_ipld_encoding::DAG_CBOR;
 use parity_db::{CompressionType, Db, Operation, Options};
@@ -220,10 +219,6 @@ impl EthMappingsStore for ParityDb {
     }
 }
 
-fn has_subscribers<T>(tx: &tokio::sync::broadcast::Sender<T>) -> bool {
-    tx.closed().now_or_never().is_none()
-}
-
 impl Blockstore for ParityDb {
     fn get(&self, k: &Cid) -> anyhow::Result<Option<Vec<u8>>> {
         let column = Self::choose_column(k);

src/utils/broadcast/mod.rs

@@ -0,0 +1,12 @@
+// Copyright 2019-2026 ChainSafe Systems
+// SPDX-License-Identifier: Apache-2.0, MIT
+
+#[cfg(test)]
+mod tests;
+
+use futures::FutureExt as _;
+
+/// Returns `true` if there are any active subscribers to the given broadcast channel.
+pub fn has_subscribers<T>(tx: &tokio::sync::broadcast::Sender<T>) -> bool {
+    tx.closed().now_or_never().is_none()
+}

src/utils/broadcast/tests.rs

@@ -0,0 +1,25 @@
+// Copyright 2019-2026 ChainSafe Systems
+// SPDX-License-Identifier: Apache-2.0, MIT
+
+use super::*;
+
+#[tokio::test]
+async fn test_has_subscribers() {
+    let (tx, mut rx1) = tokio::sync::broadcast::channel::<u32>(16);
+    let mut rx2 = tx.subscribe();
+    tx.send(10).unwrap();
+    assert_eq!(rx1.recv().await.unwrap(), 10);
+    drop(rx1);
+    assert!(has_subscribers(&tx));
+
+    assert_eq!(rx2.recv().await.unwrap(), 10);
+    drop(rx2);
+    assert!(!has_subscribers(&tx));
+
+    let mut rx3 = tx.subscribe();
+    tx.send(10).unwrap();
+    assert_eq!(rx3.recv().await.unwrap(), 10);
+    assert!(has_subscribers(&tx));
+    drop(rx3);
+    assert!(!has_subscribers(&tx));
+}

src/utils/mod.rs

@@ -1,6 +1,7 @@
 // Copyright 2019-2026 ChainSafe Systems
 // SPDX-License-Identifier: Apache-2.0, MIT
 
+pub mod broadcast;
 pub mod cache;
 pub mod cid;
 pub mod db;