coordinode-lsm-tree 7.4.2

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
// SPDX-License-Identifier: Apache-2.0
// Copyright (c) 2026-present, Dmitry Prudnikov

//! The tree's compression dictionaries on disk.
//!
//! A table records the id of the dictionary it was compressed with, and that id
//! is all it records. This module is what turns the id back into bytes across a
//! reopen: one file per dictionary under [`DICTS_FOLDER`](crate::file::DICTS_FOLDER), named by the id.
//!
//! ## Why the name is the checksum
//!
//! A dictionary id is the truncated xxh3 of its own bytes, so re-hashing the
//! file and comparing against its name is a full integrity check: a bit flip
//! changes the hash and the file stops answering to the name it is filed under.
//! That is why no separate digest is stored. Detecting corruption matters more
//! here than elsewhere in the tree, because a silently altered dictionary does
//! not fail a read, it decompresses every block written against it into
//! plausible-looking garbage.
//!
//! ## Encrypted trees
//!
//! A dictionary keeps literal stretches of the records it was trained on, so on
//! a tree with an encryption provider the file is sealed with that provider,
//! the same way the manifest is. The name check runs over the decrypted bytes,
//! and a seal that no longer opens is the same damage, caught one step earlier.
//! A tree is encrypted throughout or not at all, so the provider alone says
//! which form a file is in.

use crate::compression::{ZstdDictionaries, ZstdDictionary};
use crate::encryption::EncryptionProvider;
use crate::file::{DICT_TMP_SUFFIX, DictDirEntry, DictId};
use crate::fs::{Fs, FsFile, FsOpenOptions, SyncMode};
#[cfg(not(feature = "std"))]
use crate::io::{Read, Write};
use crate::path::{Path, PathBuf};
use alloc::sync::Arc;
use alloc::vec::Vec;
#[cfg(feature = "std")]
use std::io::{Read, Write};

/// The file `id` is stored under.
#[must_use]
pub fn path_of(folder: &Path, id: DictId) -> PathBuf {
    folder.join(id.to_string())
}

/// Writes `dict` into the tree's dictionary folder, creating the folder if this
/// is the first one, sealed with `encryption` when the tree has a provider.
///
/// Published by an atomic rename, so a crash mid-write leaves a `.tmp` the
/// sweep disposes of rather than a half-written dictionary under a live name.
/// Writing an id the folder already holds is a no-op ONLY when the bytes match.
/// The id is the truncated xxh3 of the content, so that is the overwhelmingly
/// common case; a DIFFERENT dictionary under the same id is a collision of the
/// truncation, and is refused.
///
/// # Errors
///
/// [`crate::Error::ZstdDictMismatch`] when the folder already holds a different
/// dictionary under this id. Accepting it silently would be the worst outcome
/// the store can produce: the caller's new dictionary would be selected for
/// WRITES while every read resolved the id back to the old bytes, so new blocks
/// would decompress into plausible garbage. Otherwise propagates the provider's
/// encryption failure and the create / write / sync / rename failures of the
/// backend.
pub fn write(
    fs: &dyn Fs,
    folder: &Path,
    dict: &ZstdDictionary,
    encryption: Option<&dyn EncryptionProvider>,
    sync_mode: SyncMode,
) -> crate::Result<()> {
    let final_path = path_of(folder, dict.id());
    if fs.exists(&final_path)? {
        // Same id: prove it is the same DICTIONARY before treating the write as
        // done. `read_one` re-hashes, so a corrupt file is caught here too.
        match read_one(fs, folder, dict.id(), encryption) {
            Ok(held) => {
                if held.raw() != dict.raw() {
                    return Err(crate::Error::ZstdDictMismatch {
                        expected: dict.id(),
                        got: Some(held.id()),
                    });
                }
                // Sync even though the bytes were already there. The file may
                // be a PREVIOUS attempt that died between its rename and its
                // syncs, leaving the entries not yet durable; returning without
                // syncing would let the caller durably register the id on top
                // of a file a power loss can still lose.
                return sync_entries(fs, folder, sync_mode);
            }
            // The file no longer hashes to its name, so it is damaged bytes,
            // not a different dictionary; the one being written hashes to the
            // name by construction and replaces them below.
            Err(e) if is_damage(&e, dict.id()) => {}
            Err(e) => return Err(e),
        }
    }

    if !fs.exists(folder)? {
        fs.create_dir_all(folder)?;
    }

    // Sealed before the temp file exists, so a provider failure leaves nothing
    // behind to clean up.
    let sealed = encryption.map(|e| e.encrypt(dict.raw())).transpose()?;
    let bytes = match &sealed {
        Some(ciphertext) => ciphertext.as_slice(),
        None => dict.raw(),
    };

    let tmp_path = folder.join(format!("{}{DICT_TMP_SUFFIX}", dict.id()));
    // `create(true)` rather than `create_new(true)`: a `.tmp` left by a crashed
    // registration is disposable by definition (it is referenced by no version),
    // so overwriting it is the recovery, not a hazard.
    let mut file = fs.open(
        &tmp_path,
        &FsOpenOptions::new()
            .write(true)
            .create(true)
            .truncate(true)
            .write_origin(crate::fs::WriteOrigin::Metadata),
    )?;
    let written = file
        .write_all(bytes)
        .map_err(crate::io::Error::from)
        .and_then(|()| file.flush().map_err(crate::io::Error::from))
        .and_then(|()| FsFile::sync_all_with(&*file, sync_mode));
    drop(file);
    if let Err(e) = written {
        let _ = fs.remove_file(&tmp_path);
        return Err(e.into());
    }

    if let Err(e) = fs.rename(&tmp_path, &final_path) {
        let _ = fs.remove_file(&tmp_path);
        return Err(e.into());
    }
    sync_entries(fs, folder, sync_mode)
}

/// Makes the file's entry in `dicts/` durable, and the entry of `dicts/` in the
/// tree root.
///
/// Both, on every write, whether or not this call created the folder: an
/// earlier attempt may have created it and died before its own syncs, and a
/// retry that took the existing folder as durable would let the caller register
/// the id over a folder a power loss can still take away. Syncing an already
/// durable directory costs one call; this runs once per registration.
fn sync_entries(fs: &dyn Fs, folder: &Path, sync_mode: SyncMode) -> crate::Result<()> {
    // The version edit naming the dictionary goes to the tree root above.
    fs.sync_directory_with(folder, crate::file::folder_sync_mode(fs, folder, sync_mode))?;
    if let Some(parent) = folder.parent() {
        fs.sync_directory_with(parent, sync_mode)?;
    }
    Ok(())
}

/// Reads the dictionary stored under `id`, opening the seal with `encryption`
/// when the tree has a provider.
///
/// # Errors
///
/// [`crate::Error::ZstdDictCorrupt`] when the bytes on disk do not hash to the
/// id they are filed under, which is this store's integrity check, or when the
/// seal no longer opens under `encryption`. Otherwise propagates the open /
/// read failures of the backend, including `NotFound` when the tree does not
/// hold that dictionary.
pub fn read_one(
    fs: &dyn Fs,
    folder: &Path,
    id: DictId,
    encryption: Option<&dyn EncryptionProvider>,
) -> crate::Result<ZstdDictionary> {
    let mut file = fs.open(&path_of(folder, id), &FsOpenOptions::new().read(true))?;
    let mut raw = Vec::new();
    file.read_to_end(&mut raw)?;
    if let Some(e) = encryption {
        // A seal that does not open is the authenticated form of the name check
        // failing, and is reported as damage rather than as a wrong key. That is
        // safe because a wrong key never gets to act on the verdict: an open of
        // an existing tree verifies the manifest under the key before it reads
        // a dictionary, and a repair under a wrong key stops on the first sealed
        // table, before its commit, so it never sets a healthy file aside. A
        // repair that commits with no sealed file to prove the key writes the
        // manifest under this key, which makes it the tree's.
        raw = match e.decrypt_vec(raw) {
            Ok(plain) => plain,
            Err(crate::Error::Decrypt(_)) => {
                return Err(crate::Error::ZstdDictCorrupt { id, got: None });
            }
            Err(other) => return Err(other),
        };
    }

    let dict = ZstdDictionary::new(&raw);
    if dict.id() != id {
        // The name IS the digest, so a mismatch is corruption of the bytes (or
        // a file placed under a name it does not own), never a stale name.
        return Err(crate::Error::ZstdDictCorrupt {
            id,
            got: Some(dict.id()),
        });
    }
    Ok(dict)
}

/// Loads every dictionary the folder holds.
///
/// The READ side scans rather than following the version's id list, so a
/// dictionary whose file landed before the version edit that registers it (a
/// crash in between) still resolves, and an open never fails on an id the list
/// and the folder disagree about. The list is what says which files are still
/// OWED to a reader, which is a collection question, not a read one.
///
/// # Errors
///
/// Propagates the directory read failure, and [`read_one`]'s integrity check:
/// a corrupt dictionary fails the open rather than silently dropping out of the
/// set, which would turn into "unknown dictionary id" on the first table that
/// needs it.
pub fn read_all(
    fs: &dyn Fs,
    folder: &Path,
    encryption: Option<&dyn EncryptionProvider>,
) -> crate::Result<ZstdDictionaries> {
    if !fs.exists(folder)? {
        return Ok(ZstdDictionaries::new());
    }
    let mut set = ZstdDictionaries::new();
    for dirent in fs.read_dir(folder)? {
        if let DictDirEntry::Dict(id) = DictDirEntry::classify(&dirent.file_name) {
            set = set.with(Arc::new(read_one(fs, folder, id, encryption)?));
        }
    }
    Ok(set)
}

/// Whether `err` is [`read_one`]'s integrity failure for `id`: the file filed
/// under `id` holds bytes that hash elsewhere.
fn is_damage(err: &crate::Error, id: DictId) -> bool {
    matches!(err, crate::Error::ZstdDictCorrupt { id: filed, .. } if *filed == id)
}

/// Loads every intact dictionary the folder holds, skipping each one whose
/// bytes no longer hash to its name, and returns the skipped ids.
///
/// The repair's counterpart of [`read_all`], which fails on the first damaged
/// file: a repair exists for exactly that tree, and an orphaned damaged
/// dictionary must not stop it before it looks at a table. Nothing is changed
/// on disk here; see [`set_aside_if_damaged`]. A table that needs a skipped
/// dictionary fails to open naming it, as one whose dictionary is missing does.
///
/// # Errors
///
/// Propagates the directory read and every read failure other than the
/// integrity check: a transient I/O fault or a wrong key has to fail the
/// repair, not cost the tree a dictionary.
pub fn read_all_skipping_damaged(
    fs: &dyn Fs,
    folder: &Path,
    encryption: Option<&dyn EncryptionProvider>,
) -> crate::Result<(ZstdDictionaries, Vec<DictId>)> {
    let mut set = ZstdDictionaries::new();
    let mut damaged = Vec::new();
    if !fs.exists(folder)? {
        return Ok((set, damaged));
    }
    for dirent in fs.read_dir(folder)? {
        let DictDirEntry::Dict(id) = DictDirEntry::classify(&dirent.file_name) else {
            continue;
        };
        match read_one(fs, folder, id, encryption) {
            Ok(dict) => set = set.with(Arc::new(dict)),
            Err(e) if is_damage(&e, id) => damaged.push(id),
            Err(e) => return Err(e),
        }
    }
    Ok((set, damaged))
}

/// Moves the dictionary filed under `id` to `{id}.damaged` if its bytes still
/// fail the integrity check, returning where it went.
///
/// `{id}.damaged` is outside the names an open reads, so the tree opens, and
/// outside the names a sweep removes, so the bytes stay for an operator. A file
/// that reads intact again (a correct copy was written over it meanwhile) or is
/// gone is left alone.
///
/// # Errors
///
/// Propagates the read failures other than the integrity check, and the
/// rename and its directory sync.
pub fn set_aside_if_damaged(
    fs: &dyn Fs,
    folder: &Path,
    id: DictId,
    encryption: Option<&dyn EncryptionProvider>,
    sync_mode: SyncMode,
) -> crate::Result<Option<PathBuf>> {
    if !is_still_damaged(fs, folder, id, encryption)? {
        return Ok(None);
    }
    let aside = damaged_path(folder, id);
    fs.rename(&path_of(folder, id), &aside)?;
    fs.sync_directory_with(folder, sync_mode)?;
    Ok(Some(aside))
}

/// Whether the file filed under `id` exists and still fails the integrity
/// check, so [`set_aside_if_damaged`] would move it.
///
/// # Errors
///
/// Propagates the read failures other than the integrity check and `NotFound`.
pub fn is_still_damaged(
    fs: &dyn Fs,
    folder: &Path,
    id: DictId,
    encryption: Option<&dyn EncryptionProvider>,
) -> crate::Result<bool> {
    match read_one(fs, folder, id, encryption) {
        Ok(_) => Ok(false),
        Err(crate::Error::Io(e)) if e.kind() == crate::io::ErrorKind::NotFound => Ok(false),
        Err(e) if is_damage(&e, id) => Ok(true),
        Err(e) => Err(e),
    }
}

/// Where [`set_aside_if_damaged`] moves the dictionary filed under `id`.
#[must_use]
pub fn damaged_path(folder: &Path, id: DictId) -> PathBuf {
    folder.join(format!("{id}{}", crate::file::DICT_DAMAGED_SUFFIX))
}

/// Lists the ids the folder holds, without reading a single dictionary.
///
/// The collector's view: it decides what to UNLINK, and for that the names are
/// the whole answer. Reading the bytes would make it hash every dictionary in
/// the tree, and would fail the pass on a corrupt one it may be about to remove.
///
/// # Errors
///
/// Propagates the directory read failure.
pub fn list_ids(fs: &dyn Fs, folder: &Path) -> crate::Result<Vec<DictId>> {
    if !fs.exists(folder)? {
        return Ok(Vec::new());
    }
    let mut ids = Vec::new();
    for dirent in fs.read_dir(folder)? {
        if let DictDirEntry::Dict(id) = DictDirEntry::classify(&dirent.file_name) {
            ids.push(id);
        }
    }
    Ok(ids)
}

/// Removes the dictionary stored under `id`, treating an already-absent file as
/// success.
///
/// # Errors
///
/// Propagates the backend's removal failures other than `NotFound`.
pub fn remove(fs: &dyn Fs, folder: &Path, id: DictId, sync_mode: SyncMode) -> crate::Result<()> {
    match fs.remove_file(&path_of(folder, id)) {
        Ok(()) => {}
        Err(e) if e.kind() == crate::io::ErrorKind::NotFound => return Ok(()),
        Err(e) => return Err(e.into()),
    }
    fs.sync_directory_with(folder, sync_mode)?;
    Ok(())
}

/// Deletes every unpublished `.tmp` in the dictionary folder.
///
/// A `.tmp` survives only a crash between write and rename, and is referenced
/// by no version, so it is always disposable. Foreign names are never touched:
/// the folder may hold an operator's files and they are not engine state.
///
/// # Errors
///
/// Propagates the directory read failure. A file that cannot be removed is
/// logged and skipped rather than failing the open: a leftover temp costs
/// space, not correctness.
pub fn sweep_temps(fs: &dyn Fs, folder: &Path) -> crate::Result<()> {
    if !fs.exists(folder)? {
        return Ok(());
    }
    for dirent in fs.read_dir(folder)? {
        let name = dirent.file_name;
        if let DictDirEntry::Tmp(id) = DictDirEntry::classify(&name) {
            let path = folder.join(&name);
            if let Err(e) = fs.remove_file(&path) {
                log::warn!("Could not remove unpublished dictionary temp {id}: {e:?}");
            }
        }
    }
    Ok(())
}

#[cfg(test)]
#[expect(clippy::unwrap_used, reason = "test code")]
mod tests;