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    /// Every write, in memory: for the inputs of one provider batch, which its operations already hold.
252    pub(crate) fn collect(
253        &self,
254        control: &StorageReadControl,
255    ) -> VersionResult<BudgetedVec<PreparedRecordWrite>> {
256        let mut writes = BudgetedVec::new(control.memory());
257        writes.reserve(self.len())?;
258        let mut cursor = self.writes();
259        while let Some(write) = cursor.next(control)? {
260            writes.push(write)?;
261        }
262        Ok(writes)
263    }
264
265    /// Whether some write replaces `key`.
266    pub(crate) fn contains_key(
267        &self,
268        key: &[u8],
269        control: &StorageReadControl,
270    ) -> VersionResult<bool> {
271        match &self.writes {
272            PreparedWrites::Resident(writes) => Ok(writes.iter().any(|write| write.key() == key)),
273            PreparedWrites::Spilled(run) => Ok(run.get(key, control)?.is_some()),
274        }
275    }
276
277    /// Whether some write is not canonical: such a batch has derived effects to resolve.
278    pub(crate) fn typed(&self) -> bool {
279        self.writes.typed()
280    }
281
282    /// Whether some write is of `kind`.
283    pub(crate) fn has_kind(&self, kind: RecordWriteKind) -> bool {
284        match &self.writes {
285            PreparedWrites::Resident(writes) => writes.iter().any(|write| write.kind == kind),
286            PreparedWrites::Spilled(run) => run.has_kind(kind),
287        }
288    }
289
290    /// The caller supplies exactly one final replacement for each identity.
291    pub(super) fn from_unique_owned(
292        writes: BudgetedVec<PreparedRecordWrite>,
293        control: &StorageReadControl,
294    ) -> VersionResult<Self> {
295        let typed = writes
296            .iter()
297            .any(|write| write.kind != RecordWriteKind::Canonical);
298        let mut digest = fingerprint_header(typed, writes.len() as u64);
299        for write in writes.iter() {
300            control.cancellation().check()?;
301            fingerprint_write(
302                &mut digest,
303                typed,
304                write.kind,
305                write.key(),
306                write.expected(),
307                write.value().map(|value| value.len() as u64),
308                control,
309            )?;
310            if let Some(value) = write.value() {
311                hash_bytes(&mut digest, value, control)?;
312            }
313        }
314        control.cancellation().check()?;
315        Ok(Self::with_fingerprint(
316            PreparedWrites::Resident(writes),
317            digest.finalize().into(),
318        ))
319    }
320
321    /// The writes of `run`, which holds exactly one final replacement for each identity, fingerprinted as the same writes in memory are.
322    pub(super) fn from_spilled_run(
323        run: super::overlay::run::SpilledRun,
324        control: &StorageReadControl,
325    ) -> VersionResult<Self> {
326        let run = Arc::new(run);
327        let typed = run.typed();
328        let mut digest = fingerprint_header(typed, run.len());
329        let mut cursor = run.cursor(std::ops::Bound::Unbounded);
330        while let Some(entry) = cursor.next(control)? {
331            fingerprint_write(
332                &mut digest,
333                typed,
334                entry.kind,
335                entry.key.bytes(),
336                entry.expected,
337                entry.value.map(|location| location.len),
338                control,
339            )?;
340            if let Some(location) = entry.value {
341                run.copy_value(
342                    location,
343                    &mut |chunk| {
344                        digest.update(chunk);
345                        Ok(())
346                    },
347                    control,
348                )?;
349            }
350        }
351        control.cancellation().check()?;
352        Ok(Self::with_fingerprint(
353            PreparedWrites::Spilled(run),
354            digest.finalize().into(),
355        ))
356    }
357
358    fn with_fingerprint(writes: PreparedWrites, fingerprint: CommitFingerprint) -> Self {
359        Self {
360            writes,
361            fingerprint,
362            graph: None,
363            vector: None,
364            populations: None,
365            notification: None,
366            resolved_at: None,
367            requirements: None,
368            reclamation_epoch: None,
369        }
370    }
371
372    pub(super) fn with_graph_effects(
373        mut self,
374        base: CommitSequence,
375        operations: &[OwnedGraphMutation],
376        control: &StorageReadControl,
377    ) -> VersionResult<Self> {
378        if operations.is_empty()
379            && !self.has_kind(RecordWriteKind::GraphCache)
380            && !self.has_kind(RecordWriteKind::GraphPreview)
381        {
382            return Ok(self);
383        }
384        let mut owned = BudgetedVec::new(control.memory());
385        owned.reserve(operations.len())?;
386        let mut digest = Sha256::new();
387        digest.update(b"UQA prepared graph effects 1");
388        digest.update(self.fingerprint);
389        digest.update(base.as_u64().to_be_bytes());
390        digest.update((operations.len() as u64).to_be_bytes());
391        let mut text = |bytes: &[u8]| -> VersionResult<()> {
392            digest.update((bytes.len() as u64).to_be_bytes());
393            hash_bytes(&mut digest, bytes, control)
394        };
395        for operation in operations {
396            control.cancellation().check()?;
397            match operation.borrowed() {
398                GraphMutation::InvalidateGraph(graph) => {
399                    text(b"graph")?;
400                    text(graph.as_bytes())?;
401                }
402                GraphMutation::InvalidatePath(index) => {
403                    text(b"invalidate-path")?;
404                    text(index.as_bytes())?;
405                }
406                GraphMutation::InvalidateEntity(kind, id) => {
407                    text(b"entity")?;
408                    text(kind.as_str().as_bytes())?;
409                    text(&id.to_be_bytes())?;
410                }
411                GraphMutation::PublishPath {
412                    index,
413                    graph,
414                    definition,
415                } => {
416                    text(b"path")?;
417                    text(index.as_bytes())?;
418                    text(graph.as_bytes())?;
419                    text(definition.as_bytes())?;
420                }
421            }
422            owned.push(operation.clone())?;
423        }
424        self.fingerprint = digest.finalize().into();
425        self.graph = Some(GraphEffects {
426            base,
427            operations: owned,
428        });
429        Ok(self)
430    }
431
432    pub(super) fn with_notification_effect(
433        mut self,
434        effect: Option<&Arc<super::notifications::NotificationEffect>>,
435    ) -> Self {
436        if let Some(effect) = effect {
437            self.fingerprint = effect.seal(self.fingerprint);
438            self.notification = Some(Arc::clone(effect));
439        }
440        self
441    }
442
443    pub(crate) fn resolved(mut self, original: &Self, sequence: CommitSequence) -> Self {
444        self.fingerprint = original.fingerprint;
445        self.resolved_at = Some(sequence);
446        self.requirements.clone_from(&original.requirements);
447        self.reclamation_epoch = original.reclamation_epoch;
448        self
449    }
450
451    pub(super) fn seal_vector_effects(
452        &mut self,
453        fingerprint: CommitFingerprint,
454        effects: super::vector::VectorEffects,
455    ) {
456        self.fingerprint = fingerprint;
457        self.vector = Some(effects);
458    }
459
460    pub(super) fn seal_population_effects(
461        &mut self,
462        fingerprint: CommitFingerprint,
463        effects: super::populations::PopulationEffects,
464    ) {
465        self.fingerprint = fingerprint;
466        self.populations = Some(effects);
467    }
468
469    pub(crate) fn retain_graph_effects(
470        self,
471        original: &Self,
472        control: &StorageReadControl,
473    ) -> VersionResult<Self> {
474        match &original.graph {
475            Some(effects) => self.with_graph_effects(effects.base, &effects.operations, control),
476            None => Ok(self),
477        }
478    }
479
480    /// 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.
481    pub fn validate_snapshot(&self, current: CommitSequence) -> VersionResult<()> {
482        if self.graph.is_some()
483            || self.vector.is_some()
484            || self.populations.is_some()
485            || self.notification.is_some()
486            || self.writes.typed()
487        {
488            return Err(VersionError::InvalidEncoding(
489                "unresolved storage commit effects",
490            ));
491        }
492        if let Some(expected) = self.resolved_at {
493            if expected != current {
494                return Err(VersionError::CommitSnapshotChanged {
495                    expected,
496                    actual: current,
497                });
498            }
499        }
500        Ok(())
501    }
502
503    pub fn fingerprint(&self) -> CommitFingerprint {
504        self.fingerprint
505    }
506
507    /// Check all preconditions under the provider's exclusive commit boundary. The callback reads current committed heads, not the caller's old snapshot.
508    pub fn validate(
509        &self,
510        control: &StorageReadControl,
511        mut head_revision: impl FnMut(&[u8]) -> VersionResult<Option<CommitSequence>>,
512    ) -> VersionResult<()> {
513        control.cancellation().check()?;
514        if self.graph.is_some()
515            || self.vector.is_some()
516            || self.populations.is_some()
517            || self.notification.is_some()
518            || self.writes.typed()
519        {
520            return Err(VersionError::InvalidEncoding(
521                "unresolved storage commit effects",
522            ));
523        }
524        let mut writes = self.writes();
525        let mut index = 0;
526        while let Some(write) = writes.next_metadata(control)? {
527            let actual = head_revision(write.key())?;
528            if write.expected() != actual {
529                return Err(VersionError::WriteConflict {
530                    mutation: index,
531                    expected: write.expected(),
532                    actual,
533                });
534            }
535            index += 1;
536        }
537        self.validate_requirements(control.cancellation(), head_revision)?;
538        control.cancellation().check()?;
539        Ok(())
540    }
541}
542
543fn fingerprint_header(typed: bool, len: u64) -> Sha256 {
544    let mut digest = Sha256::new();
545    if typed {
546        digest.update(b"UQA prepared record kinds 1");
547    } else {
548        digest.update(b"UQA prepared records 1");
549    }
550    digest.update(len.to_be_bytes());
551    digest
552}
553
554/// Hash one write up to its value, whose `value_len` bytes the caller hashes next.
555fn fingerprint_write(
556    digest: &mut Sha256,
557    typed: bool,
558    kind: RecordWriteKind,
559    key: &[u8],
560    expected: Option<CommitSequence>,
561    value_len: Option<u64>,
562    control: &StorageReadControl,
563) -> VersionResult<()> {
564    if typed {
565        digest.update([kind.code()]);
566    }
567    digest.update((key.len() as u64).to_be_bytes());
568    hash_bytes(digest, key, control)?;
569    digest.update(expected.map_or(0, CommitSequence::as_u64).to_be_bytes());
570    digest.update([u8::from(expected.is_some()), u8::from(value_len.is_some())]);
571    if let Some(len) = value_len {
572        digest.update(len.to_be_bytes());
573    }
574    Ok(())
575}
576
577fn hash_bytes(
578    digest: &mut Sha256,
579    bytes: &[u8],
580    control: &StorageReadControl,
581) -> VersionResult<()> {
582    for chunk in bytes.chunks(65536) {
583        control.cancellation().check()?;
584        digest.update(chunk);
585    }
586    Ok(())
587}
uqa_storage/mvcc/commit.rs

uqa_storage/mvcc/
commit.rs
