acyclic_fs/kernel/
attribute_mutation.rs

1//! Sparse persistent named-attribute mutation over the shared B+tree engine.
2
3use super::allocation::AllocationError;
4use super::attribute::{
5    attribute_internal_item_encoded_length, attribute_internal_page_encoded_length,
6    attribute_leaf_item_encoded_length, attribute_leaf_page_encoded_length,
7    attribute_page_decode_shape, encode_attribute_internal_children, encode_attribute_leaf_entries,
8};
9use super::codec::DecodedPageShape;
10use super::persistent_btree::{self, Child, Format, Page, PageRef};
11use super::{
12    AttributeEntry, AttributeName, AttributePage, CanonicalDecodeError, DecodeLimits,
13    decode_attribute_page,
14};
15use crate::performance::{OperationFailure, WorkBudget, WorkCounters, WorkError};
16use crate::storage::{ObjectId, ObjectKind, ObjectStoreError};
17use thiserror::Error;
18
19pub(crate) struct AttributeFormat;
20
21impl Format for AttributeFormat {
22    type Key = AttributeName;
23    type Value = AttributeEntry;
24
25    fn kind() -> ObjectKind {
26        ObjectKind::AttributePage
27    }
28
29    fn key(value: &Self::Value) -> &Self::Key {
30        &value.name
31    }
32
33    fn decode(bytes: &[u8], limits: DecodeLimits) -> Result<Page<Self>, CanonicalDecodeError> {
34        Ok(match decode_attribute_page(bytes, limits)? {
35            AttributePage::Leaf(entries) => Page::Leaf(entries),
36            AttributePage::Internal(children) => Page::Internal(
37                children
38                    .into_iter()
39                    .map(|child| Child {
40                        first: child.first_name,
41                        page: child.page,
42                    })
43                    .collect(),
44            ),
45        })
46    }
47
48    fn decode_shape(
49        bytes: &[u8],
50        limits: DecodeLimits,
51    ) -> Result<DecodedPageShape, CanonicalDecodeError> {
52        attribute_page_decode_shape(bytes, limits)
53    }
54
55    fn encode(
56        page: &PageRef<'_, Self>,
57        maximum_items: u32,
58    ) -> Result<Vec<u8>, CanonicalDecodeError> {
59        match page {
60            PageRef::Leaf(entries) => encode_attribute_leaf_entries(entries, maximum_items),
61            PageRef::Internal(children) => encode_attribute_internal_children(
62                children.iter().map(|child| (&child.first, child.page)),
63                maximum_items,
64            ),
65        }
66    }
67
68    fn page_encoded_length(
69        page: &PageRef<'_, Self>,
70        maximum_items: u32,
71    ) -> Result<usize, CanonicalDecodeError> {
72        match page {
73            PageRef::Leaf(entries) => attribute_leaf_page_encoded_length(entries, maximum_items),
74            PageRef::Internal(children) => attribute_internal_page_encoded_length(
75                children.iter().map(|child| (&child.first, child.page)),
76                maximum_items,
77            ),
78        }
79    }
80
81    fn leaf_item_encoded_length(value: &Self::Value) -> Result<usize, CanonicalDecodeError> {
82        Ok(attribute_leaf_item_encoded_length(value))
83    }
84
85    fn internal_item_encoded_length(key: &Self::Key) -> Result<usize, CanonicalDecodeError> {
86        Ok(attribute_internal_item_encoded_length(key))
87    }
88
89    fn key_nested_bytes(key: &Self::Key) -> u64 {
90        u64::try_from(key.as_bytes().len()).unwrap_or(u64::MAX)
91    }
92
93    fn value_nested_bytes(value: &Self::Value) -> u64 {
94        Self::key_nested_bytes(&value.name)
95    }
96
97    fn try_clone_key(
98        key: &Self::Key,
99        maximum_bytes: u32,
100    ) -> Result<Self::Key, CanonicalDecodeError> {
101        let mut bytes = Vec::new();
102        bytes
103            .try_reserve_exact(key.as_bytes().len())
104            .map_err(|_| CanonicalDecodeError::AllocationFailed)?;
105        bytes.extend_from_slice(key.as_bytes());
106        AttributeName::new(key.class(), bytes, maximum_bytes)
107            .map_err(|error| CanonicalDecodeError::Invariant(error.to_string()))
108    }
109
110    fn try_clone_value(
111        value: &Self::Value,
112        maximum_bytes: u32,
113    ) -> Result<Self::Value, CanonicalDecodeError> {
114        Ok(AttributeEntry {
115            name: Self::try_clone_key(&value.name, maximum_bytes)?,
116            value_bytes: value.value_bytes,
117            value: value.value,
118        })
119    }
120}
121
122/// One ordered named-attribute mutation.
123#[derive(Clone, Debug, Eq, PartialEq)]
124pub enum AttributeMutation {
125    /// Inserts an attribute only when its name is absent.
126    Insert(AttributeEntry),
127    /// Removes an attribute with an optional complete-state precondition.
128    Remove {
129        /// Exact canonical attribute name.
130        name: AttributeName,
131        /// Required prior entry, or any entry with this name.
132        expected: Option<AttributeEntry>,
133    },
134    /// Replaces an attribute after checking its complete prior state.
135    Replace {
136        /// Required prior entry.
137        expected: AttributeEntry,
138        /// Complete replacement with the same name.
139        replacement: AttributeEntry,
140    },
141}
142
143impl AttributeMutation {
144    fn validate(&self) -> Result<(), AttributeSemanticError> {
145        match self {
146            Self::Insert(entry) => validate_entry(entry),
147            Self::Remove {
148                name,
149                expected: Some(expected),
150            } => {
151                validate_entry(expected)?;
152                if &expected.name != name {
153                    return Err(AttributeSemanticError::NameMismatch);
154                }
155                Ok(())
156            }
157            Self::Replace {
158                expected,
159                replacement,
160            } => {
161                validate_entry(expected)?;
162                validate_entry(replacement)?;
163                if expected.name != replacement.name {
164                    return Err(AttributeSemanticError::NameMismatch);
165                }
166                Ok(())
167            }
168            Self::Remove { expected: None, .. } => Ok(()),
169        }
170    }
171}
172
173fn validate_entry(entry: &AttributeEntry) -> Result<(), AttributeSemanticError> {
174    if entry.value.kind != ObjectKind::Blob {
175        return Err(AttributeSemanticError::InvalidEntry);
176    }
177    Ok(())
178}
179
180impl persistent_btree::Mutation<AttributeFormat> for AttributeMutation {
181    type Error = AttributeSemanticError;
182
183    fn key(&self) -> &AttributeName {
184        match self {
185            Self::Insert(entry) => &entry.name,
186            Self::Remove { name, .. } => name,
187            Self::Replace { replacement, .. } => &replacement.name,
188        }
189    }
190
191    fn changes_cardinality(&self) -> bool {
192        matches!(self, Self::Insert(_) | Self::Remove { .. })
193    }
194
195    fn apply_current(&self, current: &mut Option<AttributeEntry>) -> Result<(), Self::Error> {
196        match self {
197            Self::Insert(_) if current.is_some() => Err(AttributeSemanticError::AlreadyExists),
198            Self::Insert(entry) => {
199                *current = Some(entry.clone());
200                Ok(())
201            }
202            Self::Remove { .. } | Self::Replace { .. } if current.is_none() => {
203                Err(AttributeSemanticError::Missing)
204            }
205            Self::Remove { expected, .. } => {
206                if expected
207                    .as_ref()
208                    .is_some_and(|expected| current.as_ref() != Some(expected))
209                {
210                    return Err(AttributeSemanticError::StateConflict);
211                }
212                *current = None;
213                Ok(())
214            }
215            Self::Replace {
216                expected,
217                replacement,
218            } => {
219                if current.as_ref() != Some(expected) {
220                    return Err(AttributeSemanticError::StateConflict);
221                }
222                *current = Some(replacement.clone());
223                Ok(())
224            }
225        }
226    }
227}
228
229/// Candidate attribute root and exact sparse path-copy work.
230#[derive(Clone, Copy, Debug, Eq, PartialEq)]
231pub struct AttributeMutationReceipt {
232    /// Candidate immutable attribute-tree root.
233    pub root: ObjectId,
234    /// Exact work; authority publication is not included.
235    pub work: WorkCounters,
236}
237
238/// Attribute mutation failure retaining work and safe orphan writes.
239pub type AttributeMutationFailure = OperationFailure<AttributeMutationError>;
240
241/// Applies one bounded ordered attribute mutation batch.
242///
243/// # Errors
244///
245/// Rejects malformed entries and preconditions before backend access, then
246/// fails closed on semantic, routing, storage, allocation, or work failures.
247pub fn apply_attribute_mutations<S: crate::ImmediateObjectStore>(
248    store: &S,
249    root: ObjectId,
250    mutations: Vec<AttributeMutation>,
251    maximum_mutations: u32,
252    limits: DecodeLimits,
253    budget: WorkBudget,
254) -> Result<AttributeMutationReceipt, AttributeMutationFailure> {
255    for mutation in &mutations {
256        mutation
257            .validate()
258            .map_err(|error| OperationFailure::before_work(error.into()))?;
259    }
260    let receipt = persistent_btree::apply::<S, AttributeFormat, AttributeMutation>(
261        store,
262        root,
263        mutations,
264        maximum_mutations,
265        limits,
266        budget,
267    )
268    .map_err(|failure| OperationFailure::new(map_error(failure.error), *failure.work))?;
269    Ok(AttributeMutationReceipt {
270        root: receipt.root,
271        work: receipt.work,
272    })
273}
274
275/// Asynchronously applies the same iterative attribute path-copy machine.
276///
277/// # Errors
278///
279/// Returns the same failures as [`apply_attribute_mutations`], including
280/// cooperative cancellation at object-store boundaries.
281pub async fn apply_attribute_mutations_async<S: crate::AsyncObjectStore>(
282    store: &S,
283    root: ObjectId,
284    mutations: Vec<AttributeMutation>,
285    maximum_mutations: u32,
286    limits: DecodeLimits,
287    budget: WorkBudget,
288    cancellation: &crate::CancellationToken,
289) -> Result<AttributeMutationReceipt, AttributeMutationFailure> {
290    for mutation in &mutations {
291        mutation
292            .validate()
293            .map_err(|error| OperationFailure::before_work(error.into()))?;
294    }
295    let receipt = persistent_btree::apply_async::<S, AttributeFormat, AttributeMutation>(
296        store,
297        root,
298        mutations,
299        maximum_mutations,
300        limits,
301        budget,
302        cancellation,
303    )
304    .await
305    .map_err(|failure| OperationFailure::new(map_error(failure.error), *failure.work))?;
306    Ok(AttributeMutationReceipt {
307        root: receipt.root,
308        work: receipt.work,
309    })
310}
311
312fn map_error(error: persistent_btree::Error<AttributeSemanticError>) -> AttributeMutationError {
313    match error {
314        persistent_btree::Error::Empty => AttributeMutationError::Empty,
315        persistent_btree::Error::TooManyMutations => AttributeMutationError::TooManyMutations,
316        persistent_btree::Error::WrongRootKind => AttributeMutationError::WrongRootKind,
317        persistent_btree::Error::InvalidLimits => AttributeMutationError::InvalidLimits,
318        persistent_btree::Error::PageItemTooLarge => AttributeMutationError::PageItemTooLarge,
319        persistent_btree::Error::HeightExceeded => AttributeMutationError::HeightExceeded,
320        persistent_btree::Error::CycleOrAlias => AttributeMutationError::CycleOrAlias,
321        persistent_btree::Error::ChildBoundsMismatch => AttributeMutationError::ChildBoundsMismatch,
322        persistent_btree::Error::MutationContract => AttributeMutationError::MutationContract,
323        persistent_btree::Error::Allocation(AllocationError::Work(error)) => {
324            AttributeMutationError::Work(error)
325        }
326        persistent_btree::Error::AllocationFailed | persistent_btree::Error::Allocation(_) => {
327            AttributeMutationError::AllocationFailed
328        }
329        persistent_btree::Error::Semantic(error) => error.into(),
330        persistent_btree::Error::Storage(error) => AttributeMutationError::Storage(error),
331        persistent_btree::Error::Decode(error) => AttributeMutationError::Decode(error),
332        persistent_btree::Error::Work(error) => AttributeMutationError::Work(error),
333    }
334}
335
336/// Attribute mutation semantic failures.
337#[derive(Clone, Copy, Debug, Eq, Error, PartialEq)]
338pub enum AttributeSemanticError {
339    /// Entry references a non-blob payload.
340    #[error("attribute entry is invalid")]
341    InvalidEntry,
342    /// Mutation precondition and replacement names differ.
343    #[error("attribute mutation names differ")]
344    NameMismatch,
345    /// Insert expected an absent name.
346    #[error("attribute already exists")]
347    AlreadyExists,
348    /// Remove or replace expected an existing name.
349    #[error("attribute is missing")]
350    Missing,
351    /// Existing complete entry differs from the precondition.
352    #[error("attribute state precondition failed")]
353    StateConflict,
354}
355
356/// Persistent attribute path-copy failures.
357#[derive(Debug, Error)]
358pub enum AttributeMutationError {
359    /// Batch is empty.
360    #[error("attribute mutation batch is empty")]
361    Empty,
362    /// Batch exceeds its admitted operation count.
363    #[error("attribute mutation count exceeds its bound")]
364    TooManyMutations,
365    /// Root has the wrong object kind.
366    #[error("attribute root has the wrong object kind")]
367    WrongRootKind,
368    /// Page or traversal limits are invalid.
369    #[error("attribute mutation limits are invalid")]
370    InvalidLimits,
371    /// One item or required fanout cannot fit the page-byte bound.
372    #[error("attribute item exceeds its page-byte bound")]
373    PageItemTooLarge,
374    /// Traversal exceeds the page-height bound.
375    #[error("attribute tree exceeds its height bound")]
376    HeightExceeded,
377    /// Authenticated page graph cycles or aliases.
378    #[error("attribute tree contains a cycle or alias")]
379    CycleOrAlias,
380    /// Parent routing bounds mismatch child contents.
381    #[error("attribute child bounds mismatch")]
382    ChildBoundsMismatch,
383    /// Mutation implementation violated its cardinality contract.
384    #[error("attribute mutation contract was violated")]
385    MutationContract,
386    /// Fallible bounded scratch allocation failed.
387    #[error("attribute mutation scratch allocation failed")]
388    AllocationFailed,
389    /// Semantic mutation failed.
390    #[error(transparent)]
391    Semantic(#[from] AttributeSemanticError),
392    /// Immutable object storage failed.
393    #[error(transparent)]
394    Storage(#[from] ObjectStoreError),
395    /// Canonical page encoding or decoding failed.
396    #[error(transparent)]
397    Decode(#[from] CanonicalDecodeError),
398    /// Work accounting overflowed or exceeded its budget.
399    #[error(transparent)]
400    Work(#[from] WorkError),
401}
402
403#[cfg(all(test, feature = "memory"))]
404#[path = "tests/attribute_mutation.rs"]
405mod tests;
acyclic_fs/kernel/attribute_mutation.rs

acyclic_fs/kernel/
attribute_mutation.rs
