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
//! 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::db::ShardReclaimAnswer;
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 what
/// each shard resumed (R3). 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.
pub fn release(spine: &EventStore, began: Instant) {
let database = spine.database();
if let Err(error) = database.release_reclaim() {
tracing::warn!(error = %error, "spine reclaim release refused");
return;
}
let status = database.reclaim_status();
let phases: Vec<String> = status
.shards
.iter()
.map(|shard| match shard {
ShardReclaimAnswer::Answered { shard, status } => {
format!("{shard}:{:?}", status.phase)
}
ShardReclaimAnswer::NotAnswering { shard, reason } => {
format!("{shard}:not-answering({reason})")
}
})
.collect();
tracing::info!(
boot_elapsed_ms = u64::try_from(began.elapsed().as_millis()).unwrap_or(u64::MAX),
phases = %phases.join(" "),
"spine reclaim released"
);
for shard in &status.shards {
if let ShardReclaimAnswer::Answered { shard, status } = shard {
log_resume(*shard, status.resume.as_deref());
}
}
}
/// The note haematite gives a shard that took up a pass a stop left
/// (`reclaim_resume_taken: N marked, ...`). Any other note (a record found
/// corrupt, ignored or already completed) is not a resumed pass.
const RESUME_TAKEN: &str = "reclaim_resume_taken";
/// Says, for every shard, whether it took up a pass a stop left or starts
/// from nothing (R3), so an operator reads on the log that an unclean stop
/// lost no pass, and a refused record is never read as a resumed one.
fn log_resume(shard: usize, resume: Option<&str>) {
match resume {
Some(note) if note.starts_with(RESUME_TAKEN) => {
tracing::info!(shard, resume = note, "spine reclaim resumed");
}
note => tracing::info!(
shard,
resume = note.unwrap_or("none"),
"spine reclaim starts from nothing"
),
}
}