uqa_storage/mvcc/session/
vector_fields.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Empty vector fields retire reference markers without changing their lifetime observations.
8
9use std::sync::Arc;
10
11use uqa_core::memory::{BudgetedVec, MemoryReservation};
12
13use super::VersionedKeyValueStore;
14use crate::mvcc::{CommittedRecordSnapshot, VectorFieldGuard, VersionError};
15use crate::read_control::StorageReadControl;
16use crate::{KeyValueStore, StorageBackendError, StorageBackendResult};
17
18/// A finite key-only discovery pass. Each step revisits one current field and conditionally deletes its reference marker. Retained readers keep the old records; ordinary document writers retain their lifetime observation and mergeable markers. Uncertain commits keep the current key and original attempt until explicit completion.
19pub struct VectorFieldGuardMaintenance {
20    writer: VersionedKeyValueStore,
21    read: Option<Arc<dyn CommittedRecordSnapshot>>,
22    prefix: BudgetedVec<u8>,
23    after: BudgetedVec<u8>,
24    current: BudgetedVec<u8>,
25    guard: Option<VectorFieldGuard>,
26    completed: Option<bool>,
27    control: StorageReadControl,
28    _memory: MemoryReservation,
29}
30
31impl VectorFieldGuardMaintenance {
32    pub fn start(
33        store: &VersionedKeyValueStore,
34        control: &StorageReadControl,
35    ) -> StorageBackendResult<Self> {
36        control.check()?;
37        let memory = control.memory().reserve(std::mem::size_of::<Self>())?;
38        let prefix = store
39            .persistence
40            .vector_field_guard_layout()
41            .prefix(control)
42            .map_err(VersionError::into_storage_error)?;
43        let read = store
44            .persistence
45            .snapshot(control)
46            .map_err(VersionError::into_storage_error)?;
47        Ok(Self {
48            writer: store.new_controlled_session(control),
49            read: Some(read),
50            prefix,
51            after: BudgetedVec::new(control.memory()),
52            current: BudgetedVec::new(control.memory()),
53            guard: None,
54            completed: None,
55            control: control.clone(),
56            _memory: memory,
57        })
58    }
59
60    /// `None` completes discovery. `Some(true)` deletes the current empty field's live marker; `Some(false)` leaves a live field, a tombstone, or a racing writer for later maintenance.
61    pub fn step(&mut self) -> StorageBackendResult<Option<bool>> {
62        self.control.check()?;
63        if self.writer.in_transaction() {
64            return Err(VersionError::InvalidEncoding(
65                "resolve the original field-guard maintenance attempt before advancing",
66            )
67            .into_storage_error());
68        }
69        if self.current.is_empty() {
70            let Some(read) = &self.read else {
71                return Ok(None);
72            };
73            let mut selected = BudgetedVec::new(self.control.memory());
74            let mut guard = None;
75            read.visit_keys(
76                &self.prefix,
77                (!self.after.is_empty()).then_some(&*self.after),
78                1,
79                &self.control,
80                &mut |key, metadata| {
81                    selected.extend_from_slice(key)?;
82                    if metadata.live {
83                        guard = self
84                            .writer
85                            .persistence
86                            .vector_field_guard_layout()
87                            .reference(key, &self.control)?;
88                    }
89                    Ok(false)
90                },
91            )
92            .map_err(VersionError::into_storage_error)?;
93            self.control.check()?;
94            self.current = selected;
95            self.guard = guard;
96            if self.current.is_empty() {
97                self.read = None;
98                return Ok(None);
99            }
100        }
101        let mut removed = false;
102        if let Some(guard) = self.guard.as_ref().filter(|_| self.completed.is_none()) {
103            let result = self.writer.with_mutation(&mut |read, batch| {
104                let mut present = false;
105                read.visit_value_budgeted(&guard.references, &self.control, &mut |value| {
106                    if let Some(value) = value {
107                        if value != &*guard.reference_value {
108                            return Err(VersionError::InvalidEncoding(
109                                "vector field reference marker payload changed",
110                            )
111                            .into_storage_error());
112                        }
113                        present = true;
114                    }
115                    Ok(())
116                })?;
117                if present && !read.contains_prefix_budgeted(&guard.vectors, &self.control)? {
118                    read.visit_value_budgeted(&guard.lifetime, &self.control, &mut |value| {
119                        if value.is_some() {
120                            return Err(VersionError::InvalidEncoding(
121                                "vector field lifetime guard has a live payload",
122                            )
123                            .into_storage_error());
124                        }
125                        Ok(())
126                    })?;
127                    // Cleanup changes no canonical vectors, so it must not manufacture a structural lifetime transition that rejects an ordinary writer.
128                    batch.require_unchanged(&guard.lifetime)?;
129                    batch.delete(&guard.references)?;
130                    removed = true;
131                }
132                Ok(())
133            });
134            match result {
135                Err(StorageBackendError::Backend { source, .. })
136                    if matches!(
137                        source.downcast_ref::<VersionError>(),
138                        Some(
139                            VersionError::WriteConflict { .. } | VersionError::ReadConflict { .. }
140                        )
141                    ) =>
142                {
143                    self.writer.rollback_transaction()?;
144                    removed = false;
145                }
146                result => result?,
147            }
148            self.completed = Some(removed);
149        }
150        let removed = self.completed.unwrap_or(false);
151        self.after.clear();
152        self.after.extend_from_slice(&self.current)?;
153        self.current.clear();
154        self.guard = None;
155        self.completed = None;
156        Ok(Some(removed))
157    }
158
159    /// Resolve the retained original commit; the next step advances the cursor without reevaluating the completed field.
160    pub fn commit_pending(&mut self) -> StorageBackendResult<()> {
161        self.writer.commit_transaction()?;
162        self.completed = Some(true);
163        Ok(())
164    }
165
166    pub fn rollback_pending(&self) -> StorageBackendResult<()> {
167        self.writer.rollback_transaction()
168    }
169
170    pub fn run(
171        store: &VersionedKeyValueStore,
172        control: &StorageReadControl,
173    ) -> StorageBackendResult<()> {
174        let mut pass = Self::start(store, control)?;
175        while pass.step()?.is_some() {}
176        Ok(())
177    }
178}
uqa_storage/mvcc/session/vector_fields.rs

uqa_storage/mvcc/session/
vector_fields.rs
