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
137
138
139
140
141
// 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, recover_sig},
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);
for msg in umsg {
let msg_cid = msg.cid();
let Ok(smsg) = recover_sig(&self.caches.bls_sig, msg) else {
tracing::debug!("could not recover signature for bls message {}", msg_cid);
continue;
};
msgs.push(smsg)
}
}
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;
}
}
}
*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(())
}
}