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
//! DUR-021: the spine's reclaim pass runs behind the open door. The store
//! opens with its reclaim held; the runtime releases it once the listeners
//! are bound; every pass ending is logged with its figures by a watch thread
//! that ends when the store closes; and what each shard resumed after an
//! unclean stop is on the log. Nothing here quiesces a pass: that is an
//! operator's act on the health route, never a boot's or a stop's.
use std::sync::Arc;
#[cfg(test)]
use std::sync::{Mutex, PoisonError};
use std::time::Instant;
use haematite::{Database, EventStore, ReclaimPassOutcome};
#[path = "reclaim_release.rs"]
mod release_owner;
pub use release_owner::ReclaimRelease;
/// The watch thread's name.
pub const WATCH_THREAD: &str = "liminal-spine-reclaim-watch";
/// The release thread's name.
pub const RELEASE_THREAD: &str = "liminal-spine-reclaim-release";
/// Releases the held reclaim ([`release`]) on a thread of its own, started
/// the moment the listeners are bound, and returns at once. The release asks
/// every shard for its status and waits for each answer, and a shard answers
/// only between two of its reclaim steps: on 12 October the seats' server
/// waited 1,625 ms for shard 4's one step (1,538 ms) before it logged
/// started and started its services. Nothing the start does needs the
/// answers, so nothing on the start path waits for them. The thread holds
/// the store for the release alone, at most one step per shard.
/// # Errors
/// Returns a named startup failure when the release worker cannot start.
pub fn release_after_bind(
spine: &Arc<EventStore>,
began: Instant,
) -> Result<ReclaimRelease, crate::ServerError> {
let spine = Arc::clone(spine);
ReclaimRelease::start(move || release(&spine, began))
}
/// Runs `work` on the release thread; a thread that cannot start runs it
/// here, said by name, so the release is never lost, only waited for.
#[cfg(test)]
pub fn off_the_start_path<F: FnOnce() + Send + 'static>(work: F) {
let shared = Arc::new(Mutex::new(Some(work)));
let taken = Arc::clone(&shared);
let spawned = std::thread::Builder::new()
.name(RELEASE_THREAD.to_owned())
.spawn(move || run_once(&taken));
if let Err(error) = spawned {
tracing::warn!(
error = %error,
"spine reclaim release thread did not start; released on the start path"
);
run_once(&shared);
}
}
/// Takes the work out and runs it, once, whichever side gets it.
#[cfg(test)]
fn run_once<F: FnOnce()>(work: &Mutex<Option<F>>) {
let taken = work.lock().unwrap_or_else(PoisonError::into_inner).take();
if let Some(work) = taken {
work();
}
}
/// Arms the pass watch at open with a capacity of two reports per shard and
/// drains it on a thread of its own that logs each ended pass. The thread
/// holds the receiver only, so it ends when the store closes and never holds
/// the store open. A refused watch is logged and the boot goes on: no pass
/// is held back by its own reporting.
pub fn arm_watch(database: &Database) {
let capacity = database.shard_count().saturating_mul(2).max(1);
let watch = match database.watch_reclaim_passes(capacity) {
Ok(watch) => watch,
Err(error) => {
tracing::warn!(error = %error, "spine reclaim pass watch refused");
return;
}
};
let armed_at = Instant::now();
let spawned = std::thread::Builder::new()
.name(WATCH_THREAD.to_owned())
.spawn(move || {
while let Ok(report) = watch.recv() {
let (outcome, failed_reason) = match &report.outcome {
ReclaimPassOutcome::Passed => ("passed", None),
ReclaimPassOutcome::Failed { reason } => ("failed", Some(reason.clone())),
};
tracing::info!(
shard = report.shard_id,
outcome,
passes_finished = report.passes_finished,
// Time since this watch was armed at boot, not what the pass
// took: haematite 0.18.5's report carries no start.
since_start_ms =
u64::try_from(armed_at.elapsed().as_millis()).unwrap_or(u64::MAX),
unmarked_bytes = report.unmarked_bytes,
retained_bytes = report.retained_bytes,
peak_bytes = report.peak,
ended_at_ms = report.ended_at_ms,
failed_steps = failed_reason,
"spine reclaim pass ended"
);
}
});
if let Err(error) = spawned {
tracing::warn!(error = %error, "spine reclaim pass watch thread did not start");
}
}
/// Releases the held reclaim once the listeners are bound (R1) and logs the
/// time it took (R5). A second release is a no-op by the store's own word. A
/// refused release is logged by name and the server serves on: the pass is
/// owed, not lost, and the next open resumes it. No shard is asked for its
/// status here: haematite 0.33.0's `reclaim_status` boots every shard that has
/// not started, so R3's per-shard resume log waits for
/// `reclaim_status_published` (RECLAIM-009), which asks none.
pub fn release(spine: &EventStore, began: Instant) {
if let Err(error) = spine.database().release_reclaim() {
tracing::warn!(error = %error, "spine reclaim release refused");
return;
}
tracing::info!(
boot_elapsed_ms = u64::try_from(began.elapsed().as_millis()).unwrap_or(u64::MAX),
"spine reclaim released"
);
}