uqa_storage/mvcc/session/
mod.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Logical Key/Value sessions retain private changes rather than a physical writer.
8
9mod batch;
10mod notifications;
11mod read;
12mod serializable;
13pub use serializable::{
14    SerializableReadContext, SerializableSession, SerializableSnapshotCapture,
15    SerializableSnapshotOptions,
16};
17mod transaction;
18
19use std::sync::Arc;
20
21use parking_lot::Mutex;
22
23use crate::read_control::{KeyValueReadVisitor, StorageReadControl, ValueReadVisitor};
24use crate::{
25    KeyValueBatch, KeyValueStore, PersistentStorageIdentity, StorageBackendError,
26    StorageBackendResult, StorageSessionAffinity,
27};
28
29use super::{
30    CommitFailure, MergedRecordSnapshot, StorageTransactionId, VersionError, VersionResult,
31    VersionedPersistence,
32};
33use transaction::Transaction;
34
35/// Retention allowance for one logical session, including private history, savepoints, batches and retained views. Exhaustion is an error; this implementation does not spill.
36#[derive(Debug, Clone, Copy)]
37pub struct VersionedSessionOptions {
38    pub retained_bytes: usize,
39}
40
41impl Default for VersionedSessionOptions {
42    fn default() -> Self {
43        Self {
44            retained_bytes: 64 * 1024 * 1024,
45        }
46    }
47}
48
49/// A provider-independent session with pinned reads and conditional atomic publication. Common storage resolves typed index changes before physical admission; SQL isolation and tuple locks remain above this byte-record contract.
50pub struct VersionedKeyValueStore {
51    affinity: StorageSessionAffinity,
52    persistence: Arc<dyn VersionedPersistence>,
53    identity: Option<PersistentStorageIdentity>,
54    options: VersionedSessionOptions,
55    control: StorageReadControl,
56    write_cancellation: uqa_core::CancellationToken,
57    retained: Option<MergedRecordSnapshot>,
58    active: Mutex<Option<Transaction>>,
59}
60
61impl VersionedKeyValueStore {
62    /// Reclaim committed history without changing an active transaction or its retained readers. The persistence owner supplies atomic snapshot admission and physical deletion.
63    pub fn reclaim_versions(&self) -> StorageBackendResult<u64> {
64        self.persistence
65            .reclaim_versions(&self.write_control())
66            .map_err(VersionError::into_storage_error)
67    }
68
69    /// Autonomous watermark visibility follows physical reservations, including reservations made after this session pinned its record snapshot. No write capability is required and no logical transaction is started or finished.
70    pub fn identifier_watermark(&self, namespace: &[u8]) -> StorageBackendResult<Option<u64>> {
71        self.persistence
72            .identifier_watermark(namespace, &self.control)
73            .map_err(VersionError::into_storage_error)
74    }
75
76    /// Reserve identities without publishing this session's private records. Read-only sessions and unresolved sealed commits cannot allocate; rollback never reclaims a successful reservation.
77    pub fn allocate_identifiers(
78        &self,
79        namespace: &[u8],
80        request: super::IdentifierRequest,
81    ) -> StorageBackendResult<super::IdentifierAllocation> {
82        self.require_mutable_session()?;
83        let active = self.active.lock();
84        if let Some(transaction) = active.as_ref() {
85            transaction
86                .writable()
87                .map_err(VersionError::into_storage_error)?;
88        }
89        self.persistence
90            .allocate_identifiers(namespace, request, &self.write_control())
91            .map_err(VersionError::into_storage_error)
92    }
93
94    pub fn new(
95        persistence: Arc<dyn VersionedPersistence>,
96        identity: Option<PersistentStorageIdentity>,
97        options: VersionedSessionOptions,
98    ) -> Self {
99        Self::new_with_cancellation(
100            persistence,
101            identity,
102            options,
103            uqa_core::CancellationToken::new(),
104        )
105    }
106
107    /// Share the invoking execution's cancellation during autonomous allocation and publication. Retained reads and rollback cleanup keep their independent control, so cancelling a statement cannot prevent its undo.
108    pub fn new_with_cancellation(
109        persistence: Arc<dyn VersionedPersistence>,
110        identity: Option<PersistentStorageIdentity>,
111        options: VersionedSessionOptions,
112        write_cancellation: uqa_core::CancellationToken,
113    ) -> Self {
114        Self {
115            affinity: StorageSessionAffinity::new(),
116            persistence,
117            identity,
118            options,
119            control: StorageReadControl::with_limit(options.retained_bytes),
120            write_cancellation,
121            retained: None,
122            active: Mutex::new(None),
123        }
124    }
125
126    /// Create an independent logical session with the same persistence and retention limit.
127    pub fn new_session(&self) -> Self {
128        Self::new(
129            Arc::clone(&self.persistence),
130            self.identity.clone(),
131            self.options,
132        )
133    }
134
135    /// Keep a separate transaction context and retention budget while inheriting the invoking writer's cancellation.
136    pub fn new_session_with_cancellation(
137        &self,
138        cancellation: &uqa_core::CancellationToken,
139    ) -> Self {
140        Self::new_with_cancellation(
141            Arc::clone(&self.persistence),
142            self.identity.clone(),
143            self.options,
144            cancellation.clone(),
145        )
146    }
147
148    /// Open an independently owned read-only session at this exact committed/private view. Its original participant, snapshot leases and retention allowance remain shared, while read cancellation stays independent. Completing this reader never completes the originating transaction.
149    pub fn new_retained_read_session(
150        &self,
151        cancellation: &uqa_core::CancellationToken,
152    ) -> StorageBackendResult<Self> {
153        cancellation.check()?;
154        let retained = self.view().map_err(VersionError::into_storage_error)?;
155        let mut session = self.new_session_with_cancellation(cancellation);
156        session.control =
157            StorageReadControl::new(self.control.memory(), session.control.cancellation());
158        session.retained = Some(retained);
159        Ok(session)
160    }
161
162    fn require_mutable_session(&self) -> StorageBackendResult<()> {
163        if self.retained.is_some() {
164            return Err(StorageBackendError::Other(
165                "cannot write through a retained read-only session".into(),
166            ));
167        }
168        Ok(())
169    }
170
171    fn write_control(&self) -> StorageReadControl {
172        StorageReadControl::new(self.control.memory(), &self.write_cancellation)
173    }
174
175    pub fn options(&self) -> VersionedSessionOptions {
176        self.options
177    }
178
179    /// Retain the current committed and private boundaries for a compound provider read. Later session mutations, rollback and commit do not advance this view.
180    pub fn record_snapshot(&self) -> VersionResult<MergedRecordSnapshot> {
181        self.view()
182    }
183
184    /// Advance an active transaction to the current committed boundary while retaining and rebasing its evaluated private changes. Retained readers keep their previous boundary; savepoint undo restores its matching private changes and committed base. SQL chooses when its isolation level allows this transition. A sealed commit cannot advance, and failed preparation leaves the active view unchanged.
185    pub fn refresh_transaction_snapshot(
186        &self,
187        cancellation: &uqa_core::CancellationToken,
188    ) -> StorageBackendResult<()> {
189        cancellation.check()?;
190        let control = StorageReadControl::new(self.control.memory(), cancellation);
191        let mut active = self.active.lock();
192        if self.retained.is_some() {
193            active.as_ref().ok_or_else(no_transaction)?;
194            return Ok(());
195        }
196        active
197            .as_mut()
198            .ok_or_else(no_transaction)?
199            .refresh(&*self.persistence, &control)
200            .map_err(VersionError::into_storage_error)
201    }
202
203    /// Share this session's retention allowance with provider codecs and retained record readers. Creating a control clone does not create another allowance.
204    pub fn retention_control(&self) -> StorageReadControl {
205        self.control.clone()
206    }
207
208    pub fn session_affinity(&self) -> StorageSessionAffinity {
209        self.affinity.clone()
210    }
211
212    /// Identify a sealed attempt after a failed commit. Retrying `commit_transaction` resubmits only the identical evaluated batch, and never reevaluates application code.
213    pub fn pending_commit(&self) -> Option<StorageTransactionId> {
214        self.active
215            .lock()
216            .as_ref()
217            .and_then(|transaction| transaction.allocation)
218    }
219
220    /// Identify a retained physical attempt or uncertain logical completion, including an empty/read-only SSI transaction without a write receipt. Completion retries never replay evaluated application work.
221    pub fn pending_transaction_completion(&self) -> Option<super::TransactionOutcomeId> {
222        self.active
223            .lock()
224            .as_ref()
225            .and_then(Transaction::pending_completion)
226    }
227
228    /// Establish the original SSI participant and fixed record boundary before this transaction's first logical access. SQL must call this at its first snapshot-bearing statement, then supply logical observations for every participating access path. Capability presence alone does not enable SQL SSI or fence legacy publishers.
229    pub fn establish_serializable_snapshot(&self) -> StorageBackendResult<SerializableReadContext> {
230        self.require_mutable_session()?;
231        self.active
232            .lock()
233            .as_mut()
234            .ok_or_else(no_transaction)?
235            .establish_serializable(&self.persistence, &self.write_control())
236            .map_err(VersionError::into_storage_error)
237    }
238
239    /// Register a logical write before staging its records. SQL statement/savepoint ownership must encompass both this observation and the evaluated mutation. Retained record readers expose only read observation capability.
240    pub fn observe_serializable_write(
241        &self,
242        predicate: super::SerializablePredicate<'_>,
243    ) -> StorageBackendResult<()> {
244        let mut active = self.active.lock();
245        let transaction = active.as_mut().ok_or_else(no_transaction)?;
246        transaction
247            .writable()
248            .map_err(VersionError::into_storage_error)?;
249        let context = transaction.serializable_context().ok_or_else(|| {
250            StorageBackendError::Other("no serializable participant in this session".into())
251        })?;
252        let control = self.write_control();
253        context
254            .with_graph(&control, |graph| {
255                graph.observe_write(context.id(), predicate, &control)
256            })
257            .map_err(VersionError::into_storage_error)
258    }
259
260    fn begin(&self, read_only: bool) -> StorageBackendResult<()> {
261        if !read_only {
262            self.require_mutable_session()?;
263        }
264        let mut active = self.active.lock();
265        if active.is_some() {
266            return Err(StorageBackendError::Other(
267                "a KeyValue transaction is already active".into(),
268            ));
269        }
270        *active = Some(match self.retained.as_ref() {
271            Some(view) => Transaction::at_snapshot(view.retain_committed(), true, &self.control),
272            None => Transaction::new(&*self.persistence, read_only, &self.control)
273                .map_err(VersionError::into_storage_error)?,
274        });
275        Ok(())
276    }
277
278    fn view(&self) -> VersionResult<MergedRecordSnapshot> {
279        if let Some(view) = self.retained.as_ref() {
280            return view.try_clone();
281        }
282        let active = self.active.lock();
283        if let Some(transaction) = active.as_ref() {
284            transaction.view()
285        } else {
286            Transaction::new(&*self.persistence, true, &self.control)?.view()
287        }
288    }
289
290    fn write<T>(
291        &self,
292        operation: impl FnOnce(&mut Transaction) -> VersionResult<T>,
293    ) -> StorageBackendResult<T> {
294        self.require_mutable_session()?;
295        let mut active = self.active.lock();
296        if let Some(transaction) = active.as_mut() {
297            return transaction
298                .atomic(operation)
299                .map_err(VersionError::into_storage_error);
300        }
301        let mut transaction = Transaction::new(&*self.persistence, false, &self.control)
302            .map_err(VersionError::into_storage_error)?;
303        let result = operation(&mut transaction).map_err(VersionError::into_storage_error)?;
304        // Retain even an autocommit attempt until its durable outcome is known.
305        *active = Some(transaction);
306        active
307            .as_mut()
308            .expect("retained attempt")
309            .commit(&*self.persistence, &self.write_control())?;
310        active
311            .as_ref()
312            .expect("retained attempt")
313            .acknowledge_completion(&*self.persistence, &self.control)?;
314        *active = None;
315        Ok(result)
316    }
317
318    fn savepoint_action(
319        &self,
320        operation: impl FnOnce(&mut Transaction) -> VersionResult<()>,
321    ) -> StorageBackendResult<()> {
322        let mut active = self.active.lock();
323        let transaction = active.as_mut().ok_or_else(no_transaction)?;
324        transaction
325            .unsealed()
326            .map_err(VersionError::into_storage_error)?;
327        operation(transaction).map_err(VersionError::into_storage_error)
328    }
329}
330
331impl super::IdentifierAllocator for VersionedKeyValueStore {
332    fn identifier_watermark(&self, namespace: &[u8]) -> StorageBackendResult<Option<u64>> {
333        Self::identifier_watermark(self, namespace)
334    }
335
336    fn allocate_identifiers(
337        &self,
338        namespace: &[u8],
339        request: super::IdentifierRequest,
340    ) -> StorageBackendResult<super::IdentifierAllocation> {
341        Self::allocate_identifiers(self, namespace, request)
342    }
343}
344
345impl KeyValueStore for VersionedKeyValueStore {
346    fn retention_control(&self) -> Option<StorageReadControl> {
347        Some(Self::retention_control(self))
348    }
349
350    fn notification_publications(
351        &self,
352    ) -> Option<&dyn crate::notifications::NotificationPublicationStore> {
353        Some(self)
354    }
355
356    fn serializable_session(&self) -> Option<&dyn SerializableSession> {
357        self.persistence
358            .serializable_coordinator()
359            .map(|_| self as &dyn SerializableSession)
360    }
361
362    fn vacuum(&self) -> StorageBackendResult<()> {
363        let active = self.active.lock();
364        if active.is_some() {
365            return Err(StorageBackendError::Other(
366                "vacuum requires an inactive logical session".into(),
367            ));
368        }
369        self.reclaim_versions().map(|_| ())
370    }
371
372    fn write_cancellation(&self) -> Option<uqa_core::CancellationToken> {
373        Some(self.write_cancellation.clone())
374    }
375
376    fn open_session_with_cancellation(
377        &self,
378        cancellation: &uqa_core::CancellationToken,
379    ) -> StorageBackendResult<Arc<dyn KeyValueStore>> {
380        cancellation.check()?;
381        Ok(Arc::new(self.new_session_with_cancellation(cancellation)))
382    }
383
384    fn open_retained_read_session(
385        &self,
386        cancellation: &uqa_core::CancellationToken,
387    ) -> StorageBackendResult<Arc<dyn KeyValueStore>> {
388        Ok(Arc::new(self.new_retained_read_session(cancellation)?))
389    }
390
391    fn transaction_model(&self) -> crate::StorageTransactionModel {
392        crate::StorageTransactionModel::VersionedConcurrent {
393            database: self.persistence.database_id(),
394        }
395    }
396
397    fn refresh_transaction_snapshot(
398        &self,
399        cancellation: &uqa_core::CancellationToken,
400    ) -> StorageBackendResult<()> {
401        Self::refresh_transaction_snapshot(self, cancellation)
402    }
403
404    fn identifier_allocator(&self) -> Option<&dyn super::IdentifierAllocator> {
405        Some(self)
406    }
407
408    fn with_read_view(
409        &self,
410        operation: &mut crate::key_value::KeyValueReadScope<'_>,
411    ) -> StorageBackendResult<()> {
412        self.control.check()?;
413        let view = self.view().map_err(VersionError::into_storage_error)?;
414        operation(&read::RecordRead {
415            view: &view,
416            database: self.persistence.database_id(),
417            control: &self.control,
418        })?;
419        self.control.check()
420    }
421
422    fn with_mutation(
423        &self,
424        operation: &mut crate::key_value::KeyValueMutation<'_>,
425    ) -> StorageBackendResult<()> {
426        self.control.check()?;
427        self.write(|transaction| {
428            let view = transaction.view()?;
429            let read = read::RecordRead {
430                view: &view,
431                database: self.persistence.database_id(),
432                control: &self.control,
433            };
434            let mut batch = batch::Batch::new(
435                self,
436                transaction
437                    .serializable_context()
438                    .map(SerializableReadContext::id),
439            );
440            operation(&read, &mut batch)?;
441            self.control.check()?;
442            batch.apply(transaction)
443        })
444    }
445
446    fn transaction_affinity(&self) -> Option<StorageSessionAffinity> {
447        Some(self.session_affinity())
448    }
449
450    fn storage_identity(&self) -> StorageBackendResult<Option<PersistentStorageIdentity>> {
451        Ok(self.identity.clone())
452    }
453
454    fn open_session(&self) -> StorageBackendResult<Arc<dyn KeyValueStore>> {
455        Ok(Arc::new(self.new_session()))
456    }
457
458    fn get(&self, key: &[u8]) -> StorageBackendResult<Option<Vec<u8>>> {
459        let mut value = None;
460        self.visit_value(key, &self.control, &mut |bytes| {
461            value = bytes.map(<[u8]>::to_vec);
462            Ok(())
463        })?;
464        Ok(value)
465    }
466
467    fn contains_key(&self, key: &[u8]) -> StorageBackendResult<bool> {
468        Ok(self
469            .view()
470            .and_then(|view| view.metadata(key, &self.control))
471            .map_err(VersionError::into_storage_error)?
472            .is_some_and(|record| record.live))
473    }
474
475    fn contains_prefix_budgeted(
476        &self,
477        prefix: &[u8],
478        control: &StorageReadControl,
479    ) -> StorageBackendResult<bool> {
480        control.check()?;
481        let mut found = false;
482        self.view()
483            .map_err(VersionError::into_storage_error)?
484            .visit_keys(prefix, None, usize::MAX, control, &mut |_, record| {
485                found = record.live;
486                Ok(!found)
487            })
488            .map_err(VersionError::into_storage_error)?;
489        Ok(found)
490    }
491
492    fn visit_value(
493        &self,
494        key: &[u8],
495        control: &StorageReadControl,
496        visit: &mut ValueReadVisitor<'_>,
497    ) -> StorageBackendResult<()> {
498        control.check()?;
499        self.view()
500            .map_err(VersionError::into_storage_error)?
501            .visit_value(key, control, &mut |record| {
502                visit(record.and_then(|record| record.value)).map_err(VersionError::from)
503            })
504            .map_err(VersionError::into_storage_error)
505    }
506
507    fn visit_prefix_after(
508        &self,
509        prefix: &[u8],
510        after: Option<&[u8]>,
511        limit: usize,
512        control: &StorageReadControl,
513        visit: &mut KeyValueReadVisitor<'_>,
514    ) -> StorageBackendResult<()> {
515        control.check()?;
516        if limit == 0 {
517            return Ok(());
518        }
519        let view = self.view().map_err(VersionError::into_storage_error)?;
520        read::visit_live(&view, prefix, after, limit, control, visit)
521            .map_err(VersionError::into_storage_error)
522    }
523
524    fn put(&self, key: &[u8], value: &[u8]) -> StorageBackendResult<()> {
525        self.write(|transaction| transaction.replace(key, Some(value), &self.control))
526    }
527    fn delete(&self, key: &[u8]) -> StorageBackendResult<()> {
528        self.write(|transaction| transaction.replace(key, None, &self.control))
529    }
530
531    fn scan_prefix(&self, prefix: &[u8]) -> StorageBackendResult<Vec<(Vec<u8>, Vec<u8>)>> {
532        self.scan_prefix_after(prefix, None, usize::MAX)
533    }
534
535    fn scan_prefix_after(
536        &self,
537        prefix: &[u8],
538        after: Option<&[u8]>,
539        limit: usize,
540    ) -> StorageBackendResult<Vec<(Vec<u8>, Vec<u8>)>> {
541        let mut rows = Vec::new();
542        self.visit_prefix_after(prefix, after, limit, &self.control, &mut |key, value| {
543            rows.push((key.to_vec(), value.to_vec()));
544            Ok(())
545        })?;
546        Ok(rows)
547    }
548
549    fn scan_prefix_keys_after(
550        &self,
551        prefix: &[u8],
552        after: Option<&[u8]>,
553        limit: usize,
554    ) -> StorageBackendResult<Vec<Vec<u8>>> {
555        let mut keys = Vec::new();
556        if limit != 0 {
557            self.view()
558                .map_err(VersionError::into_storage_error)?
559                .visit_keys(
560                    prefix,
561                    after,
562                    usize::MAX,
563                    &self.control,
564                    &mut |key, record| {
565                        if record.live {
566                            keys.push(key.to_vec());
567                        }
568                        Ok(keys.len() < limit)
569                    },
570                )
571                .map_err(VersionError::into_storage_error)?;
572        }
573        Ok(keys)
574    }
575
576    fn first_prefix_after(
577        &self,
578        prefix: &[u8],
579        after: Option<&[u8]>,
580    ) -> StorageBackendResult<Option<(Vec<u8>, Vec<u8>)>> {
581        Ok(self.scan_prefix_after(prefix, after, 1)?.pop())
582    }
583    fn delete_prefix(&self, prefix: &[u8]) -> StorageBackendResult<usize> {
584        self.write(|transaction| transaction.delete_prefix(prefix, &self.control))
585    }
586    fn batch(&self) -> Box<dyn KeyValueBatch + '_> {
587        let participant = self
588            .active
589            .lock()
590            .as_ref()
591            .and_then(Transaction::serializable_context)
592            .map(SerializableReadContext::id);
593        Box::new(batch::Batch::new(self, participant))
594    }
595    fn begin_transaction(&self) -> StorageBackendResult<()> {
596        self.begin(false)
597    }
598    fn begin_read_transaction(&self) -> StorageBackendResult<()> {
599        self.begin(true)
600    }
601    fn begin_upgradeable_transaction(&self) -> StorageBackendResult<()> {
602        self.begin(self.retained.is_some())
603    }
604    fn in_transaction(&self) -> bool {
605        self.active.lock().is_some()
606    }
607    fn transaction_has_written(&self) -> StorageBackendResult<bool> {
608        Ok(self.active.lock().as_ref().is_some_and(|transaction| {
609            transaction.changes.has_written() || transaction.has_derived_changes()
610        }))
611    }
612    fn change_version(&self) -> StorageBackendResult<Option<u64>> {
613        Ok(Some(
614            self.view()
615                .map_err(VersionError::into_storage_error)?
616                .sequence()
617                .as_u64(),
618        ))
619    }
620
621    fn commit_transaction(&self) -> StorageBackendResult<()> {
622        let mut active = self.active.lock();
623        active
624            .as_mut()
625            .ok_or_else(no_transaction)?
626            .commit(&*self.persistence, &self.write_control())?;
627        active
628            .as_ref()
629            .expect("completed attempt")
630            .acknowledge_completion(&*self.persistence, &self.control)?;
631        *active = None;
632        Ok(())
633    }
634
635    fn rollback_transaction(&self) -> StorageBackendResult<()> {
636        let mut active = self.active.lock();
637        active
638            .as_mut()
639            .ok_or_else(no_transaction)?
640            .abort(&*self.persistence, &self.control)?;
641        active
642            .as_ref()
643            .expect("completed attempt")
644            .acknowledge_completion(&*self.persistence, &self.control)?;
645        *active = None;
646        Ok(())
647    }
648
649    fn savepoint(&self, name: &str) -> StorageBackendResult<()> {
650        self.savepoint_action(|transaction| transaction.savepoint(name, &self.control))
651    }
652    fn release_savepoint(&self, name: &str) -> StorageBackendResult<()> {
653        self.savepoint_action(|transaction| transaction.release(name))
654    }
655    fn rollback_to_savepoint(&self, name: &str) -> StorageBackendResult<()> {
656        self.savepoint_action(|transaction| transaction.rollback_to(name, &self.control))
657    }
658}
659
660fn no_transaction() -> StorageBackendError {
661    StorageBackendError::Other("no active KeyValue transaction".into())
662}
663
664fn commit_error(error: CommitFailure) -> StorageBackendError {
665    match error {
666        CommitFailure::Rejected(error) => error.into_storage_error(),
667        error @ CommitFailure::Indeterminate { .. } => StorageBackendError::backend("MVCC", error),
668    }
669}
uqa_storage/mvcc/session/mod.rs

uqa_storage/mvcc/session/
mod.rs
