uqa_storage/mvcc/
persistence.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Atomic persistence and durable transaction outcomes, independent of physical drivers.
8
9use std::sync::Arc;
10
11use uqa_core::memory::BudgetedVec;
12
13use crate::read_control::StorageReadControl;
14use crate::StorageBackendError;
15
16use super::{
17    CommitSequence, CommittedRecordSnapshot, PreparedRecordCommit, ScannedRecord, VersionError,
18    VersionResult,
19};
20
21#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
22pub struct DatabaseId([u8; 16]);
23
24impl DatabaseId {
25    pub const fn from_bytes(bytes: [u8; 16]) -> Self {
26        Self(bytes)
27    }
28    pub const fn as_bytes(self) -> [u8; 16] {
29        self.0
30    }
31}
32
33/// A durable, non-reused allocation in one database incarnation; it is not a SQL-visible XID.
34#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
35pub struct StorageTransactionId {
36    database: DatabaseId,
37    allocation: u64,
38}
39
40impl StorageTransactionId {
41    pub fn new(database: DatabaseId, allocation: u64) -> VersionResult<Self> {
42        if allocation == 0 {
43            return Err(VersionError::InvalidTransactionId);
44        }
45        Ok(Self {
46            database,
47            allocation,
48        })
49    }
50    pub const fn database(self) -> DatabaseId {
51        self.database
52    }
53    pub const fn allocation(self) -> u64 {
54        self.allocation
55    }
56}
57
58pub type CommitFingerprint = [u8; 32];
59pub type RecordPage = BudgetedVec<ScannedRecord>;
60
61#[derive(Debug, Clone, Copy, PartialEq, Eq)]
62pub struct CommitReceipt {
63    pub transaction: StorageTransactionId,
64    pub sequence: CommitSequence,
65    pub fingerprint: CommitFingerprint,
66}
67
68#[derive(Debug, Clone, Copy, PartialEq, Eq)]
69pub enum CommitStatus {
70    /// No retained authoritative outcome exists. This is not proof of rollback.
71    Unknown,
72    Pending,
73    Aborted,
74    Committed(CommitReceipt),
75}
76
77/// Durable outcome information retained in a storage error, including through provider wrappers. Absence of this information does not prove that a transaction aborted.
78#[derive(Debug, Clone, Copy, PartialEq, Eq)]
79pub enum CommitErrorOutcome {
80    Indeterminate(StorageTransactionId),
81    Committed(CommitReceipt),
82    Aborted(StorageTransactionId),
83}
84
85impl StorageBackendError {
86    /// Classify typed commit evidence without interpreting provider diagnostic text.
87    pub fn commit_outcome(&self) -> Option<CommitErrorOutcome> {
88        let mut cause: Option<&(dyn std::error::Error + 'static)> = Some(self);
89        while let Some(error) = cause {
90            if let Some(outcome) = error
91                .downcast_ref::<super::TransactionCompletionError>()
92                .and_then(|error| error.publication)
93                .or_else(|| error_commit_outcome(error))
94            {
95                return Some(outcome);
96            }
97            cause = error.source();
98        }
99        None
100    }
101}
102
103pub(super) fn error_commit_outcome(
104    error: &(dyn std::error::Error + 'static),
105) -> Option<CommitErrorOutcome> {
106    if let Some(CommitFailure::Indeterminate { transaction, .. }) =
107        error.downcast_ref::<CommitFailure>()
108    {
109        Some(CommitErrorOutcome::Indeterminate(*transaction))
110    } else if let Some(VersionError::AlreadyCommitted(receipt)) =
111        error.downcast_ref::<VersionError>()
112    {
113        Some(CommitErrorOutcome::Committed(*receipt))
114    } else if let Some(VersionError::AlreadyAborted(transaction)) =
115        error.downcast_ref::<VersionError>()
116    {
117        Some(CommitErrorOutcome::Aborted(*transaction))
118    } else {
119        None
120    }
121}
122
123#[derive(Debug, thiserror::Error)]
124pub enum CommitFailure {
125    /// The attempt published no changes; any prior successful attempt still has its receipt.
126    #[error(transparent)]
127    Rejected(#[from] VersionError),
128    #[error("commit outcome for {transaction:?} is indeterminate: {source}")]
129    Indeterminate {
130        transaction: StorageTransactionId,
131        #[source]
132        source: StorageBackendError,
133    },
134}
135
136pub type CommitResult = Result<CommitReceipt, CommitFailure>;
137
138/// Versioned record persistence. Every mutation of the current sequence, records and receipt must share one physical commit. Implementations retain no physical writer between calls and must distinguish rejection from an uncertain native commit outcome.
139pub trait VersionedPersistence: Send + Sync {
140    fn database_id(&self) -> DatabaseId;
141
142    /// Shared participant admission and recovery for this exact physical database. Wrappers must forward the capability; its presence alone does not enable public SQL SSI.
143    fn serializable_coordinator(&self) -> Option<&dyn super::SerializableCoordinator> {
144        None
145    }
146
147    /// Read the latest autonomous watermark in one physical read transaction. Absence must remain absent; this must not acquire write admission, create records or change allocation state. Validate the database incarnation and retain the caller's resource and cancellation controls.
148    fn identifier_watermark(
149        &self,
150        namespace: &[u8],
151        control: &StorageReadControl,
152    ) -> VersionResult<Option<u64>>;
153
154    /// Atomically observe or reserve identifiers under the database's physical admission. Persist the high watermark before returning; an error may consume identifiers but must never permit their reuse. This operation advances neither record visibility nor transaction allocation, publishes no private records, and survives transaction/savepoint rollback. Implementations and wrappers must preserve the caller's resource/cancellation control and reject a different database incarnation.
155    fn allocate_identifiers(
156        &self,
157        namespace: &[u8],
158        request: super::IdentifierRequest,
159        control: &StorageReadControl,
160    ) -> VersionResult<super::IdentifierAllocation>;
161
162    fn graph_record_layout(&self) -> Option<&dyn super::GraphRecordLayout> {
163        None
164    }
165
166    /// Common Key/Value addressing is the default. Native record adapters and their wrappers must return their own physical occurrence layout.
167    fn occurrence_record_layout(&self) -> &dyn super::OccurrenceRecordLayout {
168        &crate::key_value::KeyValueOccurrenceRecords
169    }
170
171    fn notification_record_layout(&self) -> &dyn super::NotificationRecordLayout {
172        &crate::key_value::KeyValueNotificationRecords
173    }
174
175    fn maintenance_record_layout(&self) -> &dyn super::MaintenanceRecordLayout {
176        &crate::key_value::KeyValueMaintenanceRecords
177    }
178
179    fn ivf_record_layout(&self) -> &dyn super::IVFRecordLayout {
180        &crate::key_value::KeyValueIVFRecords
181    }
182
183    /// Native adapters must override this when their physical node and edge families differ.
184    fn hnsw_record_layout(&self) -> Option<&dyn super::HNSWRecordLayout> {
185        Some(&crate::key_value::KeyValueHNSWRecords)
186    }
187
188    /// Persist a new transaction allocation before returning it, without advancing record visibility.
189    fn allocate_transaction(
190        &self,
191        control: &StorageReadControl,
192    ) -> VersionResult<StorageTransactionId>;
193
194    /// Allocate a receipt with explicit managed resolution ownership. Publish its lease before the Pending row becomes durable under the same receipt admission used by recovery. Only an authoritatively absent managed lease permits abandoned-owner recovery; raw allocations retain manual ownership. The default preserves legacy indefinite receipt retention.
195    fn allocate_managed_transaction(
196        &self,
197        control: &StorageReadControl,
198    ) -> VersionResult<super::RetainedTransactionAllocation> {
199        self.allocate_transaction(control)
200            .map(super::RetainedTransactionAllocation::untracked)
201    }
202
203    fn snapshot(
204        &self,
205        control: &StorageReadControl,
206    ) -> VersionResult<Arc<dyn CommittedRecordSnapshot>>;
207
208    /// Atomically remove obsolete historical revisions under snapshot admission. Preserve the newest revision at/before every live snapshot, all newer revisions, head tombstones and every transaction receipt. Neither visibility nor transaction allocation advances. Wrappers must forward the capability and its resource/cancellation control.
209    fn reclaim_versions(&self, control: &StorageReadControl) -> VersionResult<u64>;
210
211    /// Retry the same sealed logical changes and fingerprint. Return a durable matching receipt before validating snapshots or record heads. After verifying a pending receipt under exclusive admission, call `PreparedRecordCommit::validate_snapshot` before validating heads; only its rejection permits common storage to materialize derived effects again. Canonical changes and application callbacks must never be replayed, and mismatched fingerprint reuse is rejected.
212    fn commit(
213        &self,
214        transaction: StorageTransactionId,
215        prepared: &PreparedRecordCommit,
216        control: &StorageReadControl,
217    ) -> CommitResult;
218
219    fn commit_status(
220        &self,
221        transaction: StorageTransactionId,
222        control: &StorageReadControl,
223    ) -> VersionResult<CommitStatus>;
224
225    /// Durably release caller resolution ownership only after preserving its confirmed terminal outcome. This never releases a retained SSI publication. The default keeps historical receipts indefinitely; providers with bounded receipts override it. Errors preserve the caller's completed retry state, and a retry after durable acknowledgement remains valid even after reclamation.
226    fn acknowledge_transaction(
227        &self,
228        _acknowledgement: super::ReceiptAcknowledgement,
229        _control: &StorageReadControl,
230    ) -> VersionResult<()> {
231        Ok(())
232    }
233
234    /// Reclaim explicitly acknowledged receipts only after their final durable SSI publication reference has disappeared. Pending and unacknowledged receipts survive, including unknown completion owners. This is separate from record-history reclamation and does not advance allocation or visibility.
235    fn reclaim_transaction_receipts(&self, _control: &StorageReadControl) -> VersionResult<u64> {
236        Ok(0)
237    }
238
239    /// Atomically end a pending allocation. Return an existing committed receipt unchanged; abort must never overwrite a commit.
240    fn abort(
241        &self,
242        transaction: StorageTransactionId,
243        control: &StorageReadControl,
244    ) -> VersionResult<CommitStatus>;
245}
246
247/// Resolve an idempotent commit under the provider's exclusive commit boundary before validating or changing records.
248pub fn resolve_prepared_receipt(
249    status: CommitStatus,
250    transaction: StorageTransactionId,
251    fingerprint: CommitFingerprint,
252) -> VersionResult<Option<CommitReceipt>> {
253    match status {
254        CommitStatus::Pending => Ok(None),
255        CommitStatus::Committed(receipt)
256            if receipt.transaction == transaction && receipt.fingerprint == fingerprint =>
257        {
258            Ok(Some(receipt))
259        }
260        CommitStatus::Committed(_) => Err(VersionError::CommitMismatch),
261        CommitStatus::Aborted => Err(VersionError::TransactionFinished),
262        CommitStatus::Unknown => Err(VersionError::UnknownTransaction),
263    }
264}
uqa_storage/mvcc/persistence.rs

uqa_storage/mvcc/
persistence.rs
