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, recover_sig},
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                for msg in umsg {
55                    let msg_cid = msg.cid();
56                    let Ok(smsg) = recover_sig(&self.caches.bls_sig, msg) else {
57                        tracing::debug!("could not recover signature for bls message {}", msg_cid);
58                        continue;
59                    };
60                    msgs.push(smsg)
61                }
62            }
63
64            for msg in msgs {
65                utils::add_to_selected_msgs(msg, &mut rmsgs);
66            }
67        }
68
69        for ts in apply {
70            for b in ts.block_headers() {
71                let Ok((msgs, smsgs)) = self.api.messages_for_block(b) else {
72                    tracing::error!("error retrieving messages for block");
73                    continue;
74                };
75
76                for msg in smsgs {
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                for msg in msgs {
84                    self.remove_applied_from_pool(&msg.from, msg.sequence, &mut rmsgs, &ts)
85                        .await?;
86                    if !repub && self.republish.was_republished(&msg.cid()) {
87                        repub = true;
88                    }
89                }
90            }
91            *self.cur_tipset.write() = ts;
92        }
93        if repub {
94            self.republish.trigger()?;
95        }
96
97        let cur_ts = self.cur_tipset.read().shallow_clone();
98        for (_, hm) in rmsgs {
99            for (_, msg) in hm {
100                if let Err(e) = self
101                    .add_to_pool_unchecked(
102                        &cur_ts,
103                        msg,
104                        TrustPolicy::Trusted,
105                        StrictnessPolicy::Relaxed,
106                    )
107                    .await
108                {
109                    tracing::error!("Failed to read message from reorg to mpool: {}", e);
110                }
111            }
112        }
113        self.pending.shrink_to_fit();
114        Ok(())
115    }
116
117    /// Remove a message from the in-progress `rmsgs` scratch map. If the
118    /// message isn't there, fall back to removing it from the real pending
119    /// pool. Used by [`Self::apply_head_change`] when an applied tipset
120    /// includes a message that we hadn't yet seen reverted.
121    async fn remove_applied_from_pool(
122        &self,
123        from: &Address,
124        sequence: u64,
125        rmsgs: &mut HashMap<Address, HashMap<u64, SignedMessage>>,
126        ts: &Tipset,
127    ) -> Result<(), Error> {
128        if rmsgs
129            .get_mut(from)
130            .and_then(|temp| temp.remove(&sequence))
131            .is_none()
132            && let Ok(resolved) = self
133                .resolve_to_key(from, ts)
134                .await
135                .inspect_err(|e| tracing::debug!(%from, "remove: failed to resolve address: {e:#}"))
136        {
137            let _ = self.pending.remove(&resolved, sequence, true);
138        }
139        Ok(())
140    }
141}
forest/message_pool/msgpool/reorg.rs

forest/message_pool/msgpool/
reorg.rs
