forest/message_pool/msgpool/
republish.rs

1// Copyright 2019-2026 ChainSafe Systems
2// SPDX-License-Identifier: Apache-2.0, MIT
3
4//! Tracks which CIDs were already broadcast in the current republish cycle
5//! and exposes a trigger to wake the republish task early.
6
7use crate::message::{MessageRead as _, SignedMessage};
8use crate::message_pool::{
9    Error,
10    msg_chain::{Chains, create_message_chains},
11    msgpool::{MIN_GAS, msg_pool::MessagePool},
12    provider::Provider,
13    utils::get_base_fee_lower_bound,
14};
15use crate::prelude::*;
16use crate::shim::address::Address;
17use ahash::{HashMap, HashSet};
18use parking_lot::RwLock as SyncRwLock;
19
20const REPUB_TRIGGER_CAPACITY: usize = 1;
21const BASE_FEE_LOWER_BOUND_FACTOR: i64 = 10;
22const REPUB_MSG_LIMIT: usize = 30;
23
24pub(in crate::message_pool) struct RepublishState {
25    republished: SyncRwLock<HashSet<Cid>>,
26    trigger: flume::Sender<()>,
27}
28
29impl RepublishState {
30    pub(in crate::message_pool) fn new() -> (Self, flume::Receiver<()>) {
31        let (trigger, rx) = flume::bounded(REPUB_TRIGGER_CAPACITY);
32        (
33            Self {
34                republished: SyncRwLock::default(),
35                trigger,
36            },
37            rx,
38        )
39    }
40
41    /// Returns `true` if `cid` was seen by the republished state.
42    pub(in crate::message_pool) fn was_republished(&self, cid: &Cid) -> bool {
43        self.republished.read().contains(cid)
44    }
45
46    /// Wake the republish task early.
47    pub(in crate::message_pool) fn trigger(&self) -> Result<(), Error> {
48        match self.trigger.try_send(()) {
49            Ok(()) | Err(flume::TrySendError::Full(_)) => Ok(()),
50            Err(flume::TrySendError::Disconnected(_)) => {
51                Err(Error::Other("republish receiver dropped".into()))
52            }
53        }
54    }
55
56    pub(in crate::message_pool) fn replace_with<I: IntoIterator<Item = Cid>>(&self, cids: I) {
57        let mut set = self.republished.write();
58        set.clear();
59        set.extend(cids);
60    }
61}
62
63impl<T: Provider> MessagePool<T> {
64    pub(in crate::message_pool) async fn run_republish_cycle(&self) -> Result<(), Error> {
65        let ts = self.cur_tipset.read().shallow_clone();
66
67        // Only republish messages from local addresses, i.e., transactions which
68        // were sent to this node directly.
69        let local: Vec<Address> = self.local_addrs.read().iter().copied().collect();
70        let mut pending_map: HashMap<Address, HashMap<u64, SignedMessage>> =
71            HashMap::with_capacity(local.len());
72        for actor in &local {
73            if let Some(mset) = self.pending.snapshot_for(actor)
74                && !mset.msgs.is_empty()
75            {
76                pending_map.insert(*actor, mset.msgs);
77            }
78        }
79
80        let msgs =
81            select_messages_to_republish(self.api.as_ref(), &self.chain_config, &ts, pending_map)?;
82
83        for m in msgs.iter() {
84            self.publish_pubsub(m).await?;
85        }
86
87        self.republish.replace_with(msgs.iter().map(|m| m.cid()));
88
89        Ok(())
90    }
91}
92
93/// Score local senders' pending message chains for the republish broadcast.
94///
95/// Distinct from the block-producer selection path (`selection.rs`): uses
96/// the aggressive [`BASE_FEE_LOWER_BOUND_FACTOR`] of 10 (vs. 100 in the add
97/// path) and caps the result at [`REPUB_MSG_LIMIT`] messages.
98fn select_messages_to_republish<T>(
99    api: &T,
100    chain_config: &crate::networks::ChainConfig,
101    base: &crate::blocks::Tipset,
102    pending: HashMap<Address, HashMap<u64, SignedMessage>>,
103) -> Result<Vec<SignedMessage>, Error>
104where
105    T: Provider,
106{
107    let mut msgs: Vec<SignedMessage> = vec![];
108
109    let base_fee = api.chain_compute_base_fee(base)?;
110    let base_fee_lower_bound = get_base_fee_lower_bound(&base_fee, BASE_FEE_LOWER_BOUND_FACTOR);
111
112    if pending.is_empty() {
113        return Ok(msgs);
114    }
115
116    let mut chains = Chains::new();
117    for (actor, mset) in pending.iter() {
118        create_message_chains(
119            api,
120            actor,
121            mset,
122            &base_fee_lower_bound,
123            base,
124            &mut chains,
125            chain_config,
126        )?;
127    }
128
129    if chains.is_empty() {
130        return Ok(msgs);
131    }
132
133    chains.sort(false);
134
135    let mut gas_limit = crate::shim::econ::BLOCK_GAS_LIMIT;
136    let mut i = 0;
137    'l: while let Some(chain) = chains.get_mut_at(i) {
138        // we can exceed this if we have picked (some) longer chain already
139        if msgs.len() > REPUB_MSG_LIMIT {
140            break;
141        }
142
143        if gas_limit <= MIN_GAS {
144            break;
145        }
146
147        // check if chain has been invalidated
148        if !chain.valid {
149            i += 1;
150            continue;
151        }
152
153        // check if fits in block
154        if chain.gas_limit <= gas_limit {
155            // check the baseFee lower bound -- only republish messages that can be included
156            // in the chain within the next 20 blocks.
157            for m in chain.msgs.iter() {
158                if m.gas_fee_cap() < base_fee_lower_bound {
159                    let key = chains.get_key_at(i);
160                    chains.invalidate(key);
161                    continue 'l;
162                }
163                gas_limit = gas_limit.saturating_sub(m.gas_limit());
164                msgs.push(m.clone());
165            }
166
167            i += 1;
168            continue;
169        }
170
171        // we can't fit the current chain but there is gas to spare
172        // trim it and push it down
173        chains.trim_msgs_at(i, gas_limit, REPUB_MSG_LIMIT, &base_fee);
174        chains.bubble_down_after_trim(i);
175    }
176
177    if msgs.len() > REPUB_MSG_LIMIT {
178        msgs.truncate(REPUB_MSG_LIMIT);
179    }
180
181    Ok(msgs)
182}
183
184#[cfg(test)]
185mod tests {
186    use super::*;
187    use crate::message_pool::msg_chain::MsgChainNode;
188    use crate::shim::econ::TokenAmount;
189
190    fn chains_from_perfs(perfs: &[f64]) -> Chains {
191        let mut chains = Chains::new();
192        let mut key_vec = Vec::with_capacity(perfs.len());
193        for (i, &p) in perfs.iter().enumerate() {
194            let node = MsgChainNode {
195                gas_perf: p,
196                gas_reward: TokenAmount::from_atto(i as u64 + 1),
197                ..Default::default()
198            };
199            chains.push_with(node, &mut key_vec);
200        }
201        chains.key_vec = key_vec;
202        chains
203    }
204
205    #[test]
206    fn bubble_down_after_trim_restores_compare_order() {
207        let mut chains = chains_from_perfs(&[1.0, 5.0, 3.0, 4.0]);
208        chains.bubble_down_after_trim(1);
209        let perfs: Vec<f64> = (0..chains.len()).map(|i| chains[i].gas_perf).collect();
210        assert_eq!(perfs, vec![1.0, 3.0, 4.0, 5.0]);
211    }
212
213    #[test]
214    fn was_republished_reflects_replace_with() {
215        let (state, _rx) = RepublishState::new();
216        let cid = Cid::default();
217
218        assert!(
219            !state.was_republished(&cid),
220            "fresh state should not contain any CIDs",
221        );
222
223        state.replace_with([cid]);
224        assert!(
225            state.was_republished(&cid),
226            "replace_with should populate the set",
227        );
228
229        state.replace_with(std::iter::empty());
230        assert!(
231            !state.was_republished(&cid),
232            "replace_with with empty iter should clear the set",
233        );
234    }
235
236    #[test]
237    fn trigger_succeeds_when_receiver_is_alive() {
238        let (state, rx) = RepublishState::new();
239        state.trigger().expect("send should succeed");
240        rx.try_recv()
241            .expect("trigger should be observable on the receiver");
242    }
243
244    #[test]
245    fn trigger_drops_silently_when_buffer_full() {
246        let (state, _rx) = RepublishState::new();
247        state.trigger().expect("first trigger should send");
248        // Buffer (capacity 1) is now full; a second trigger must coalesce
249        // silently instead of failing head_change.
250        state
251            .trigger()
252            .expect("overflow trigger should be dropped silently");
253    }
254
255    #[test]
256    fn trigger_errors_when_receiver_disconnected() {
257        let (state, rx) = RepublishState::new();
258        drop(rx);
259        let err = state
260            .trigger()
261            .expect_err("disconnected receiver should surface as an error");
262        assert!(matches!(err, Error::Other(_)));
263    }
264}
forest/message_pool/msgpool/republish.rs

forest/message_pool/msgpool/
republish.rs
