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