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
// SPDX-License-Identifier: BUSL-1.1
//! Segment-level restore helpers: flushed timeseries partitions and
//! plain-columnar engines.
//!
//! Split from `restore.rs` to keep each file under the 500-line limit.
//! All methods are `pub(super)` — called only from `restore.rs`.
use crate::data::executor::core_loop::CoreLoop;
use crate::types::TsFlushedCollectionBlob;
impl CoreLoop {
/// Restore flushed on-disk timeseries segment directories from snapshot blobs.
///
/// For each captured partition:
/// - Collision handling is fail-closed (no silent overwrites):
/// - If the partition dir already exists AND the registry already tracks a
/// partition at the same `min_ts`, we compare `row_count` and
/// `last_flushed_wal_lsn` to determine idempotency:
/// - Identical metadata → skip (idempotent re-apply).
/// - Different metadata → return `Storage` error (would clobber live data).
/// - Otherwise: create the directory, write all files, register in
/// `ts_registries` mirroring `flush_ts_collection`'s exact registration.
///
/// `replace_mode` (Raft InstallSnapshot apply) SKIPS the fail-closed
/// collision checks and OVERWRITES the partition directory and registry
/// entry with the snapshot's version. `!replace_mode` (user RESTORE) keeps
/// the fail-closed behavior described above.
pub(super) fn restore_flushed_ts_segments(
&mut self,
blobs: &[TsFlushedCollectionBlob],
replace_mode: bool,
) -> crate::Result<()> {
for coll_blob in blobs {
let (database_id, tenant_id, collection) =
super::restore::parse_timeseries_snapshot_key(&coll_blob.collection_key);
let segment_dir = super::super::timeseries::paths::ts_collection_dir(
&self.data_dir,
database_id,
tenant_id,
&collection,
);
let reg_key = (
nodedb_types::DatabaseId::new(database_id),
crate::types::TenantId::new(tenant_id),
collection.clone(),
);
for part_blob in &coll_blob.partitions {
// Deserialize PartitionMeta from the embedded msgpack bytes.
let meta: nodedb_types::timeseries::PartitionMeta =
zerompk::from_msgpack(&part_blob.meta_bytes).map_err(|e| {
crate::Error::Serialization {
format: "msgpack".into(),
detail: format!(
"restore: deserialize PartitionMeta for {}/{}: {e}",
collection, part_blob.dir_name
),
}
})?;
let partition_dir = segment_dir.join(&part_blob.dir_name);
// Collision check: registry already knows this min_ts key.
// Skipped under `replace_mode` (Raft install): the snapshot's
// partition OVERWRITES the local one — `registry.import` replaces
// the entry keyed by min_ts and the directory is wiped + rewritten
// below.
if !replace_mode
&& let Some(registry) = self.ts_registries.get(®_key)
&& let Some(existing) = registry.get(meta.min_ts)
{
let is_identical = existing.meta.row_count == meta.row_count
&& existing.meta.last_flushed_wal_lsn == meta.last_flushed_wal_lsn;
if is_identical {
// Idempotent: same partition already present, skip.
continue;
}
return Err(crate::Error::Storage {
engine: "timeseries".into(),
detail: format!(
"restore: partition collision for collection '{}' min_ts={}: \
existing (rows={}, lsn={}) differs from snapshot \
(rows={}, lsn={}); refusing to overwrite live data",
collection,
meta.min_ts,
existing.meta.row_count,
existing.meta.last_flushed_wal_lsn,
meta.row_count,
meta.last_flushed_wal_lsn,
),
});
}
// Also check the filesystem: if the directory already exists
// and is non-empty, treat it as a collision. Skipped under
// `replace_mode` — the directory is removed and rewritten below.
if !replace_mode && partition_dir.exists() {
let is_empty = std::fs::read_dir(&partition_dir)
.map_err(crate::Error::Io)
.map(|mut d| d.next().is_none())?;
if !is_empty {
return Err(crate::Error::Storage {
engine: "timeseries".into(),
detail: format!(
"restore: partition directory '{}' already exists for \
collection '{}'; refusing to overwrite live data",
part_blob.dir_name, collection,
),
});
}
}
// Under replace_mode, wipe any stale partition directory so a
// changed file layout cannot leave orphaned segment files behind.
if replace_mode && partition_dir.exists() {
std::fs::remove_dir_all(&partition_dir)?;
}
// Create the partition directory and write all captured files.
std::fs::create_dir_all(&partition_dir)?;
for (filename, bytes) in &part_blob.files {
std::fs::write(partition_dir.join(filename), bytes)?;
}
// Register the restored partition in ts_registries, mirroring
// exactly the registration step in flush_ts_collection.
let registry = self
.ts_registries
.entry(reg_key.clone())
.or_insert_with(|| {
crate::engine::timeseries::partition_registry::PartitionRegistry::new(
nodedb_types::timeseries::TieredPartitionConfig::origin_defaults(),
)
});
let pe = crate::engine::timeseries::partition_registry::PartitionEntry {
meta,
dir_name: part_blob.dir_name.clone(),
};
registry.import(vec![(pe.meta.min_ts, pe)]);
}
}
Ok(())
}
/// Restore plain-columnar (and spatial) engine state from snapshot entries.
///
/// For each `(collection_key, msgpack_bytes)` entry:
/// - Deserialises the `ColumnarEngineSnapshot` from `msgpack_bytes`.
/// - Reconstructs the `MutationEngine` via `MutationEngine::from_snapshot`.
/// - Inserts the engine into `columnar_engines` and any returned flushed
/// segment blobs into `columnar_flushed_segments`.
///
/// **Collision handling depends on `replace_mode`:**
/// - `!replace_mode` (user RESTORE): fail-closed — if either
/// `columnar_engines` or `columnar_flushed_segments` already contains an
/// entry for the key, return `Error::Storage` rather than silently
/// overwriting live data.
/// - `replace_mode` (Raft InstallSnapshot apply): SKIP the guards and
/// OVERWRITE — the engine entry is replaced, and the flushed segments /
/// surrogates maps are SET to the snapshot's (the stale entries are removed
/// when the snapshot carries none) so they are never appended to stale
/// state.
pub(super) fn restore_columnar_engines(
&mut self,
entries: &[(String, Vec<u8>)],
replace_mode: bool,
) -> crate::Result<()> {
for (collection_key, bytes) in entries {
let (database_id, tenant_id, collection) =
super::restore::parse_timeseries_snapshot_key(collection_key);
let engine_key = (
nodedb_types::DatabaseId::new(database_id),
crate::types::TenantId::new(tenant_id),
collection.clone(),
);
// Fail-closed (user RESTORE only): refuse to overwrite any live
// engine or flushed segment state that was present before this
// restore call. Skipped under `replace_mode` (Raft install).
if !replace_mode {
if self.columnar_engines.contains_key(&engine_key) {
return Err(crate::Error::Storage {
engine: "columnar".into(),
detail: format!(
"restore: columnar engine already exists for collection '{collection}' \
(db={database_id}, tenant={tenant_id}); refusing to overwrite live data"
),
});
}
if self.columnar_flushed_segments.contains_key(&engine_key) {
return Err(crate::Error::Storage {
engine: "columnar".into(),
detail: format!(
"restore: flushed segment state already exists for collection \
'{collection}' (db={database_id}, tenant={tenant_id}); \
refusing to overwrite live data"
),
});
}
}
let snap: nodedb_columnar::ColumnarEngineSnapshot = zerompk::from_msgpack(bytes)
.map_err(|e| crate::Error::Serialization {
format: "msgpack".into(),
detail: format!(
"restore: deserialize ColumnarEngineSnapshot for '{collection}': {e}"
),
})?;
let (engine, flushed, flushed_surrogates) =
nodedb_columnar::MutationEngine::from_snapshot(snap).map_err(|e| {
crate::Error::Storage {
engine: "columnar".into(),
detail: format!(
"restore: from_snapshot for collection '{collection}': {e}"
),
}
})?;
self.columnar_engines.insert(engine_key.clone(), engine);
if !flushed.is_empty() {
self.columnar_flushed_segments
.insert(engine_key.clone(), flushed);
} else if replace_mode {
// Replace: the snapshot carries no flushed segments, so any stale
// local entry must be dropped (not left behind to mismatch the
// overwritten engine).
self.columnar_flushed_segments.remove(&engine_key);
}
// Re-attach the cross-engine surrogate sidecar under the SAME key so
// prefiltered scans see flushed rows post-restore. Old snapshots
// carry empty surrogates (non-empty segments): we skip populating
// the sidecar, so those rows read as `None`-surrogate and are
// conservatively excluded under an active prefilter — the correct
// backward-compat behavior. Order is preserved so segment_id ==
// index + 1 holds for the sidecar too.
if !flushed_surrogates.is_empty() {
self.columnar_flushed_surrogates
.insert(engine_key, flushed_surrogates);
} else if replace_mode {
// Replace: drop any stale surrogate sidecar when the snapshot
// carries none, keeping the sidecar consistent with the
// overwritten engine + flushed segments.
self.columnar_flushed_surrogates.remove(&engine_key);
}
}
Ok(())
}
}