coordinode-lsm-tree 7.3.0

Embedded LSM-tree storage engine in pure Rust, no C/C++ dependency. MVCC snapshots, BuRR filters, zstd dictionary compression, columnar PAX blocks, AES-256-GCM at rest, self-healing per-block ECC, compaction on a near-full disk, no_std support.
Documentation
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
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
// SPDX-License-Identifier: Apache-2.0
// Copyright (c) 2026-present, Dmitry Prudnikov

//! Sidecar recording a tight-space-punched SST's restriction lower bound.
//!
//! Tight-space compaction reclaims a table's consumed prefix data blocks in
//! place (hole-punched, reading back as zeros) and restricts the surviving view
//! to keys `>= bound`. The bound lives in the manifest, but manifest repair
//! rebuilds the manifest from the on-disk files alone and so cannot recover it.
//!
//! This sidecar makes the bound recoverable WITHOUT mutating the SST. Writing the
//! bound into the SST's own meta would change the SST bytes while the manifest
//! still references it by a whole-file checksum, opening a crash window where the
//! two disagree (a scrub then reports false corruption and a checkpoint can
//! hard-link the modified bytes under the stale digest). A separate sidecar
//! avoids that entirely: the SST is never touched, so its manifest checksum stays
//! valid in every crash window, and the sidecar is published through a single
//! atomic `temp + fsync + rename` — never two mirrors that can diverge.
//!
//! # Threat model
//!
//! For an ENCRYPTED table the payload is AEAD-sealed with the table's provider,
//! so an offline attacker without the key cannot forge a bound `decrypt` accepts.
//! For an UNENCRYPTED table the payload carries an XXH3-128 checksum that detects
//! corruption; forging a *valid* sidecar needs the same directory write access
//! that could tamper with the SST or manifest directly, so it opens no new attack
//! surface. Repair uses the physical punch to grade the bound, not to gate it:
//! when the prefix below a valid bound reads as fully hole-punched the bound is
//! exact; when it does NOT (the crash window between a durable install and the
//! punch that follows it, or a stale sidecar over a never-restricted file) the two
//! are indistinguishable on disk, so recovery follows the resurrection policy. By
//! default it HONORS the bound (restricting, dropping the ambiguous prefix, which
//! never resurrects a superseded row); with `allow_resurrection` it keeps the
//! whole table. See `docs/manifest-recovery.md`.

use crate::encryption::EncryptionProvider;
use crate::fs::{Fs, FsOpenOptions, SyncMode};
use alloc::vec::Vec;
use std::path::{Path, PathBuf};

/// Fixed header length: `table_id` (u64) + `bound_len` (u32).
const HEADER_LEN: usize = 8 + 4;
/// Trailing integrity checksum length: XXH3-128.
const CHECKSUM_LEN: usize = 16;

/// The sidecar sits next to the SST with a `.restrict-bound` suffix APPENDED (not
/// a replaced extension), so it never collides with the SST's own numeric name.
#[must_use]
pub fn sidecar_path(table_path: &Path) -> PathBuf {
    let mut name = table_path.as_os_str().to_os_string();
    name.push(".restrict-bound");
    PathBuf::from(name)
}

/// The table id a sidecar FILE NAME claims.
///
/// `None` when the name is not a sidecar, or names no numeric table. Both the
/// live `{id}.restrict-bound` and its staging `{id}.restrict-bound.tmp` claim
/// the id: an id whose sidecar survives must not be handed to a different
/// table, or that table inherits a restriction belonging to another file.
#[must_use]
pub fn table_id_from_sidecar_name(file_name: &str) -> Option<u64> {
    let stem = file_name.strip_suffix(".tmp").unwrap_or(file_name);
    stem.strip_suffix(".restrict-bound")?.parse().ok()
}

/// The staging path the sidecar is written + synced to before its atomic rename
/// onto [`sidecar_path`]. A crash between the temp write and the rename leaves
/// this `{id}.restrict-bound.tmp` behind; tree recovery sweeps it (it is
/// disposable — either the live sidecar it would have replaced still exists, or
/// the compaction is simply re-run).
#[must_use]
fn sidecar_tmp_path(table_path: &Path) -> PathBuf {
    let mut name = sidecar_path(table_path).into_os_string();
    name.push(".tmp");
    PathBuf::from(name)
}

/// Serializes `(table_id, bound)` with a trailing XXH3-128 checksum over the
/// header + bound (the checksum detects accidental corruption; an encrypted
/// payload is additionally AEAD-authenticated by the provider).
fn serialize(table_id: u64, bound: &[u8]) -> Vec<u8> {
    // `bound` is a user key; its length is bounded by the same limits the
    // writer enforces on stored keys (u16::MAX), well within u32. Fail loudly
    // here rather than fall back to a length that lies about the payload —
    // a clamped length would serialize a self-inconsistent frame.
    #[expect(
        clippy::expect_used,
        reason = "user keys are length-capped at u16::MAX by the writer"
    )]
    let bound_len = u32::try_from(bound.len()).expect("restriction bound length exceeds u32");
    let mut out = Vec::with_capacity(HEADER_LEN + bound.len() + CHECKSUM_LEN);
    out.extend_from_slice(&table_id.to_le_bytes());
    out.extend_from_slice(&bound_len.to_le_bytes());
    out.extend_from_slice(bound);
    let checksum = crate::hash::hash128(&out);
    out.extend_from_slice(&checksum.to_le_bytes());
    out
}

/// Parses a serialized payload, validating the checksum and length framing.
/// Returns `(table_id, bound)` or `None` on any malformation.
fn deserialize(plain: &[u8]) -> Option<(u64, Vec<u8>)> {
    if plain.len() < HEADER_LEN + CHECKSUM_LEN {
        return None;
    }
    let (body, checksum_bytes) = plain.split_at(plain.len() - CHECKSUM_LEN);
    let stored = u128::from_le_bytes(checksum_bytes.try_into().ok()?);
    if crate::hash::hash128(body) != stored {
        return None;
    }
    let table_id = u64::from_le_bytes(body.get(0..8)?.try_into().ok()?);
    let bound_len = u32::from_le_bytes(body.get(8..12)?.try_into().ok()?) as usize;
    let bound = body.get(HEADER_LEN..)?;
    if bound.len() != bound_len {
        return None;
    }
    Some((table_id, bound.to_vec()))
}

/// Publishes the restriction bound for `table_id` next to `table_path`.
///
/// Written through an atomic synced `temp + rename`, and MUST be called (and
/// returned) BEFORE the prefix is hole-punched, so a crash leaves the sidecar
/// either fully present or absent — never partial — beside an SST whose bytes were
/// never touched.
///
/// # Errors
///
/// Propagates encryption or filesystem failures from the atomic write.
pub fn write(
    fs: &dyn Fs,
    table_path: &Path,
    encryption: Option<&dyn EncryptionProvider>,
    table_id: u64,
    bound: &[u8],
    sync_mode: SyncMode,
) -> crate::Result<()> {
    let content = encode(encryption, table_id, bound)?;
    publish_raw(fs, table_path, &content, sync_mode)
}

/// The on-disk bytes of the sidecar for `table_id` restricted at `bound`.
fn encode(
    encryption: Option<&dyn EncryptionProvider>,
    table_id: u64,
    bound: &[u8],
) -> crate::Result<Vec<u8>> {
    // A bound is a user key, at most `u16::MAX` bytes. `read` rejects a sidecar
    // whose payload exceeds that limit as corrupt, so reject an oversized bound
    // HERE rather than publishing a sidecar a later repair would read as corrupt,
    // silently dropping the restriction and resurrecting the punched prefix.
    if bound.len() > u16::MAX as usize {
        return Err(crate::Error::InvalidHeader(
            "restriction bound exceeds the maximum key length",
        ));
    }
    let plain = serialize(table_id, bound);
    match encryption {
        Some(enc) => Ok(enc.encrypt(&plain)?),
        None => Ok(plain),
    }
}

/// Atomically publishes already-encoded sidecar `content` beside `table_path`
/// through the synced `temp + rename` protocol [`write()`] uses. Also the
/// re-publish path for raw bytes captured from an existing sidecar, which must
/// land verbatim — they are already serialized (and possibly encrypted).
///
/// # Errors
///
/// Propagates filesystem failures from the atomic write; a failed publish
/// leaves no partial sidecar (the temp is removed best-effort).
pub(crate) fn publish_raw(
    fs: &dyn Fs,
    table_path: &Path,
    content: &[u8],
    sync_mode: SyncMode,
) -> crate::Result<()> {
    publish_many(&[(fs, table_path, content)], sync_mode)
        .pop()
        .unwrap_or(Ok(()))
}

/// One sidecar [`write_many`] publishes: the table's file, its encryption,
/// its id and its bound.
pub(crate) struct SidecarWrite<'a> {
    pub fs: &'a dyn Fs,
    pub table_path: &'a Path,
    pub encryption: Option<&'a dyn EncryptionProvider>,
    pub table_id: u64,
    pub bound: &'a [u8],
}

/// [`write()`] for several tables at once, as one tight-space slice restricts
/// them: each sidecar is written and synced, then renamed, and each folder
/// they share synced once. A sidecar is durable once its folder's sync
/// returns, so each write's result is its own; no punch may rely on one that
/// failed.
pub(crate) fn write_many(
    writes: &[SidecarWrite<'_>],
    sync_mode: SyncMode,
) -> Vec<crate::Result<()>> {
    let encoded: Vec<crate::Result<Vec<u8>>> = writes
        .iter()
        .map(|w| encode(w.encryption, w.table_id, w.bound))
        .collect();
    // One publish per encoded sidecar, in order.
    let published = {
        let items: Vec<(&dyn Fs, &Path, &[u8])> = writes
            .iter()
            .zip(&encoded)
            .filter_map(|(w, content)| {
                content
                    .as_ref()
                    .ok()
                    .map(|content| (w.fs, w.table_path, content.as_slice()))
            })
            .collect();
        publish_many(&items, sync_mode)
    };
    let mut published = published.into_iter();
    encoded
        .into_iter()
        .map(|content| match content {
            Err(e) => Err(e),
            Ok(_) => published.next().unwrap_or(Ok(())),
        })
        .collect()
}

/// A folder [`publish_many`] synced, and the failure its sync ended in.
struct FolderSync<'a> {
    fs: &'a dyn Fs,
    folder: &'a Path,
    failure: Option<(crate::io::ErrorKind, alloc::string::String)>,
}

/// Publishes each `(fs, table_path, content)` through the synced
/// `temp + rename` protocol: every temp written and synced, then every one
/// renamed, then each folder the renames share synced once. The result of
/// each is its own: a failure at any step fails that sidecar, and a failed
/// folder sync fails every sidecar renamed into that folder. A failed publish
/// leaves no partial sidecar (the temp is removed best-effort).
fn publish_many(items: &[(&dyn Fs, &Path, &[u8])], sync_mode: SyncMode) -> Vec<crate::Result<()>> {
    // A sidecar-specific `.restrict-bound.tmp`, atomically renamed onto the
    // live path. A dedicated temp suffix (not a generic `.tmp_*`) lets tree
    // recovery recognize and sweep an abandoned temp in the tables folder by
    // name, instead of failing to parse it as a table id.
    let mut results: Vec<crate::Result<()>> = items
        .iter()
        .map(|&(fs, table_path, content)| {
            let mut file = fs.open(
                &sidecar_tmp_path(table_path),
                &FsOpenOptions::new()
                    .write(true)
                    .create(true)
                    .truncate(true)
                    .write_origin(crate::fs::WriteOrigin::Metadata),
            )?;
            file.write_all(content)?;
            file.flush()?;
            crate::fs::FsFile::sync_all_with(&*file, sync_mode)?;
            Ok(())
        })
        .collect();
    for (result, &(fs, table_path, _)) in results.iter_mut().zip(items) {
        if result.is_ok() {
            // `std::fs::rename` REPLACES an existing destination FILE on every
            // supported platform (on Windows via MOVEFILE_REPLACE_EXISTING /
            // POSIX-semantics rename; only a destination DIRECTORY fails
            // there), so re-publishing over an earlier slice's sidecar is
            // atomic everywhere, with no remove-then-rename window.
            *result = fs
                .rename(&sidecar_tmp_path(table_path), &sidecar_path(table_path))
                .map_err(crate::Error::from);
        }
    }
    // Each folder once, on the backend that holds it, with how its sync ended.
    let mut synced: Vec<FolderSync<'_>> = Vec::new();
    for (result, &(fs, table_path, _)) in results.iter_mut().zip(items) {
        let Some(parent) = table_path.parent().filter(|_| result.is_ok()) else {
            continue;
        };
        let failure = if let Some(seen) = synced
            .iter()
            .find(|seen| core::ptr::addr_eq(seen.fs, fs) && seen.folder == parent)
        {
            seen.failure.clone()
        } else {
            let failure = fs
                .sync_directory_with(parent, sync_mode)
                .err()
                .map(|e| (e.kind(), alloc::format!("{e}")));
            synced.push(FolderSync {
                fs,
                folder: parent,
                failure: failure.clone(),
            });
            failure
        };
        if let Some((kind, message)) = failure {
            *result = Err(crate::Error::from(crate::io::Error::new(kind, message)));
        }
    }
    for (result, &(fs, table_path, _)) in results.iter().zip(items) {
        if result.is_err() {
            let _ = fs.remove_file(&sidecar_tmp_path(table_path));
        }
    }
    results
}

/// The largest on-disk size a VALID sidecar can have: header + bound + checksum
/// (plaintext), plus the provider's AEAD overhead when encrypted. A user key is
/// at most `u16::MAX` bytes (the writer's key-size limit). Everything that
/// reads sidecar bytes — [`read`] and the repair restore's raw rescue capture —
/// rejects anything larger BEFORE allocating, so a corrupt / attacker-padded
/// sidecar cannot force a huge allocation.
pub(crate) fn max_encoded_len(encryption: Option<&dyn EncryptionProvider>) -> u64 {
    HEADER_LEN as u64
        + u64::from(u16::MAX)
        + CHECKSUM_LEN as u64
        + u64::from(encryption.map_or(0, EncryptionProvider::max_overhead))
}

/// Outcome of reading the sidecar.
///
/// A genuine I/O failure (open other than not-found, or a read failure) is
/// returned as `Err` from [`read`] so the caller can classify it (transient →
/// retry, persistent → salvage); these variants cover the conclusive on-disk
/// states.
pub enum SidecarRead {
    /// Decoded and checksum-validated `(table_id, bound)`.
    Present(u64, Vec<u8>),
    /// Conclusively absent (the file does not exist): the SST is unrestricted.
    Missing,
    /// The sidecar was read but its payload is malformed — a failed checksum,
    /// bad length framing, an AEAD rejection, or an over-long file. PERSISTENT
    /// (a retry reads the same bytes): the bound is unrecoverable, so a caller
    /// facing a genuinely punched SST must fail closed rather than guess.
    Corrupt,
}

/// Reads and validates the restriction sidecar beside `table_path`.
///
/// # Errors
///
/// Returns the underlying I/O error when the sidecar exists but its open / read
/// fails (not a not-found, which is [`SidecarRead::Missing`]). The caller
/// classifies it — transient (retry) vs persistent (salvage). A payload that
/// reads but does not validate is [`SidecarRead::Corrupt`], not an error.
pub fn read(
    fs: &dyn Fs,
    table_path: &Path,
    encryption: Option<&dyn EncryptionProvider>,
) -> crate::Result<SidecarRead> {
    let path = sidecar_path(table_path);
    let mut file = match fs.open(&path, &FsOpenOptions::new().read(true)) {
        Ok(file) => file,
        Err(e) if e.kind() == crate::io::ErrorKind::NotFound => return Ok(SidecarRead::Missing),
        Err(e) => return Err(crate::Error::from(e)),
    };
    let max_len = max_encoded_len(encryption);
    if file.metadata()?.len > max_len {
        return Ok(SidecarRead::Corrupt);
    }
    let mut content = Vec::new();
    // Cap the reader at the same bound as the metadata probe, so the allocation
    // limit holds even if the file grew after the probe (a concurrent writer, or a
    // backend whose metadata disagrees with its data). Reading `max_len + 1` bytes
    // lets an over-long payload surface as `Corrupt` below.
    std::io::Read::read_to_end(
        &mut std::io::Read::take(&mut file, max_len.saturating_add(1)),
        &mut content,
    )
    .map_err(crate::Error::from)?;
    if content.len() as u64 > max_len {
        return Ok(SidecarRead::Corrupt);
    }
    let plain = match encryption {
        Some(enc) => match enc.decrypt(&content) {
            Ok(plain) => plain,
            Err(_) => return Ok(SidecarRead::Corrupt),
        },
        None => content,
    };
    match deserialize(&plain) {
        Some((table_id, bound)) => Ok(SidecarRead::Present(table_id, bound)),
        None => Ok(SidecarRead::Corrupt),
    }
}

/// Removes the sidecar (best-effort), syncing the parent so the removal is durable.
///
/// Called when a restricted table is compacted away / deleted, so a stale sidecar
/// does not linger to restrict an unrelated later file that reuses the id.
pub fn remove(fs: &dyn Fs, table_path: &Path, sync_mode: SyncMode) {
    let path = sidecar_path(table_path);
    if fs.remove_file(&path).is_ok()
        && let Some(parent) = path.parent()
    {
        let _ = fs.sync_directory_with(parent, sync_mode);
    }
}

/// True when a `.restrict-bound` sidecar exists beside `table_path`.
///
/// # Errors
///
/// Propagates the underlying [`Fs::exists`] error.
pub fn exists(fs: &dyn Fs, table_path: &Path) -> crate::io::Result<bool> {
    fs.exists(&sidecar_path(table_path))
}

#[cfg(test)]
mod tests;