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
309
310
311
312
313
314
315
316
317
318
// SPDX-License-Identifier: Apache-2.0
//! Lite identity and CRDT state restore.
use std::sync::Arc;
use nodedb_types::Namespace;
use nodedb_types::error::{NodeDbError, NodeDbResult};
use crate::engine::crdt::CrdtEngine;
/// Rows per collection that updates replayed on top of their snapshot
/// changed. Those updates can postdate every derived-index checkpoint.
pub(in crate::nodedb::core::open) type ReplayedRows =
std::collections::BTreeMap<String, std::collections::BTreeSet<String>>;
use crate::storage::engine::StorageEngine;
use crate::nodedb::core::types::{
META_CRDT_DELTAS, META_CSR_LEGACY, META_LAST_FLUSHED_MID, NodeDbLite,
};
impl<S: StorageEngine> NodeDbLite<S> {
/// Restore Lite identity (lite_id + epoch) and CRDT state.
///
/// Loads or creates the Lite identity, restores per-collection CRDT
/// snapshots, backfills the registered-collection set and `LatestVersion`
/// index from persisted bitemporal flags, restores pending deltas, checks
/// partial-flush safety, and deletes the legacy single-CSR checkpoint if
/// present.
pub(in crate::nodedb::core::open) async fn restore_identity_and_crdt(
storage: &Arc<S>,
policy: crate::storage::corruption::CorruptionPolicy,
) -> NodeDbResult<(CrdtEngine, crate::identity::LiteIdentity, ReplayedRows)> {
// ── Load or create Lite identity (lite_id + epoch + peer id) ──
//
// This must happen before any outbound sync so the handshake carries a
// non-empty lite_id and epoch ≥ 1, enabling Origin's idempotent-producer
// gate. The epoch is incremented on every open, so a new process
// incarnation fences out writes from the previous one. The peer id
// comes from the same record, which is what binds the identity every
// local operation is authored under to the store holding them.
let lite_identity = crate::identity::LiteIdentity::load_or_create(&**storage)
.await
.map_err(|e| {
// Preserve corruption typing so a corrupt identity read is
// routed to the post-open recovery driver rather than
// crash-looping as a generic storage error.
let detail = format!("lite identity load failed: {e}");
if crate::error::is_corruption(&e) {
NodeDbError::segment_corrupted(detail)
} else {
NodeDbError::storage(detail)
}
})?;
// ── Restore CRDT state, one Loro document per collection ──
// Snapshots are stored under `loro_snapshot:<collection>`, so the whole
// set is recovered with a single prefix scan. A collection whose
// snapshot fails its CRC32C check is dropped individually — the other
// collections stay intact instead of the whole engine resetting.
let mut crdt = CrdtEngine::new(lite_identity.peer_id)
.map_err(|e| NodeDbError::storage(format!("CRDT init failed: {e}")))?;
let snapshot_entries = storage
.scan_prefix(Namespace::LoroState, CrdtEngine::snapshot_key_prefix())
.await?;
let mut base_bytes: std::collections::HashMap<String, usize> =
std::collections::HashMap::new();
for (key, envelope) in &snapshot_entries {
let Some(collection) = CrdtEngine::collection_from_snapshot_key(key) else {
tracing::error!(
"CRDT snapshot key is not a valid `loro_snapshot:<collection>` entry — \
skipping; its collection cannot be determined without guessing."
);
continue;
};
match crate::storage::checksum::unwrap(envelope) {
Some(snapshot) => {
base_bytes.insert(collection.to_string(), snapshot.len());
crdt.import_snapshot(collection, &snapshot).map_err(|e| {
NodeDbError::storage(format!("CRDT restore of '{collection}' failed: {e}"))
})?;
}
None => {
// The collection's entire history is in this blob. Dropping
// it is the same decision as dropping the store, at a
// smaller scale, so it needs the same consent — and the
// bytes stay put either way until the caller has decided,
// because a deleted snapshot cannot be handed to a forensic
// tool or recovered from.
if !policy.may_discard() {
return Err(NodeDbError::segment_corrupted(format!(
"CRDT snapshot for collection '{collection}' failed its CRC32C \
check. The snapshot has been left in place; opening past it would \
silently empty the collection."
)));
}
tracing::error!(
collection = %collection,
"CRDT snapshot CRC32C mismatch — the caller opted into discarding \
corrupted state, so this collection is being dropped. A full re-sync \
from Origin is needed for it."
);
// Delete the corrupted snapshot so we don't re-read it. A
// failed delete leaves it to be re-read and re-reported on
// the next open, which is recoverable — but it must not be
// silent, or the collection looks cleanly dropped.
if let Err(e) = storage.delete(Namespace::LoroState, key).await {
tracing::error!(
collection = %collection,
error = %e,
"failed to delete the corrupted CRDT snapshot; it will be \
re-read and discarded again on the next open"
);
}
}
}
}
// ── Replay the incremental updates written on top of each snapshot ──
// Flush writes a full snapshot only periodically; between checkpoints it
// appends `loro_delta:<collection>:<seq>` entries. A prefix scan returns
// them in key order, which the zero-padded sequence makes replay order.
// Skipping an update whose base is present would silently roll that
// collection back to its last checkpoint, so a corrupt one is an error
// rather than a warning.
//
// An update whose base is *absent* is the opposite case. Every update is
// exported as the operations since a base that was written in an earlier
// batch, so without that base its causal predecessors are missing and
// Loro buffers it as pending — which `import_local` reports as an error.
// Replaying one would therefore fail the whole open, and since nothing
// removes these keys, it would fail every subsequent open too: a
// collection whose snapshot was discarded one line above would take the
// entire store down with it, permanently, instead of being the isolated
// re-sync the discard is there to make it. They are deleted instead.
let mut delta_bytes: std::collections::HashMap<String, usize> =
std::collections::HashMap::new();
let mut next_delta_seq: std::collections::HashMap<String, u64> =
std::collections::HashMap::new();
let delta_entries = storage
.scan_prefix(Namespace::LoroState, CrdtEngine::state_delta_key_prefix())
.await?;
let mut orphaned_keys: Vec<&[u8]> = Vec::new();
let mut replayed = ReplayedRows::new();
for (key, envelope) in &delta_entries {
let Some((collection, seq)) = CrdtEngine::state_delta_from_key(key) else {
tracing::error!(
"CRDT update key is not a valid `loro_delta:<collection>:<seq>` entry — \
skipping; its collection cannot be determined without guessing."
);
continue;
};
if !base_bytes.contains_key(collection) {
// These carry writes that exist nowhere else once deleted.
if !policy.may_discard() {
return Err(NodeDbError::segment_corrupted(format!(
"CRDT update '{collection}' #{seq} has no base snapshot to apply to, so \
the writes it carries cannot be replayed. It has been left in place; \
discarding it would lose them."
)));
}
tracing::error!(
collection = %collection,
seq,
"CRDT update has no base snapshot to apply to — discarding it. Its base \
was corrupt or missing, so the operations it carries have no causal \
predecessors and cannot be replayed. A full re-sync from Origin is \
needed for this collection."
);
orphaned_keys.push(key);
continue;
}
let Some(update) = crate::storage::checksum::unwrap(envelope) else {
return Err(NodeDbError::segment_corrupted(format!(
"CRDT update '{collection}' #{seq} failed its CRC32C check; the writes it \
carries are not in the snapshot behind it, so opening without it would \
silently roll the collection back"
)));
};
delta_bytes
.entry(collection.to_string())
.and_modify(|n| *n += update.len())
.or_insert(update.len());
next_delta_seq.insert(collection.to_string(), seq + 1);
let imported = crdt
.import_local_tracked(collection, &update)
.map_err(|e| {
NodeDbError::storage(format!(
"CRDT update replay for '{collection}' #{seq} failed: {e}"
))
})?;
replayed
.entry(collection.to_string())
.or_default()
.extend(imported.changed_rows);
}
for key in orphaned_keys {
// A failed delete is recoverable — the entry is skipped by the same
// base-absent check on the next open — but never silent.
if let Err(e) = storage.delete(Namespace::LoroState, key).await {
tracing::error!(
error = %e,
"failed to delete an orphaned CRDT update; it will be skipped and \
reported again on the next open"
);
}
}
// Seed the checkpoint accounting from what was on disk, so the first
// flush after open does not rewrite a base that is already current.
for (collection, base) in &base_bytes {
let Some(version) = crdt.state(collection).map(|s| s.oplog_version_vector()) else {
continue;
};
crdt.adopt_persisted_state(
collection,
version,
*base,
delta_bytes.get(collection).copied().unwrap_or(0),
next_delta_seq.get(collection).copied().unwrap_or(0),
);
}
// Rebuild the CRDT's registered-collection set from persisted bitemporal
// flags so that SELECT queries on bitemporal collections work immediately
// after open, even for collections with no inserted documents yet.
// Also backfill the LatestVersion index for collections written before
// the index was introduced — safe on fresh DBs and idempotent otherwise.
const BITEMPORAL_PREFIX: &[u8] = b"document_bitemporal:";
let bitemporal_entries = storage
.scan_prefix(Namespace::Meta, BITEMPORAL_PREFIX)
.await
.unwrap_or_default();
for (key, value) in &bitemporal_entries {
// Only process collections where the flag byte is 0x01 (enabled).
if value.first().copied() != Some(1) {
continue;
}
if let Ok(key_str) = std::str::from_utf8(key)
&& let Some(name) = key_str.strip_prefix("document_bitemporal:")
{
crdt.register_collection(name);
if let Err(e) = crate::engine::document::history::ops::backfill_latest_version(
storage.as_ref(),
name,
)
.await
{
tracing::warn!(
collection = name,
error = %e,
"LatestVersion backfill failed — bitemporal reads will \
fall back to prefix scan for this collection"
);
}
}
}
// Restore pending deltas — prefer incremental entries over legacy bulk blob.
let incremental_entries = storage.scan_prefix(Namespace::Crdt, b"delta:").await?;
if !incremental_entries.is_empty() {
// Use incremental entries (append-only format).
crdt.restore_pending_deltas_incremental(&incremental_entries, policy.may_discard())
.map_err(|e| NodeDbError::segment_corrupted(e.to_string()))?;
} else if let Some(delta_bytes) = storage.get(Namespace::Crdt, META_CRDT_DELTAS).await? {
// Fall back to legacy bulk blob.
crdt.restore_pending_deltas(&delta_bytes);
}
// Partial flush safety: check if the last-flushed mutation_id matches.
if crdt.pending_count() > 0
&& let Some(last_flushed_bytes) =
storage.get(Namespace::Meta, META_LAST_FLUSHED_MID).await?
&& last_flushed_bytes.len() == 8
{
let last_flushed = u64::from_le_bytes(last_flushed_bytes.try_into().unwrap_or([0; 8]));
let max_pending = crdt
.pending_deltas()
.iter()
.map(|d| d.mutation_id)
.max()
.unwrap_or(0);
if max_pending > 0 && last_flushed > 0 && max_pending != last_flushed {
// Clearing the queue throws away mutations that may not be in
// the CRDT state behind it — the "CRDT state is authoritative"
// assumption is exactly what a partial flush puts in doubt.
if !policy.may_discard() {
return Err(NodeDbError::segment_corrupted(format!(
"partial flush detected: the last flushed mutation is {last_flushed} but \
the queue reaches {max_pending}, so the queued mutations and the CRDT \
state behind them disagree. The queue has been left intact."
)));
}
tracing::warn!(
last_flushed,
max_pending,
"partial flush detected — pending deltas may be inconsistent. \
Clearing pending queue; CRDT state is authoritative."
);
crdt.clear_pending_deltas();
}
}
// ── Delete legacy single-CSR checkpoint if present ──
if storage
.get(Namespace::Graph, META_CSR_LEGACY)
.await?
.is_some()
{
let _ = storage.delete(Namespace::Graph, META_CSR_LEGACY).await;
}
Ok((crdt, lite_identity, replayed))
}
}