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

uqa_storage/mvcc/session/
mod.rs
