loonfs_core/metadata/
rows.rs

1//! Metadata row records plus the append and accounting plumbing that keeps
2//! [`MetadataState`]'s derived indexes and decoded-size totals in step with
3//! its append-only rows.
4
5use super::indexes::MetadataIndexes;
6use loonfs_api::wire::manifest::lookup_keys;
7use loonfs_api::{
8    ChangeSeq, CommitId, ContentRef, DisplayName, InodeId, InodeKind, NameKey, RevisionNo,
9};
10use serde::{Deserialize, Serialize};
11use std::mem::{size_of, size_of_val};
12
13#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
14pub struct MetadataState {
15    #[serde(default)]
16    pub(super) inodes: Vec<InodeRecord>,
17    #[serde(default)]
18    pub(super) direntry_binds: Vec<DirentryBindRecord>,
19    #[serde(default)]
20    pub(super) direntry_unbinds: Vec<DirentryUnbindRecord>,
21    #[serde(default)]
22    pub(super) revisions: Vec<RevisionRecord>,
23    #[serde(default)]
24    pub(super) subtree_tombstones: Vec<SubtreeTombstoneRecord>,
25    #[serde(default)]
26    pub(super) commit_receipts: Vec<CommitReceiptRecord>,
27    #[serde(skip)]
28    pub(super) row_count: usize,
29    #[serde(skip)]
30    pub(super) decoded_bytes: usize,
31    #[serde(skip)]
32    pub(super) indexes: MetadataIndexes,
33}
34
35#[derive(Debug, Clone, PartialEq, Eq, Default, Deserialize)]
36struct MetadataStateRows {
37    #[serde(default)]
38    inodes: Vec<InodeRecord>,
39    #[serde(default)]
40    direntry_binds: Vec<DirentryBindRecord>,
41    #[serde(default)]
42    direntry_unbinds: Vec<DirentryUnbindRecord>,
43    #[serde(default)]
44    revisions: Vec<RevisionRecord>,
45    #[serde(default)]
46    subtree_tombstones: Vec<SubtreeTombstoneRecord>,
47    #[serde(default)]
48    commit_receipts: Vec<CommitReceiptRecord>,
49}
50
51impl Default for MetadataState {
52    fn default() -> Self {
53        Self::from_rows(
54            Vec::new(),
55            Vec::new(),
56            Vec::new(),
57            Vec::new(),
58            Vec::new(),
59            Vec::new(),
60        )
61    }
62}
63
64impl<'de> Deserialize<'de> for MetadataState {
65    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
66    where
67        D: serde::Deserializer<'de>,
68    {
69        let rows = MetadataStateRows::deserialize(deserializer)?;
70        Ok(Self::from_rows(
71            rows.inodes,
72            rows.direntry_binds,
73            rows.direntry_unbinds,
74            rows.revisions,
75            rows.subtree_tombstones,
76            rows.commit_receipts,
77        ))
78    }
79}
80
81#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
82pub struct InodeRecord {
83    pub inode_id: InodeId,
84    pub inode_kind: InodeKind,
85    pub created_seq: ChangeSeq,
86}
87
88#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
89pub struct DirentryBindRecord {
90    pub parent_inode_id: InodeId,
91    pub name_key: NameKey,
92    pub display_name: DisplayName,
93    pub child_inode_id: InodeId,
94    pub bind_seq: ChangeSeq,
95    pub bind_delta_index: u32,
96}
97
98#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
99pub struct DirentryUnbindRecord {
100    pub parent_inode_id: InodeId,
101    pub name_key: NameKey,
102    /// User-facing spelling the retired binding carried: while the unbind
103    /// row is retained, it is the durable home of a deleted name.
104    pub display_name: DisplayName,
105    pub child_inode_id: InodeId,
106    pub bind_seq: ChangeSeq,
107    pub bind_delta_index: u32,
108    pub unbind_seq: ChangeSeq,
109    pub unbind_delta_index: u32,
110}
111
112#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
113pub struct RevisionRecord {
114    pub inode_id: InodeId,
115    pub revision_no: RevisionNo,
116    pub committed_seq: ChangeSeq,
117    /// Observational wall-clock stamp of the owning commit; never a
118    /// validity input — `committed_seq` is the order.
119    pub committed_at_ms: u64,
120    pub revision_delta_index: u32,
121    pub content_ref: ContentRef,
122}
123
124#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
125pub struct SubtreeTombstoneRecord {
126    pub root_inode_id: InodeId,
127    /// The recording event's sequence: the delete's committed seq for a
128    /// `Set`, the undelete's committed seq for a `Revoke`.
129    pub tombstone_seq: ChangeSeq,
130    pub tombstone_delta_index: u32,
131    /// Wall-clock stamp of the recording commit.
132    pub deleted_at_ms: u64,
133    /// Deleted-binding identity for `Set` rows from path deletes; the
134    /// tombstone row is immortal, so the deleted name lives here after
135    /// unbind rows age out.
136    pub parent_inode_id: Option<InodeId>,
137    pub name_key: Option<NameKey>,
138    pub display_name: Option<DisplayName>,
139    /// What this event did. Newest-by-(seq, delta) wins at every read
140    /// site: a `Set` newest means that deletion is active, a `Revoke`
141    /// newest means none is, and a later re-delete supersedes the revoke.
142    pub action: SubtreeTombstoneAction,
143}
144
145/// The tombstone family's event vocabulary — an explicit state machine
146/// instead of a boolean, so every reader matches exhaustively and a revoke
147/// names the exact deletion generation it cancels.
148#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
149pub enum SubtreeTombstoneAction {
150    /// The subtree rooted here is deleted.
151    Set,
152    /// The deletion recorded at `(target_seq, target_delta_index)` is
153    /// revoked. Validation guarantees the target was the active generation
154    /// when the revoke committed.
155    Revoke {
156        target_seq: ChangeSeq,
157        target_delta_index: u32,
158    },
159}
160
161/// The newest-event-wins active-tombstone rule, shared by every aggregation
162/// site: among records at or below `visible_seq`, the newest by
163/// `(tombstone_seq, tombstone_delta_index)` speaks for the root — a `Set`
164/// newest means that deletion is active, a `Revoke` newest means none is.
165/// The newest event is authoritative WITHOUT consulting the revoke's
166/// target: commit validation guarantees a revoke only ever lands against
167/// the generation that was active, so for valid histories the two rules
168/// agree, and the recorded target serves as audit metadata and the
169/// projection contract (change-feed consumers reduce with it and can flag
170/// a mismatch as corruption). Keep every reader on this helper — a site
171/// with its own copy of the rule is how visibility splits from the durable
172/// truth.
173pub(crate) fn active_tombstone_from_records(
174    records: impl IntoIterator<Item = SubtreeTombstoneRecord>,
175    visible_seq: ChangeSeq,
176) -> Option<SubtreeTombstoneRecord> {
177    records
178        .into_iter()
179        .filter(|tombstone| tombstone.tombstone_seq <= visible_seq)
180        .max_by_key(|tombstone| (tombstone.tombstone_seq, tombstone.tombstone_delta_index))
181        .filter(|tombstone| matches!(tombstone.action, SubtreeTombstoneAction::Set))
182}
183
184/// One row of the derived `ActiveDeletions` family: the current-state view of
185/// a single deletion generation.
186///
187/// Nothing appends these — [`active_deletion_from_tombstone`] derives one from
188/// every tombstone event, so the family is a projection of the tombstone
189/// family and never a second source of truth.
190#[derive(Debug, Clone, PartialEq, Eq)]
191pub(crate) struct ActiveDeletionRecord {
192    pub(crate) root_inode_id: InodeId,
193    /// The deletion's committed sequence. Together with `root_inode_id` this
194    /// is the handle `undelete` addresses and the trash entry renders.
195    pub(crate) deleted_at_seq: ChangeSeq,
196    pub(crate) action: ActiveDeletionAction,
197}
198
199/// What one active-deletion row says about its generation.
200#[derive(Debug, Clone, PartialEq, Eq)]
201pub(crate) enum ActiveDeletionAction {
202    /// The deletion is recoverable, and the trash lists it with these
203    /// details.
204    Listed {
205        deleted_at_ms: u64,
206        parent_inode_id: Option<InodeId>,
207        name_key: Option<NameKey>,
208        display_name: Option<DisplayName>,
209    },
210    /// An undelete cancelled the deletion, so the listing skips the key.
211    Removed { revoked_at_seq: ChangeSeq },
212}
213
214/// The `ActiveDeletions` reducer, and the only mapping from a tombstone event
215/// to its derived row: a `set` adds the deletion to the listing, and a
216/// `revoke` removes the exact generation it names.
217///
218/// It reduces target-aware where the newest-event-wins rule in
219/// [`active_tombstone_from_records`] reduces target-blind. Commit validation
220/// only ever lands a revoke against the generation that was active, so the two
221/// agree on every history a writer can produce; the target is what lets a
222/// removal be derived one event at a time instead of by re-reading a root's
223/// whole history.
224pub(crate) fn active_deletion_from_tombstone(
225    tombstone: &SubtreeTombstoneRecord,
226) -> ActiveDeletionRecord {
227    match &tombstone.action {
228        SubtreeTombstoneAction::Set => ActiveDeletionRecord {
229            root_inode_id: tombstone.root_inode_id,
230            deleted_at_seq: tombstone.tombstone_seq,
231            action: ActiveDeletionAction::Listed {
232                deleted_at_ms: tombstone.deleted_at_ms,
233                parent_inode_id: tombstone.parent_inode_id,
234                name_key: tombstone.name_key.clone(),
235                display_name: tombstone.display_name.clone(),
236            },
237        },
238        SubtreeTombstoneAction::Revoke { target_seq, .. } => ActiveDeletionRecord {
239            root_inode_id: tombstone.root_inode_id,
240            deleted_at_seq: *target_seq,
241            action: ActiveDeletionAction::Removed {
242                revoked_at_seq: tombstone.tombstone_seq,
243            },
244        },
245    }
246}
247
248impl ActiveDeletionRecord {
249    /// This row's durable key, which is also its position in the trash
250    /// listing's order.
251    pub(crate) fn row_key(&self) -> String {
252        lookup_keys::active_deletion_row_key(
253            self.deleted_at_seq,
254            self.root_inode_id,
255            match &self.action {
256                ActiveDeletionAction::Listed { .. } => lookup_keys::ACTIVE_DELETION_RANK_LISTED,
257                ActiveDeletionAction::Removed { .. } => lookup_keys::ACTIVE_DELETION_RANK_REMOVED,
258            },
259        )
260    }
261
262    /// The deletion this row lists, or `None` when the row is an undelete's
263    /// removal marker.
264    pub(crate) fn into_recoverable(self) -> Option<RecoverableDeletion> {
265        match self.action {
266            ActiveDeletionAction::Listed {
267                deleted_at_ms,
268                parent_inode_id,
269                name_key,
270                display_name,
271            } => Some(RecoverableDeletion {
272                root_inode_id: self.root_inode_id,
273                deleted_at_seq: self.deleted_at_seq,
274                deleted_at_ms,
275                parent_inode_id,
276                name_key,
277                display_name,
278            }),
279            ActiveDeletionAction::Removed { .. } => None,
280        }
281    }
282}
283
284/// One recoverable deletion as the trash listing renders it.
285#[derive(Debug, Clone, PartialEq, Eq)]
286pub(crate) struct RecoverableDeletion {
287    pub(crate) root_inode_id: InodeId,
288    pub(crate) deleted_at_seq: ChangeSeq,
289    pub(crate) deleted_at_ms: u64,
290    pub(crate) parent_inode_id: Option<InodeId>,
291    pub(crate) name_key: Option<NameKey>,
292    pub(crate) display_name: Option<DisplayName>,
293}
294
295#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
296pub struct CommitReceiptRecord {
297    pub commit_id: CommitId,
298    pub semantic_commit_fingerprint: String,
299    pub committed_seq: ChangeSeq,
300    /// Observational wall-clock stamp of the commit; never a validity
301    /// input — `committed_seq` is the order.
302    pub committed_at_ms: u64,
303    #[serde(default, skip_serializing_if = "Option::is_none")]
304    pub message: Option<String>,
305}
306
307impl MetadataState {
308    pub(crate) fn from_rows(
309        inodes: Vec<InodeRecord>,
310        direntry_binds: Vec<DirentryBindRecord>,
311        direntry_unbinds: Vec<DirentryUnbindRecord>,
312        revisions: Vec<RevisionRecord>,
313        subtree_tombstones: Vec<SubtreeTombstoneRecord>,
314        commit_receipts: Vec<CommitReceiptRecord>,
315    ) -> Self {
316        let mut state = Self {
317            inodes,
318            direntry_binds,
319            direntry_unbinds,
320            revisions,
321            subtree_tombstones,
322            commit_receipts,
323            row_count: 0,
324            decoded_bytes: 0,
325            indexes: MetadataIndexes::default(),
326        };
327        state.rebuild_indexes();
328        state
329    }
330
331    /// Highest sequence carried by any indexed record.
332    ///
333    /// No record carries a larger seq, so a query at `base_seq >=
334    /// indexed_seq()` passes every `seq <= base_seq` filter and is equivalent
335    /// to an at-head query. The seq-gated read methods below rely on this to
336    /// route to the indexes; only queries strictly below `indexed_seq()` need
337    /// the historical scans.
338    pub fn indexed_seq(&self) -> ChangeSeq {
339        self.indexes.indexed_seq()
340    }
341
342    pub fn inodes(&self) -> &[InodeRecord] {
343        &self.inodes
344    }
345
346    pub fn direntry_binds(&self) -> &[DirentryBindRecord] {
347        &self.direntry_binds
348    }
349
350    pub fn direntry_unbinds(&self) -> &[DirentryUnbindRecord] {
351        &self.direntry_unbinds
352    }
353
354    pub fn revisions(&self) -> &[RevisionRecord] {
355        &self.revisions
356    }
357
358    pub fn subtree_tombstones(&self) -> &[SubtreeTombstoneRecord] {
359        &self.subtree_tombstones
360    }
361
362    pub fn commit_receipts(&self) -> &[CommitReceiptRecord] {
363        &self.commit_receipts
364    }
365
366    pub fn row_count(&self) -> usize {
367        self.row_count
368    }
369
370    pub fn decoded_bytes(&self) -> usize {
371        self.decoded_bytes
372    }
373
374    pub fn find_commit_receipt(&self, commit_id: &CommitId) -> Option<&CommitReceiptRecord> {
375        self.indexes.commit_receipt(commit_id)
376    }
377
378    fn rebuild_indexes(&mut self) {
379        self.row_count = metadata_row_count(
380            &self.inodes,
381            &self.direntry_binds,
382            &self.direntry_unbinds,
383            &self.revisions,
384            &self.subtree_tombstones,
385            &self.commit_receipts,
386        );
387        self.decoded_bytes = metadata_decoded_bytes(
388            &self.inodes,
389            &self.direntry_binds,
390            &self.direntry_unbinds,
391            &self.revisions,
392            &self.subtree_tombstones,
393            &self.commit_receipts,
394        );
395        self.indexes = MetadataIndexes::rebuild(
396            &self.inodes,
397            &self.direntry_binds,
398            &self.direntry_unbinds,
399            &self.revisions,
400            &self.subtree_tombstones,
401            &self.commit_receipts,
402        );
403    }
404
405    pub(crate) fn push_inode_record(&mut self, record: InodeRecord) {
406        self.indexes.record_inode(&record);
407        self.record_row_weight(size_of::<InodeRecord>());
408        self.inodes.push(record);
409    }
410
411    pub(crate) fn push_direntry_bind_record(&mut self, record: DirentryBindRecord) {
412        self.indexes.record_bind(&record);
413        self.record_row_weight(direntry_bind_decoded_bytes(&record));
414        self.direntry_binds.push(record);
415    }
416
417    pub(crate) fn push_direntry_unbind_record(&mut self, record: DirentryUnbindRecord) {
418        self.indexes.record_unbind(&record);
419        self.record_row_weight(direntry_unbind_decoded_bytes(&record));
420        self.direntry_unbinds.push(record);
421    }
422
423    pub(crate) fn push_revision_record(&mut self, record: RevisionRecord) {
424        self.indexes.record_revision(&record);
425        self.record_row_weight(revision_decoded_bytes(&record));
426        self.revisions.push(record);
427    }
428
429    pub(crate) fn push_subtree_tombstone_record(&mut self, record: SubtreeTombstoneRecord) {
430        self.indexes.record_tombstone(&record);
431        self.record_row_weight(size_of::<SubtreeTombstoneRecord>());
432        self.subtree_tombstones.push(record);
433    }
434
435    pub(crate) fn push_commit_receipt_record(&mut self, record: CommitReceiptRecord) {
436        self.indexes.record_commit_receipt(&record);
437        self.record_row_weight(commit_receipt_decoded_bytes(&record));
438        self.commit_receipts.push(record);
439    }
440
441    fn record_row_weight(&mut self, decoded_bytes: usize) {
442        self.row_count = self.row_count.saturating_add(1);
443        self.decoded_bytes = self.decoded_bytes.saturating_add(decoded_bytes);
444    }
445}
446
447#[cfg(test)]
448#[derive(Debug, Default)]
449pub(crate) struct MetadataStateBuilder {
450    state: MetadataState,
451}
452
453#[cfg(test)]
454impl MetadataStateBuilder {
455    pub(crate) fn push_inode(&mut self, record: InodeRecord) {
456        self.state.push_inode_record(record);
457    }
458
459    pub(crate) fn push_direntry_bind(&mut self, record: DirentryBindRecord) {
460        self.state.push_direntry_bind_record(record);
461    }
462
463    pub(crate) fn push_direntry_unbind(&mut self, record: DirentryUnbindRecord) {
464        self.state.push_direntry_unbind_record(record);
465    }
466
467    pub(crate) fn push_revision(&mut self, record: RevisionRecord) {
468        self.state.push_revision_record(record);
469    }
470
471    pub(crate) fn push_subtree_tombstone(&mut self, record: SubtreeTombstoneRecord) {
472        self.state.push_subtree_tombstone_record(record);
473    }
474
475    pub(crate) fn push_commit_receipt(&mut self, record: CommitReceiptRecord) {
476        self.state.push_commit_receipt_record(record);
477    }
478
479    pub(crate) fn finish(mut self) -> MetadataState {
480        self.state.rebuild_indexes();
481        self.state
482    }
483}
484
485fn metadata_row_count(
486    inodes: &[InodeRecord],
487    direntry_binds: &[DirentryBindRecord],
488    direntry_unbinds: &[DirentryUnbindRecord],
489    revisions: &[RevisionRecord],
490    subtree_tombstones: &[SubtreeTombstoneRecord],
491    commit_receipts: &[CommitReceiptRecord],
492) -> usize {
493    inodes
494        .len()
495        .saturating_add(direntry_binds.len())
496        .saturating_add(direntry_unbinds.len())
497        .saturating_add(revisions.len())
498        .saturating_add(subtree_tombstones.len())
499        .saturating_add(commit_receipts.len())
500}
501
502fn metadata_decoded_bytes(
503    inodes: &[InodeRecord],
504    direntry_binds: &[DirentryBindRecord],
505    direntry_unbinds: &[DirentryUnbindRecord],
506    revisions: &[RevisionRecord],
507    subtree_tombstones: &[SubtreeTombstoneRecord],
508    commit_receipts: &[CommitReceiptRecord],
509) -> usize {
510    size_of_val(inodes)
511        .saturating_add(
512            direntry_binds
513                .iter()
514                .map(direntry_bind_decoded_bytes)
515                .sum::<usize>(),
516        )
517        .saturating_add(
518            direntry_unbinds
519                .iter()
520                .map(direntry_unbind_decoded_bytes)
521                .sum::<usize>(),
522        )
523        .saturating_add(revisions.iter().map(revision_decoded_bytes).sum::<usize>())
524        .saturating_add(size_of_val(subtree_tombstones))
525        .saturating_add(
526            commit_receipts
527                .iter()
528                .map(commit_receipt_decoded_bytes)
529                .sum::<usize>(),
530        )
531}
532
533fn direntry_bind_decoded_bytes(record: &DirentryBindRecord) -> usize {
534    size_of::<DirentryBindRecord>()
535        + record.name_key.as_str().len()
536        + record.display_name.as_str().len()
537}
538
539fn direntry_unbind_decoded_bytes(record: &DirentryUnbindRecord) -> usize {
540    size_of::<DirentryUnbindRecord>() + record.name_key.as_str().len()
541}
542
543fn revision_decoded_bytes(record: &RevisionRecord) -> usize {
544    size_of::<RevisionRecord>() + content_ref_decoded_bytes(&record.content_ref)
545}
546
547fn commit_receipt_decoded_bytes(record: &CommitReceiptRecord) -> usize {
548    size_of::<CommitReceiptRecord>()
549        + record.commit_id.as_str().len()
550        + record.semantic_commit_fingerprint.len()
551        + record.message.as_ref().map_or(0, String::len)
552}
553
554fn content_ref_decoded_bytes(content_ref: &ContentRef) -> usize {
555    size_of::<ContentRef>() + content_ref_evidence_bytes(content_ref)
556}
557
558/// Heap bytes a content reference owns beyond its struct: the identity and
559/// the checksum strings.
560pub(crate) fn content_ref_evidence_bytes(content_ref: &ContentRef) -> usize {
561    content_ref.content_id.as_str().len()
562        + content_ref.storage_checksum.value.len()
563        + content_ref
564            .whole_file_sha256
565            .as_ref()
566            .map_or(0, String::len)
567}
loonfs_core/metadata/rows.rs

loonfs_core/metadata/
rows.rs
