Skip to main content

uqa_storage/mvcc/
overlay.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Transaction-private evaluated replacements, retained command views and undo.
8
9use std::num::NonZeroU64;
10use std::sync::atomic::{AtomicU64, Ordering};
11use std::sync::Arc;
12
13use parking_lot::Mutex;
14use triomphe::Arc as StrongArc;
15use uqa_core::memory::{BudgetedSharedMap, BudgetedVec, MemoryBudget, MemoryReservation};
16
17use crate::read_control::StorageReadControl;
18use crate::StorageSavepointId;
19
20use super::key::RecordKey;
21use super::{PreparedRecordCommit, PreparedRecordWrite, RecordWrite, VersionError, VersionResult};
22
23mod cursor;
24pub use cursor::{PrivateRecordCursor, PrivateRecordEntry};
25mod selection;
26pub(super) use selection::PrivateRecordSelection;
27mod retained;
28mod scopes;
29pub use scopes::PrivateRevisionScope;
30use scopes::RevisionScopes;
31pub(in crate::mvcc) mod run;
32mod run_set;
33mod tiers;
34use retained::{RetainedSources, Sources};
35use run::SpilledRunWriter;
36use run_set::RunSet;
37use tiers::{TieredChange, TieredCursor};
38
39/// The resident bytes of a memory-tier change beyond its key and value, as an estimate: its tree node and entry, the shared key and value handles and their allocation headers.
40const CHANGE_OVERHEAD: usize = 192;
41/// The memory tier spills once the transaction's allowance is more than this fraction used.
42const PRESSURE_DIVISOR: usize = 2;
43/// The memory tier spills only once it holds at least this fraction of the allowance, so that memory held elsewhere does not spill a run for every write.
44const RESIDENT_DIVISOR: usize = 16;
45
46struct Change {
47    write: PreparedRecordWrite,
48    identity: PrivateRecordRevision,
49}
50
51type Records = BudgetedSharedMap<RecordKey, Change>;
52
53/// Process-local identity of an evaluated private batch. Identities are never reused across transactions or undo branches; they are not durable commit sequences.
54#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
55pub struct PrivateRecordRevision(NonZeroU64);
56
57impl PrivateRecordRevision {
58    pub(in crate::mvcc) fn allocate() -> VersionResult<Self> {
59        static NEXT: AtomicU64 = AtomicU64::new(1);
60        uqa_core::atomic::try_update_u64(&NEXT, Ordering::Relaxed, Ordering::Relaxed, |id| {
61            id.checked_add(1)
62        })
63        .map(|id| Self(NonZeroU64::new(id).expect("private revisions start at one")))
64        .map_err(|_| VersionError::PrivateRevisionExhausted)
65    }
66
67    #[cfg(test)]
68    pub(crate) fn for_tests() -> Self {
69        Self::allocate().expect("test private revision")
70    }
71
72    pub fn as_u64(self) -> u64 {
73        self.0.get()
74    }
75
76    /// Decode an identity written to a spill file of this process.
77    fn from_u64(value: u64) -> VersionResult<Self> {
78        NonZeroU64::new(value)
79            .map(Self)
80            .ok_or(VersionError::InvalidEncoding(
81                "zero private record revision",
82            ))
83    }
84}
85
86/// A changed key and its private batch identity, without retaining or copying its payload.
87pub struct PrivateRecordKey {
88    key: RecordKey,
89    revision: PrivateRecordRevision,
90}
91
92impl PrivateRecordKey {
93    pub fn key(&self) -> &[u8] {
94        self.key.bytes()
95    }
96
97    pub fn revision(&self) -> PrivateRecordRevision {
98        self.revision
99    }
100}
101
102struct Savepoint {
103    id: StorageSavepointId,
104    records: Records,
105    resident: usize,
106    runs: RunSet,
107    sources: RetainedSources,
108    scopes: Option<RevisionScopes>,
109    revision: Option<PrivateRecordRevision>,
110}
111
112/// The changes of one transaction in two tiers: the newest in a persistent ordered root in memory, older ones in immutable spilled runs. A key's change in memory shadows its changes in runs, and a newer run shadows an older one.
113struct State {
114    records: Records,
115    /// The estimated bytes the memory tier holds, which decide when it spills.
116    resident: usize,
117    runs: RunSet,
118    sources: Sources,
119    scopes: Option<RevisionScopes>,
120    revision: Option<PrivateRecordRevision>,
121    savepoints: BudgetedVec<Savepoint>,
122}
123
124impl State {
125    /// The newest change of `key`, from either tier.
126    fn change(
127        &self,
128        key: &[u8],
129        control: &StorageReadControl,
130    ) -> VersionResult<Option<TieredChange<'_>>> {
131        tiers::lookup(&self.records, &self.runs, key, control)
132    }
133
134    /// The smallest memory tier that spills, which is also the size of the smallest runs.
135    fn least_spill(&self) -> usize {
136        self.records.budget().limit() / RESIDENT_DIVISOR
137    }
138
139    /// Move the memory tier into a new run when the transaction's allowance is under pressure: more than half of it is used and the memory tier holds a share of it worth a run. The allowance measures what this tier actually holds, so changes that share their values with an earlier owner do not spill on their own. Failure leaves both tiers unchanged.
140    fn make_room(&mut self, control: &StorageReadControl) -> VersionResult<()> {
141        let budget = self.records.budget();
142        if self.records.is_empty()
143            || self.resident < self.least_spill()
144            || budget.used() <= budget.limit() / PRESSURE_DIVISOR
145        {
146            return Ok(());
147        }
148        let memory = self.records.budget().clone();
149        let key_bytes = self.records.iter().fold(0_u64, |bytes, (key, _)| {
150            bytes.saturating_add(key.bytes().len() as u64)
151        });
152        let mut writer = SpilledRunWriter::new(self.records.len() as u64, key_bytes, &memory)?;
153        for (_, change) in &self.records {
154            writer.push(
155                change.write.key(),
156                change.write.expected(),
157                change.write.kind(),
158                change.identity,
159                change.write.value(),
160                control,
161            )?;
162        }
163        let run = writer
164            .finish()?
165            .expect("a nonempty memory tier spills a nonempty run");
166        let base = self.least_spill() as u64;
167        self.runs = self.runs.with_run(run, base, &memory, control)?;
168        self.records = Records::new(&memory);
169        self.resident = 0;
170        Ok(())
171    }
172
173    fn savepoint_position(&self, id: StorageSavepointId) -> VersionResult<usize> {
174        self.savepoints
175            .iter()
176            .rposition(|savepoint| savepoint.id == id)
177            .ok_or(VersionError::SavepointMissing(id))
178    }
179
180    fn truncate_savepoints(&mut self, len: usize) {
181        if len == 0 {
182            self.savepoints = BudgetedVec::new(self.records.budget());
183        } else {
184            self.savepoints.truncate(len);
185        }
186    }
187}
188
189struct Owner {
190    state: Mutex<State>,
191}
192
193/// One transaction's record changes. No provider lock or native transaction is retained between operations.
194///
195/// Command views and savepoints share immutable ordered roots and the spilled runs beneath them. Replacements copy only their search paths; obsolete payloads are released as soon as the last referencing root or returned record is dropped. Rollback restores a root without allocating. Once more than half of the transaction's allowance is used, the memory tier moves into a run in an encrypted temporary file, so the size of a transaction is bounded by disk rather than by its allowance.
196pub struct PrivateRecordChanges {
197    owner: StrongArc<Owner>,
198}
199
200impl PrivateRecordChanges {
201    /// An independent writable owner over the same evaluated roots. Later writes and savepoint operations do not change the original owner; unchanged records and spill runs keep their existing allocation leases.
202    pub fn fork(&self) -> Self {
203        let state = self.owner.state.lock();
204        Self {
205            owner: StrongArc::new(Owner {
206                state: Mutex::new(State {
207                    records: state.records.clone(),
208                    resident: state.resident,
209                    runs: state.runs.clone(),
210                    sources: state.sources.clone(),
211                    scopes: state.scopes.as_ref().map(RevisionScopes::fork),
212                    revision: state.revision,
213                    savepoints: BudgetedVec::new(state.records.budget()),
214                }),
215            }),
216        }
217    }
218
219    pub(super) fn share_owner(&self) -> Self {
220        Self {
221            owner: StrongArc::clone(&self.owner),
222        }
223    }
224
225    pub(super) fn write_kind(
226        &self,
227        key: &[u8],
228        control: &StorageReadControl,
229    ) -> VersionResult<Option<super::commit::RecordWriteKind>> {
230        control.check()?;
231        let state = self.owner.state.lock();
232        Ok(state.change(key, control)?.map(|change| change.kind()))
233    }
234
235    /// A mutation needs both the original condition and the prior write kind.
236    /// Read that one private entry once without materializing its value.
237    pub(in crate::mvcc) fn write_metadata(
238        &self,
239        key: &[u8],
240        control: &StorageReadControl,
241    ) -> VersionResult<Option<(super::RecordMetadata, super::commit::RecordWriteKind)>> {
242        control.check()?;
243        let state = self.owner.state.lock();
244        Ok(state
245            .change(key, control)?
246            .map(|change| (change.metadata(), change.kind())))
247    }
248
249    /// Reuse immutable decoded blocks across the point lookups of one evaluated
250    /// batch. Releasing the scope releases its cached bytes, even when a
251    /// savepoint still owns the runs. Later changes keep their ordinary lookup
252    /// precedence; this scope retains no record view.
253    pub(in crate::mvcc) fn read_scope(
254        &self,
255        control: &StorageReadControl,
256    ) -> VersionResult<BudgetedVec<run::RunCacheReader>> {
257        control.check()?;
258        let state = self.owner.state.lock();
259        let mut readers = BudgetedVec::new(control.memory());
260        for run in state.runs.newest_first() {
261            readers.push(run.cache_reader())?;
262        }
263        Ok(readers)
264    }
265
266    pub fn new(memory: &MemoryBudget) -> Self {
267        Self::with_revision_scope(memory, None)
268    }
269
270    /// Adopt a sorted derived run without copying its payloads. Its writes share `revision`; subsequent changes and savepoints use the ordinary private-root lifecycle.
271    pub(in crate::mvcc) fn from_spilled_run(
272        run: Option<run::SpilledRun>,
273        revision: PrivateRecordRevision,
274        scope: Option<PrivateRevisionScope>,
275        control: &StorageReadControl,
276    ) -> VersionResult<Self> {
277        Self::from_shared_spilled_run(run.map(Arc::new), revision, scope, control)
278    }
279
280    /// Retain immutable entries in an independent owner with its own root and scope identities.
281    fn from_shared_spilled_run(
282        run: Option<Arc<run::SpilledRun>>,
283        revision: PrivateRecordRevision,
284        scope: Option<PrivateRevisionScope>,
285        control: &StorageReadControl,
286    ) -> VersionResult<Self> {
287        control.check()?;
288        let changes = Self::with_revision_scope(control.memory(), scope);
289        if let Some(run) = run {
290            let mut state = changes.owner.state.lock();
291            if let Some(mut scopes) = state.scopes.take() {
292                let mut cursor = run.cursor(std::ops::Bound::Unbounded);
293                while let Some(entry) = cursor.next(control)? {
294                    let key = PreparedRecordWrite::from_shared(entry.key, None, None);
295                    scopes = scopes.with_writes(&[key], revision, control)?;
296                }
297                state.scopes = Some(scopes);
298            }
299            state.runs = RunSet::from_run(run);
300            state.revision = Some(revision);
301        }
302        control.check()?;
303        Ok(changes)
304    }
305
306    pub(in crate::mvcc) fn with_revision_scope(
307        memory: &MemoryBudget,
308        scope: Option<PrivateRevisionScope>,
309    ) -> Self {
310        Self {
311            owner: StrongArc::new(Owner {
312                state: Mutex::new(State {
313                    records: Records::new(memory),
314                    resident: 0,
315                    runs: RunSet::empty(),
316                    sources: Sources::new(memory),
317                    scopes: scope.map(|scope| RevisionScopes::new(memory, scope)),
318                    revision: None,
319                    savepoints: BudgetedVec::new(memory),
320                }),
321            }),
322        }
323    }
324
325    /// Stage one atomic batch of already evaluated replacements. Repeated changes to a private record must preserve its original committed precondition.
326    pub fn apply(
327        &self,
328        writes: &[RecordWrite<'_>],
329        control: &StorageReadControl,
330    ) -> VersionResult<()> {
331        let memory = self.owner.state.lock().records.budget().clone();
332        let owned_control = StorageReadControl::new(&memory, control.cancellation());
333        let prepared = PreparedRecordCommit::new(writes, &owned_control)?;
334        let writes = prepared.resident().ok_or(VersionError::InvalidEncoding(
335            "a supplied batch is in memory",
336        ))?;
337        self.apply_owned(writes, control)
338    }
339
340    pub(crate) fn apply_owned(
341        &self,
342        writes: &[PreparedRecordWrite],
343        control: &StorageReadControl,
344    ) -> VersionResult<()> {
345        if writes.is_empty() {
346            return Ok(());
347        }
348        let mut state = self.owner.state.lock();
349        for (index, write) in writes.iter().enumerate() {
350            control.cancellation().check()?;
351            if let Some(previous) = state.change(write.key(), control)? {
352                if previous.expected() != write.expected() {
353                    return Err(VersionError::WriteConflict {
354                        mutation: index,
355                        expected: write.expected(),
356                        actual: previous.expected(),
357                    });
358                }
359            }
360        }
361        state.make_room(control)?;
362        let identity = PrivateRecordRevision::allocate()?;
363        let scopes = state
364            .scopes
365            .as_ref()
366            .map(|scopes| scopes.with_writes(writes, identity, control))
367            .transpose()?;
368        let invalidates_source = writes
369            .iter()
370            .any(|write| state.sources.get(write.key()).is_some_and(Option::is_some));
371        if writes.len() == 1 && !invalidates_source {
372            let write = &writes[0];
373            control.cancellation().check()?;
374            let resident = replaced_resident(&state.records, state.resident, write);
375            state.records.try_insert(
376                write.shared_key(),
377                Change {
378                    write: write.clone(),
379                    identity,
380                },
381            )?;
382            state.resident = resident;
383            state.scopes = scopes;
384            state.revision = Some(identity);
385            return Ok(());
386        }
387        let mut records = state.records.clone();
388        let mut sources = state.sources.clone();
389        let mut resident = state.resident;
390        for write in writes {
391            control.cancellation().check()?;
392            resident = replaced_resident(&records, resident, write);
393            records.try_insert(
394                write.shared_key(),
395                Change {
396                    write: write.clone(),
397                    identity,
398                },
399            )?;
400            if sources.get(write.key()).is_some_and(Option::is_some) {
401                sources.try_insert(write.shared_key(), None)?;
402            }
403        }
404        control.cancellation().check()?;
405        // Candidate roots own every reservation before this single atomic publication.
406        state.records = records;
407        state.sources = sources;
408        state.scopes = scopes;
409        state.resident = resident;
410        state.revision = Some(identity);
411        Ok(())
412    }
413
414    /// Adopt prepared replacements in an independent private root. Immutable runs keep their original allowance; crossing allowances copies them into a fresh run. Failure exposes no partially staged root.
415    pub(in crate::mvcc) fn from_prepared(
416        prepared: &PreparedRecordCommit,
417        scope: Option<PrivateRevisionScope>,
418        control: &StorageReadControl,
419    ) -> VersionResult<Self> {
420        if let Some(writes) = prepared.resident() {
421            let changes = Self::with_revision_scope(control.memory(), scope);
422            for write in writes {
423                changes.apply_owned(std::slice::from_ref(write), control)?;
424            }
425            return Ok(changes);
426        }
427        if let Some(run) = prepared.shared_spilled_run(control.memory()) {
428            return Self::from_shared_spilled_run(
429                Some(run),
430                PrivateRecordRevision::allocate()?,
431                scope,
432                control,
433            );
434        }
435        let mut builder = super::commit::PreparedWritesBuilder::like(prepared, control)?;
436        let mut writes = prepared.writes();
437        while let Some(write) = writes.next(control)? {
438            builder.push(write, control)?;
439        }
440        builder.finish_changes(scope, control)
441    }
442
443    pub fn has_written(&self) -> bool {
444        let state = self.owner.state.lock();
445        !state.records.is_empty() || !state.runs.is_empty()
446    }
447
448    pub fn snapshot(&self) -> VersionResult<PrivateRecordSnapshot> {
449        let state = self.owner.state.lock();
450        let memory = state
451            .records
452            .budget()
453            .reserve(std::mem::size_of::<PrivateRecordSnapshot>())?;
454        Ok(PrivateRecordSnapshot {
455            records: state.records.clone(),
456            runs: state.runs.clone(),
457            sources: state.sources.snapshot(),
458            scopes: state
459                .scopes
460                .as_ref()
461                .map(RevisionScopes::snapshot)
462                .transpose()?,
463            revision: state.revision,
464            _memory: memory,
465        })
466    }
467
468    pub fn savepoint(&self, id: StorageSavepointId) -> VersionResult<()> {
469        let mut state = self.owner.state.lock();
470        let records = state.records.clone();
471        let resident = state.resident;
472        let runs = state.runs.clone();
473        let sources = state.sources.snapshot();
474        let scopes = state.scopes.as_ref().map(RevisionScopes::fork);
475        let revision = state.revision;
476        state.savepoints.push(Savepoint {
477            id,
478            records,
479            resident,
480            runs,
481            sources,
482            scopes,
483            revision,
484        })?;
485        Ok(())
486    }
487
488    /// Release the nearest matching identity and its nested savepoints, retaining every write.
489    pub fn release_savepoint(&self, id: StorageSavepointId) -> VersionResult<()> {
490        let mut state = self.owner.state.lock();
491        let position = state.savepoint_position(id)?;
492        state.truncate_savepoints(position);
493        Ok(())
494    }
495
496    /// Undo only this owner's changes, retaining the target savepoint for another rollback.
497    pub fn rollback_to_savepoint(&self, id: StorageSavepointId) -> VersionResult<()> {
498        let mut state = self.owner.state.lock();
499        let position = state.savepoint_position(id)?;
500        state.records = state.savepoints[position].records.clone();
501        state.resident = state.savepoints[position].resident;
502        state.runs = state.savepoints[position].runs.clone();
503        let sources = state.savepoints[position].sources.clone();
504        state.sources.restore(&sources);
505        state.scopes = state.savepoints[position]
506            .scopes
507            .as_ref()
508            .map(RevisionScopes::fork);
509        state.revision = state.savepoints[position].revision;
510        state.truncate_savepoints(position + 1);
511        Ok(())
512    }
513
514    pub fn rollback(&self) -> VersionResult<()> {
515        let mut state = self.owner.state.lock();
516        let memory = state.records.budget().clone();
517        state.records = Records::new(&memory);
518        state.resident = 0;
519        state.runs = RunSet::empty();
520        state.sources = Sources::new(&memory);
521        state.scopes = state.scopes.as_ref().map(RevisionScopes::empty);
522        state.revision = None;
523        state.truncate_savepoints(0);
524        Ok(())
525    }
526
527    /// Capture immutable final replacements without reevaluating or copying their payloads. Changes that spilled are merged into one spilled run of final replacements, which commits read one block at a time.
528    pub fn prepare(&self, control: &StorageReadControl) -> VersionResult<PreparedRecordCommit> {
529        control.cancellation().check()?;
530        let state = self.owner.state.lock();
531        if state.records.is_empty() {
532            if let Some(run) = state
533                .runs
534                .only_run()
535                .filter(|run| run.shares_allowance(control.memory()))
536            {
537                return PreparedRecordCommit::from_shared_spilled_run(Arc::clone(run), control);
538            }
539        }
540        let mut changes = TieredCursor::new(
541            Some(&state.records),
542            &state.runs,
543            std::ops::Bound::Unbounded,
544            control,
545        )?;
546        if state.runs.is_empty() {
547            let mut writes = BudgetedVec::new(control.memory());
548            writes.reserve(state.records.len())?;
549            while let Some(change) = changes.next(control)? {
550                writes.push(change.write(control)?)?;
551            }
552            return PreparedRecordCommit::from_unique_owned(writes, control);
553        }
554        let (entries, key_bytes) = state.runs.size();
555        let key_bytes = state.records.iter().fold(key_bytes, |bytes, (key, _)| {
556            bytes.saturating_add(key.bytes().len() as u64)
557        });
558        let mut writer = SpilledRunWriter::new(
559            entries.saturating_add(state.records.len() as u64),
560            key_bytes,
561            control.memory(),
562        )?;
563        while let Some(change) = changes.next(control)? {
564            match &change {
565                TieredChange::Resident(resident) => writer.push(
566                    resident.write.key(),
567                    resident.write.expected(),
568                    resident.write.kind(),
569                    resident.identity,
570                    resident.write.value(),
571                    control,
572                )?,
573                TieredChange::Spilled { entry, run } => writer.push_spilled(entry, run, control)?,
574            }
575        }
576        let run = writer.finish()?.expect("a spilled transaction has changes");
577        PreparedRecordCommit::from_spilled_run(run, control)
578    }
579}
580
581/// Fixed private visibility for a command or retained source cursor; tombstones remain distinguishable from an unchanged key.
582pub struct PrivateRecordSnapshot {
583    records: Records,
584    runs: RunSet,
585    sources: RetainedSources,
586    scopes: Option<Box<PrivateRecordSnapshot>>,
587    revision: Option<PrivateRecordRevision>,
588    _memory: MemoryReservation,
589}
590
591impl PrivateRecordSnapshot {
592    /// Identity of this complete private root. Undo restores the saved identity; successful replacement batches receive identities never used by another root or transaction.
593    pub fn revision(&self) -> Option<PrivateRecordRevision> {
594        self.revision
595    }
596
597    pub(super) fn last_before(
598        &self,
599        prefix: &[u8],
600        before: Option<&[u8]>,
601        control: &StorageReadControl,
602    ) -> VersionResult<Option<PreparedRecordWrite>> {
603        control.check()?;
604        let mut upper = BudgetedVec::new(control.memory());
605        upper.extend_from_slice(prefix)?;
606        let upper = match upper.iter().rposition(|byte| *byte != u8::MAX) {
607            Some(last) => {
608                upper[last] += 1;
609                upper.truncate(last + 1);
610                Some(upper)
611            }
612            None => None,
613        };
614        let end = match (before, upper.as_deref()) {
615            (Some(before), Some(upper)) => Some(before.min(upper)),
616            (before, upper) => before.or(upper),
617        };
618        let bound = end.map_or(std::ops::Bound::Unbounded, std::ops::Bound::Excluded);
619        let resident = self
620            .records
621            .last_before::<[u8]>(bound)
622            .map(|(_, change)| TieredChange::Resident(change));
623        let spilled = self.runs.last_before(bound, control)?;
624        // The memory tier is newer, so it wins a tie.
625        let last = match (resident, spilled) {
626            (Some(resident), Some(spilled)) if spilled.key() > resident.key() => Some(spilled),
627            (Some(resident), _) => Some(resident),
628            (None, spilled) => spilled,
629        };
630        last.filter(|change| change.key().starts_with(prefix))
631            .map(|change| change.write(control))
632            .transpose()
633    }
634
635    /// Retain this exact private revision without following later writes or copying its values.
636    pub(super) fn try_clone(&self) -> VersionResult<Self> {
637        let memory = self.records.budget().reserve(std::mem::size_of::<Self>())?;
638        Ok(Self {
639            records: self.records.clone(),
640            runs: self.runs.clone(),
641            sources: self.sources.clone(),
642            scopes: self
643                .scopes
644                .as_ref()
645                .map(|scopes| scopes.try_clone().map(Box::new))
646                .transpose()?,
647            revision: self.revision,
648            _memory: memory,
649        })
650    }
651
652    /// Select changed keys on this retained command boundary, including deletions. Undo restores their earlier identities, while later branches and other transactions receive distinct identities.
653    pub fn scan_keys(
654        &self,
655        prefix: &[u8],
656        after: Option<&[u8]>,
657        limit: usize,
658        control: &StorageReadControl,
659    ) -> VersionResult<BudgetedVec<PrivateRecordKey>> {
660        control.check()?;
661        let mut result = BudgetedVec::new(control.memory());
662        if limit == 0 {
663            return Ok(result);
664        }
665        let start = after
666            .filter(|after| *after >= prefix)
667            .map_or(std::ops::Bound::Included(prefix), std::ops::Bound::Excluded);
668        let mut changes =
669            TieredCursor::for_prefix(Some(&self.records), &self.runs, prefix, start, control)?;
670        while let Some(change) = changes.next(control)? {
671            control.check()?;
672            let key = change.key();
673            if !key.starts_with(prefix) {
674                break;
675            }
676            if after.is_some_and(|after| key <= after) {
677                continue;
678            }
679            result.push(PrivateRecordKey {
680                key: change.shared_key(),
681                revision: change.identity(),
682            })?;
683            if result.len() == limit {
684                break;
685            }
686        }
687        Ok(result)
688    }
689
690    pub(super) fn visit_merged<P: super::projection::Projection>(
691        &self,
692        committed: &dyn super::CommittedRecordSnapshot,
693        prefix: &[u8],
694        after: Option<&[u8]>,
695        limit: usize,
696        control: &StorageReadControl,
697        visit: &mut super::projection::Visitor<'_, P>,
698    ) -> VersionResult<()> {
699        control.cancellation().check()?;
700        if limit == 0 {
701            return Ok(());
702        }
703        let mut changes = self.cursor(prefix, after, control)?;
704        let mut pending = changes.next(control)?;
705        let mut count = 0;
706        let mut running = true;
707        let mut emit = |key: &[u8], record: P::Record<'_>| -> VersionResult<bool> {
708            control.cancellation().check()?;
709            count += 1;
710            let more = visit(key, record)?;
711            control.cancellation().check()?;
712            Ok(more && count < limit)
713        };
714        P::visit(committed, prefix, after, control, &mut |key, record| {
715            while let Some(entry) = pending.as_ref().filter(|entry| entry.key() < key) {
716                running = P::private(entry, control, &mut emit)?;
717                if !running {
718                    return Ok(false);
719                }
720                pending = changes.next(control)?;
721            }
722            if let Some(entry) = pending.as_ref().filter(|entry| entry.key() == key) {
723                running = P::private(entry, control, &mut emit)?;
724                if running {
725                    pending = changes.next(control)?;
726                }
727            } else {
728                running = emit(key, record)?;
729            }
730            Ok(running)
731        })?;
732        while running {
733            let Some(entry) = pending.as_ref() else {
734                break;
735            };
736            running = P::private(entry, control, &mut emit)?;
737            if running {
738                pending = changes.next(control)?;
739            }
740        }
741        Ok(())
742    }
743
744    /// Select the original revision and presence from the entry without reading a spilled value.
745    pub(super) fn metadata(
746        &self,
747        key: &[u8],
748        control: &StorageReadControl,
749    ) -> VersionResult<Option<super::RecordMetadata>> {
750        Ok(self
751            .revision_metadata(key, control)?
752            .map(|(metadata, _)| metadata))
753    }
754
755    /// Select one private entry's presence, original committed revision and exact private identity in the same point lookup, without loading its value.
756    pub(super) fn revision_metadata(
757        &self,
758        key: &[u8],
759        control: &StorageReadControl,
760    ) -> VersionResult<Option<(super::RecordMetadata, PrivateRecordRevision)>> {
761        control.cancellation().check()?;
762        Ok(tiers::lookup(&self.records, &self.runs, key, control)?
763            .map(|change| (change.metadata(), change.identity())))
764    }
765
766    pub fn get(
767        &self,
768        key: &[u8],
769        control: &StorageReadControl,
770    ) -> VersionResult<Option<PreparedRecordWrite>> {
771        control.cancellation().check()?;
772        tiers::lookup(&self.records, &self.runs, key, control)?
773            .map(|change| change.write(control))
774            .transpose()
775    }
776
777    /// Visit the private replacements whose keys start with `prefix` and follow `after`, including deletions, in key order until `visit` returns false. Each replacement's value is loaded only when it is visited, so a caller can bound what it keeps by bytes.
778    pub fn visit(
779        &self,
780        prefix: &[u8],
781        after: Option<&[u8]>,
782        control: &StorageReadControl,
783        visit: &mut dyn FnMut(&PreparedRecordWrite) -> VersionResult<bool>,
784    ) -> VersionResult<()> {
785        control.cancellation().check()?;
786        let start = after
787            .filter(|after| *after >= prefix)
788            .map_or(std::ops::Bound::Included(prefix), std::ops::Bound::Excluded);
789        let mut changes =
790            TieredCursor::for_prefix(Some(&self.records), &self.runs, prefix, start, control)?;
791        while let Some(change) = changes.next(control)? {
792            control.cancellation().check()?;
793            let key = change.key();
794            if !key.starts_with(prefix) {
795                break;
796            }
797            if after.is_some_and(|after| key <= after) {
798                continue;
799            }
800            if !visit(&change.write(control)?)? {
801                break;
802            }
803        }
804        Ok(())
805    }
806
807    /// Return a bounded ordered page of private replacements, including deletions.
808    pub fn scan(
809        &self,
810        prefix: &[u8],
811        after: Option<&[u8]>,
812        limit: usize,
813        control: &StorageReadControl,
814    ) -> VersionResult<BudgetedVec<PreparedRecordWrite>> {
815        control.cancellation().check()?;
816        let mut result = BudgetedVec::new(control.memory());
817        if limit == 0 {
818            return Ok(result);
819        }
820        let start = after
821            .filter(|after| *after >= prefix)
822            .map_or(std::ops::Bound::Included(prefix), std::ops::Bound::Excluded);
823        let mut changes =
824            TieredCursor::for_prefix(Some(&self.records), &self.runs, prefix, start, control)?;
825        while let Some(change) = changes.next(control)? {
826            control.cancellation().check()?;
827            let key = change.key();
828            if !key.starts_with(prefix) {
829                break;
830            }
831            if after.is_some_and(|after| key <= after) {
832                continue;
833            }
834            result.push(change.write(control)?)?;
835            if result.len() == limit {
836                break;
837            }
838        }
839        Ok(result)
840    }
841}
842
843/// The estimated memory a change takes in the memory tier.
844fn resident_bytes(write: &PreparedRecordWrite) -> usize {
845    write
846        .key()
847        .len()
848        .saturating_add(write.value().map_or(0, <[u8]>::len))
849        .saturating_add(CHANGE_OVERHEAD)
850}
851
852fn replaced_resident(records: &Records, resident: usize, write: &PreparedRecordWrite) -> usize {
853    let previous = records
854        .get(write.key())
855        .map_or(0, |change| resident_bytes(&change.write));
856    resident
857        .saturating_sub(previous)
858        .saturating_add(resident_bytes(write))
859}
860
861#[cfg(test)]
862mod tests;