uqa_storage/mvcc/session/
serializable.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Original participant attribution shared by session and retained record views.
8
9use std::sync::Arc;
10
11use crate::mvcc::{
12    SafeSnapshot, SerializableGraph, SerializableParticipant, SerializablePredicate,
13    SerializableTransactionId, VersionError, VersionResult, VersionedPersistence,
14};
15use crate::read_control::StorageReadControl;
16use crate::StorageBackendResult;
17
18/// Logical transaction characteristics, independent of whether the physical session permits private writes. Deferrable waiting applies only to a read-only participant.
19#[derive(Clone, Copy, Debug, Default)]
20pub struct SerializableSnapshotOptions {
21    pub read_only: bool,
22    pub deferrable: bool,
23}
24
25/// Scope one candidate snapshot capture together with the caller's publication baseline. Invoke the supplied operation exactly once, without executing application work. Common storage may request another candidate before returning a safe deferrable snapshot; each call must release its guards before returning so overlapping writers can finish during the wait. An already admitted session retains its original participant without invoking this scope.
26pub type SerializableSnapshotCapture<'a> =
27    dyn FnMut(&mut dyn FnMut() -> VersionResult<()>) -> VersionResult<()> + 'a;
28
29/// Logical SSI capabilities of one original or retained storage session. Providers forward this boundary without deriving predicates from physical record keys. SQL must cover its access paths before enabling automatic admission.
30pub trait SerializableSession: Send + Sync {
31    fn establish_serializable_snapshot(&self) -> StorageBackendResult<SerializableReadContext>;
32    /// Select the first snapshot using logical characteristics and a caller-scoped capture. Existing participants retain their original classification without another capture. Only read-only deferrable admission waits for a proven safe snapshot.
33    fn establish_serializable_snapshot_with(
34        &self,
35        options: SerializableSnapshotOptions,
36        capture: &mut SerializableSnapshotCapture<'_>,
37    ) -> StorageBackendResult<SerializableReadContext>;
38    fn serializable_read_context(&self) -> StorageBackendResult<Option<SerializableReadContext>>;
39    fn observe_serializable_write(
40        &self,
41        predicate: SerializablePredicate<'_>,
42    ) -> StorageBackendResult<()>;
43}
44
45impl SerializableSession for super::VersionedKeyValueStore {
46    fn establish_serializable_snapshot(&self) -> StorageBackendResult<SerializableReadContext> {
47        Self::establish_serializable_snapshot(self)
48    }
49
50    fn establish_serializable_snapshot_with(
51        &self,
52        options: SerializableSnapshotOptions,
53        capture: &mut SerializableSnapshotCapture<'_>,
54    ) -> StorageBackendResult<SerializableReadContext> {
55        self.require_mutable_session()?;
56        self.active
57            .lock()
58            .as_mut()
59            .ok_or_else(super::no_transaction)?
60            .establish_serializable_with(&self.persistence, options, capture, &self.write_control())
61            .map_err(VersionError::into_storage_error)
62    }
63
64    fn serializable_read_context(&self) -> StorageBackendResult<Option<SerializableReadContext>> {
65        if let Some(view) = &self.retained {
66            return Ok(view.serializable().cloned());
67        }
68        let active = self.active.lock();
69        let Some(transaction) = active.as_ref() else {
70            return Ok(None);
71        };
72        transaction
73            .unsealed()
74            .map_err(VersionError::into_storage_error)?;
75        Ok(transaction.serializable_context().cloned())
76    }
77
78    fn observe_serializable_write(
79        &self,
80        predicate: SerializablePredicate<'_>,
81    ) -> StorageBackendResult<()> {
82        self.require_mutable_session()?;
83        Self::observe_serializable_write(self, predicate)
84    }
85}
86
87/// A retained reader attributes observations to its original transaction, even after session refresh or nested view cloning. This does not create another participant or give the reader ownership of transaction completion.
88#[derive(Clone)]
89pub struct SerializableReadContext {
90    pub(super) persistence: Arc<dyn VersionedPersistence>,
91    pub(super) participant: SerializableParticipant,
92    pub(super) memory: uqa_core::memory::MemoryBudget,
93    pub(super) safe: bool,
94}
95
96impl SerializableReadContext {
97    pub fn id(&self) -> SerializableTransactionId {
98        self.participant.id()
99    }
100
101    /// Retain the original session's allowance while observing cancellation from the invoking reader. Cloning a context or attaching a nested reader does not grant another observation budget.
102    pub fn read_control(&self, cancellation: &uqa_core::CancellationToken) -> StorageReadControl {
103        StorageReadControl::new(&self.memory, cancellation)
104    }
105
106    /// Observe the logical predicate before exposing its result, including an empty, cached or index-only result. Proven safe snapshots need only the original participant's lifetime check. Physical record keys are not a substitute for logical object/row/index identities.
107    pub fn observe_read(
108        &self,
109        predicate: SerializablePredicate<'_>,
110        control: &StorageReadControl,
111    ) -> VersionResult<()> {
112        self.with_graph(control, |graph| {
113            if self.safe {
114                // A proven safe snapshot cannot contribute to a serialization anomaly. Retained readers still validate the original participant's lifetime.
115                graph.check_active(self.id())
116            } else {
117                graph.observe_read(self.id(), predicate, control)
118            }
119        })
120    }
121
122    pub fn safe_snapshot(&self, control: &StorageReadControl) -> VersionResult<SafeSnapshot> {
123        self.with_graph(control, |graph| graph.safe_snapshot(self.id(), control))
124    }
125
126    pub(super) fn with_graph<T>(
127        &self,
128        control: &StorageReadControl,
129        operation: impl FnOnce(&mut SerializableGraph) -> VersionResult<T>,
130    ) -> VersionResult<T> {
131        let coordinator =
132            self.persistence
133                .serializable_coordinator()
134                .ok_or(VersionError::InvalidEncoding(
135                    "serializable session lost its coordinator",
136                ))?;
137        let mut operation = Some(operation);
138        let mut result = None;
139        coordinator.with_serializable_admission(control, &mut |graph, leases| {
140            let operation = operation.take().ok_or(VersionError::InvalidEncoding(
141                "serializable session operation was replayed",
142            ))?;
143            crate::mvcc::serializable::recover(coordinator, graph, leases, control)?;
144            result = Some(operation(graph)?);
145            Ok(())
146        })?;
147        result.ok_or(VersionError::InvalidEncoding(
148            "serializable session operation was not invoked",
149        ))
150    }
151}
uqa_storage/mvcc/session/serializable.rs

uqa_storage/mvcc/session/
serializable.rs
