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
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
use super::*;
impl MVCCEngine {
/// Closes the engine (inherent method)
pub fn close_engine(&self) -> Result<()> {
const SHUTDOWN_ACTIVE_TXN_DRAIN_TIMEOUT: std::time::Duration =
std::time::Duration::from_secs(30);
let _startup_guard = self
.startup_mutex
.lock()
.map_err(|_| Error::LockAcquisitionFailed("engine lifecycle".to_string()))?;
let retrying_close = matches!(
&*self.lifecycle.read().unwrap(),
EngineLifecycleState::Closing | EngineLifecycleState::CloseFailed(_)
);
if !retrying_close {
// Use CAS to atomically close admission. A failed shutdown remains
// `Closing`, so a later call retries the drain instead of claiming
// that the engine is already closed.
if self
.open
.compare_exchange(true, false, Ordering::AcqRel, Ordering::Acquire)
.is_err()
{
if matches!(
&*self.lifecycle.read().unwrap(),
EngineLifecycleState::Closed
) {
return Ok(());
}
return Err(Error::internal(
"engine is not open and has no retryable close transition",
));
}
*self.lifecycle.write().unwrap() = EngineLifecycleState::Closing;
self.shutdown_checkpoint_complete
.store(false, Ordering::Release);
} else {
*self.lifecycle.write().unwrap() = EngineLifecycleState::Closing;
}
// Stop background warmup before final checkpoint can replace its
// immutable generation inputs.
{
let mut warmup_handle = self.page_cache_warmup_handle.lock().unwrap();
if let Some(mut handle) = warmup_handle.take() {
if let Err(error) = handle.stop() {
return Err(self.record_close_failure(Error::internal(format!(
"page-cache warmup worker failed: {error}"
))));
}
}
}
// Stop the cleanup thread first (before stopping transactions)
{
let mut cleanup_handle = self.cleanup_handle.lock().unwrap();
if let Some(mut handle) = cleanup_handle.take() {
if let Err(error) = handle.stop() {
return Err(self.record_close_failure(error));
}
}
}
// Stop accepting new transactions
self.registry.stop_accepting_transactions();
let remaining_active = self
.registry
.wait_for_active_transactions(SHUTDOWN_ACTIVE_TXN_DRAIN_TIMEOUT);
if remaining_active > 0 {
let error = Error::internal(format!(
"close_engine timed out with {} active transaction(s) after {:?}; resources remain owned and shutdown may be retried",
remaining_active, SHUTDOWN_ACTIVE_TXN_DRAIN_TIMEOUT
));
return Err(self.record_close_failure(error));
}
// Run a final checkpoint to seal ALL remaining hot rows into volumes.
// Use force_seal=true to bypass thresholds — on close, we want all data
// in volumes so startup is fast and doesn't depend on WAL replay.
// Skipped when checkpoint_on_close is false (crash simulation in tests).
let checkpoint_on_close = self.config.read().unwrap().persistence.checkpoint_on_close;
if !self.shutdown_checkpoint_complete.load(Ordering::Acquire) {
if checkpoint_on_close {
if let Some(pm) = self.persistence() {
if pm.is_enabled() {
// Retry checkpoint until all hot buffers are empty.
// After stop_accepting_transactions(), no new writes can start,
// but in-flight commits may still add rows between seal passes.
// Retry ensures WAL truncation happens and startup is fast.
let mut drained = false;
for attempt in 0..5 {
self.checkpoint_cycle_inner(true)
.map_err(|error| self.record_close_failure(error))?;
let pass_empty = self
.version_stores
.read()
.unwrap()
.values()
.all(|s| s.committed_row_count() == 0);
if pass_empty {
drained = true;
break;
}
if attempt < 4 {
std::thread::sleep(std::time::Duration::from_millis(10));
}
}
if !drained {
let error = Error::internal(
"final checkpoint did not drain all committed hot rows; resources remain owned and shutdown may be retried",
);
return Err(self.record_close_failure(error));
}
self.compact_after_checkpoint_forced()
.map_err(|error| self.record_close_failure(error))?;
}
}
} // checkpoint_on_close
self.shutdown_checkpoint_complete
.store(true, Ordering::Release);
}
// Wait for any background compaction to finish before releasing
// resources. Without this, a detached compaction thread could still
// be rewriting manifests and deleting volume files after close returns.
while self.compaction_running.load(Ordering::Acquire) {
std::thread::sleep(std::time::Duration::from_millis(10));
}
// Stop persistence before closing version stores. If WAL drain/fsync
// fails, close_engine returns with stores and file lock still owned so
// the same close transition can be retried safely.
if let Some(pm) = self.persistence() {
if pm.is_enabled() {
pm.stop()
.map_err(|error| self.record_close_failure(error))?;
}
}
// Close all version stores only after persistence is durably closed.
let stores = self.version_stores.read().unwrap();
for store in stores.values() {
store.close();
}
drop(stores);
// The publisher owns a clone of the same lock. Drop it before the
// engine's final clone so close really releases writer ownership.
// Revoke the shared publisher first: a retained internal Arc must not
// keep mutation authority or the OS lock alive beyond close.
if let Some(publisher) = self.physical_generation.load_full() {
publisher.release_writer_lock();
}
self.physical_generation.store(None);
// Release file lock (drops the final lock owner, allowing another open)
{
let mut file_lock = self.file_lock.lock().unwrap();
*file_lock = None;
}
*self.lifecycle.write().unwrap() = EngineLifecycleState::Closed;
Ok(())
}
/// Returns whether the engine is open
pub fn is_open(&self) -> bool {
self.open.load(Ordering::Acquire)
}
/// Per-table, per-volume statistics for PRAGMA VOLUME_STATS.
/// Returns table identity, total resident ownership, its disjoint
/// components, idle cycles and tombstones.
#[allow(clippy::type_complexity)]
pub fn volume_stats(
&self,
) -> Vec<(
String,
u64,
&'static str,
usize,
usize,
usize,
usize,
usize,
usize,
usize,
usize,
u64,
usize,
)> {
let mgrs = self.segment_managers.read().unwrap();
let mut result = Vec::new();
let mut table_names: Vec<&String> = mgrs.keys().collect();
table_names.sort();
for table_name in table_names {
if let Some(mgr) = mgrs.get(table_name) {
let tombstone_count = mgr.tombstone_count();
for (
seg_id,
tier,
row_count,
mem,
metadata,
row_ids,
exact_indices,
ordered_indices,
descriptor,
column_payload,
idle,
) in mgr.volume_stats()
{
result.push((
table_name.clone(),
seg_id,
tier,
row_count,
mem,
metadata,
row_ids,
exact_indices,
ordered_indices,
descriptor,
column_payload,
idle,
tombstone_count,
));
}
}
}
result
}
/// Returns the database path
pub fn get_path(&self) -> &str {
&self.path
}
/// Returns a copy of the configuration
pub fn config(&self) -> Config {
self.config.read().unwrap().clone()
}
/// Updates the engine configuration
pub fn update_engine_config(&self, config: Config) -> Result<()> {
let current = self.config.read().unwrap();
if config.path != current.path {
return Err(Error::internal("cannot change database path after opening"));
}
if config.persistence.storage_cpu_workers != current.persistence.storage_cpu_workers {
return Err(Error::invalid_argument(
"storage_cpu_workers is fixed when the database opens; restart with the new value",
));
}
drop(current);
*self.config.write().unwrap() = config;
Ok(())
}
/// Returns the transaction registry
pub fn registry(&self) -> Arc<TransactionRegistry> {
Arc::clone(&self.registry)
}
/// Replay an ALTER TABLE operation from WAL
/// Check if we should skip WAL writes (during recovery replay)
pub(super) fn should_skip_wal(&self) -> bool {
self.loading_from_disk.load(Ordering::Acquire)
}
#[doc(hidden)]
pub fn truncate_table_under_ddl_fence(&self, table_name: &str, txn_id: i64) -> Result<i32> {
if !self.is_open() {
return Err(Error::EngineNotOpen);
}
let table_name_lower = table_name.to_lowercase();
let store = self.get_version_store(&table_name_lower)?;
// Commits hold the same table-local fence from index mutation through
// hot-version publication. TRUNCATE must own it exclusively as well;
// otherwise a pure INSERT (which has no pre-existing row claim) can
// publish an index entry on one side of the clear and its row version
// on the other.
let membership_fence = store.membership_fence();
let _membership_guard = membership_fence.write();
let manager = self
.segment_managers
.read()
.unwrap()
.get(&table_name_lower)
.cloned();
let cold_rows = manager
.as_ref()
.map(|manager| manager.total_row_count() as i32)
.unwrap_or(0);
let hot_rows = store.truncate_all_after(|| {
if let Some(manager) = manager.as_ref() {
manager.rollback_pending_tombstones(txn_id);
}
<Self as Engine>::record_truncate_table(self, &table_name_lower)
})?;
Ok(hot_rows.saturating_add(cold_rows))
}
}