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
195
196
197
198
199
200
201
202
203
204
205
206
207
// SPDX-License-Identifier: BUSL-1.1
//! Idle maintenance loop: checkpoint coordinator, KV expiry wheel, idle
//! flush of timeseries memtables, and the periodic compaction trigger.
//!
//! Driven by the runtime event loop on every idle wake; rate-limited via
//! `compaction_interval` so the heavy `run_compaction` path runs at most
//! once per interval.
use tracing::info;
use crate::data::executor::core_loop::CoreLoop;
impl CoreLoop {
/// Run maintenance tasks if enough time has elapsed.
///
/// Called from the runtime event loop on every idle wake. Tracks the
/// last maintenance time internally and skips if the interval hasn't
/// elapsed. Returns `true` if maintenance was executed.
pub fn maybe_run_maintenance(&mut self) -> bool {
// Checkpoint coordinator tick: incremental dirty page flushing.
// Runs on its own interval (independent from compaction interval).
let flush_plan = self.checkpoint_coordinator.tick();
for (engine, pages) in &flush_plan {
match engine.as_str() {
// A maintenance flush deliberately does NOT advance
// `vector_durable_lsn` / `crdt_durable_lsn`, even on success.
// Those fields are the floor the coordinated checkpoint clamps
// to, and they may only record what a flush ordered against the
// truncation it authorises has made durable. Raising them from
// an unordered timer would let a later failed checkpoint clamp
// to a point this path claimed — the exact "a flush that is not
// ordered against the truncation it authorises is not a
// checkpoint" mistake that moved the sparse-vector flush out of
// `data/runtime.rs`. Leaving them alone costs nothing: the next
// `execute_checkpoint` re-flushes and reports for itself.
"vector" => match self.checkpoint_vector_indexes() {
Ok(outcome) => {
self.checkpoint_coordinator
.record_flush("vector", outcome.files_written.min(*pages));
}
Err(e) => {
tracing::warn!(
core = self.core_id,
error = %e,
"maintenance vector checkpoint failed; pages stay dirty for the \
next tick and the coordinated checkpoint will clamp its \
reported LSN if it fails there too"
);
}
},
"crdt" => match self.checkpoint_crdt_engines() {
Ok(outcome) => {
self.checkpoint_coordinator
.record_flush("crdt", outcome.files_written.min(*pages));
}
Err(e) => {
tracing::warn!(
core = self.core_id,
error = %e,
"maintenance CRDT checkpoint failed; pages stay dirty for the \
next tick and the coordinated checkpoint will clamp its \
reported LSN if it fails there too"
);
}
},
// Same rule as vector/crdt: the flushed point is deliberately
// NOT recorded into `columnar_durable_lsn`. This flush is
// ordered against a timer, not against the truncation the
// coordinated checkpoint authorises, so it may not raise the
// floor that checkpoint clamps to. It exists only to keep the
// backlog from arriving at that checkpoint whole.
"columnar" => match self.checkpoint_columnar_engines() {
Ok(_) => {
self.checkpoint_coordinator.record_flush("columnar", *pages);
}
Err(e) => {
tracing::warn!(
core = self.core_id,
error = %e,
"maintenance columnar checkpoint failed; pages stay dirty for the \
next tick and the coordinated checkpoint will clamp its \
reported LSN if it fails there too"
);
}
},
"sparse" => {
// redb is ACID — writes are already durable.
self.checkpoint_coordinator.record_flush("sparse", *pages);
}
"timeseries" => {
// Idle flush: if no ingest for 5 seconds, flush all
// non-empty memtables so data becomes queryable.
let idle_threshold = std::time::Duration::from_secs(5);
let is_idle = self
.last_ts_ingest
.map(|t| t.elapsed() >= idle_threshold)
.unwrap_or(false);
if is_idle {
let now_ms = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_millis() as i64)
.unwrap_or(0);
let collections: Vec<(
nodedb_types::DatabaseId,
crate::types::TenantId,
String,
)> = self
.columnar_memtables
.iter()
.filter(|(_, mt)| !mt.is_empty())
.map(|(k, _)| k.clone())
.collect();
let mut flushed = 0usize;
for (db, tid, collection) in &collections {
match self.flush_ts_collection(*tid, *db, collection, now_ms) {
Ok(()) => flushed += 1,
Err(e) => {
tracing::error!(
collection = %collection,
error = %e,
"idle ts flush failed — segment write error; \
collection skipped this maintenance cycle"
);
}
}
}
if flushed > 0 {
info!(
core = self.core_id,
flushed, "idle flush: timeseries memtables flushed"
);
}
// Reset so we don't re-flush until next ingest.
self.last_ts_ingest = None;
self.checkpoint_coordinator
.record_flush("timeseries", flushed.max(*pages));
} else {
self.checkpoint_coordinator
.record_flush("timeseries", *pages);
}
}
// `tick()` only ever plans engines from `TRACKED_ENGINES`, so
// reaching this arm means that list gained an entry without an
// arm here. Silently ignoring it would leave the engine planned
// every tick and flushed never, its dirty count only growing,
// so it is reported rather than dropped.
other => {
tracing::warn!(
core = self.core_id,
engine = other,
pages = *pages,
"checkpoint tick planned a flush for an engine with no maintenance \
flush path; its backlog cannot be worked off"
);
}
}
}
// KV expiry wheel tick: process expired keys on every maintenance call.
// Bounded by the per-tick reap budget internally — safe for the reactor.
// Expired keys are emitted as structured log events for CDC visibility.
{
let now_ms = crate::engine::kv::current_ms();
let expired_keys = self.kv_engine.tick_expiry(now_ms);
if !expired_keys.is_empty() {
tracing::debug!(
core = self.core_id,
reaped = expired_keys.len(),
backlog = self.kv_engine.expiry_backlog(),
"kv expiry wheel tick"
);
for ek in &expired_keys {
info!(
target: "nodedb::kv::expired",
database_id = ek.database_id,
tenant_id = ek.tenant_id,
collection = %ek.collection,
key_len = ek.key.len(),
"kv key expired"
);
}
}
}
// Compaction: periodic tombstone removal + segment merge.
let now = std::time::Instant::now();
if let Some(last) = self.last_maintenance
&& now.duration_since(last) < self.compaction_interval
{
return !flush_plan.is_empty();
}
self.last_maintenance = Some(now);
// Horizon-GC the per-core last-write-LSN version index: evict entries
// far below the watermark and enforce the entry-count backstop. Rides
// the compaction interval — no dedicated timer.
self.gc_write_index();
// Lease-GC abandoned per-transaction staging overlays (client vanished,
// teardown dispatch failed, or vShard leader moved mid-txn). Bounded by
// a per-tick budget internally. Rides the compaction interval.
self.reap_expired_overlays();
self.run_compaction(false);
true
}
}