loonfs_core/metadata/
apply.rs

1//! Application of committed WAL deltas and commit records onto
2//! [`MetadataState`] rows.
3
4use super::{
5    CommitReceiptRecord, DirentryBindRecord, DirentryUnbindRecord, InodeRecord, MetadataState,
6    RevisionRecord, SubtreeTombstoneAction, SubtreeTombstoneRecord,
7};
8use loonfs_api::wire::wal::{WalCommitDelta, WalCommitPayload, WalDelta};
9use loonfs_api::{ChangeSeq, CommitId};
10
11impl MetadataState {
12    pub fn apply_committed_wal_deltas(
13        &self,
14        committed_seq: ChangeSeq,
15        committed_at_ms: u64,
16        deltas: &[WalDelta],
17    ) -> MetadataState {
18        let mut metadata_state = self.clone();
19        metadata_state.apply_committed_wal_deltas_mut(committed_seq, committed_at_ms, deltas);
20        metadata_state
21    }
22
23    pub fn apply_committed_wal_deltas_mut(
24        &mut self,
25        committed_seq: ChangeSeq,
26        committed_at_ms: u64,
27        deltas: &[WalDelta],
28    ) {
29        for delta in deltas {
30            self.apply_committed_wal_delta_mut(committed_seq, committed_at_ms, delta);
31        }
32    }
33
34    /// Appends the metadata row encoded by one committed WAL delta.
35    ///
36    /// This is the only WAL-delta to metadata-row mapping in the crate:
37    /// durable replay ([`Self::apply_committed_wal_deltas_mut`]) and the
38    /// commit validation overlay (`commit::metadata_overlay`) both append
39    /// rows through it, so the effects a batch validates against cannot
40    /// diverge from what replay later persists.
41    pub(crate) fn apply_committed_wal_delta_mut(
42        &mut self,
43        committed_seq: ChangeSeq,
44        committed_at_ms: u64,
45        delta: &WalDelta,
46    ) {
47        match delta {
48            WalDelta::CreateInode {
49                delta_index: _,
50                inode_id,
51                inode_kind,
52            } => {
53                self.push_inode_record(InodeRecord {
54                    inode_id: *inode_id,
55                    inode_kind: *inode_kind,
56                    created_seq: committed_seq,
57                });
58            }
59            WalDelta::BindDirentry {
60                delta_index,
61                parent_inode_id,
62                name_key,
63                display_name,
64                child_inode_id,
65            } => {
66                self.push_direntry_bind_record(DirentryBindRecord {
67                    parent_inode_id: *parent_inode_id,
68                    name_key: name_key.clone(),
69                    display_name: display_name.clone(),
70                    child_inode_id: *child_inode_id,
71                    bind_seq: committed_seq,
72                    bind_delta_index: *delta_index,
73                });
74            }
75            WalDelta::UnbindDirentry {
76                delta_index,
77                parent_inode_id,
78                name_key,
79                display_name,
80                child_inode_id,
81                bind_seq,
82                bind_delta_index,
83            } => {
84                self.push_direntry_unbind_record(DirentryUnbindRecord {
85                    parent_inode_id: *parent_inode_id,
86                    name_key: name_key.clone(),
87                    display_name: display_name.clone(),
88                    child_inode_id: *child_inode_id,
89                    bind_seq: *bind_seq,
90                    bind_delta_index: *bind_delta_index,
91                    unbind_seq: committed_seq,
92                    unbind_delta_index: *delta_index,
93                });
94            }
95            WalDelta::AppendFileRevision {
96                delta_index,
97                inode_id,
98                revision_no,
99                content_ref,
100            } => {
101                self.push_revision_record(RevisionRecord {
102                    inode_id: *inode_id,
103                    revision_no: *revision_no,
104                    committed_seq,
105                    committed_at_ms,
106                    revision_delta_index: *delta_index,
107                    content_ref: content_ref.clone(),
108                });
109            }
110            WalDelta::TombstoneSubtree {
111                delta_index,
112                root_inode_id,
113                parent_inode_id,
114                name_key,
115                display_name,
116            } => {
117                self.push_subtree_tombstone_record(SubtreeTombstoneRecord {
118                    root_inode_id: *root_inode_id,
119                    tombstone_seq: committed_seq,
120                    tombstone_delta_index: *delta_index,
121                    deleted_at_ms: committed_at_ms,
122                    parent_inode_id: *parent_inode_id,
123                    name_key: name_key.clone(),
124                    display_name: display_name.clone(),
125                    action: SubtreeTombstoneAction::Set,
126                });
127            }
128            WalDelta::RevokeSubtreeTombstone {
129                delta_index,
130                root_inode_id,
131                target_seq,
132                target_delta_index,
133            } => {
134                self.push_subtree_tombstone_record(SubtreeTombstoneRecord {
135                    root_inode_id: *root_inode_id,
136                    tombstone_seq: committed_seq,
137                    tombstone_delta_index: *delta_index,
138                    deleted_at_ms: committed_at_ms,
139                    parent_inode_id: None,
140                    name_key: None,
141                    display_name: None,
142                    action: SubtreeTombstoneAction::Revoke {
143                        target_seq: *target_seq,
144                        target_delta_index: *target_delta_index,
145                    },
146                });
147            }
148        }
149    }
150
151    pub fn apply_committed_wal_record_mut(&mut self, record: &WalCommitPayload) {
152        self.apply_committed_wal_record_parts_mut(
153            record.seq,
154            record.committed_at_ms,
155            &record.commit_id,
156            &record.semantic_commit_fingerprint,
157            record.message.as_deref(),
158            &record.deltas,
159        )
160    }
161
162    pub(crate) fn apply_committed_wal_record_parts_mut(
163        &mut self,
164        seq: ChangeSeq,
165        committed_at_ms: u64,
166        commit_id: &CommitId,
167        semantic_commit_fingerprint: &str,
168        message: Option<&str>,
169        deltas: &[WalCommitDelta],
170    ) {
171        for delta in deltas {
172            self.apply_committed_wal_delta_mut(seq, committed_at_ms, &delta.delta);
173        }
174        self.push_commit_receipt_record(CommitReceiptRecord {
175            commit_id: commit_id.clone(),
176            semantic_commit_fingerprint: semantic_commit_fingerprint.to_owned(),
177            committed_seq: seq,
178            committed_at_ms,
179            message: message.map(str::to_owned),
180        });
181    }
182}
loonfs_core/metadata/apply.rs

loonfs_core/metadata/
apply.rs
