forest/message_pool/msgpool/
reorg.rs

1// Copyright 2019-2026 ChainSafe Systems
2// SPDX-License-Identifier: Apache-2.0, MIT
3
4//! Reorg handling: revert + apply tipsets against the pending pool.
5
6use crate::blocks::Tipset;
7use crate::message::{MessageRead as _, SignedMessage};
8use crate::message_pool::msgpool::utils;
9use crate::message_pool::{
10    Error,
11    msg_pool::{StrictnessPolicy, TrustPolicy},
12    msgpool::{msg_pool::MessagePool, recovered_bls_messages},
13    provider::Provider,
14};
15use crate::shim::address::Address;
16use crate::utils::ShallowClone as _;
17use ahash::{HashMap, HashMapExt};
18
19impl<T> MessagePool<T>
20where
21    T: Provider + Send + Sync + 'static,
22{
23    /// Revert and/or apply tipsets to the message pool.
24    ///
25    /// - **Apply**: messages included in the new tipset are removed from the
26    ///   pending pool with `applied = true`.
27    /// - **Revert**: messages from the reverted tipset are re-added to the
28    ///   pool with [`StrictnessPolicy::Relaxed`] and [`TrustPolicy::Trusted`],
29    ///   allowing them back without nonce-gap restrictions.
30    ///
31    /// The state-nonce cache is naturally invalidated when the tipset
32    /// changes, since it is keyed by `(TipsetKey, Address)`.
33    pub(in crate::message_pool) async fn apply_head_change(
34        &self,
35        revert: Vec<Tipset>,
36        apply: Vec<Tipset>,
37    ) -> Result<(), Error> {
38        let mut repub = false;
39        let mut rmsgs: HashMap<Address, HashMap<u64, SignedMessage>> = HashMap::new();
40        for ts in revert {
41            let Ok(pts) = self.api.load_tipset(ts.parents()) else {
42                tracing::error!("error loading reverted tipset parent");
43                continue;
44            };
45            *self.cur_tipset.write() = pts;
46
47            let mut msgs: Vec<SignedMessage> = Vec::new();
48            for block in ts.block_headers() {
49                let Ok((umsg, smsgs)) = self.api.messages_for_block(block) else {
50                    tracing::error!("error retrieving messages for reverted block");
51                    continue;
52                };
53                msgs.extend(smsgs);
54                msgs.extend(recovered_bls_messages(&self.caches.bls_sig, umsg));
55            }
56
57            for msg in msgs {
58                utils::add_to_selected_msgs(msg, &mut rmsgs);
59            }
60        }
61
62        for ts in apply {
63            for b in ts.block_headers() {
64                let Ok((msgs, smsgs)) = self.api.messages_for_block(b) else {
65                    tracing::error!("error retrieving messages for block");
66                    continue;
67                };
68
69                for msg in smsgs {
70                    self.remove_applied_from_pool(&msg.from(), msg.sequence(), &mut rmsgs, &ts)
71                        .await?;
72                    if !repub && self.republish.was_republished(&msg.cid()) {
73                        repub = true;
74                    }
75                }
76                for msg in msgs {
77                    self.remove_applied_from_pool(&msg.from, msg.sequence, &mut rmsgs, &ts)
78                        .await?;
79                    if !repub && self.republish.was_republished(&msg.cid()) {
80                        repub = true;
81                    }
82                }
83            }
84            // Must stay after the removals above: `pending` relies on this order to
85            // avoid pairing a stale pool with a newer tipset.
86            *self.cur_tipset.write() = ts;
87        }
88        if repub {
89            self.republish.trigger()?;
90        }
91
92        let cur_ts = self.cur_tipset.read().shallow_clone();
93        for (_, hm) in rmsgs {
94            for (_, msg) in hm {
95                if let Err(e) = self
96                    .add_to_pool_unchecked(
97                        &cur_ts,
98                        msg,
99                        TrustPolicy::Trusted,
100                        StrictnessPolicy::Relaxed,
101                    )
102                    .await
103                {
104                    tracing::error!("Failed to read message from reorg to mpool: {}", e);
105                }
106            }
107        }
108        self.pending.shrink_to_fit();
109        Ok(())
110    }
111
112    /// Remove a message from the in-progress `rmsgs` scratch map. If the
113    /// message isn't there, fall back to removing it from the real pending
114    /// pool. Used by [`Self::apply_head_change`] when an applied tipset
115    /// includes a message that we hadn't yet seen reverted.
116    async fn remove_applied_from_pool(
117        &self,
118        from: &Address,
119        sequence: u64,
120        rmsgs: &mut HashMap<Address, HashMap<u64, SignedMessage>>,
121        ts: &Tipset,
122    ) -> Result<(), Error> {
123        if rmsgs
124            .get_mut(from)
125            .and_then(|temp| temp.remove(&sequence))
126            .is_none()
127            && let Ok(resolved) = self
128                .resolve_to_key(from, ts)
129                .await
130                .inspect_err(|e| tracing::debug!(%from, "remove: failed to resolve address: {e:#}"))
131        {
132            let _ = self.pending.remove(&resolved, sequence, true);
133        }
134        Ok(())
135    }
136}
forest/message_pool/msgpool/reorg.rs

forest/message_pool/msgpool/
reorg.rs
