fix: head change publisher

[LesnyRumcajs]

Summary of changes

Changes introduced in this pull request:

• replace lossy tokio::broadcast with a flume-based publisher. Each subscriber gets its own queue. This also made the consumer code more ergonomic. The change proposed in the linked issue wouldn't really help; it'd just push the loss elsewhere.
• ChainNotify sampled the current head the subscribed, then skipped the first head-change message. Not sure why this behavior was introduced in the first place, but it doesn't match Lotus (tested on live nodes). It now subscribes before sampling and forwards every change. Added some regression tests around it. Pretty sure it'd mess curio integration.

Reference issue to close (if applicable)

Closes #7442

Other information and links

Change checklist

• I have performed a self-review of my own code,
• I have made corresponding changes to the documentation. All new code adheres to the team's documentation standards,
• I have added tests that prove my fix is effective or that my feature works (if possible),
• I have made sure the CHANGELOG is up-to-date. All user-facing changes should be reflected in this document.

Outside contributions

• This pull request is based on an issue that a maintainer has accepted (see Before Opening a Pull Request).
• I have read and agree to the CONTRIBUTING document.
• I have read and agree to the AI Policy document. I understand that failure to comply with the guidelines will lead to rejection of the pull request.

Summary by CodeRabbit

• Bug Fixes

  • Chain notifications no longer skip the first update and now match Lotus behavior.
  • Improved delivery of head changes, including reorg revert and apply events.
  • Prevented reverted-chain messages from appearing after reorganization.
  • Reduced duplicate or empty chain notifications.

• Improvements

  • Added reliable subscription handling for chain and head-change updates.
  • Added bounded buffering for Ethereum newHeads subscriptions to protect against slow clients.

• Documentation

  • Added a changelog entry describing the corrected notification behavior.

[coderabbitai]

No actionable comments were generated in the recent review. 🎉

ℹ️ Recent review info

⚙️ Run configuration

Configuration used: Repository UI

Review profile: CHILL

Plan: Pro

Run ID: 13001af4-fea6-41fc-b602-c7d2312dc132

📥 Commits

Reviewing files that changed from the base of the PR and between a873e05 and ae3a4b3.

📒 Files selected for processing (12)

• CHANGELOG.md
• src/chain/store/chain_store.rs
• src/daemon/db_util.rs
• src/daemon/mod.rs
• src/message_pool/msgpool/msg_pool.rs
• src/message_pool/msgpool/provider.rs
• src/message_pool/msgpool/test_provider.rs
• src/rpc/methods/chain.rs
• src/rpc/methods/eth/pubsub.rs
• src/state_manager/message_search.rs
• src/utils/mod.rs
• src/utils/publisher.rs

🔗 Linked repositories identified

CodeRabbit considers these linked repositories for cross-repo context during reviews:

• filecoin-project/lotus (manual)

---

Walkthrough

The PR adds a thread-safe Publisher, migrates head-change delivery from Tokio broadcast to flume receivers, updates all consumers, and fixes ChainNotify subscription ordering. Bounded Ethereum subscriptions can drop events for slow clients.

Changes

Head-change delivery

Layer / File(s)	Summary
Publisher implementation src/utils/mod.rs, src/utils/publisher.rs	Adds lossless and bounded subscriptions, non-blocking publication, disconnected-subscriber pruning, and coverage tests.
ChainStore publication and subscriptions src/chain/store/chain_store.rs, src/message_pool/msgpool/provider.rs, src/message_pool/msgpool/test_provider.rs	Replaces the broadcast sender with Publisher. Publishes only nonempty changes after actual head changes. Exposes unbounded and bounded receivers.
Head-change consumer migration src/daemon/db_util.rs, src/daemon/mod.rs, src/message_pool/msgpool/msg_pool.rs, src/rpc/methods/eth/pubsub.rs, src/state_manager/message_search.rs	Updates indexer, message-pool, Ethereum subscription, and message-search consumers to use asynchronous or iterator-based reception.
ChainNotify ordering and validation src/rpc/methods/chain.rs, CHANGELOG.md	Subscribes before sampling the current head and forwards every received change. Tests cover ordering, deduplication, reorgs, and initial delivery.

Estimated code review effort: 4 (Complex) | ~45 minutes

Merge Risk: ⚪ Minimal · up to ae3a4

The change is merge-ready after normal checks and review; no actionable merge-blocking risk remains.

Sequence Diagram(s)

sequenceDiagram
  participant Client
  participant chain_notify_inner
  participant ChainStore
  participant Publisher
  Client->>chain_notify_inner: request chain notifications
  chain_notify_inner->>ChainStore: subscribe before sampling current head
  ChainStore-->>chain_notify_inner: return flume receiver
  chain_notify_inner->>Client: send current head
  Publisher-->>chain_notify_inner: deliver HeadChanges
  chain_notify_inner->>Client: forward head changes

Loading

🚥 Pre-merge checks | ✅ 4 | ❌ 1

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Linked Issues check	⚠️ Warning	The PR addresses lossless delivery and consumer updates, but it does not migrate the channel to the required async-broadcast implementation.	Use async-broadcast with overflow disabled, and add or confirm tests for slow NoSQL and SQLite consumers and clean shutdown.

✅ Passed checks (4 passed)

Check name	Status	Explanation
Out of Scope Changes check	✅ Passed	The changes stay focused on head-change publication, subscription consumers, ChainNotify behavior, and related regression tests.
Docstring Coverage	✅ Passed	Docstring coverage is 100.00% which is sufficient. The required threshold is 80.00%.
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The title clearly identifies the main change: replacing the head-change publisher with a new implementation.

✨ Finishing Touches

📝 Generate docstrings

• Create stacked PR
• Commit on current branch

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch head-change-publisher-fix

✨ Simplify code

• Create PR with simplified code
• Commit simplified code in branch head-change-publisher-fix

---

_{Comment @coderabbitai help to get the list of available commands.}

[codecov]

Codecov Report

❌ Patch coverage is 83.94161% with 44 lines in your changes missing coverage. Please review.
✅ Project coverage is 65.59%. Comparing base (a873e05) to head (ae3a4b3).
✅ All tests successful. No failed tests found.

Files with missing lines	Patch %	Lines
src/state_manager/message_search.rs	86.58%	10 Missing and 1 partial ⚠️
src/daemon/db_util.rs	0.00%	8 Missing ⚠️
src/daemon/mod.rs	0.00%	8 Missing ⚠️
src/rpc/methods/chain.rs	92.85%	5 Missing and 1 partial ⚠️
src/rpc/methods/eth/pubsub.rs	0.00%	6 Missing ⚠️
src/message_pool/msgpool/msg_pool.rs	40.00%	2 Missing and 1 partial ⚠️
src/chain/store/chain_store.rs	80.00%	2 Missing ⚠️

Additional details and impacted files

Files with missing lines	Coverage Δ
src/message_pool/msgpool/provider.rs	69.84% <100.00%> (ø)
src/message_pool/msgpool/test_provider.rs	99.02% <100.00%> (-0.02%)	⬇️
src/utils/mod.rs	77.68% <ø> (ø)
src/utils/publisher.rs	100.00% <100.00%> (ø)
src/chain/store/chain_store.rs	73.09% <80.00%> (-0.43%)	⬇️
src/message_pool/msgpool/msg_pool.rs	88.90% <40.00%> (+0.23%)	⬆️
src/rpc/methods/chain.rs	53.80% <92.85%> (+4.72%)	⬆️
src/rpc/methods/eth/pubsub.rs	58.36% <0.00%> (-0.44%)	⬇️
src/daemon/db_util.rs	56.23% <0.00%> (+0.70%)	⬆️
src/daemon/mod.rs	24.46% <0.00%> (+0.10%)	⬆️
... and 1 more

... and 9 files with indirect coverage changes

---

Continue to review full report in Codecov by Harness.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update a873e05...ae3a4b3. Read the comment docs.

🚀 New features to boost your workflow:

• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.