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

uqa_storage/mvcc/
commit.rs
