uqa_storage/mvcc/session/
sequence_values.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Sequence consumers retain durability independently of savepoint undo. Publication admission closes before the owner's completion barrier, so a retained child cannot publish an unsynchronized suffix after its consumer completes.
8
9use super::{
10    Arc, StorageBackendError, StorageBackendResult, StorageReadControl, VersionError,
11    VersionedKeyValueStore,
12};
13use parking_lot::{RwLock, RwLockWriteGuard};
14use std::sync::atomic::{AtomicBool, Ordering};
15
16pub(super) struct SequenceDurability {
17    state: RwLock<SequenceBoundary>,
18    required: AtomicBool,
19}
20
21pub(super) struct SequenceBoundary {
22    generation: u64,
23    open: bool,
24}
25
26impl Default for SequenceDurability {
27    fn default() -> Self {
28        Self {
29            state: RwLock::new(SequenceBoundary {
30                generation: 0,
31                open: false,
32            }),
33            required: AtomicBool::new(false),
34        }
35    }
36}
37
38impl SequenceDurability {
39    pub(super) fn begin(&self) -> StorageBackendResult<()> {
40        let mut state = self.state.write();
41        state.generation = state.generation.checked_add(1).ok_or_else(|| {
42            StorageBackendError::Other("sequence completion generations exhausted".into())
43        })?;
44        state.open = true;
45        self.required.store(false, Ordering::Release);
46        Ok(())
47    }
48
49    pub(super) fn generation(&self) -> u64 {
50        self.state.read().generation
51    }
52
53    pub(super) fn close(&self) -> RwLockWriteGuard<'_, SequenceBoundary> {
54        let mut state = self.state.write();
55        state.open = false;
56        state
57    }
58
59    pub(super) fn clear(&self) {
60        self.required.store(false, Ordering::Release);
61    }
62
63    fn lease(self: &Arc<Self>, expected: Option<u64>) -> Option<SequencePublicationLease> {
64        let state = self.state.read();
65        if !state.open || expected.is_some_and(|generation| generation != state.generation) {
66            return None;
67        }
68        self.required.store(true, Ordering::Release);
69        Some(SequencePublicationLease {
70            owner: Arc::clone(self),
71            generation: state.generation,
72        })
73    }
74}
75
76/// A sequence publisher's exact consuming transaction. Its admission lasts through physical publication; owner completion closes admission and waits for every admitted publication before flushing. A child retained past completion or reused after another transaction begins must publish synchronously.
77#[derive(Clone)]
78pub struct SequencePublicationLease {
79    owner: Arc<SequenceDurability>,
80    generation: u64,
81}
82
83impl SequencePublicationLease {
84    /// Hold publication admission while `operation` runs. `true` permits sequence-only WAL deferral; `false` requires the ordinary synchronous path. Acquire this before physical writer admission or connection checkout.
85    pub fn with_publication<T>(&self, operation: impl FnOnce(bool) -> T) -> T {
86        let state = self.owner.state.read();
87        operation(state.open && state.generation == self.generation)
88    }
89}
90
91impl VersionedKeyValueStore {
92    /// Select autonomous sequence publication only while its consuming transaction remains open. Unsupported providers and calls without an active completion owner retain synchronous behavior.
93    pub fn new_sequence_value_session(&self, cancellation: &uqa_core::CancellationToken) -> Self {
94        let persistence = self
95            .sequence_durability
96            .lease(self.sequence_generation)
97            .and_then(|owner| self.persistence.sequence_value_persistence(owner))
98            .unwrap_or_else(|| Arc::clone(&self.persistence));
99        Self::new_with_cancellation(
100            persistence,
101            self.identity.clone(),
102            self.options,
103            cancellation.clone(),
104        )
105    }
106
107    /// Sequence values can outlive transaction and savepoint undo; retain their original consumer's completion obligation, including through a retained reader.
108    pub fn require_sequence_value_durability(&self) -> StorageBackendResult<()> {
109        self.sequence_durability
110            .lease(self.sequence_generation)
111            .map(|_| ())
112            .ok_or_else(|| {
113                StorageBackendError::Other(
114                    "sequence values require an open consuming transaction".into(),
115                )
116            })
117    }
118
119    pub(super) fn synchronize_sequence_values(
120        &self,
121        control: &StorageReadControl,
122    ) -> StorageBackendResult<()> {
123        if self.retained.is_none() && self.sequence_durability.required.load(Ordering::Acquire) {
124            self.persistence
125                .synchronize_sequence_values(control)
126                .map_err(VersionError::into_storage_error)?;
127        }
128        Ok(())
129    }
130}
131
132impl Drop for VersionedKeyValueStore {
133    fn drop(&mut self) {
134        if self.retained.is_none() {
135            // An abandoned consumer has no successful completion to acknowledge. Its surviving children must nevertheless stop deferring new publications.
136            drop(self.sequence_durability.close());
137        }
138    }
139}
uqa_storage/mvcc/session/sequence_values.rs

uqa_storage/mvcc/session/
sequence_values.rs
