uqa_storage/mvcc/commit/
requirements.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Commit preconditions share definition revisions without manufacturing writes to those definitions.
8
9use std::sync::Arc;
10
11use sha2::{Digest, Sha256};
12use uqa_core::memory::BudgetedVec;
13use uqa_core::CancellationToken;
14
15use crate::mvcc::key::RecordKey;
16use crate::mvcc::{CommitSequence, VersionError, VersionResult};
17use crate::read_control::StorageReadControl;
18
19use super::{hash_bytes, PreparedRecordCommit};
20
21#[derive(Clone)]
22pub(in crate::mvcc) struct RecordRequirement {
23    pub(in crate::mvcc) key: RecordKey,
24    pub(in crate::mvcc) expected: Option<CommitSequence>,
25}
26
27impl PreparedRecordCommit {
28    pub(in crate::mvcc) fn with_requirements(
29        mut self,
30        requirements: &[RecordRequirement],
31        control: &StorageReadControl,
32    ) -> VersionResult<Self> {
33        if requirements.is_empty() {
34            return Ok(self);
35        }
36        let mut owned = BudgetedVec::new(control.memory());
37        owned.reserve(requirements.len())?;
38        let mut digest = Sha256::new();
39        digest.update(b"UQA prepared record requirements 1");
40        digest.update(self.fingerprint);
41        digest.update((requirements.len() as u64).to_be_bytes());
42        for requirement in requirements {
43            control.cancellation().check()?;
44            digest.update((requirement.key.bytes().len() as u64).to_be_bytes());
45            hash_bytes(&mut digest, requirement.key.bytes(), control)?;
46            digest.update([u8::from(requirement.expected.is_some())]);
47            digest.update(
48                requirement
49                    .expected
50                    .map_or(0, CommitSequence::as_u64)
51                    .to_be_bytes(),
52            );
53            owned.push(requirement.clone())?;
54        }
55        self.fingerprint = digest.finalize().into();
56        self.requirements = Some(Arc::new(owned));
57        Ok(self)
58    }
59
60    pub(crate) fn has_requirements(&self) -> bool {
61        self.requirements.is_some()
62    }
63
64    /// Keys validated without being written. Providers must charge any temporary driver bindings for these keys as well as replacement records.
65    pub fn required_keys(&self) -> impl Iterator<Item = &[u8]> {
66        self.requirements.iter().flat_map(|requirements| {
67            requirements
68                .iter()
69                .map(|requirement| requirement.key.bytes())
70        })
71    }
72
73    pub(crate) fn validate_requirements(
74        &self,
75        cancellation: &CancellationToken,
76        mut head_revision: impl FnMut(&[u8]) -> VersionResult<Option<CommitSequence>>,
77    ) -> VersionResult<()> {
78        cancellation.check()?;
79        if let Some(requirements) = &self.requirements {
80            for (index, requirement) in requirements.iter().enumerate() {
81                cancellation.check()?;
82                let actual = head_revision(requirement.key.bytes())?;
83                if requirement.expected != actual {
84                    return Err(VersionError::ReadConflict {
85                        dependency: index,
86                        expected: requirement.expected,
87                        actual,
88                    });
89                }
90            }
91        }
92        Ok(())
93    }
94}
uqa_storage/mvcc/commit/requirements.rs

uqa_storage/mvcc/commit/
requirements.rs
