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
//! A command's blocking state writes, run off the runtime thread in the order
//! they are queued.
//!
//! A fetch's lanes share the command's task (a bundle pull runs every entry
//! and lane as one future), so a durable write made inline there stops every
//! lane for as long as the disk takes (#2211). These writes are also order
//! sensitive across the whole command: an open that landed after a later
//! fault would clear that fault's stamp, and a voucher watermark rebase
//! replaces the lane row outright. One [`OrderedWrites`] per command
//! therefore runs a lane's open and fault peer records and its watermark
//! commits on the blocking pool, one at a time, in queue order. Other state
//! writes do not go through it: `spawn_harvest`'s `record_sample`,
//! `resolve_bootstrap`'s identity upserts, and the pool `record` /
//! `add_deposit` writes of an open or top-up.
//!
//! A worker runs on `spawn_blocking` only while writes are queued. It takes
//! them in order and exits when the queue is empty; the next queued write
//! starts a new one. The queue and the running flag share one lock, so a write
//! is never left queued with no worker to take it, and at most one worker runs
//! at a time. An idle command holds no blocking-pool thread. A worker is a
//! `spawn_blocking` writer, not a `tokio::spawn` of pull work, so a pull still
//! runs as one future. A write that panics is logged and the queue drains on,
//! so no write queued behind it is stranded. A worker the runtime drops
//! without running it, as a runtime that is shutting down does, drains the
//! queue on the thread that drops it.
use std::collections::VecDeque;
use std::panic::AssertUnwindSafe;
use std::sync::{Arc, Mutex, MutexGuard, PoisonError};
use std::time::Duration;
/// How long a command waits, before it returns, for the writes it queued
/// ([`OrderedWrites::settle`]).
pub const WRITE_DRAIN: Duration = Duration::from_secs(30);
/// One queued write.
type Write = Box<dyn FnOnce() + Send>;
/// A command-wide queue of blocking writes, run in queue order. Cloning shares
/// the queue.
#[derive(Clone, Default)]
pub struct OrderedWrites {
inner: Arc<Inner>,
}
impl std::fmt::Debug for OrderedWrites {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("OrderedWrites").finish_non_exhaustive()
}
}
#[derive(Default)]
struct Inner {
state: Mutex<State>,
}
#[derive(Default)]
struct State {
queue: VecDeque<Write>,
/// A worker is taking writes from `queue`.
running: bool,
}
impl Inner {
/// The queue state. No write runs under this lock, and each critical
/// section leaves the queue whole, so a poisoned lock is safe to reuse.
fn lock(&self) -> MutexGuard<'_, State> {
self.state.lock().unwrap_or_else(PoisonError::into_inner)
}
/// The worker body: run every queued write in order, then exit. A write
/// that panics is logged, and the next one runs.
fn drain(&self) {
loop {
let next = {
let mut state = self.lock();
let next = state.queue.pop_front();
if next.is_none() {
state.running = false;
}
next
};
let Some(write) = next else {
break;
};
if let Err(panic) = std::panic::catch_unwind(AssertUnwindSafe(write)) {
let reason = panic
.downcast_ref::<&str>()
.copied()
.or_else(|| panic.downcast_ref::<String>().map(String::as_str))
.unwrap_or("a non-string payload");
tracing::error!("a queued state write panicked: {reason}");
}
}
}
}
/// Drains the queue when it drops: inside the worker that runs it, or, when
/// the runtime drops the worker without running it, on the dropping thread.
struct DrainOnDrop(Arc<Inner>);
impl Drop for DrainOnDrop {
fn drop(&mut self) {
self.0.drain();
}
}
impl OrderedWrites {
/// Queue `write` behind every write queued before it.
///
/// A running worker takes it in turn. With none running, this call starts
/// one on the blocking pool. Outside a Tokio runtime, or when a runtime
/// that is shutting down drops the new worker, the queue drains on this
/// thread before this returns.
pub fn queue(&self, write: impl FnOnce() + Send + 'static) {
let start = {
let mut state = self.inner.lock();
state.queue.push_back(Box::new(write));
!std::mem::replace(&mut state.running, true)
};
if !start {
return;
}
let drain = DrainOnDrop(Arc::clone(&self.inner));
match tokio::runtime::Handle::try_current() {
Ok(runtime) => drop(runtime.spawn_blocking(move || drop(drain))),
Err(_) => drop(drain),
}
}
/// Queue `write` like [`Self::queue`], and return a receiver that resolves
/// with its result once it ran. The write is queued before this returns,
/// so its place in the order is fixed at the call, not at the first poll.
/// The receiver fails if the write panicked.
pub fn queue_awaitable<T: Send + 'static>(
&self,
write: impl FnOnce() -> T + Send + 'static,
) -> tokio::sync::oneshot::Receiver<T> {
let (done, landed) = tokio::sync::oneshot::channel();
self.queue(move || {
// The caller may have stopped waiting; the write still ran.
let _ = done.send(write());
});
landed
}
/// Wait, for at most [`WRITE_DRAIN`], until every write queued so far has
/// run. A command calls this before it returns, on the interrupted path
/// too, so a drop guard's watermark write is durable before the process
/// exits.
pub async fn settle(&self) {
let barrier = self.queue_awaitable(|| ());
match tokio::time::timeout(WRITE_DRAIN, barrier).await {
Ok(Ok(())) => {}
Ok(Err(_)) => tracing::warn!(
"the queued state writes did not finish; the next run may re-sign a stale \
voucher watermark"
),
Err(_) => tracing::warn!(
"the queued state writes were still running {WRITE_DRAIN:?} after the command \
ended; the next run may re-sign a stale voucher watermark"
),
}
}
/// Whether a worker is running.
#[cfg(test)]
fn running(&self) -> bool {
self.inner.lock().running
}
}
#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used)] // tests
mod tests;