uqa_storage/mvcc/
commit.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Private evaluated changes and provider-independent revision validation.
8
9mod builder;
10mod cursor;
11mod lookup;
12mod metadata;
13mod reclamation;
14mod requirements;
15mod writes;
16pub(super) use builder::PreparedWritesBuilder;
17pub use cursor::PreparedWriteCursor;
18pub(super) use lookup::PreparedLookup;
19pub use metadata::PreparedWriteMetadata;
20pub(super) use requirements::RecordRequirement;
21use writes::PreparedWrites;
22
23use std::collections::BTreeMap;
24use std::sync::Arc;
25
26use sha2::{Digest, Sha256};
27use uqa_core::memory::{BudgetedVec, MemoryError};
28
29use crate::read_control::StorageReadControl;
30
31use super::graph::{GraphEffects, OwnedGraphMutation};
32use super::key::RecordKey;
33use super::{
34    CommitFingerprint, CommitSequence, GraphMutation, RecordWrite, VersionError, VersionResult,
35};
36
37#[derive(Clone)]
38pub struct PreparedRecordWrite {
39    key: RecordKey,
40    expected: Option<CommitSequence>,
41    value: Option<Arc<BudgetedVec<u8>>>,
42    kind: RecordWriteKind,
43}
44
45#[derive(Clone, Copy, PartialEq, Eq)]
46pub(crate) enum RecordWriteKind {
47    Canonical,
48    GraphCache,
49    GraphPreview,
50    Occurrence,
51    OccurrenceCache,
52    IVFPreview,
53    HNSWPreview,
54    Marker,
55    StatisticsMaintenance,
56    IdempotentDelete,
57    // A value is a complete typed replacement; absence is a raw invalidation that must not leave a stale population.
58    DiskANNOrigin,
59    DiskANNPopulationPreview,
60}
61
62impl RecordWriteKind {
63    /// The kind's code, which the commit fingerprint and spill files record.
64    pub(crate) const fn code(self) -> u8 {
65        match self {
66            Self::Canonical => 0,
67            Self::GraphCache => 1,
68            Self::GraphPreview => 2,
69            Self::Occurrence => 3,
70            Self::OccurrenceCache => 4,
71            Self::IVFPreview => 5,
72            Self::HNSWPreview => 6,
73            Self::Marker => 7,
74            Self::StatisticsMaintenance => 8,
75            Self::IdempotentDelete => 9,
76            Self::DiskANNOrigin => 10,
77            Self::DiskANNPopulationPreview => 11,
78        }
79    }
80
81    /// The kind whose code is `code`.
82    pub(crate) fn from_code(code: u8) -> VersionResult<Self> {
83        Ok(match code {
84            0 => Self::Canonical,
85            1 => Self::GraphCache,
86            2 => Self::GraphPreview,
87            3 => Self::Occurrence,
88            4 => Self::OccurrenceCache,
89            5 => Self::IVFPreview,
90            6 => Self::HNSWPreview,
91            7 => Self::Marker,
92            8 => Self::StatisticsMaintenance,
93            9 => Self::IdempotentDelete,
94            10 => Self::DiskANNOrigin,
95            11 => Self::DiskANNPopulationPreview,
96            _ => return Err(VersionError::InvalidEncoding("unknown record write kind")),
97        })
98    }
99}
100
101impl PreparedRecordWrite {
102    pub(super) fn from_shared(
103        key: RecordKey,
104        expected: Option<CommitSequence>,
105        value: Option<Arc<BudgetedVec<u8>>>,
106    ) -> Self {
107        Self {
108            key,
109            expected,
110            value,
111            kind: RecordWriteKind::Canonical,
112        }
113    }
114
115    pub(super) fn copy_bytes(
116        key: &[u8],
117        expected: Option<CommitSequence>,
118        value: Option<&[u8]>,
119        control: &StorageReadControl,
120    ) -> VersionResult<Self> {
121        control.cancellation().check()?;
122        let key = RecordKey::new(key, control.memory())?;
123        let value = value
124            .map(|value| {
125                let mut owned = BudgetedVec::new(control.memory());
126                owned.extend_from_slice(value)?;
127                Ok::<_, VersionError>(Arc::new(owned))
128            })
129            .transpose()?;
130        Ok(Self::from_shared(key, expected, value))
131    }
132
133    pub fn key(&self) -> &[u8] {
134        self.key.bytes()
135    }
136
137    pub fn expected(&self) -> Option<CommitSequence> {
138        self.expected
139    }
140
141    pub fn value(&self) -> Option<&[u8]> {
142        self.value.as_ref().map(|value| &***value)
143    }
144
145    pub(crate) fn shared_value(&self) -> Option<Arc<BudgetedVec<u8>>> {
146        self.value.clone()
147    }
148
149    pub(super) fn shared_key(&self) -> RecordKey {
150        self.key.clone()
151    }
152
153    pub(super) fn rebase(mut self, expected: Option<CommitSequence>) -> Self {
154        self.expected = expected;
155        self
156    }
157
158    pub(crate) fn kind(&self) -> RecordWriteKind {
159        self.kind
160    }
161    pub(crate) fn with_kind(mut self, kind: RecordWriteKind) -> Self {
162        self.kind = kind;
163        self
164    }
165
166    pub(super) fn with_value(mut self, value: BudgetedVec<u8>) -> Self {
167        self.value = Some(Arc::new(value));
168        self
169    }
170}
171
172/// Immutable prepared replacements. Construct before opening a physical writer. The writes of a transaction larger than its allowance are in a spilled run ordered by key; the others are in memory in the order they were supplied.
173pub struct PreparedRecordCommit {
174    writes: PreparedWrites,
175    fingerprint: CommitFingerprint,
176    pub(super) graph: Option<GraphEffects>,
177    pub(super) vector: Option<super::vector::VectorEffects>,
178    pub(super) populations: Option<super::populations::PopulationEffects>,
179    pub(super) notification: Option<Arc<super::notifications::NotificationEffect>>,
180    pub(super) resolved_at: Option<CommitSequence>,
181    requirements: Option<Arc<BudgetedVec<RecordRequirement>>>,
182    reclamation_epoch: Option<u64>,
183}
184
185impl PreparedRecordCommit {
186    pub fn new(writes: &[RecordWrite<'_>], control: &StorageReadControl) -> VersionResult<Self> {
187        control.cancellation().check()?;
188        let slots = writes
189            .len()
190            .checked_mul(std::mem::size_of::<(&[u8], usize)>())
191            .ok_or(MemoryError::SizeOverflow)?;
192        // Charge logical tree entries; allocator node bookkeeping is separate.
193        let seen_memory = control.memory().reserve(slots)?;
194        let mut seen = BTreeMap::new();
195        for (index, write) in writes.iter().enumerate() {
196            control.cancellation().check()?;
197            if let Some(first) = seen.insert(write.key, index) {
198                return Err(VersionError::DuplicateRecord {
199                    first,
200                    second: index,
201                });
202            }
203        }
204        drop(seen);
205        drop(seen_memory);
206
207        let mut prepared = BudgetedVec::new(control.memory());
208        prepared.reserve(writes.len())?;
209        for write in writes {
210            prepared.push(PreparedRecordWrite::copy_bytes(
211                write.key,
212                write.expected,
213                write.value,
214                control,
215            )?)?;
216        }
217        control.cancellation().check()?;
218        Self::from_unique_owned(prepared, control)
219    }
220
221    /// The number of writes.
222    pub fn len(&self) -> usize {
223        self.writes.len()
224    }
225
226    pub fn is_empty(&self) -> bool {
227        self.writes.len() == 0
228    }
229
230    /// Visit the writes in order.
231    pub fn writes(&self) -> PreparedWriteCursor<'_> {
232        self.writes.cursor()
233    }
234
235    /// Visit the writes of a spilled batch in key order from `start`; `None` for a batch in memory, whose writes `resident` returns in the order they were supplied.
236    pub fn spilled_writes_from(&self, start: &[u8]) -> Option<PreparedWriteCursor<'_>> {
237        match &self.writes {
238            PreparedWrites::Resident(_) => None,
239            PreparedWrites::Spilled(run) => Some(PreparedWriteCursor::spilled_from(run, start)),
240        }
241    }
242
243    /// The writes, when they are in memory; a transaction larger than its allowance has spilled them.
244    pub fn resident(&self) -> Option<&[PreparedRecordWrite]> {
245        match &self.writes {
246            PreparedWrites::Resident(writes) => Some(writes),
247            PreparedWrites::Spilled(_) => None,
248        }
249    }
250
251    /// Whether some write replaces `key`.
252    pub(crate) fn contains_key(
253        &self,
254        key: &[u8],
255        control: &StorageReadControl,
256    ) -> VersionResult<bool> {
257        match &self.writes {
258            PreparedWrites::Resident(writes) => Ok(writes.iter().any(|write| write.key() == key)),
259            PreparedWrites::Spilled(run) => Ok(run.get(key, control)?.is_some()),
260        }
261    }
262
263    /// Whether some write is not canonical: such a batch has derived effects to resolve.
264    pub(crate) fn typed(&self) -> bool {
265        self.writes.typed()
266    }
267
268    /// Whether some write is of `kind`.
269    pub(crate) fn has_kind(&self, kind: RecordWriteKind) -> bool {
270        match &self.writes {
271            PreparedWrites::Resident(writes) => writes.iter().any(|write| write.kind == kind),
272            PreparedWrites::Spilled(run) => run.has_kind(kind),
273        }
274    }
275
276    /// The caller supplies exactly one final replacement for each identity.
277    pub(super) fn from_unique_owned(
278        writes: BudgetedVec<PreparedRecordWrite>,
279        control: &StorageReadControl,
280    ) -> VersionResult<Self> {
281        let typed = writes
282            .iter()
283            .any(|write| write.kind != RecordWriteKind::Canonical);
284        let mut digest = fingerprint_header(typed, writes.len() as u64);
285        for write in writes.iter() {
286            control.cancellation().check()?;
287            fingerprint_write(
288                &mut digest,
289                typed,
290                write.kind,
291                write.key(),
292                write.expected(),
293                write.value().map(|value| value.len() as u64),
294                control,
295            )?;
296            if let Some(value) = write.value() {
297                hash_bytes(&mut digest, value, control)?;
298            }
299        }
300        control.cancellation().check()?;
301        Ok(Self::with_fingerprint(
302            PreparedWrites::Resident(writes),
303            digest.finalize().into(),
304        ))
305    }
306
307    /// The writes of `run`, which holds exactly one final replacement for each identity, fingerprinted as the same writes in memory are.
308    pub(super) fn from_spilled_run(
309        run: super::overlay::run::SpilledRun,
310        control: &StorageReadControl,
311    ) -> VersionResult<Self> {
312        let run = Arc::new(run);
313        let typed = run.typed();
314        let mut digest = fingerprint_header(typed, run.len());
315        let mut cursor = run.cursor(std::ops::Bound::Unbounded);
316        while let Some(entry) = cursor.next(control)? {
317            fingerprint_write(
318                &mut digest,
319                typed,
320                entry.kind,
321                entry.key.bytes(),
322                entry.expected,
323                entry.value.map(|location| location.len),
324                control,
325            )?;
326            if let Some(location) = entry.value {
327                run.copy_value(
328                    location,
329                    &mut |chunk| {
330                        digest.update(chunk);
331                        Ok(())
332                    },
333                    control,
334                )?;
335            }
336        }
337        control.cancellation().check()?;
338        Ok(Self::with_fingerprint(
339            PreparedWrites::Spilled(run),
340            digest.finalize().into(),
341        ))
342    }
343
344    fn with_fingerprint(writes: PreparedWrites, fingerprint: CommitFingerprint) -> Self {
345        Self {
346            writes,
347            fingerprint,
348            graph: None,
349            vector: None,
350            populations: None,
351            notification: None,
352            resolved_at: None,
353            requirements: None,
354            reclamation_epoch: None,
355        }
356    }
357
358    pub(super) fn with_graph_effects(
359        mut self,
360        base: CommitSequence,
361        operations: &[OwnedGraphMutation],
362        control: &StorageReadControl,
363    ) -> VersionResult<Self> {
364        if operations.is_empty()
365            && !self.has_kind(RecordWriteKind::GraphCache)
366            && !self.has_kind(RecordWriteKind::GraphPreview)
367        {
368            return Ok(self);
369        }
370        let mut owned = BudgetedVec::new(control.memory());
371        owned.reserve(operations.len())?;
372        let mut digest = Sha256::new();
373        digest.update(b"UQA prepared graph effects 1");
374        digest.update(self.fingerprint);
375        digest.update(base.as_u64().to_be_bytes());
376        digest.update((operations.len() as u64).to_be_bytes());
377        let mut text = |bytes: &[u8]| -> VersionResult<()> {
378            digest.update((bytes.len() as u64).to_be_bytes());
379            hash_bytes(&mut digest, bytes, control)
380        };
381        for operation in operations {
382            control.cancellation().check()?;
383            match operation.borrowed() {
384                GraphMutation::InvalidateGraph(graph) => {
385                    text(b"graph")?;
386                    text(graph.as_bytes())?;
387                }
388                GraphMutation::InvalidatePath(index) => {
389                    text(b"invalidate-path")?;
390                    text(index.as_bytes())?;
391                }
392                GraphMutation::InvalidateEntity(kind, id) => {
393                    text(b"entity")?;
394                    text(kind.as_str().as_bytes())?;
395                    text(&id.to_be_bytes())?;
396                }
397                GraphMutation::PublishPath {
398                    index,
399                    graph,
400                    definition,
401                } => {
402                    text(b"path")?;
403                    text(index.as_bytes())?;
404                    text(graph.as_bytes())?;
405                    text(definition.as_bytes())?;
406                }
407            }
408            owned.push(operation.clone())?;
409        }
410        self.fingerprint = digest.finalize().into();
411        self.graph = Some(GraphEffects {
412            base,
413            operations: owned,
414        });
415        Ok(self)
416    }
417
418    pub(super) fn with_notification_effect(
419        mut self,
420        effect: Option<&Arc<super::notifications::NotificationEffect>>,
421    ) -> Self {
422        if let Some(effect) = effect {
423            self.fingerprint = effect.seal(self.fingerprint);
424            self.notification = Some(Arc::clone(effect));
425        }
426        self
427    }
428
429    pub(crate) fn resolved(mut self, original: &Self, sequence: CommitSequence) -> Self {
430        self.fingerprint = original.fingerprint;
431        self.resolved_at = Some(sequence);
432        self.requirements.clone_from(&original.requirements);
433        self.reclamation_epoch = original.reclamation_epoch;
434        self
435    }
436
437    pub(super) fn seal_vector_effects(
438        &mut self,
439        fingerprint: CommitFingerprint,
440        effects: super::vector::VectorEffects,
441    ) {
442        self.fingerprint = fingerprint;
443        self.vector = Some(effects);
444    }
445
446    pub(super) fn seal_population_effects(
447        &mut self,
448        fingerprint: CommitFingerprint,
449        effects: super::populations::PopulationEffects,
450    ) {
451        self.fingerprint = fingerprint;
452        self.populations = Some(effects);
453    }
454
455    pub(crate) fn retain_graph_effects(
456        self,
457        original: &Self,
458        control: &StorageReadControl,
459    ) -> VersionResult<Self> {
460        match &original.graph {
461            Some(effects) => self.with_graph_effects(effects.base, &effects.operations, control),
462            None => Ok(self),
463        }
464    }
465
466    /// Check the snapshot used to discover derived dependencies under exclusive admission, after resolving any existing receipt and before validating record heads. A mismatch proves this pending attempt has not published and permits pure effect preparation to restart.
467    pub fn validate_snapshot(&self, current: CommitSequence) -> VersionResult<()> {
468        if self.graph.is_some()
469            || self.vector.is_some()
470            || self.populations.is_some()
471            || self.notification.is_some()
472            || self.writes.typed()
473        {
474            return Err(VersionError::InvalidEncoding(
475                "unresolved storage commit effects",
476            ));
477        }
478        if let Some(expected) = self.resolved_at {
479            if expected != current {
480                return Err(VersionError::CommitSnapshotChanged {
481                    expected,
482                    actual: current,
483                });
484            }
485        }
486        Ok(())
487    }
488
489    pub fn fingerprint(&self) -> CommitFingerprint {
490        self.fingerprint
491    }
492
493    /// Check all preconditions under the provider's exclusive commit boundary. The callback reads current committed heads, not the caller's old snapshot.
494    pub fn validate(
495        &self,
496        control: &StorageReadControl,
497        mut head_revision: impl FnMut(&[u8]) -> VersionResult<Option<CommitSequence>>,
498    ) -> VersionResult<()> {
499        control.cancellation().check()?;
500        if self.graph.is_some()
501            || self.vector.is_some()
502            || self.populations.is_some()
503            || self.notification.is_some()
504            || self.writes.typed()
505        {
506            return Err(VersionError::InvalidEncoding(
507                "unresolved storage commit effects",
508            ));
509        }
510        let mut writes = self.writes();
511        let mut index = 0;
512        while let Some(write) = writes.next_metadata(control)? {
513            let actual = head_revision(write.key())?;
514            if write.expected() != actual {
515                return Err(VersionError::WriteConflict {
516                    mutation: index,
517                    expected: write.expected(),
518                    actual,
519                });
520            }
521            index += 1;
522        }
523        self.validate_requirements(control.cancellation(), head_revision)?;
524        control.cancellation().check()?;
525        Ok(())
526    }
527}
528
529fn fingerprint_header(typed: bool, len: u64) -> Sha256 {
530    let mut digest = Sha256::new();
531    if typed {
532        digest.update(b"UQA prepared record kinds 1");
533    } else {
534        digest.update(b"UQA prepared records 1");
535    }
536    digest.update(len.to_be_bytes());
537    digest
538}
539
540/// Hash one write up to its value, whose `value_len` bytes the caller hashes next.
541fn fingerprint_write(
542    digest: &mut Sha256,
543    typed: bool,
544    kind: RecordWriteKind,
545    key: &[u8],
546    expected: Option<CommitSequence>,
547    value_len: Option<u64>,
548    control: &StorageReadControl,
549) -> VersionResult<()> {
550    if typed {
551        digest.update([kind.code()]);
552    }
553    digest.update((key.len() as u64).to_be_bytes());
554    hash_bytes(digest, key, control)?;
555    digest.update(expected.map_or(0, CommitSequence::as_u64).to_be_bytes());
556    digest.update([u8::from(expected.is_some()), u8::from(value_len.is_some())]);
557    if let Some(len) = value_len {
558        digest.update(len.to_be_bytes());
559    }
560    Ok(())
561}
562
563fn hash_bytes(
564    digest: &mut Sha256,
565    bytes: &[u8],
566    control: &StorageReadControl,
567) -> VersionResult<()> {
568    for chunk in bytes.chunks(65536) {
569        control.cancellation().check()?;
570        digest.update(chunk);
571    }
572    Ok(())
573}
uqa_storage/mvcc/commit.rs

uqa_storage/mvcc/
commit.rs
