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
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
//! Cycle loop. `start_with` drives [`Mempool::tick_with`] every
//! [`PERIOD`]. Each cycle is wrapped in `catch_unwind` so a panic
//! doesn't freeze the snapshot. `parking_lot` locks don't poison.
use std::{
any::Any,
panic::{AssertUnwindSafe, catch_unwind},
sync::atomic::Ordering,
thread,
time::{Duration, Instant},
};
use brk_error::Result;
use brk_types::{TxOut, Txid, Vout};
use rustc_hash::FxHashMap;
use tracing::error;
use crate::{
Inner, Mempool,
cycle::{Cycle, CycleDiff},
steps::{Applier, Fetched, Fetcher, Preparer, Prevouts},
};
const PERIOD: Duration = Duration::from_millis(1000);
impl Mempool {
/// Infinite update loop with a 1s interval. Resolves
/// confirmed-parent prevouts via the default `getrawtransaction`
/// resolver. Requires bitcoind started with `txindex=1`. Discards
/// per-cycle [`Cycle`] events - use [`Mempool::tick`] to consume them.
pub fn start(&self) {
self.start_with(Prevouts::rpc_resolver(self.0.client.clone()));
}
/// Variant of `start` that uses a caller-supplied resolver for
/// confirmed-parent prevouts (typically backed by an indexer).
///
/// Sleep is `PERIOD - work_duration`, so a 350ms cycle followed by
/// a 100ms cycle still ticks roughly every `PERIOD`. When work
/// overruns `PERIOD`, the next cycle starts immediately.
///
/// # Panics
///
/// Panics if a driver is already running on this `Mempool` instance.
/// One `Mempool` may host at most one driver. Spawn another instance
/// for additional loops.
pub fn start_with<F>(&self, resolver: F)
where
F: Fn(&[(Txid, Vout)]) -> FxHashMap<(Txid, Vout), TxOut> + Send,
{
if self
.0
.started
.compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
.is_err()
{
panic!("Mempool::start_with already running on this instance");
}
loop {
let started = Instant::now();
let outcome = catch_unwind(AssertUnwindSafe(|| {
if let Err(e) = self.tick_with(&resolver) {
error!("update failed: {e}");
}
}));
if let Err(payload) = outcome {
error!(
"mempool update panicked, continuing loop: {}",
Self::panic_msg(&payload)
);
}
if let Some(rest) = PERIOD.checked_sub(started.elapsed()) {
thread::sleep(rest);
}
}
}
/// One sync cycle: fetch, prepare, apply, fill prevouts, rebuild.
/// Returns a [`Cycle`] reporting everything that changed. Uses the
/// default `getrawtransaction` resolver for confirmed-parent
/// prevouts (requires `txindex=1`).
///
/// # Errors
///
/// Propagates any failure from the initial RPC fetch (network drop,
/// auth, bitcoind error). Steps after `Fetcher::fetch` are infallible
/// today. The resolver itself swallows its own errors and retries
/// next cycle.
pub fn tick(&self) -> Result<Cycle> {
self.tick_with(Prevouts::rpc_resolver(self.0.client.clone()))
}
/// Variant of [`Mempool::tick`] with a caller-supplied resolver for
/// confirmed-parent prevouts. The resolver MUST resolve confirmed
/// prevouts only. Mempool-to-mempool chains are wired internally
/// and the resolver is never called for them.
///
/// # Errors
///
/// Same as [`Mempool::tick`]: only the RPC fetch is fallible.
pub fn tick_with<F>(&self, resolver: F) -> Result<Cycle>
where
F: Fn(&[(Txid, Vout)]) -> FxHashMap<(Txid, Vout), TxOut>,
{
let started = Instant::now();
let Inner {
client,
state,
rebuilder,
..
} = &*self.0;
let Fetched {
state: rpc,
new_entries,
new_txs,
block_template_txids,
} = Fetcher::fetch(client, state)?;
let pulled = Preparer::prepare(&rpc.live_txids, new_entries, new_txs, state);
let mut diff = CycleDiff::default();
let prev_snapshot = rebuilder.snapshot();
Applier::apply(state, &prev_snapshot, pulled, &mut diff);
drop(prev_snapshot);
Prevouts::fill(state, &mut diff, resolver);
rebuilder.tick(state, &block_template_txids, rpc.min_fee);
let CycleDiff {
added,
removed,
addrs,
} = diff;
let (addr_enters, addr_leaves) = addrs.into_vecs();
Ok(Cycle {
added,
removed,
addr_enters,
addr_leaves,
tip_hash: rpc.tip_hash,
tip_height: rpc.tip_height,
info: self.info(),
snapshot: rebuilder.snapshot(),
took: started.elapsed(),
})
}
fn panic_msg(payload: &(dyn Any + Send)) -> &str {
payload
.downcast_ref::<&'static str>()
.copied()
.or_else(|| payload.downcast_ref::<String>().map(String::as_str))
.unwrap_or("<non-string panic payload>")
}
}
#[cfg(test)]
mod tests {
use std::panic::catch_unwind;
use rustc_hash::FxHashMap;
use super::*;
#[test]
#[should_panic(expected = "Mempool::start_with already running on this instance")]
fn double_start_panics_with_documented_message() {
let mempool = Mempool::for_test();
// Simulate a prior `start_with` having grabbed the latch. We
// can't actually call it first because the real call enters an
// infinite loop. Flipping the atomic is what the runtime check
// observes anyway.
mempool.0.started.store(true, Ordering::Release);
mempool.start_with(|_: &[(Txid, Vout)]| FxHashMap::default());
}
#[test]
fn panic_msg_extracts_static_str_payload() {
let payload = catch_unwind(|| panic!("boom static")).unwrap_err();
assert_eq!(Mempool::panic_msg(payload.as_ref()), "boom static");
}
#[test]
fn panic_msg_extracts_string_payload() {
let payload = catch_unwind(|| panic!("boom owned {}", 42)).unwrap_err();
assert_eq!(Mempool::panic_msg(payload.as_ref()), "boom owned 42");
}
#[test]
fn panic_msg_falls_back_for_non_string_payload() {
// Payload that isn't &str or String: the helper labels it
// explicitly instead of dropping it on the floor.
let payload = catch_unwind(|| std::panic::panic_any(42u32)).unwrap_err();
assert_eq!(Mempool::panic_msg(payload.as_ref()), "<non-string panic payload>");
}
}