uqa_storage/mvcc/
serializable.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Serializable read/write dependencies, commit ordering and retained conflict summaries.
8
9mod checkpoint;
10mod conflicts;
11mod coordinator;
12mod identity;
13mod liveness;
14mod observations;
15mod predicates;
16mod publication;
17#[cfg(test)]
18mod tests;
19
20use uqa_core::memory::{BudgetedVec, MemoryBudget};
21
22use super::{DatabaseId, VersionError, VersionResult};
23use crate::read_control::StorageReadControl;
24
25pub use checkpoint::records::{SerializableCheckpointKey, SerializableCheckpointRecord};
26pub(crate) use coordinator::recover;
27pub use coordinator::{
28    admit_serializable, LocalSerializableState, SerializableCoordinator, SerializableLeases,
29    SerializableOperation,
30};
31pub use identity::SerializableTransactionId;
32pub use liveness::{LocalSerializableLeases, SerializableParticipant};
33pub use observations::SerializableWriteMark;
34pub(in crate::mvcc) use predicates::OwnedPredicate;
35pub use predicates::{SerializableKeySpace, SerializablePredicate};
36pub use publication::SerializablePublication;
37
38/// Whether a read-only snapshot can execute without predicate observations. An unsafe snapshot must be discarded and captured again; waiting cannot repair it.
39#[derive(Debug, Clone, Copy, PartialEq, Eq)]
40pub enum SafeSnapshot {
41    Pending,
42    Safe,
43    Unsafe,
44}
45
46/// Logical participant state, independent of a physical write receipt. A retained provider lease keeps terminal state available for completion retries even when the transaction wrote no records. Conflict history can be reclaimed before that lease is released.
47#[derive(Debug, Clone, Copy, PartialEq, Eq)]
48pub enum SerializableStatus {
49    Active,
50    Prepared,
51    Committed,
52    Aborted,
53    Doomed,
54}
55
56#[derive(Clone, Copy, PartialEq, Eq)]
57enum ParticipantOwner {
58    // No provider lease remains bound; active manual actors are never inferred dead.
59    Manual,
60    Leased,
61}
62
63#[derive(Clone, Copy)]
64struct Transaction {
65    id: u64,
66    snapshot: u64,
67    read_only: bool,
68    prepared: Option<u64>,
69    committed: Option<u64>,
70    aborted: bool,
71    doomed: bool,
72    owner: ParticipantOwner,
73    summarized_out: Option<u64>,
74    writes: u64,
75    publication: Option<publication::PreparedPublication>,
76}
77
78impl Transaction {
79    fn live(self) -> bool {
80        !self.aborted && self.committed.is_none()
81    }
82
83    fn relevant(self) -> bool {
84        !self.aborted && !self.doomed
85    }
86
87    fn retains_history(self, oldest: u64) -> bool {
88        !self.aborted && self.committed.is_none_or(|end| end > oldest)
89    }
90}
91
92/// Both orientations of each edge are sorted, so incident-edge traversal does not scan unrelated dependencies. IDs are non-reused transaction allocations, not vector positions.
93#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
94struct Edge(u64, u64);
95
96/// Common SSI algorithm for one database incarnation. Participants, both edge indexes, logical observations and owned predicate keys share a caller-selected retention budget. The owner serializes calls with snapshot admission, predicate registration and durable commit admission; this type does not acquire provider or SQL locks.
97///
98/// Only SERIALIZABLE transactions participate. Register a read/write edge when the reader cannot see the writer's logical change, including absent keys and predicate ranges. Physical index-sharing conflicts are not such edges. Check existing writer intents when registering reads and existing reads when registering writes; observing only committed changes loses cycles.
99///
100/// A prepared transaction has passed the final serialization check and cannot be selected as a victim. Keep it prepared across an uncertain physical outcome, and call `commit` or `rollback` only after authoritative completion. Savepoint undo must not discard the transaction's read dependencies.
101pub struct SerializableGraph {
102    database: DatabaseId,
103    coordinator: [u8; 16],
104    last_allocation: u64,
105    clock: u64,
106    pending_finishes: u64,
107    transactions: BudgetedVec<Transaction>,
108    outgoing: BudgetedVec<Edge>,
109    incoming: BudgetedVec<Edge>,
110    predicates: observations::Observations,
111    checkpoint_changed: bool,
112    checkpoint_records: BudgetedVec<SerializableCheckpointKey>,
113}
114
115impl SerializableGraph {
116    /// Create the graph for a nonzero provider-coordinated incarnation. A provider must retain this incarnation and its admission state for every overlapping participant, including across process handoff.
117    pub fn new(
118        database: DatabaseId,
119        coordinator: [u8; 16],
120        memory: &MemoryBudget,
121    ) -> VersionResult<Self> {
122        identity::validate_coordinator(coordinator)?;
123        Ok(Self {
124            database,
125            coordinator,
126            last_allocation: 0,
127            clock: 0,
128            pending_finishes: 0,
129            transactions: BudgetedVec::new(memory),
130            outgoing: BudgetedVec::new(memory),
131            incoming: BudgetedVec::new(memory),
132            predicates: observations::Observations::new(memory),
133            checkpoint_changed: true,
134            checkpoint_records: BudgetedVec::new(memory),
135        })
136    }
137
138    /// Allocate and admit a logical participant at the same boundary as its fixed data snapshot, without allocating a physical transaction or writing user data. The owner coordinates and retains admission across every provider process. Allocations survive graph reclamation and are never reused within the coordinator incarnation.
139    pub fn admit(
140        &mut self,
141        read_only: bool,
142        control: &StorageReadControl,
143    ) -> VersionResult<SerializableTransactionId> {
144        control.check()?;
145        let allocation = self
146            .last_allocation
147            .checked_add(1)
148            .ok_or(VersionError::SequenceExhausted)?;
149        let transaction =
150            SerializableTransactionId::new(self.database, self.coordinator, allocation)?;
151        self.begin(transaction, read_only, control)?;
152        Ok(transaction)
153    }
154
155    fn begin(
156        &mut self,
157        transaction: SerializableTransactionId,
158        read_only: bool,
159        control: &StorageReadControl,
160    ) -> VersionResult<()> {
161        control.check()?;
162        self.validate_identity(transaction)?;
163        let position = self
164            .transactions
165            .binary_search_by_key(&transaction.allocation(), |entry| entry.id)
166            .err()
167            .ok_or(VersionError::InvalidEncoding(
168                "duplicate serializable transaction",
169            ))?;
170        self.reserve_order(false)?;
171        self.transactions.push(Transaction {
172            id: transaction.allocation(),
173            snapshot: self.clock + 1,
174            read_only,
175            prepared: None,
176            committed: None,
177            aborted: false,
178            doomed: false,
179            owner: ParticipantOwner::Manual,
180            summarized_out: None,
181            writes: 0,
182            publication: None,
183        })?;
184        self.checkpoint_changed = true;
185        self.transactions[position..].rotate_right(1);
186        self.clock += 1;
187        self.last_allocation = self.last_allocation.max(transaction.allocation());
188        Ok(())
189    }
190
191    /// Read the original logical outcome while its participant lease is retained. Terminal outcomes take precedence over earlier victim selection; this never converts an uncertain physical publication into a completed transaction.
192    pub fn status(
193        &self,
194        transaction: SerializableTransactionId,
195    ) -> VersionResult<SerializableStatus> {
196        let entry = self.transactions[self.position(transaction)?];
197        Ok(if entry.committed.is_some() {
198            SerializableStatus::Committed
199        } else if entry.aborted {
200            SerializableStatus::Aborted
201        } else if entry.doomed {
202            SerializableStatus::Doomed
203        } else if entry.prepared.is_some() {
204            SerializableStatus::Prepared
205        } else {
206            SerializableStatus::Active
207        })
208    }
209
210    /// Fail a doomed transaction before another statement or commit. Prepared state is sealed, including while its durable outcome is unknown.
211    pub fn check_active(&self, transaction: SerializableTransactionId) -> VersionResult<()> {
212        let entry = self.transactions[self.position(transaction)?];
213        if entry.doomed {
214            return Err(VersionError::SerializationConflict { transaction });
215        }
216        if !entry.live() {
217            return Err(VersionError::TransactionFinished);
218        }
219        if entry.prepared.is_some() {
220            return Err(VersionError::TransactionSealed);
221        }
222        Ok(())
223    }
224
225    /// Register a dependency directed from reader to writer. The observing transaction must be one endpoint and still active. A peer may be marked doomed; its next active check or preparation then reports a serialization failure. A rejected local observation publishes no edge and does not doom its transaction: statement/savepoint recovery can undo that operation while retaining earlier reads. Nonoverlapping and duplicate edges are ignored.
226    pub fn observe_rw(
227        &mut self,
228        observer: SerializableTransactionId,
229        reader: SerializableTransactionId,
230        writer: SerializableTransactionId,
231        control: &StorageReadControl,
232    ) -> VersionResult<()> {
233        let Some(action) = self.plan_dependency(observer, reader, writer, control)? else {
234            return Ok(());
235        };
236        // Reserve both orientations before publishing either, so exhaustion cannot hide half an edge.
237        self.reserve_dependencies(usize::from(action.victim.is_none()))?;
238        self.publish_dependency(action)
239    }
240
241    /// Check incoming dangerous structures before durable publication, preferring an unprepared pivot as the victim. Preparation is idempotent and reserves ordering space for authoritative completion even if other transactions continue meanwhile.
242    pub fn prepare_commit(
243        &mut self,
244        transaction: SerializableTransactionId,
245        control: &StorageReadControl,
246    ) -> VersionResult<()> {
247        control.check()?;
248        let position = self.position(transaction)?;
249        let entry = self.transactions[position];
250        if entry.prepared.is_some() && entry.live() && !entry.doomed {
251            return Ok(());
252        }
253        self.check_active(transaction)?;
254        self.reserve_order(true)?;
255        let incoming = edge_range(&self.incoming, entry.id);
256        // Inspect every candidate, including cancellation, before dooming any peer.
257        for index in incoming.clone() {
258            control.check()?;
259            let pivot = self.node(self.incoming[index].1);
260            if self.dangerous_pivot(entry.id, pivot, Some(control))? && pivot.prepared.is_some() {
261                self.transactions[position].doomed = true;
262                self.checkpoint_changed = true;
263                return Err(VersionError::SerializationConflict { transaction });
264            }
265        }
266        self.checkpoint_changed = true;
267        for index in incoming {
268            let pivot = self.node(self.incoming[index].1);
269            if self.dangerous_pivot(entry.id, pivot, None)? {
270                let position = self.allocation_position(pivot.id)?;
271                self.transactions[position].doomed = true;
272            }
273        }
274        self.clock += 1;
275        self.pending_finishes += 1;
276        self.transactions[position].prepared = Some(self.clock);
277        Ok(())
278    }
279
280    /// Publish a confirmed committed outcome without allocation or cancellation. Never call this on a merely attempted or uncertain provider commit. Repeating the same confirmed outcome is harmless.
281    pub fn commit(&mut self, transaction: SerializableTransactionId) -> VersionResult<()> {
282        let position = self.position(transaction)?;
283        let entry = self.transactions[position];
284        if entry.committed.is_some() {
285            return Ok(());
286        }
287        if entry.aborted || entry.doomed || entry.prepared.is_none() {
288            return Err(VersionError::InvalidEncoding(
289                "serializable completion requires a prepared transaction",
290            ));
291        }
292        if entry
293            .publication
294            .is_some_and(|publication| !publication.committed())
295        {
296            return Err(VersionError::InvalidEncoding(
297                "durable serializable publication requires a confirmed receipt",
298            ));
299        }
300        // Each prepared transaction owns a completion slot; admission cannot consume it.
301        self.checkpoint_changed = true;
302        self.clock += 1;
303        self.pending_finishes -= 1;
304        self.transactions[position].committed = Some(self.clock);
305        Ok(())
306    }
307
308    /// Remove a confirmed aborted participant from dependency decisions. This cleanup does not allocate or observe statement cancellation. Uncertain prepared outcomes must remain retained.
309    pub fn rollback(&mut self, transaction: SerializableTransactionId) -> VersionResult<()> {
310        let position = self.position(transaction)?;
311        let entry = &mut self.transactions[position];
312        if entry.committed.is_some() {
313            return Err(VersionError::TransactionFinished);
314        }
315        if entry
316            .publication
317            .is_some_and(|publication| !publication.aborted())
318        {
319            return Err(VersionError::InvalidEncoding(
320                "durable serializable publication requires a confirmed abort",
321            ));
322        }
323        if !entry.aborted {
324            if entry.prepared.is_some() {
325                self.pending_finishes -= 1;
326            }
327            entry.aborted = true;
328            self.checkpoint_changed = true;
329        }
330        self.checkpoint_changed |= self.predicates.forget(transaction.allocation());
331        Ok(())
332    }
333
334    /// A deferrable reader waits only for read/write transactions overlapping its snapshot admission. A committed overlap with a conflict to an earlier prepared transaction makes this snapshot unsafe; the owner must capture a new snapshot before executing it.
335    pub fn safe_snapshot(
336        &self,
337        transaction: SerializableTransactionId,
338        control: &StorageReadControl,
339    ) -> VersionResult<SafeSnapshot> {
340        control.check()?;
341        self.check_active(transaction)?;
342        let reader = self.transactions[self.position(transaction)?];
343        if !reader.read_only {
344            return Err(VersionError::InvalidEncoding(
345                "safe-snapshot check requires a read-only transaction",
346            ));
347        }
348        let mut pending = false;
349        for entry in &*self.transactions {
350            control.check()?;
351            if entry.read_only
352                || !entry.relevant()
353                || entry.snapshot > reader.snapshot
354                || entry.committed.is_some_and(|end| end <= reader.snapshot)
355            {
356                continue;
357            }
358            if entry.live() {
359                pending = true;
360            } else if self
361                .earliest_out(*entry, control)?
362                .is_some_and(|order| order <= reader.snapshot)
363            {
364                return Ok(SafeSnapshot::Unsafe);
365            }
366        }
367        Ok(if pending {
368            SafeSnapshot::Pending
369        } else {
370            SafeSnapshot::Safe
371        })
372    }
373
374    /// Reclaim conflict history that cannot overlap any live snapshot, preserving earlier conflict-out order on every retained reader. Retained participant leases keep their terminal identity/outcome until owner recovery confirms release, without retaining obsolete predicates or edges. Dropping necessary summaries would allow a late read of a committed pivot to miss a serialization anomaly. No allocation or cancellation is needed for cleanup.
375    pub fn reclaim(&mut self) {
376        let oldest = self
377            .transactions
378            .iter()
379            .filter(|entry| entry.live())
380            .map(|entry| entry.snapshot)
381            .min();
382        let Some(oldest) = oldest else {
383            self.checkpoint_changed |= retain_copy(&mut self.transactions, |entry| {
384                entry.owner == ParticipantOwner::Leased
385            });
386            self.checkpoint_changed |= !self.outgoing.is_empty();
387            self.outgoing = BudgetedVec::new(self.outgoing.budget());
388            self.incoming = BudgetedVec::new(self.incoming.budget());
389            self.checkpoint_changed |= self.predicates.clear();
390            return;
391        };
392        for index in 0..self.transactions.len() {
393            let entry = self.transactions[index];
394            let mut earliest = entry.summarized_out;
395            for edge in &self.outgoing[edge_range(&self.outgoing, entry.id)] {
396                let target = self.node(edge.1);
397                if target.relevant() && target.committed.is_some_and(|end| end <= oldest) {
398                    if let Some(order) = target.prepared {
399                        earliest = Some(earliest.map_or(order, |old| old.min(order)));
400                    }
401                }
402            }
403            self.checkpoint_changed |= entry.summarized_out != earliest;
404            self.transactions[index].summarized_out = earliest;
405        }
406        self.checkpoint_changed |= retain_copy(&mut self.transactions, |entry| {
407            entry.owner == ParticipantOwner::Leased || entry.retains_history(oldest)
408        });
409        let retained = &self.transactions;
410        let retain_edge = |edge: Edge| {
411            retained
412                .binary_search_by_key(&edge.0, |entry| entry.id)
413                .is_ok_and(|position| retained[position].retains_history(oldest))
414                && retained
415                    .binary_search_by_key(&edge.1, |entry| entry.id)
416                    .is_ok_and(|position| retained[position].retains_history(oldest))
417        };
418        self.checkpoint_changed |= retain_copy(&mut self.outgoing, retain_edge);
419        retain_copy(&mut self.incoming, retain_edge);
420        self.checkpoint_changed |= self.predicates.reclaim(&self.transactions, oldest);
421    }
422
423    fn reserve_order(&self, preparing: bool) -> VersionResult<()> {
424        self.clock
425            .checked_add(self.pending_finishes)
426            .and_then(|order| order.checked_add(1 + u64::from(preparing)))
427            .ok_or(VersionError::SequenceExhausted)?;
428        Ok(())
429    }
430
431    fn validate_identity(&self, transaction: SerializableTransactionId) -> VersionResult<()> {
432        if transaction.database() != self.database {
433            return Err(VersionError::WrongDatabase);
434        }
435        if transaction.coordinator() != self.coordinator {
436            return Err(VersionError::WrongSerializableCoordinator);
437        }
438        Ok(())
439    }
440
441    fn position(&self, transaction: SerializableTransactionId) -> VersionResult<usize> {
442        self.validate_identity(transaction)?;
443        self.allocation_position(transaction.allocation())
444    }
445
446    fn allocation_position(&self, allocation: u64) -> VersionResult<usize> {
447        self.transactions
448            .binary_search_by_key(&allocation, |entry| entry.id)
449            .map_err(|_| VersionError::UnknownTransaction)
450    }
451
452    fn node(&self, allocation: u64) -> Transaction {
453        self.transactions[self
454            .allocation_position(allocation)
455            .expect("retained SSI edge must name a retained transaction")]
456    }
457}
458
459fn edge_range(edges: &[Edge], allocation: u64) -> std::ops::Range<usize> {
460    edges.partition_point(|edge| edge.0 < allocation)
461        ..edges.partition_point(|edge| edge.0 <= allocation)
462}
463
464fn insert_edge(edges: &mut BudgetedVec<Edge>, edge: Edge) -> VersionResult<()> {
465    if let Err(position) = edges.binary_search(&edge) {
466        edges.push(edge)?;
467        edges[position..].rotate_right(1);
468    }
469    Ok(())
470}
471
472fn retain_copy<T: Copy>(values: &mut BudgetedVec<T>, mut retain: impl FnMut(T) -> bool) -> bool {
473    let previous = values.len();
474    let mut kept = 0;
475    for index in 0..values.len() {
476        if retain(values[index]) {
477            values[kept] = values[index];
478            kept += 1;
479        }
480    }
481    values.truncate(kept);
482    if values.is_empty() {
483        *values = BudgetedVec::new(values.budget());
484    }
485    kept != previous
486}
uqa_storage/mvcc/serializable.rs

uqa_storage/mvcc/
serializable.rs
