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 evaluation;
11mod notifications;
12mod read;
13pub use read::RecordRead;
14mod serializable;
15pub use serializable::{
16    SerializableReadContext, SerializableSession, SerializableSnapshotCapture,
17    SerializableSnapshotOptions,
18};
19mod transaction;
20mod vector_fields;
21pub use vector_fields::VectorFieldGuardMaintenance;
22
23use std::sync::Arc;
24
25use parking_lot::Mutex;
26
27use crate::read_control::{KeyValueReadVisitor, StorageReadControl, ValueReadVisitor};
28use crate::{
29    KeyValueBatch, KeyValueStore, PersistentStorageIdentity, StorageBackendError,
30    StorageBackendResult, StorageSessionAffinity,
31};
32
33use super::{
34    CommitFailure, MergedRecordSnapshot, StorageTransactionId, VersionError, VersionResult,
35    VersionedPersistence,
36};
37use transaction::Transaction;
38
39/// Retention allowance for one logical session, including private history, savepoints, batches and retained views. Exhaustion is an error; this implementation does not spill.
40#[derive(Debug, Clone, Copy)]
41pub struct VersionedSessionOptions {
42    pub retained_bytes: usize,
43}
44
45impl Default for VersionedSessionOptions {
46    fn default() -> Self {
47        Self {
48            retained_bytes: 64 * 1024 * 1024,
49        }
50    }
51}
52
53/// 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.
54pub struct VersionedKeyValueStore {
55    affinity: StorageSessionAffinity,
56    persistence: Arc<dyn VersionedPersistence>,
57    identity: Option<PersistentStorageIdentity>,
58    options: VersionedSessionOptions,
59    control: StorageReadControl,
60    write_cancellation: uqa_core::CancellationToken,
61    retained: Option<MergedRecordSnapshot>,
62    active: Mutex<Option<Transaction>>,
63}
64
65impl VersionedKeyValueStore {
66    /// Evaluate one provider mutation against the same fixed native record view as its batch. The callback must not reenter this session or complete its batch. Common MVCC owns autocommit, failed-evaluation cleanup and receipt retry; callbacks are never replayed.
67    pub fn with_versioned_record_mutation<T>(
68        &self,
69        operation: impl FnOnce(
70            super::StorageMutationOrigin,
71            MergedRecordSnapshot,
72            &mut dyn KeyValueBatch,
73        ) -> StorageBackendResult<T>,
74    ) -> StorageBackendResult<T> {
75        self.control.check()?;
76        self.write(|transaction| {
77            let origin = transaction.mutation_origin(&*self.persistence, &self.write_control())?;
78            let view = transaction.view()?;
79            let mut batch = batch::Batch::new(
80                self,
81                transaction
82                    .serializable_context()
83                    .map(SerializableReadContext::id),
84            );
85            let result = operation(origin, view, &mut batch)?;
86            self.control.check()?;
87            batch.apply(transaction)?;
88            Ok(result)
89        })
90    }
91
92    /// Reclaim committed history without changing an active transaction or its retained readers. The persistence owner supplies atomic snapshot admission and physical deletion.
93    pub fn reclaim_versions(&self) -> StorageBackendResult<u64> {
94        let control = self.write_control();
95        VectorFieldGuardMaintenance::run(self, &control)?;
96        let removed = self
97            .persistence
98            .reclaim_versions(&control)
99            .map_err(VersionError::into_storage_error)?;
100        self.persistence
101            .reclaim_diskann_tombstones(&control)
102            .map_err(VersionError::into_storage_error)?;
103        Ok(removed)
104    }
105
106    /// 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.
107    pub fn identifier_watermark(&self, namespace: &[u8]) -> StorageBackendResult<Option<u64>> {
108        self.persistence
109            .identifier_watermark(namespace, &self.control)
110            .map_err(VersionError::into_storage_error)
111    }
112
113    /// Reserve identities without publishing this session's private records. Read-only sessions and unresolved sealed commits cannot allocate; rollback never reclaims a successful reservation.
114    pub fn allocate_identifiers(
115        &self,
116        namespace: &[u8],
117        request: super::IdentifierRequest,
118    ) -> StorageBackendResult<super::IdentifierAllocation> {
119        self.require_mutable_session()?;
120        let active = self.active.lock();
121        if let Some(transaction) = active.as_ref() {
122            transaction
123                .writable()
124                .map_err(VersionError::into_storage_error)?;
125        }
126        self.persistence
127            .allocate_identifiers(namespace, request, &self.write_control())
128            .map_err(VersionError::into_storage_error)
129    }
130
131    pub fn new(
132        persistence: Arc<dyn VersionedPersistence>,
133        identity: Option<PersistentStorageIdentity>,
134        options: VersionedSessionOptions,
135    ) -> Self {
136        Self::new_with_cancellation(
137            persistence,
138            identity,
139            options,
140            uqa_core::CancellationToken::new(),
141        )
142    }
143
144    /// 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.
145    pub fn new_with_cancellation(
146        persistence: Arc<dyn VersionedPersistence>,
147        identity: Option<PersistentStorageIdentity>,
148        options: VersionedSessionOptions,
149        write_cancellation: uqa_core::CancellationToken,
150    ) -> Self {
151        Self {
152            affinity: StorageSessionAffinity::new(),
153            persistence,
154            identity,
155            options,
156            control: StorageReadControl::with_limit(options.retained_bytes),
157            write_cancellation,
158            retained: None,
159            active: Mutex::new(None),
160        }
161    }
162
163    /// Create an independent logical session with the same persistence and retention limit.
164    pub fn new_session(&self) -> Self {
165        Self::new(
166            Arc::clone(&self.persistence),
167            self.identity.clone(),
168            self.options,
169        )
170    }
171
172    /// Keep a separate transaction context and retention budget while inheriting the invoking writer's cancellation.
173    pub fn new_session_with_cancellation(
174        &self,
175        cancellation: &uqa_core::CancellationToken,
176    ) -> Self {
177        Self::new_with_cancellation(
178            Arc::clone(&self.persistence),
179            self.identity.clone(),
180            self.options,
181            cancellation.clone(),
182        )
183    }
184
185    /// Keep an independent writer while sharing the invoking operation's memory allowance. Read/rollback cancellation stays independent so cleanup remains possible after write cancellation.
186    pub fn new_controlled_session(&self, control: &StorageReadControl) -> Self {
187        let mut session = self.new_session_with_cancellation(control.cancellation());
188        session.control = StorageReadControl::new(control.memory(), session.control.cancellation());
189        session
190    }
191
192    /// 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.
193    pub fn new_retained_read_session(
194        &self,
195        cancellation: &uqa_core::CancellationToken,
196    ) -> StorageBackendResult<Self> {
197        cancellation.check()?;
198        let retained = self.view().map_err(VersionError::into_storage_error)?;
199        let mut session = self.new_session_with_cancellation(cancellation);
200        session.control =
201            StorageReadControl::new(self.control.memory(), session.control.cancellation());
202        session.retained = Some(retained);
203        Ok(session)
204    }
205
206    fn require_mutable_session(&self) -> StorageBackendResult<()> {
207        if self.retained.is_some() {
208            return Err(StorageBackendError::Other(
209                "cannot write through a retained read-only session".into(),
210            ));
211        }
212        Ok(())
213    }
214
215    fn write_control(&self) -> StorageReadControl {
216        StorageReadControl::new(self.control.memory(), &self.write_cancellation)
217    }
218
219    pub fn options(&self) -> VersionedSessionOptions {
220        self.options
221    }
222
223    /// Retain the current committed and private boundaries for a compound provider read. Later session mutations, rollback and commit do not advance this view.
224    pub fn record_snapshot(&self) -> VersionResult<MergedRecordSnapshot> {
225        self.view()
226    }
227
228    /// 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.
229    pub fn refresh_transaction_snapshot(
230        &self,
231        cancellation: &uqa_core::CancellationToken,
232    ) -> StorageBackendResult<()> {
233        cancellation.check()?;
234        let control = StorageReadControl::new(self.control.memory(), cancellation);
235        let mut active = self.active.lock();
236        if self.retained.is_some() {
237            active.as_ref().ok_or_else(no_transaction)?;
238            return Ok(());
239        }
240        active
241            .as_mut()
242            .ok_or_else(no_transaction)?
243            .refresh(&*self.persistence, &control)
244            .map_err(VersionError::into_storage_error)
245    }
246
247    /// Share this session's retention allowance with provider codecs and retained record readers. Creating a control clone does not create another allowance.
248    pub fn retention_control(&self) -> StorageReadControl {
249        self.control.clone()
250    }
251
252    pub fn session_affinity(&self) -> StorageSessionAffinity {
253        self.affinity.clone()
254    }
255
256    /// Identify a sealed attempt after a failed commit. Retrying `commit_transaction` resubmits only the identical evaluated batch, and never reevaluates application code.
257    pub fn pending_commit(&self) -> Option<StorageTransactionId> {
258        self.active
259            .lock()
260            .as_ref()
261            .and_then(Transaction::pending_commit)
262    }
263
264    /// 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.
265    pub fn pending_transaction_completion(&self) -> Option<super::TransactionOutcomeId> {
266        self.active
267            .lock()
268            .as_ref()
269            .and_then(Transaction::pending_completion)
270    }
271
272    /// 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.
273    pub fn establish_serializable_snapshot(&self) -> StorageBackendResult<SerializableReadContext> {
274        self.require_mutable_session()?;
275        self.active
276            .lock()
277            .as_mut()
278            .ok_or_else(no_transaction)?
279            .establish_serializable(&self.persistence, &self.write_control())
280            .map_err(VersionError::into_storage_error)
281    }
282
283    /// 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.
284    pub fn observe_serializable_write(
285        &self,
286        predicate: super::SerializablePredicate<'_>,
287    ) -> StorageBackendResult<()> {
288        let mut active = self.active.lock();
289        let transaction = active.as_mut().ok_or_else(no_transaction)?;
290        transaction
291            .writable()
292            .map_err(VersionError::into_storage_error)?;
293        let context = transaction.serializable_context().ok_or_else(|| {
294            StorageBackendError::Other("no serializable participant in this session".into())
295        })?;
296        let control = self.write_control();
297        context
298            .with_graph(&control, |graph| {
299                graph.observe_write(context.id(), predicate, &control)
300            })
301            .map_err(VersionError::into_storage_error)
302    }
303
304    fn begin(&self, read_only: bool) -> StorageBackendResult<()> {
305        if !read_only {
306            self.require_mutable_session()?;
307        }
308        let mut active = self.active.lock();
309        if active.is_some() {
310            return Err(StorageBackendError::Other(
311                "a KeyValue transaction is already active".into(),
312            ));
313        }
314        *active = Some(match self.retained.as_ref() {
315            Some(view) => Transaction::at_snapshot(view.retain_committed(), true, &self.control),
316            None => Transaction::new(&*self.persistence, read_only, &self.control)
317                .map_err(VersionError::into_storage_error)?,
318        });
319        Ok(())
320    }
321
322    fn view(&self) -> VersionResult<MergedRecordSnapshot> {
323        if let Some(view) = self.retained.as_ref() {
324            return view.try_clone();
325        }
326        let active = self.active.lock();
327        if let Some(transaction) = active.as_ref() {
328            transaction.view()
329        } else {
330            Transaction::new(&*self.persistence, true, &self.control)?.view()
331        }
332    }
333
334    fn write<T>(
335        &self,
336        operation: impl FnOnce(&mut Transaction) -> VersionResult<T>,
337    ) -> StorageBackendResult<T> {
338        self.require_mutable_session()?;
339        let mut active = self.active.lock();
340        if let Some(transaction) = active.as_mut() {
341            return transaction
342                .atomic(operation)
343                .map_err(VersionError::into_storage_error);
344        }
345        let transaction = Transaction::new(&*self.persistence, false, &self.control)
346            .map_err(VersionError::into_storage_error)?;
347        let result = self.evaluate_autocommit(&mut active, transaction, operation)?;
348        active
349            .as_mut()
350            .expect("retained attempt")
351            .commit(&*self.persistence, &self.write_control())?;
352        active
353            .as_ref()
354            .expect("retained attempt")
355            .acknowledge_completion(&*self.persistence, &self.control)?;
356        *active = None;
357        Ok(result)
358    }
359
360    fn savepoint_action(
361        &self,
362        operation: impl FnOnce(&mut Transaction) -> VersionResult<()>,
363    ) -> StorageBackendResult<()> {
364        let mut active = self.active.lock();
365        let transaction = active.as_mut().ok_or_else(no_transaction)?;
366        transaction
367            .unsealed()
368            .map_err(VersionError::into_storage_error)?;
369        operation(transaction).map_err(VersionError::into_storage_error)
370    }
371}
372
373impl super::IdentifierAllocator for VersionedKeyValueStore {
374    fn identifier_watermark(&self, namespace: &[u8]) -> StorageBackendResult<Option<u64>> {
375        Self::identifier_watermark(self, namespace)
376    }
377
378    fn allocate_identifiers(
379        &self,
380        namespace: &[u8],
381        request: super::IdentifierRequest,
382    ) -> StorageBackendResult<super::IdentifierAllocation> {
383        Self::allocate_identifiers(self, namespace, request)
384    }
385}
386
387impl KeyValueStore for VersionedKeyValueStore {
388    fn resource_leases(&self) -> Option<&dyn super::ResourceLeaseProvider> {
389        self.persistence.resource_leases()
390    }
391
392    fn retention_control(&self) -> Option<StorageReadControl> {
393        Some(Self::retention_control(self))
394    }
395
396    fn notification_publications(
397        &self,
398    ) -> Option<&dyn crate::notifications::NotificationPublicationStore> {
399        Some(self)
400    }
401
402    fn serializable_session(&self) -> Option<&dyn SerializableSession> {
403        self.persistence
404            .serializable_coordinator()
405            .map(|_| self as &dyn SerializableSession)
406    }
407
408    fn reclaim_obsolete(&self) -> StorageBackendResult<()> {
409        let active = self.active.lock();
410        if active.is_some() {
411            return Err(StorageBackendError::Other(
412                "vacuum requires an inactive logical session".into(),
413            ));
414        }
415        let control = self.write_control();
416        let source: Arc<dyn KeyValueStore> =
417            Arc::new(self.new_session_with_cancellation(control.cancellation()));
418        crate::key_value::KeyValueDiskANNMaintenance::run(&source, &control)?;
419        self.reclaim_versions().map(|_| ())
420    }
421
422    fn write_cancellation(&self) -> Option<uqa_core::CancellationToken> {
423        Some(self.write_cancellation.clone())
424    }
425
426    fn open_controlled_session(
427        &self,
428        control: &StorageReadControl,
429    ) -> StorageBackendResult<Arc<dyn KeyValueStore>> {
430        control.check()?;
431        Ok(Arc::new(self.new_controlled_session(control)))
432    }
433
434    fn open_session_with_cancellation(
435        &self,
436        cancellation: &uqa_core::CancellationToken,
437    ) -> StorageBackendResult<Arc<dyn KeyValueStore>> {
438        cancellation.check()?;
439        Ok(Arc::new(self.new_session_with_cancellation(cancellation)))
440    }
441
442    fn open_retained_read_session(
443        &self,
444        cancellation: &uqa_core::CancellationToken,
445    ) -> StorageBackendResult<Arc<dyn KeyValueStore>> {
446        Ok(Arc::new(self.new_retained_read_session(cancellation)?))
447    }
448
449    fn transaction_model(&self) -> crate::StorageTransactionModel {
450        crate::StorageTransactionModel::VersionedConcurrent {
451            database: self.persistence.database_id(),
452        }
453    }
454
455    fn refresh_transaction_snapshot(
456        &self,
457        cancellation: &uqa_core::CancellationToken,
458    ) -> StorageBackendResult<()> {
459        Self::refresh_transaction_snapshot(self, cancellation)
460    }
461
462    fn identifier_allocator(&self) -> Option<&dyn super::IdentifierAllocator> {
463        Some(self)
464    }
465
466    fn with_read_view(
467        &self,
468        operation: &mut crate::key_value::KeyValueReadScope<'_>,
469    ) -> StorageBackendResult<()> {
470        self.control.check()?;
471        let view = self.view().map_err(VersionError::into_storage_error)?;
472        operation(&read::RecordRead {
473            view: &view,
474            database: self.persistence.database_id(),
475            control: &self.control,
476        })?;
477        self.control.check()
478    }
479
480    fn with_mutation(
481        &self,
482        operation: &mut crate::key_value::KeyValueMutation<'_>,
483    ) -> StorageBackendResult<()> {
484        self.control.check()?;
485        self.write(|transaction| {
486            let view = transaction.view()?;
487            let read = read::RecordRead {
488                view: &view,
489                database: self.persistence.database_id(),
490                control: &self.control,
491            };
492            let mut batch = batch::Batch::new(
493                self,
494                transaction
495                    .serializable_context()
496                    .map(SerializableReadContext::id),
497            );
498            operation(&read, &mut batch)?;
499            self.control.check()?;
500            batch.apply(transaction)
501        })
502    }
503
504    fn with_versioned_mutation(
505        &self,
506        operation: &mut crate::key_value::KeyValueVersionedMutation<'_>,
507    ) -> StorageBackendResult<()> {
508        self.with_versioned_record_mutation(|origin, view, batch| {
509            operation(
510                origin,
511                &read::RecordRead {
512                    view: &view,
513                    database: self.persistence.database_id(),
514                    control: &self.control,
515                },
516                batch,
517            )
518        })
519    }
520
521    fn transaction_affinity(&self) -> Option<StorageSessionAffinity> {
522        Some(self.session_affinity())
523    }
524
525    fn storage_identity(&self) -> StorageBackendResult<Option<PersistentStorageIdentity>> {
526        Ok(self.identity.clone())
527    }
528
529    fn open_session(&self) -> StorageBackendResult<Arc<dyn KeyValueStore>> {
530        Ok(Arc::new(self.new_session()))
531    }
532
533    fn get(&self, key: &[u8]) -> StorageBackendResult<Option<Vec<u8>>> {
534        let mut value = None;
535        self.visit_value(key, &self.control, &mut |bytes| {
536            value = bytes.map(<[u8]>::to_vec);
537            Ok(())
538        })?;
539        Ok(value)
540    }
541
542    fn contains_key(&self, key: &[u8]) -> StorageBackendResult<bool> {
543        Ok(self
544            .view()
545            .and_then(|view| view.metadata(key, &self.control))
546            .map_err(VersionError::into_storage_error)?
547            .is_some_and(|record| record.live))
548    }
549
550    fn contains_prefix_budgeted(
551        &self,
552        prefix: &[u8],
553        control: &StorageReadControl,
554    ) -> StorageBackendResult<bool> {
555        control.check()?;
556        let mut found = false;
557        self.view()
558            .map_err(VersionError::into_storage_error)?
559            .visit_keys(prefix, None, usize::MAX, control, &mut |_, record| {
560                found = record.live;
561                Ok(!found)
562            })
563            .map_err(VersionError::into_storage_error)?;
564        Ok(found)
565    }
566
567    fn visit_value(
568        &self,
569        key: &[u8],
570        control: &StorageReadControl,
571        visit: &mut ValueReadVisitor<'_>,
572    ) -> StorageBackendResult<()> {
573        control.check()?;
574        self.view()
575            .map_err(VersionError::into_storage_error)?
576            .visit_value(key, control, &mut |record| {
577                visit(record.and_then(|record| record.value)).map_err(VersionError::from)
578            })
579            .map_err(VersionError::into_storage_error)
580    }
581
582    fn visit_value_bounded(
583        &self,
584        key: &[u8],
585        max_bytes: usize,
586        control: &StorageReadControl,
587        visit: &mut ValueReadVisitor<'_>,
588    ) -> StorageBackendResult<()> {
589        control.check()?;
590        self.view()
591            .map_err(VersionError::into_storage_error)?
592            .visit_value_bounded(key, max_bytes, control, &mut |record| {
593                visit(record.and_then(|record| record.value)).map_err(VersionError::from)
594            })
595            .map_err(VersionError::into_storage_error)
596    }
597
598    fn visit_prefix_after(
599        &self,
600        prefix: &[u8],
601        after: Option<&[u8]>,
602        limit: usize,
603        control: &StorageReadControl,
604        visit: &mut KeyValueReadVisitor<'_>,
605    ) -> StorageBackendResult<()> {
606        control.check()?;
607        if limit == 0 {
608            return Ok(());
609        }
610        let view = self.view().map_err(VersionError::into_storage_error)?;
611        read::visit_live(&view, prefix, after, limit, control, visit)
612            .map_err(VersionError::into_storage_error)
613    }
614
615    fn put(&self, key: &[u8], value: &[u8]) -> StorageBackendResult<()> {
616        self.write(|transaction| transaction.replace(key, Some(value), &self.control))
617    }
618    fn delete(&self, key: &[u8]) -> StorageBackendResult<()> {
619        self.write(|transaction| transaction.replace(key, None, &self.control))
620    }
621
622    fn scan_prefix(&self, prefix: &[u8]) -> StorageBackendResult<Vec<(Vec<u8>, Vec<u8>)>> {
623        self.scan_prefix_after(prefix, None, usize::MAX)
624    }
625
626    fn scan_prefix_after(
627        &self,
628        prefix: &[u8],
629        after: Option<&[u8]>,
630        limit: usize,
631    ) -> StorageBackendResult<Vec<(Vec<u8>, Vec<u8>)>> {
632        let mut rows = Vec::new();
633        self.visit_prefix_after(prefix, after, limit, &self.control, &mut |key, value| {
634            rows.push((key.to_vec(), value.to_vec()));
635            Ok(())
636        })?;
637        Ok(rows)
638    }
639
640    fn scan_prefix_keys_after(
641        &self,
642        prefix: &[u8],
643        after: Option<&[u8]>,
644        limit: usize,
645    ) -> StorageBackendResult<Vec<Vec<u8>>> {
646        let mut keys = Vec::new();
647        if limit != 0 {
648            self.view()
649                .map_err(VersionError::into_storage_error)?
650                .visit_keys(
651                    prefix,
652                    after,
653                    usize::MAX,
654                    &self.control,
655                    &mut |key, record| {
656                        if record.live {
657                            keys.push(key.to_vec());
658                        }
659                        Ok(keys.len() < limit)
660                    },
661                )
662                .map_err(VersionError::into_storage_error)?;
663        }
664        Ok(keys)
665    }
666
667    fn first_prefix_after(
668        &self,
669        prefix: &[u8],
670        after: Option<&[u8]>,
671    ) -> StorageBackendResult<Option<(Vec<u8>, Vec<u8>)>> {
672        Ok(self.scan_prefix_after(prefix, after, 1)?.pop())
673    }
674    fn delete_prefix(&self, prefix: &[u8]) -> StorageBackendResult<usize> {
675        self.write(|transaction| transaction.delete_prefix(prefix, &self.control))
676    }
677    fn batch(&self) -> Box<dyn KeyValueBatch + '_> {
678        let participant = self
679            .active
680            .lock()
681            .as_ref()
682            .and_then(Transaction::serializable_context)
683            .map(SerializableReadContext::id);
684        Box::new(batch::Batch::new(self, participant))
685    }
686    fn begin_transaction(&self) -> StorageBackendResult<()> {
687        self.begin(false)
688    }
689    fn begin_read_transaction(&self) -> StorageBackendResult<()> {
690        self.begin(true)
691    }
692    fn begin_upgradeable_transaction(&self) -> StorageBackendResult<()> {
693        self.begin(self.retained.is_some())
694    }
695    fn in_transaction(&self) -> bool {
696        self.active.lock().is_some()
697    }
698    fn transaction_has_written(&self) -> StorageBackendResult<bool> {
699        Ok(self.active.lock().as_ref().is_some_and(|transaction| {
700            transaction.changes.has_written() || transaction.has_derived_changes()
701        }))
702    }
703    fn change_version(&self) -> StorageBackendResult<Option<u64>> {
704        Ok(Some(
705            self.view()
706                .map_err(VersionError::into_storage_error)?
707                .sequence()
708                .as_u64(),
709        ))
710    }
711
712    fn commit_transaction(&self) -> StorageBackendResult<()> {
713        let mut active = self.active.lock();
714        active
715            .as_mut()
716            .ok_or_else(no_transaction)?
717            .commit(&*self.persistence, &self.write_control())?;
718        active
719            .as_ref()
720            .expect("completed attempt")
721            .acknowledge_completion(&*self.persistence, &self.control)?;
722        *active = None;
723        Ok(())
724    }
725
726    fn rollback_transaction(&self) -> StorageBackendResult<()> {
727        let mut active = self.active.lock();
728        active
729            .as_mut()
730            .ok_or_else(no_transaction)?
731            .abort(&*self.persistence, &self.control)?;
732        active
733            .as_ref()
734            .expect("completed attempt")
735            .acknowledge_completion(&*self.persistence, &self.control)?;
736        *active = None;
737        Ok(())
738    }
739
740    fn savepoint(&self, name: &str) -> StorageBackendResult<()> {
741        self.savepoint_action(|transaction| transaction.savepoint(name, &self.control))
742    }
743    fn release_savepoint(&self, name: &str) -> StorageBackendResult<()> {
744        self.savepoint_action(|transaction| transaction.release(name))
745    }
746    fn rollback_to_savepoint(&self, name: &str) -> StorageBackendResult<()> {
747        self.savepoint_action(|transaction| transaction.rollback_to(name, &self.control))
748    }
749}
750
751fn no_transaction() -> StorageBackendError {
752    StorageBackendError::Other("no active KeyValue transaction".into())
753}
754
755fn commit_error(error: CommitFailure) -> StorageBackendError {
756    match error {
757        CommitFailure::Rejected(error) => error.into_storage_error(),
758        error @ CommitFailure::Indeterminate { .. } => StorageBackendError::backend("MVCC", error),
759    }
760}
uqa_storage/mvcc/session/mod.rs

uqa_storage/mvcc/session/
mod.rs
