1use 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
39const CHANGE_OVERHEAD: usize = 192;
41const PRESSURE_DIVISOR: usize = 2;
43const RESIDENT_DIVISOR: usize = 16;
45
46struct Change {
47 write: PreparedRecordWrite,
48 identity: PrivateRecordRevision,
49}
50
51type Records = BudgetedSharedMap<RecordKey, Change>;
52
53#[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 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
86pub 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
112struct State {
114 records: Records,
115 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 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 fn least_spill(&self) -> usize {
136 self.records.budget().limit() / RESIDENT_DIVISOR
137 }
138
139 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
193pub struct PrivateRecordChanges {
197 owner: StrongArc<Owner>,
198}
199
200impl PrivateRecordChanges {
201 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 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 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 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 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 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 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 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 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 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 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
581pub 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 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 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 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 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 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 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 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 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
843fn 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;