lsm_tree/
pin.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026-present, Dmitry Prudnikov
3
4//! A pinned version: a point-in-time view of a tree that holds its files in
5//! place without copying, linking or syncing anything, and the means to rebuild
6//! a tree from it elsewhere.
7//!
8//! [`AbstractTree::pin`](crate::AbstractTree::pin) seals the active memtable
9//! and returns a [`PinnedVersion`] holding the tree's tables, blob files and
10//! sealed memtables as they were at the call. The files stay on disk with
11//! unchanged bytes until the last pin listing them drops, so a consumer can
12//! stream them at its own pace (a Raft snapshot sent to a lagging follower);
13//! [`Config::restore_pinned`](crate::Config::restore_pinned) builds the tree
14//! on the receiving side from the streamed files, [`PinnedVersion::layout`]
15//! and the memtable contents.
16
17use crate::{
18    SeqNo, UserKey, UserValue, ValueType,
19    deletion_pause::DeletionPause,
20    encryption::EncryptionProvider,
21    fs::{Fs, FsFile, FsOpenOptions, MemFs, SyncMode},
22    path::{Path, PathBuf},
23    runtime_config::RuntimeConfig,
24    tree::sealed::SealedMemtables,
25    version::Version,
26};
27use alloc::sync::Arc;
28use std::io::Read;
29
30/// What a [`PinnedVersion::layout`] starts with.
31const LAYOUT_MAGIC: [u8; 4] = *b"LSMP";
32
33/// The shape of the layout after its magic; a restore refuses any other.
34const LAYOUT_VERSION: u8 = 1;
35
36/// A file a [`PinnedVersion`] holds in place: a table, a blob file or a
37/// compression dictionary.
38#[derive(Clone)]
39pub struct PinnedFile {
40    id: u64,
41    level: Option<usize>,
42    path: PathBuf,
43    fs: Arc<dyn Fs>,
44}
45
46impl PinnedFile {
47    /// The file's id: the table, blob file or dictionary id it is named by.
48    #[must_use]
49    pub fn id(&self) -> u64 {
50        self.id
51    }
52
53    /// The level a table sits in; `None` for a blob file or a dictionary.
54    #[must_use]
55    pub fn level(&self) -> Option<usize> {
56        self.level
57    }
58
59    /// Where the file lies on the tree's filesystem.
60    #[must_use]
61    pub fn path(&self) -> &Path {
62        &self.path
63    }
64
65    /// The file's length in bytes.
66    ///
67    /// # Errors
68    ///
69    /// Propagates the filesystem's metadata failure.
70    pub fn file_size(&self) -> crate::Result<u64> {
71        Ok(self.fs.metadata(&self.path)?.len)
72    }
73
74    /// Opens the file for reading through the filesystem the tree uses for it,
75    /// to stream it byte for byte.
76    ///
77    /// # Errors
78    ///
79    /// Propagates the filesystem's open failure.
80    pub fn open(&self) -> crate::Result<Box<dyn FsFile>> {
81        Ok(self.fs.open(&self.path, &FsOpenOptions::new().read(true))?)
82    }
83
84    /// The heal attestation lying beside a table, if it has one: the record of
85    /// an in-place heal whose digest the version does not hold yet. Shipped
86    /// beside the table (same name with `.heal-attest` appended), it lets the
87    /// restored tree reconcile that table as the source would.
88    ///
89    /// # Errors
90    ///
91    /// Propagates the filesystem's probe failure.
92    pub fn heal_attestation(&self) -> crate::Result<Option<PathBuf>> {
93        if self.level.is_none() {
94            return Ok(None);
95        }
96        let path = crate::scrub::heal_attest::attest_path(&self.path);
97        Ok(self.fs.exists(&path)?.then_some(path))
98    }
99}
100
101impl core::fmt::Debug for PinnedFile {
102    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
103        f.debug_struct("PinnedFile")
104            .field("id", &self.id)
105            .field("level", &self.level)
106            .field("path", &self.path)
107            .finish_non_exhaustive()
108    }
109}
110
111/// One entry of a pinned memtable, as a flush would write it.
112#[derive(Clone, Debug, PartialEq, Eq)]
113pub struct PinnedEntry {
114    /// The user key.
115    pub key: UserKey,
116    /// The value (empty for a tombstone).
117    pub value: UserValue,
118    /// The entry's sequence number.
119    pub seqno: SeqNo,
120    /// What the entry is: a value, a tombstone, a merge operand, a cell row.
121    pub value_type: ValueType,
122}
123
124/// A range tombstone of a pinned memtable, deleting `[start, end)` below
125/// `seqno`.
126#[derive(Clone, Debug, PartialEq, Eq)]
127pub struct PinnedRangeTombstone {
128    /// First key deleted.
129    pub start: UserKey,
130    /// First key past the deleted range.
131    pub end: UserKey,
132    /// The tombstone's sequence number.
133    pub seqno: SeqNo,
134}
135
136/// A point-in-time view of a tree that keeps its files in place.
137///
138/// Taken by [`AbstractTree::pin`](crate::AbstractTree::pin). While it lives,
139/// every file it lists stays on disk with unchanged bytes, however the tree
140/// flushes and compacts meanwhile: the removal of a superseded file, an
141/// in-place heal and a tight-space hole punch of a listed file wait for the
142/// last pin listing it. Dropping it releases them; a restart drops every pin.
143///
144/// Pins are independent: each lists its own files and releases them on its
145/// own drop.
146pub struct PinnedVersion {
147    version: Version,
148    sealed: Arc<SealedMemtables>,
149    seqno: SeqNo,
150    tables: Vec<PinnedFile>,
151    blob_files: Vec<PinnedFile>,
152    dictionaries: Vec<PinnedFile>,
153    /// The source tree's directory lock: while the pin lives, no other opener
154    /// gets the folder (with a registry of its own that knows nothing of the
155    /// pin), even after the tree drops. Declared before `pause`, so it is
156    /// released before the pin lets the tree close.
157    _directory_lock: Option<Arc<dyn FsFile>>,
158    pause: Arc<DeletionPause>,
159    comparator_name: Arc<str>,
160    runtime: Arc<RuntimeConfig>,
161    encryption: Option<Arc<dyn EncryptionProvider>>,
162}
163
164impl PinnedVersion {
165    /// Pins `tree`'s current state; see [`AbstractTree::pin`](crate::AbstractTree::pin).
166    pub(crate) fn take(tree: &crate::Tree) -> Self {
167        let pause = Arc::clone(&tree.deletion_pause);
168        // No in-place mutation may start, and none may still be running, while
169        // the version is captured and its files registered.
170        let pinning = pause.begin_pin();
171
172        // The version lock is held only to seal, capture and register the
173        // dictionaries: writers wait on it, and the listings below can be long.
174        let (seqno, version, sealed, dictionaries) = {
175            let mut history = tree.version_history.write();
176            // Every write below this seqno has reached the active memtable:
177            // the insert runs under this lock's read side.
178            let seqno = tree.config.visible_seqno.get();
179            tree.rotate_memtable_locked(&mut history);
180            let super_version = history.latest_version_ref();
181            let version = super_version.version.clone();
182            let sealed = Arc::clone(&super_version.sealed_memtables);
183
184            let folder = tree.config.path.join(crate::file::DICTS_FOLDER);
185            let dictionaries: Vec<PinnedFile> = version
186                .dicts()
187                .iter()
188                .map(|&id| PinnedFile {
189                    id: u64::from(id),
190                    level: None,
191                    path: folder.join(id.to_string()),
192                    fs: Arc::clone(&tree.config.fs),
193                })
194                .collect();
195            // Under the version lock: the dictionary collection, which runs
196            // under it too, sees the pin or has finished. A table or blob file
197            // needs no such window: the captured version holds it, and no
198            // mutation starts before `pinning` drops.
199            pause.pin_files(dictionaries.iter().map(PinnedFile::path));
200            drop(history);
201            (seqno, version, sealed, dictionaries)
202        };
203
204        let tables: Vec<PinnedFile> = version
205            .iter_levels()
206            .enumerate()
207            .flat_map(|(level, runs)| {
208                runs.iter()
209                    .flat_map(|run| run.iter())
210                    .map(move |table| PinnedFile {
211                        id: table.id(),
212                        level: Some(level),
213                        path: (*table.path).clone(),
214                        fs: Arc::clone(&table.fs),
215                    })
216            })
217            .collect();
218        let blob_files: Vec<PinnedFile> = version
219            .blob_files
220            .iter()
221            .map(|blob_file| PinnedFile {
222                id: blob_file.id(),
223                level: None,
224                path: blob_file.0.path.clone(),
225                fs: Arc::clone(&blob_file.0.fs),
226            })
227            .collect();
228        pause.pin_files(tables.iter().chain(&blob_files).map(PinnedFile::path));
229        drop(pinning);
230
231        Self {
232            version,
233            sealed,
234            seqno,
235            tables,
236            blob_files,
237            dictionaries,
238            pause,
239            _directory_lock: tree.directory_lock.clone(),
240            comparator_name: tree.config.comparator.name().into(),
241            runtime: tree.runtime_config.load_full(),
242            encryption: tree.config.encryption.clone(),
243        }
244    }
245
246    /// The sequence number the pin was taken at: it holds every write below
247    /// it, and reads at any snapshot up to it see what the tree served then.
248    #[must_use]
249    pub fn seqno(&self) -> SeqNo {
250        self.seqno
251    }
252
253    /// The tables, with their levels.
254    #[must_use]
255    pub fn tables(&self) -> &[PinnedFile] {
256        &self.tables
257    }
258
259    /// The blob files.
260    #[must_use]
261    pub fn blob_files(&self) -> &[PinnedFile] {
262        &self.blob_files
263    }
264
265    /// The compression dictionaries the tables and blob files were written
266    /// against.
267    #[must_use]
268    pub fn dictionaries(&self) -> &[PinnedFile] {
269        &self.dictionaries
270    }
271
272    /// The version descriptor a manifest would hold for the pinned files
273    /// (levels, runs, restrictions, blob files, dictionaries, the retention
274    /// floor, a repair report the tree still owes: the pinned files hold
275    /// that repair's data, so the restored tree owes it too), serialized in
276    /// memory, sealed with the tree's encryption provider when it has one.
277    /// Opaque: hand it to
278    /// [`Config::restore_pinned`](crate::Config::restore_pinned).
279    ///
280    /// # Errors
281    ///
282    /// Propagates an encoding or encryption failure.
283    pub fn layout(&self) -> crate::Result<Vec<u8>> {
284        let fs = MemFs::new();
285        let path = Path::new("/layout");
286        let mut writer = crate::manifest_blocks::writer::ManifestArchiveWriter::create(
287            path,
288            &fs,
289            Arc::clone(&self.runtime),
290            self.encryption.clone(),
291            SyncMode::Normal,
292        )?;
293        self.version
294            .encode_into(&mut writer, &self.comparator_name)?;
295        writer.finish()?;
296
297        // Named and versioned first, so a restore refuses a layout of another
298        // shape rather than misread it.
299        let mut layout = LAYOUT_MAGIC.to_vec();
300        layout.push(LAYOUT_VERSION);
301        layout.extend_from_slice(&self.version.id().to_le_bytes());
302        // The pin's seqno rides along: the restored tree's counters start at
303        // it, so a later write there lands above what the pin served.
304        layout.extend_from_slice(&self.seqno.to_le_bytes());
305        fs.open(path, &FsOpenOptions::new().read(true))?
306            .read_to_end(&mut layout)?;
307        Ok(layout)
308    }
309
310    /// The entries of the sealed memtables, every version of every key below
311    /// [`seqno`](Self::seqno), with their sequence numbers and value types.
312    pub fn memtable_entries(&self) -> impl Iterator<Item = PinnedEntry> + '_ {
313        let seqno = self.seqno;
314        self.sealed
315            .iter()
316            .flat_map(|memtable| memtable.iter())
317            .filter(move |entry| entry.key.seqno < seqno)
318            .map(|entry| PinnedEntry {
319                key: entry.key.user_key,
320                value: entry.value,
321                seqno: entry.key.seqno,
322                value_type: entry.key.value_type,
323            })
324    }
325
326    /// The range tombstones of the sealed memtables below
327    /// [`seqno`](Self::seqno).
328    #[must_use]
329    pub fn memtable_range_tombstones(&self) -> Vec<PinnedRangeTombstone> {
330        self.sealed
331            .iter()
332            .flat_map(|memtable| memtable.range_tombstones_sorted())
333            .filter(|tombstone| tombstone.seqno < self.seqno)
334            .map(|tombstone| PinnedRangeTombstone {
335                start: tombstone.start,
336                end: tombstone.end,
337                seqno: tombstone.seqno,
338            })
339            .collect()
340    }
341}
342
343impl Drop for PinnedVersion {
344    fn drop(&mut self) {
345        // Before the version's handles drop, so the files whose last view goes
346        // with them are removed the ordinary way.
347        self.pause.unpin_files(
348            self.tables
349                .iter()
350                .chain(&self.blob_files)
351                .chain(&self.dictionaries)
352                .map(PinnedFile::path),
353        );
354    }
355}
356
357/// Builds the tree at `config.path` from a pin's files, its
358/// [`layout`](PinnedVersion::layout) and its memtable contents; see
359/// [`Config::restore_pinned`](crate::Config::restore_pinned).
360pub(crate) fn restore(
361    config: crate::Config,
362    layout: &[u8],
363    entries: impl IntoIterator<Item = PinnedEntry>,
364    range_tombstones: impl IntoIterator<Item = PinnedRangeTombstone>,
365) -> crate::Result<crate::AnyTree> {
366    use crate::AbstractTree;
367    use crate::file::{BLOBS_FOLDER, DICTS_FOLDER, TABLES_FOLDER, fsync_directory};
368    use std::io::Write;
369
370    let rest = layout
371        .strip_prefix(LAYOUT_MAGIC.as_slice())
372        .ok_or(crate::Error::InvalidHeader("pinned layout"))?;
373    let (version, rest) = rest
374        .split_first()
375        .ok_or(crate::Error::InvalidHeader("pinned layout"))?;
376    if *version != LAYOUT_VERSION {
377        return Err(crate::Error::InvalidHeader("pinned layout version"));
378    }
379    let (id, rest) = rest
380        .split_first_chunk::<8>()
381        .ok_or(crate::Error::InvalidHeader("pinned layout"))?;
382    let (pin_seqno, manifest) = rest
383        .split_first_chunk::<8>()
384        .ok_or(crate::Error::InvalidHeader("pinned layout"))?;
385    let version_id = u64::from_le_bytes(*id);
386    let pin_seqno = u64::from_le_bytes(*pin_seqno);
387
388    let fs = Arc::clone(&config.fs);
389    let root = config.path.clone();
390    let sync_mode = config.sync_mode;
391
392    // Exclusive from the first write to the open: no other opener may see the
393    // folder before the memtable contents are in.
394    fs.create_dir_all(&root)?;
395    let lock = crate::config::acquire_directory_lock(&*fs, &root, config.directory_lock)?;
396    let marker = root.join(crate::file::RESTORE_MARKER_FILE);
397    // A folder an interrupted restore left is this restore's to finish: its
398    // manifest goes, and the open below sweeps what that restore's flush
399    // wrote. Any other tree is never published over: its open would sweep
400    // that tree's manifest and tables as orphans of the restored layout.
401    let resuming = fs.exists(&marker)?;
402    let dirents = fs.read_dir(&root)?;
403    use crate::version::recovery::{ManifestFile, manifest_file_of};
404    let manifest_file = |name: &str| manifest_file_of(name).is_some();
405    let installed_manifest = |name: &str| {
406        matches!(
407            manifest_file_of(name),
408            Some(ManifestFile::Snapshot(_) | ManifestFile::Log(_))
409        )
410    };
411    // Judged before anything goes: a tree found here keeps every file.
412    if !resuming
413        && let Some(entry) = dirents
414            .iter()
415            .find(|entry| installed_manifest(&entry.file_name))
416    {
417        return Err(crate::Error::from(crate::io::Error::new(
418            crate::io::ErrorKind::AlreadyExists,
419            format!(
420                "restore_pinned: {} already holds a tree ({})",
421                root.display(),
422                entry.file_name
423            ),
424        )));
425    }
426    // The interrupted restore's manifest goes.
427    for dirent in dirents {
428        if manifest_file(&dirent.file_name) {
429            fs.remove_file(&dirent.path)?;
430        }
431    }
432
433    // The streamed files first: the manifest installed below must never name a
434    // file a power loss can still take back.
435    for folder in [TABLES_FOLDER, BLOBS_FOLDER, DICTS_FOLDER] {
436        let folder = root.join(folder);
437        if !fs.exists(&folder)? {
438            continue;
439        }
440        for entry in fs.read_dir(&folder)? {
441            if entry.is_dir {
442                continue;
443            }
444            // Write access for the flush alone: Windows refuses to flush a
445            // read-only handle. Nothing is written through it.
446            let file = fs.open(
447                &entry.path,
448                &FsOpenOptions::new()
449                    .read(true)
450                    .write(true)
451                    .write_origin(crate::fs::WriteOrigin::Metadata),
452            )?;
453            FsFile::sync_all_with(&*file, sync_mode)?;
454        }
455        fsync_directory(&folder, &*fs, sync_mode)?;
456    }
457
458    // Durable before the manifest is: a crash from here until the restored
459    // memtable contents are flushed leaves a folder an ordinary open refuses.
460    if !resuming {
461        let file = fs.open(
462            &marker,
463            &FsOpenOptions::new()
464                .write(true)
465                .create_new(true)
466                .write_origin(crate::fs::WriteOrigin::Metadata),
467        )?;
468        FsFile::sync_all_with(&*file, sync_mode)?;
469        drop(file);
470        fsync_directory(&root, &*fs, sync_mode)?;
471    }
472
473    // Written under a temporary name and renamed once its bytes are synced:
474    // the name installs the snapshot, and one directory sync makes it and the
475    // empty log durable. A temporary an earlier attempt left went with the
476    // manifest files above.
477    let temporary = root.join(format!(
478        "v{version_id}{}",
479        crate::version::SNAPSHOT_TEMPORARY_SUFFIX
480    ));
481    let mut file = fs.open(
482        &temporary,
483        &FsOpenOptions::new()
484            .write(true)
485            .create_new(true)
486            .write_origin(crate::fs::WriteOrigin::Manifest),
487    )?;
488    file.write_all(manifest)?;
489    file.flush()?;
490    FsFile::sync_data_with(&*file, sync_mode)?;
491    drop(file);
492    crate::version::edit_log::create_log(&*fs, &root, version_id, sync_mode)?;
493    fs.rename(&temporary, &root.join(format!("v{version_id}")))?;
494    fsync_directory(&root, &*fs, sync_mode)?;
495
496    let seqno = Arc::clone(&config.seqno);
497    let visible_seqno = Arc::clone(&config.visible_seqno);
498    let (tree, _) = config.open_with(
499        crate::config::DirectoryLock::Held(lock),
500        crate::tree::NewFolderNames::Synced,
501    )?;
502
503    // A restricted table keeps its bound beside it, as on the source, so a
504    // manifest repair of the restored tree recovers the restriction.
505    let version = tree.current_version();
506    for table in version.iter_tables() {
507        if let Some(bound) = table.restrict_lower_bound() {
508            crate::restrict_bound::write(
509                &*table.fs,
510                &table.path,
511                table.encryption.as_deref(),
512                table.id(),
513                bound,
514                sync_mode,
515            )?;
516        }
517    }
518
519    let index = match &tree {
520        crate::AnyTree::Standard(tree) => tree,
521        crate::AnyTree::Blob(tree) => &tree.index,
522    };
523    for entry in entries {
524        // The tree is not shared yet: the memtable size needs no backpressure,
525        // and the flush below takes everything.
526        let _ = index.append_entry(crate::InternalValue::from_components(
527            entry.key,
528            entry.value,
529            entry.seqno,
530            entry.value_type,
531        ));
532    }
533    let memtable = index.active_memtable();
534    for tombstone in range_tombstones {
535        let _ = memtable.insert_range_tombstone(tombstone.start, tombstone.end, tombstone.seqno);
536    }
537    drop(memtable);
538    // Collect nothing: the restored tree serves every snapshot the source did.
539    tree.flush_active_memtable(0)?;
540    // Under deferred durability the flush is installed in memory only; the
541    // marker below may go only once it is on disk.
542    tree.persist_flushes()?;
543
544    // A write on the restored tree must land above every restored version, and
545    // reads see what the pin served. The tables may hold a write at or above
546    // the pin's seqno (flushed before its seqno was published).
547    let next = tree
548        .get_highest_seqno()
549        .map_or(pin_seqno, |highest| pin_seqno.max(highest + 1));
550    seqno.fetch_max(next);
551    visible_seqno.fetch_max(pin_seqno);
552
553    // The flush made the restored contents durable: the tree is whole now.
554    fs.remove_file(&marker)?;
555    fsync_directory(&root, &*fs, sync_mode)?;
556    Ok(tree)
557}
lsm_tree/pin.rs

lsm_tree/
pin.rs
