reddb_engine/transaction/
read.rs

1//! Explicit snapshot visibility for entity reads. Hosts retain auth, tenant,
2//! moderation and ambient statement-frame policy around these helpers.
3use std::sync::Arc;
4
5/// Snapshot + manager pair used for read-path visibility checks.
6///
7/// The manager is needed in addition to the snapshot because `aborted`
8/// state mutates after the snapshot is captured — a ROLLBACK by a
9/// committed-at-capture-time writer must still hide its tuples. Keeping
10/// the Arc around is O(pointer) and the RwLock reads on `is_aborted`
11/// are cheap (HashSet lookup under a parking_lot read guard).
12///
13/// `own_xids` (Phase 2.3.2e) lists the xids belonging to the current
14/// connection's transaction — the parent xid plus open and released
15/// savepoint sub-xids. The visibility rule promotes rows stamped with
16/// these xids to "always visible (unless aborted)" so the writer sees
17/// its own nested-savepoint writes even though their xids exceed
18/// `snapshot.xid`.
19#[derive(Clone)]
20pub struct SnapshotContext {
21    pub snapshot: super::snapshot::Snapshot,
22    pub manager: Arc<super::snapshot::SnapshotManager>,
23    pub own_xids: std::collections::HashSet<super::snapshot::Xid>,
24    pub requires_index_fallback: bool,
25    pub serializable_reader: Option<super::snapshot::Xid>,
26}
27
28/// Record an admitted table-row read after the host has applied visibility and
29/// policy gates. SSI keys use logical identity, shared by all physical versions.
30pub fn record_serializable_read(
31    ctx: &SnapshotContext,
32    entity: &crate::storage::entity::UnifiedEntity,
33) {
34    let Some(reader) = ctx.serializable_reader else {
35        return;
36    };
37    if !matches!(
38        &entity.kind,
39        crate::storage::entity::EntityKind::TableRow { .. }
40    ) {
41        return;
42    }
43    ctx.manager
44        .record_serializable_read(reader, entity.kind.collection(), entity.logical_id());
45}
46
47/// Apply captured MVCC visibility, including own/aborted savepoint writers.
48/// This predicate supplies no authorization, tenant or moderation policy.
49#[inline]
50pub fn visibility_check(ctx: &SnapshotContext, xmin: u64, xmax: u64) -> bool {
51    // Writer aborted → tuple never existed from any future reader's view.
52    // Checked *before* the own-xids fast path so an aborted own-sub-xid
53    // (rolled-back savepoint) stays hidden from the parent.
54    if xmin != 0 && ctx.manager.is_aborted(xmin) {
55        return false;
56    }
57    // Deleter aborted → treat xmax as unset; fall back to xmin-only check.
58    let effective_xmax = if xmax != 0 && ctx.manager.is_aborted(xmax) {
59        0
60    } else {
61        xmax
62    };
63    // Phase 2.3.2e: own-tx writes are always visible to the connection
64    // that stamped them, even when xmin/xmax exceed `snapshot.xid` (as
65    // happens for sub-xids allocated by SAVEPOINT after BEGIN).
66    let own_xmin = xmin != 0 && ctx.own_xids.contains(&xmin);
67    let own_xmax = effective_xmax != 0 && ctx.own_xids.contains(&effective_xmax);
68    if own_xmax {
69        // This connection deleted the row via this xid — hide it from self.
70        return false;
71    }
72    if own_xmin {
73        return true;
74    }
75    ctx.snapshot.sees(xmin, effective_xmax)
76}
77
78/// Candidate access for the canonical table-row version selector. The store
79/// supplies only physical versions of the requested collection/logical identity.
80pub trait TableRowVersionStore {
81    fn table_row_versions_by_logical_id(
82        &self,
83        collection: &str,
84        logical_id: crate::storage::entity::EntityId,
85    ) -> Vec<crate::storage::entity::UnifiedEntity>;
86
87    fn get_table_row_by_logical_id(
88        &self,
89        collection: &str,
90        logical_id: crate::storage::entity::EntityId,
91    ) -> Option<crate::storage::entity::UnifiedEntity>;
92}
93
94/// Select the newest visible physical version, or use the current-row lookup
95/// when the host has no snapshot requiring history. The host's predicate retains
96/// table-kind, authorization and moderation gates before materialization.
97pub fn resolve_logical_id(
98    store: &impl TableRowVersionStore,
99    collection: &str,
100    logical_id: crate::storage::entity::EntityId,
101    select_versions: bool,
102    mut visible: impl FnMut(&crate::storage::entity::UnifiedEntity) -> bool,
103) -> Option<crate::storage::entity::UnifiedEntity> {
104    if select_versions {
105        return store
106            .table_row_versions_by_logical_id(collection, logical_id)
107            .into_iter()
108            .filter(|entity| visible(entity))
109            .max_by_key(|entity| entity.xmin);
110    }
111    store
112        .get_table_row_by_logical_id(collection, logical_id)
113        .filter(|entity| visible(entity))
114}
115
116/// Fallible version access streams candidates instead of retaining a history Vec.
117/// Hosts supply only versions of the requested collection/logical identity.
118pub trait FallibleTableRowVersionStore {
119    type Error;
120
121    fn try_get_table_row_by_logical_id(
122        &self,
123        collection: &str,
124        logical_id: crate::storage::entity::EntityId,
125    ) -> Result<Option<crate::storage::entity::UnifiedEntity>, Self::Error>;
126
127    fn try_for_each_table_row_version(
128        &self,
129        collection: &str,
130        logical_id: crate::storage::entity::EntityId,
131        visit: impl FnMut(&crate::storage::entity::UnifiedEntity) -> bool,
132    ) -> Result<(), Self::Error>;
133}
134
135/// Retain one selected version; a later read failure invalidates the selection.
136pub fn try_resolve_logical_id<S: FallibleTableRowVersionStore>(
137    store: &S,
138    collection: &str,
139    logical_id: crate::storage::entity::EntityId,
140    select_versions: bool,
141    mut visible: impl FnMut(&crate::storage::entity::UnifiedEntity) -> bool,
142) -> Result<Option<crate::storage::entity::UnifiedEntity>, S::Error> {
143    if !select_versions {
144        return store
145            .try_get_table_row_by_logical_id(collection, logical_id)
146            .map(|entity| entity.filter(|entity| visible(entity)));
147    }
148    let mut selected: Option<crate::storage::entity::UnifiedEntity> = None;
149    store.try_for_each_table_row_version(collection, logical_id, |entity| {
150        if visible(entity) && selected.as_ref().is_none_or(|old| entity.xmin >= old.xmin) {
151            selected = Some(entity.clone());
152        }
153        true
154    })?;
155    Ok(selected)
156}
reddb_engine/transaction/read.rs

reddb_engine/transaction/
read.rs
