uqa_storage/mvcc/memory/
reclamation.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7use std::ops::Bound::{Excluded, Included, Unbounded};
8
9use uqa_core::memory::BudgetedVec;
10
11use crate::mvcc::{
12    TombstoneReclamationRequest, TombstoneReclamationStep, VersionError, VersionResult,
13    TOMBSTONE_RECLAMATION_PAGE,
14};
15use crate::read_control::StorageReadControl;
16
17use super::{MemoryReservation, MemoryVersionStore, RecordKey};
18
19impl MemoryVersionStore {
20    /// Reference implementation of atomic prefix retirement, sharing snapshot and write admission. Every allocation completes before published state is changed.
21    pub fn reclaim_tombstones(
22        &self,
23        request: &TombstoneReclamationRequest<'_>,
24        control: &StorageReadControl,
25    ) -> VersionResult<TombstoneReclamationStep> {
26        request.validate(control)?;
27        let mut state = self.database.state.lock();
28        if !state.snapshots.is_empty() {
29            return Ok(TombstoneReclamationStep::Retained);
30        }
31        if request.through > state.sequence {
32            return Err(VersionError::InvalidEncoding(
33                "tombstone cutoff exceeds committed visibility",
34            ));
35        }
36        let mut keys = BudgetedVec::new(control.memory());
37        keys.reserve(TOMBSTONE_RECLAMATION_PAGE)?;
38        let mut inspected = 0;
39        let mut after = BudgetedVec::new(control.memory());
40        let start = request.after.map_or(Included(request.prefix), Excluded);
41        for (key, entry) in state.records.range::<[u8], _>((start, Unbounded)) {
42            control.check()?;
43            if !key.bytes().starts_with(request.prefix) {
44                break;
45            }
46            let head = entry
47                .history
48                .head()
49                .ok_or(VersionError::InvalidEncoding("record head has no history"))?;
50            if head.sequence() > request.through {
51                continue;
52            }
53            inspected += 1;
54            after.clear();
55            after.extend_from_slice(key.bytes())?;
56            if head.value().is_none() && entry.history.len() == 1 {
57                keys.push(key.clone())?;
58            }
59            if inspected == TOMBSTONE_RECLAMATION_PAGE {
60                break;
61            }
62        }
63        let removed = keys.len();
64        if removed != 0 {
65            let epoch = state
66                .reclamation_epoch
67                .checked_add(1)
68                .ok_or(VersionError::IdentifiersExhausted)?;
69            let domain = if state.reclamation_domains.contains_key(request.prefix) {
70                None
71            } else {
72                Some((
73                    RecordKey::new(request.prefix, &self.database.memory)?,
74                    self.database
75                        .memory
76                        .reserve(std::mem::size_of::<(RecordKey, (u64, MemoryReservation))>())?,
77                ))
78            };
79            control.check()?;
80            if let Some((prefix, memory)) = domain {
81                state.reclamation_domains.insert(prefix, (epoch, memory));
82            } else {
83                state
84                    .reclamation_domains
85                    .get_mut(request.prefix)
86                    .expect("existing domain")
87                    .0 = epoch;
88            }
89            state.reclamation_epoch = epoch;
90            for key in keys.iter() {
91                state.records.remove(key.bytes());
92            }
93        }
94        Ok(if inspected == TOMBSTONE_RECLAMATION_PAGE {
95            TombstoneReclamationStep::More { after, removed }
96        } else {
97            TombstoneReclamationStep::Complete { removed }
98        })
99    }
100}
uqa_storage/mvcc/memory/reclamation.rs

uqa_storage/mvcc/memory/
reclamation.rs
