loonfs_core/checkpoint/
files.rs

1//! Enumerating the files a checkpoint pins.
2//!
3//! A checkpoint pins one immutable manifest. This module walks that
4//! manifest's inode family in ascending inode-id order and answers, for
5//! every file visible in exactly that pinned state, the revision and content
6//! reference the checkpoint recorded for it. Nothing here consults the live
7//! head, the WAL, or any later manifest: a consumer that wants what changed
8//! after the pinned sequence reads the change feed.
9
10use super::cache::MetadataTableCache;
11use super::error::ManifestLoadError;
12use super::load::load_verified_manifest_tables_with_cache;
13use super::record::read_checkpoint_record;
14use crate::error::{CoreError, MetadataProjectionLoadError, Result};
15use crate::metadata::MetadataView;
16use loonfs_api::wire::control::{CheckpointRecordLifecycle, CheckpointRecordState};
17use loonfs_api::wire::manifest::{lookup_keys, MetadataRow, MetadataTableFamily};
18use loonfs_api::wire::sst_blocks::string_prefix_upper_bound;
19use loonfs_api::{
20    ChangeSeq, CheckpointId, ContentRef, InodeId, InodeKind, NamespaceId, PageRequest, RevisionNo,
21};
22use loonfs_objectstore::ObjectStore;
23
24/// Inode rows read per scan wave. Directories and invisible inodes are
25/// filtered out after the read, so a page of files may need several waves;
26/// the floor keeps a small page size from paying one scan per row.
27const INODE_SCAN_WAVE_ROWS: usize = 64;
28
29/// Where a file enumeration resumes: strictly after this inode id.
30///
31/// Enumeration order is ascending inode id, which is also the durable row
32/// order of the inode family, so a resume is one range bound — it never
33/// re-reads a row at or before `after_inode_id`.
34#[derive(Debug, Clone, Copy, PartialEq, Eq)]
35pub struct CheckpointFilesPageCursor {
36    /// Last inode id returned by the previous page.
37    pub after_inode_id: InodeId,
38}
39
40/// One file visible in the checkpointed state, with the content the
41/// checkpoint pinned for it.
42#[derive(Debug, Clone, PartialEq, Eq)]
43pub struct CheckpointFile {
44    /// The file's inode identity, stable across renames and replacements.
45    pub inode_id: InodeId,
46    /// The file's current revision as of the checkpointed sequence.
47    pub revision_no: RevisionNo,
48    /// Immutable bytes that revision published.
49    pub content_ref: ContentRef,
50    /// Byte length carried by `content_ref`, lifted out for planners that
51    /// size work before reading anything.
52    pub size_bytes: u64,
53}
54
55/// One page of the files a checkpoint pins.
56#[derive(Debug, Clone, PartialEq, Eq)]
57pub struct CheckpointFilesPage {
58    /// The sequence the checkpoint pins. Every file in the page is the
59    /// state at this sequence, and the change feed after it is exactly what
60    /// this page does not cover.
61    pub checkpoint_seq: ChangeSeq,
62    /// Files in ascending inode-id order.
63    pub files: Vec<CheckpointFile>,
64    /// Resume position when more files remain.
65    pub next_cursor: Option<CheckpointFilesPageCursor>,
66}
67
68/// Reads one page of the files visible in the state `checkpoint_id` pins.
69///
70/// The checkpoint's own manifest is the only state read: no WAL tail is
71/// replayed over it, and no later manifest is consulted, so the answer is
72/// the namespace exactly as of `checkpoint_seq`. Visibility is the same rule
73/// every read applies — the inode exists, and no subtree tombstone covers it
74/// or any of its ancestors — evaluated against that manifest. Directories
75/// are not returned.
76///
77/// A record that is missing or released answers `checkpoint_unavailable`,
78/// and so does a pinned manifest that is already gone. There is no fallback
79/// to current state: a consumer that lost its checkpoint takes a new one and
80/// starts over.
81///
82/// Release is the whole authority here — no clock. An active record whose
83/// expiry has passed still pins its basis and still serves: garbage
84/// collection is what turns a passed expiry into a release, and until it
85/// does, the record is a root, so answering from it is answering from state
86/// that is provably still there.
87pub(crate) async fn list_checkpoint_files_page<S: ObjectStore + ?Sized>(
88    store: &S,
89    table_cache: Option<&MetadataTableCache>,
90    namespace_id: &NamespaceId,
91    checkpoint_id: &CheckpointId,
92    request: PageRequest<CheckpointFilesPageCursor>,
93) -> Result<CheckpointFilesPage> {
94    let record = read_pinning_checkpoint_record(store, namespace_id, checkpoint_id).await?;
95    let tables = load_verified_manifest_tables_with_cache(
96        store,
97        table_cache,
98        namespace_id,
99        &record.manifest_object_id,
100    )
101    .await
102    .map_err(|error| match error {
103        ManifestLoadError::MissingManifest { object_key } => CoreError::CheckpointUnavailable(
104            format!("checkpoint `{checkpoint_id}` pins manifest `{object_key}`, which is gone"),
105        ),
106        other => CoreError::MetadataProjection(MetadataProjectionLoadError::ManifestLoad(other)),
107    })?;
108    let manifest = tables.manifest();
109    if manifest.payload_checksum != record.manifest_payload_checksum
110        || manifest.payload.head_seq != record.manifest_head_seq
111    {
112        return Err(CoreError::NamespaceCorrupt(format!(
113            "checkpoint `{checkpoint_id}` basis does not match its manifest"
114        )));
115    }
116
117    let checkpoint_seq = record.manifest_head_seq;
118    let view = MetadataView::over_manifest_tables(&tables, checkpoint_seq);
119    let mut session = view.session();
120
121    // One row past the page proves whether another file exists, so a cursor
122    // is handed out only when it will answer something.
123    let wanted = request.limit.limit_plus_one();
124    let wave_rows = wanted.max(INODE_SCAN_WAVE_ROWS);
125    let mut lower_bound = match request.cursor {
126        Some(cursor) => lookup_keys::inode_key_after(cursor.after_inode_id),
127        None => lookup_keys::INODE_ROW_PREFIX.to_owned(),
128    };
129    let upper_bound = string_prefix_upper_bound(lookup_keys::INODE_ROW_PREFIX);
130    let mut files = Vec::with_capacity(wanted);
131    while files.len() < wanted {
132        let rows = tables
133            .scan_range_page_with_keys(
134                MetadataTableFamily::Inodes,
135                &lower_bound,
136                upper_bound.as_deref(),
137                wave_rows,
138            )
139            .await
140            .map_err(|error| {
141                CoreError::MetadataProjection(MetadataProjectionLoadError::ManifestLoad(error))
142            })?;
143        let family_exhausted = rows.len() < wave_rows;
144        // Advance before consuming the wave: the next wave starts strictly
145        // after the last row this one read, whatever the filters keep.
146        match rows.last() {
147            Some((row_key, _)) => lower_bound = format!("{row_key}\0"),
148            None => break,
149        }
150        for (row_key, row) in rows {
151            let MetadataRow::Inode {
152                inode_id,
153                inode_kind,
154                ..
155            } = row
156            else {
157                return Err(CoreError::NamespaceCorrupt(format!(
158                    "inodes family returned a non-inode row at `{row_key}`"
159                )));
160            };
161            if inode_kind != InodeKind::File {
162                continue;
163            }
164            if session.visible_inode(inode_id).await?.is_none() {
165                continue;
166            }
167            let Some(revision) = session.latest_revision_head_of_visible(inode_id).await? else {
168                continue;
169            };
170            files.push(CheckpointFile {
171                inode_id,
172                revision_no: revision.revision_no,
173                size_bytes: revision.content_ref.size_bytes,
174                content_ref: revision.content_ref,
175            });
176            if files.len() == wanted {
177                break;
178            }
179        }
180        if family_exhausted {
181            break;
182        }
183    }
184
185    let has_more = files.len() > request.limit.as_usize();
186    if has_more {
187        files.truncate(request.limit.as_usize());
188    }
189    let next_cursor = has_more.then(|| CheckpointFilesPageCursor {
190        after_inode_id: files
191            .last()
192            .expect("a non-zero page limit with more files must return a file")
193            .inode_id,
194    });
195    Ok(CheckpointFilesPage {
196        checkpoint_seq,
197        files,
198        next_cursor,
199    })
200}
201
202/// Loads the record and refuses the one lifecycle that no longer pins its
203/// basis, so a caller never reads state garbage collection may already be
204/// reclaiming.
205async fn read_pinning_checkpoint_record<S: ObjectStore + ?Sized>(
206    store: &S,
207    namespace_id: &NamespaceId,
208    checkpoint_id: &CheckpointId,
209) -> Result<CheckpointRecordState> {
210    let Some(record) = read_checkpoint_record(store, namespace_id, checkpoint_id)
211        .await?
212        .map(|loaded| loaded.state)
213    else {
214        return Err(CoreError::CheckpointUnavailable(format!(
215            "checkpoint `{checkpoint_id}` does not exist in namespace `{namespace_id}`"
216        )));
217    };
218    if record.state != (CheckpointRecordLifecycle::Active {}) {
219        return Err(CoreError::CheckpointUnavailable(format!(
220            "checkpoint `{checkpoint_id}` is `{}` and no longer pins its basis",
221            record.state
222        )));
223    }
224    Ok(record)
225}
loonfs_core/checkpoint/files.rs

loonfs_core/checkpoint/
files.rs
