sui_rpc_store/indexer/object_version_by_checkpoint.rs
1// Copyright (c) Mysten Labs, Inc.
2// SPDX-License-Identifier: Apache-2.0
3
4//! Sequential pipeline that populates the
5//! [`schema::object_version_by_checkpoint`](crate::schema::object_version_by_checkpoint)
6//! CF, which resolves an object's version as of a checkpoint.
7//!
8//! It writes three kinds of rows:
9//!
10//! - **Change rows** `(id, c) -> final version` -- one per object that
11//! changed in checkpoint `c`, carrying its final version at the end
12//! of `c` (a live version, or the tombstone version for an object
13//! deleted or wrapped and not re-created). Only the final version is
14//! recorded; intra-checkpoint intermediate versions stay addressable
15//! through the version-keyed [`objects`](crate::schema::objects) CF.
16//! - **Restore floor rows** `(id, T) -> version`, marked `from_restore`
17//! -- one per live object at the restore anchor `T`, bulk-loaded by
18//! the restore impl. A read below `T` for an object that never
19//! changed in the available window falls back to these.
20//! - **Synthetic floor rows** `(id, 0) -> window-entry version` -- for
21//! an object that existed before the available window `[L, T]` and
22//! first changes within it, this records the version it entered the
23//! window with, so a read in `[L, first-change)` resolves to that
24//! instead of the newer restore floor. Written during the embedded
25//! backfill only: past `T` the restore floor already covers
26//! pre-window objects, so the dedup read is skipped. The row is keyed
27//! at checkpoint 0 (below the window, where `L > 0`) so the floor
28//! scan finds it, and the effects-driven pruner retracts it once the
29//! object's first in-window change ages out.
30
31use std::collections::BTreeMap;
32use std::num::NonZeroUsize;
33use std::ops::Bound;
34use std::sync::Arc;
35use std::sync::Mutex;
36
37use anyhow::Context as _;
38use async_trait::async_trait;
39use lru::LruCache;
40use sui_consistent_store::Batch;
41use sui_consistent_store::Db;
42use sui_consistent_store::DbMap;
43use sui_consistent_store::PipelineTaskKey;
44use sui_consistent_store::Restore;
45use sui_consistent_store::error::Error;
46use sui_consistent_store::reader::Reader;
47use sui_consistent_store::restore_state;
48use sui_indexer_alt_framework::pipeline::Processor;
49use sui_indexer_alt_framework::pipeline::sequential;
50use sui_types::base_types::ObjectID;
51use sui_types::base_types::SequenceNumber;
52use sui_types::effects::TransactionEffectsAPI;
53use sui_types::full_checkpoint_content::Checkpoint;
54use sui_types::object::Object;
55
56use crate::RpcStoreSchema;
57use crate::indexer::Schema;
58use crate::indexer::Store;
59use crate::indexer::checkpoint_input_objects;
60use crate::indexer::checkpoint_output_objects;
61use crate::schema::object_version_by_checkpoint;
62
63/// Upper bound on the number of object ids the [floor cache] holds.
64///
65/// The cache only needs to keep the objects that recur as input floor
66/// candidates -- the hot set (`0x5`, `0x6`, popular shared objects and
67/// packages) that would otherwise pay a redundant `iter_rev` scan on
68/// every checkpoint they appear in. An LRU keeps exactly that hot set
69/// resident and evicts one-off objects, so this bounds the cache to
70/// order-of-a-hundred-MB regardless of how wide the backfill window
71/// `(L, T]` is (an unbounded set over a large window could reach several
72/// GB). A miss (including every miss after an eviction or a restart)
73/// simply falls back to the durable scan, so the capacity is a pure
74/// performance knob with no bearing on correctness.
75///
76/// [floor cache]: ObjectVersionByCheckpoint::floored
77const FLOOR_CACHE_CAPACITY: usize = 1_000_000;
78
79/// Pipeline marker for `object_version_by_checkpoint`.
80///
81/// `anchor` is the restore anchor `T`, used two ways: the restore impl
82/// writes its `from_restore` floor rows at `T`, and the processor only
83/// produces floor candidates within the backfill window `[L, T]`
84/// (checkpoints at or below `T`). It is `None` for a from-genesis build
85/// (no restore), which has no window and needs no floor rows.
86pub struct ObjectVersionByCheckpoint {
87 anchor: Option<u64>,
88
89 /// Objects already resolved as input floor candidates during this
90 /// run: each necessarily has a row at a checkpoint strictly below
91 /// any checkpoint still to be committed (either the synthetic floor
92 /// this pipeline wrote at `(id, 0)`, or the prior in-window row the
93 /// fallback scan found). Consulted in [`commit`](Self::commit) to
94 /// skip the `iter_rev` first-appearance scan for objects that
95 /// recur as inputs -- overwhelmingly the frequently-touched objects
96 /// (`0x5`, `0x6`, ...). Bounded by an LRU (see
97 /// [`FLOOR_CACHE_CAPACITY`]). The [`Mutex`] only guards interior
98 /// mutation behind `&self`; commits are sequential and `process`
99 /// never touches the cache, so it is locked once per batch and never
100 /// contended.
101 floored: Mutex<LruCache<ObjectID, ()>>,
102}
103
104impl Default for ObjectVersionByCheckpoint {
105 fn default() -> Self {
106 Self::with_anchor(None)
107 }
108}
109
110/// One staged write produced by [`process`](ObjectVersionByCheckpoint::process).
111pub enum Row {
112 /// Object `id`'s final version at the end of `checkpoint` -- a live
113 /// version, or a tombstone version for a removal.
114 Change {
115 id: ObjectID,
116 checkpoint: u64,
117 version: SequenceNumber,
118 },
119 /// Object `id` existed before `checkpoint` and was an input to it,
120 /// entering with `version`. A synthetic floor row is written at
121 /// `(id, 0)` iff this is the object's first appearance in the
122 /// backfill window (so it predates the window).
123 Floor {
124 id: ObjectID,
125 checkpoint: u64,
126 version: SequenceNumber,
127 },
128}
129
130impl ObjectVersionByCheckpoint {
131 /// Marker for the restore-driver registration: writes `from_restore`
132 /// floor rows at the anchor `checkpoint`.
133 pub fn for_restore(checkpoint: u64) -> Self {
134 Self::with_anchor(Some(checkpoint))
135 }
136
137 /// Marker for the tip/backfill registration, carrying the restore
138 /// anchor `T` (or `None` for a from-genesis build) so the processor
139 /// scopes floor candidates to the backfill window `[L, T]`.
140 pub fn with_anchor(anchor: Option<u64>) -> Self {
141 Self {
142 anchor,
143 floored: Mutex::new(LruCache::new(
144 NonZeroUsize::new(FLOOR_CACHE_CAPACITY).expect("FLOOR_CACHE_CAPACITY is non-zero"),
145 )),
146 }
147 }
148}
149
150#[async_trait]
151impl Processor for ObjectVersionByCheckpoint {
152 const NAME: &'static str = "object_version_by_checkpoint";
153 type Value = Row;
154
155 async fn process(&self, checkpoint: &Arc<Checkpoint>) -> anyhow::Result<Vec<Row>> {
156 let cp = checkpoint.summary.data().sequence_number;
157
158 // Change rows: objects live at the end of the checkpoint, each
159 // with its final version.
160 let outputs = checkpoint_output_objects(checkpoint)?;
161 let mut rows: Vec<Row> = outputs
162 .iter()
163 .map(|(id, (object, _))| Row::Change {
164 id: *id,
165 checkpoint: cp,
166 version: object.version(),
167 })
168 .collect();
169
170 // Objects removed (deleted or wrapped) during the checkpoint:
171 // record the tombstone version -- the removing transaction's
172 // lamport version, where the `objects` pipeline writes the
173 // tombstone row -- keeping the highest such version if an id is
174 // touched more than once. A read pinned at `cp` then resolves
175 // to the tombstone (and thus "no live object") instead of the
176 // stale prior version.
177 let mut removed: BTreeMap<ObjectID, SequenceNumber> = BTreeMap::new();
178 for tx in &checkpoint.transactions {
179 let lamport = tx.effects.lamport_version();
180 for oref in tx
181 .effects
182 .deleted()
183 .into_iter()
184 .chain(tx.effects.unwrapped_then_deleted())
185 .chain(tx.effects.wrapped())
186 {
187 removed
188 .entry(oref.0)
189 .and_modify(|v| *v = (*v).max(lamport))
190 .or_insert(lamport);
191 }
192 }
193
194 for (id, version) in removed {
195 // Removed then re-created within the same checkpoint (e.g.
196 // wrapped then unwrapped) -- it is live at the end and
197 // already covered by its output row above.
198 if outputs.contains_key(&id) {
199 continue;
200 }
201 rows.push(Row::Change {
202 id,
203 checkpoint: cp,
204 version,
205 });
206 }
207
208 // Floor candidates, only within the backfill window `[L, T]`:
209 // objects that existed *before* this checkpoint and were inputs
210 // to it (so they predate any creation this checkpoint), each
211 // carrying its incoming version. `commit` turns the first such
212 // appearance per object into a synthetic floor row. Past `T` the
213 // restore floor already covers pre-window objects, so producing
214 // these (in the worker pool) would be wasted work.
215 if self.anchor.is_some_and(|t| cp <= t) {
216 for (id, (input, _)) in checkpoint_input_objects(checkpoint)? {
217 rows.push(Row::Floor {
218 id,
219 checkpoint: cp,
220 version: input.version(),
221 });
222 }
223 }
224
225 Ok(rows)
226 }
227}
228
229impl Restore for ObjectVersionByCheckpoint {
230 type Schema = RpcStoreSchema;
231
232 fn restore(
233 &self,
234 schema: &Self::Schema,
235 object: &Object,
236 batch: &mut Batch,
237 ) -> anyhow::Result<()> {
238 // Restoration runs against a live-object snapshot with no
239 // per-checkpoint history, so every live object contributes one
240 // row at the restore anchor. The anchor is supplied at
241 // registration (`for_restore`); a tip-mode marker would never
242 // be registered with the restore driver, so its absence is a
243 // programmer error.
244 let checkpoint = self
245 .anchor
246 .context("object_version_by_checkpoint restored without a restore anchor checkpoint")?;
247 // Mark these as restore-floor rows so a checkpoint-pinned read
248 // below the anchor can tell a pre-window object (live) apart
249 // from one created in the anchor checkpoint.
250 let (key, value) =
251 object_version_by_checkpoint::store_restored(object.id(), checkpoint, object.version());
252 batch.put(&schema.object_version_by_checkpoint, &key, &value)?;
253 Ok(())
254 }
255}
256
257#[async_trait]
258impl sequential::Handler for ObjectVersionByCheckpoint {
259 type Store = Store;
260 type Batch = Vec<Row>;
261
262 fn batch(&self, batch: &mut Self::Batch, values: std::vec::IntoIter<Row>) {
263 batch.extend(values);
264 }
265
266 async fn commit<'a>(
267 &self,
268 batch: &Self::Batch,
269 conn: &mut sui_consistent_store::Connection<'a, Schema>,
270 ) -> anyhow::Result<usize> {
271 let cf = &conn.store.schema().object_version_by_checkpoint;
272
273 // Lock the floor cache once for the whole batch rather than per
274 // floor row. Commits are sequential and `process` never touches
275 // the cache, so the guard is uncontended; `commit` has no
276 // `.await`, so it never crosses a suspend point.
277 let mut floored = self.floored.lock().expect("floor cache mutex poisoned");
278
279 let mut count = 0;
280 for row in batch {
281 match row {
282 Row::Change {
283 id,
284 checkpoint,
285 version,
286 } => {
287 let (k, v) = object_version_by_checkpoint::store(*id, *checkpoint, *version);
288 conn.batch.put(cf, &k, &v)?;
289 count += 1;
290 }
291 Row::Floor {
292 id,
293 checkpoint,
294 version,
295 } => {
296 // The processor only emits floor candidates within
297 // the backfill window, so all that is left is to
298 // dedup repeated and re-indexed appearances: only the
299 // object's first appearance writes the floor row. The
300 // cache short-circuits the scan for objects that have
301 // already appeared this run.
302 if needs_floor(&mut floored, cf, *id, *checkpoint)? {
303 let (k, v) = object_version_by_checkpoint::store(*id, 0, *version);
304 conn.batch.put(cf, &k, &v)?;
305 count += 1;
306 }
307 }
308 }
309 }
310 Ok(count)
311 }
312}
313
314/// The restore anchor `T` (`__restore.restored_at`) for this pipeline,
315/// or `None` if it was never restored. Read once at registration so the
316/// processor can scope floor candidates to the backfill window `[L, T]`.
317pub(crate) fn restored_anchor(db: &Db) -> anyhow::Result<Option<u64>> {
318 let key = PipelineTaskKey::new(ObjectVersionByCheckpoint::NAME);
319 Ok(db
320 .framework()
321 .restore
322 .get(&key)?
323 .and_then(|s| match s.state {
324 Some(restore_state::State::Complete(c)) => Some(c.restored_at),
325 _ => None,
326 }))
327}
328
329/// Whether object `id`'s synthetic floor row should be written at
330/// `checkpoint`, consulting `cache` before the durable
331/// [`is_first_appearance`] scan.
332///
333/// A cache hit means the object has already been resolved as an input
334/// floor candidate earlier this run, so it necessarily has a row below
335/// `checkpoint` (commits advance monotonically) and is not a first
336/// appearance -- return `false` without touching RocksDB. A miss falls
337/// back to the scan; the object is recorded either way so its next
338/// input appearance hits.
339///
340/// The fallback keeps the result correct when the cache cannot answer:
341/// after a restart (empty cache) or an LRU eviction, the scan finds the
342/// floor row a prior run persisted at `(id, 0)` and returns `false`, so
343/// the row is never rewritten with a newer -- and wrong -- version.
344///
345/// The caller holds the cache lock for the batch, so this takes a plain
346/// `&mut` and stays oblivious to the locking.
347fn needs_floor<R: Reader>(
348 cache: &mut LruCache<ObjectID, ()>,
349 cf: &DbMap<object_version_by_checkpoint::Key, object_version_by_checkpoint::Value, R>,
350 id: ObjectID,
351 checkpoint: u64,
352) -> Result<bool, Error> {
353 // `get` rather than `contains` so a hit promotes the entry to
354 // most-recently-used, keeping the hot set (`0x5`, `0x6`, ...)
355 // resident instead of aging out under churn from one-off objects.
356 if cache.get(&id).is_some() {
357 return Ok(false);
358 }
359 let first = is_first_appearance(cf, id, checkpoint)?;
360 cache.put(id, ());
361 Ok(first)
362}
363
364/// Whether `checkpoint` is the object's first appearance in the index:
365/// it has no row strictly below `checkpoint`. The restore floor sits at
366/// `T >= checkpoint`, so it is excluded, as is the change row written
367/// for this same checkpoint. Dedups repeated and re-indexed appearances
368/// so only the first writes the synthetic floor row.
369fn is_first_appearance<R: Reader>(
370 cf: &DbMap<object_version_by_checkpoint::Key, object_version_by_checkpoint::Value, R>,
371 id: ObjectID,
372 checkpoint: u64,
373) -> Result<bool, Error> {
374 let lo = object_version_by_checkpoint::Key { id, checkpoint: 0 };
375 let hi = object_version_by_checkpoint::Key { id, checkpoint };
376 let seen = cf
377 .iter_rev((Bound::Included(lo), Bound::Excluded(hi)))?
378 .next()
379 .is_some();
380 Ok(!seen)
381}
382
383#[cfg(test)]
384mod tests {
385 use std::sync::Arc;
386
387 use sui_consistent_store::Db;
388 use sui_consistent_store::DbOptions;
389 use sui_consistent_store::FrameworkSchema;
390 use sui_consistent_store::RestoreState;
391 use sui_types::base_types::SuiAddress;
392 use sui_types::test_checkpoint_data_builder::TestCheckpointBuilder;
393
394 use super::*;
395
396 fn fresh_db() -> (tempfile::TempDir, Db, RpcStoreSchema) {
397 let dir = tempfile::tempdir().unwrap();
398 let (db, schema) = Db::open::<RpcStoreSchema>(dir.path(), DbOptions::default()).unwrap();
399 (dir, db, schema)
400 }
401
402 /// Seed this pipeline's `__restore` row as `Complete { restored_at }`.
403 fn seed_restored_at(db: &Db, restored_at: u64) {
404 let framework = FrameworkSchema::new(db.clone());
405 let mut batch = db.batch();
406 batch
407 .put(
408 &framework.restore,
409 &PipelineTaskKey::new(ObjectVersionByCheckpoint::NAME),
410 &RestoreState {
411 state: Some(restore_state::State::Complete(restore_state::Complete {
412 restored_at,
413 })),
414 },
415 )
416 .unwrap();
417 batch.commit().unwrap();
418 }
419
420 fn put(schema: &RpcStoreSchema, db: &Db, id: ObjectID, checkpoint: u64, version: u64) {
421 let (k, v) =
422 object_version_by_checkpoint::store(id, checkpoint, SequenceNumber::from_u64(version));
423 let mut batch = db.batch();
424 batch
425 .put(&schema.object_version_by_checkpoint, &k, &v)
426 .unwrap();
427 batch.commit().unwrap();
428 }
429
430 #[tokio::test]
431 async fn process_runs_against_synthetic_checkpoint() {
432 let checkpoint = Arc::new(TestCheckpointBuilder::new(1).build_checkpoint());
433 let _rows = ObjectVersionByCheckpoint::default()
434 .process(&checkpoint)
435 .await
436 .unwrap();
437 }
438
439 /// A live object created in the checkpoint gets a change row at the
440 /// checkpoint's sequence number, carrying its current version.
441 #[tokio::test]
442 async fn process_records_final_live_version() {
443 let checkpoint = Arc::new(
444 TestCheckpointBuilder::new(7)
445 .start_transaction(0)
446 .create_owned_object(0)
447 .finish_transaction()
448 .build_checkpoint(),
449 );
450 let created_id = TestCheckpointBuilder::derive_object_id(0);
451 let version = checkpoint.transactions[0].effects.lamport_version();
452
453 // Within the backfill window (anchor above this checkpoint), so
454 // the floor candidates are produced.
455 let rows = ObjectVersionByCheckpoint::with_anchor(Some(100))
456 .process(&checkpoint)
457 .await
458 .unwrap();
459 let found = rows.iter().find_map(|r| match r {
460 Row::Change {
461 id,
462 checkpoint,
463 version,
464 } if *id == created_id => Some((*checkpoint, *version)),
465 _ => None,
466 });
467 assert_eq!(found, Some((7, version)));
468 // A freshly created object is not an input to its own checkpoint,
469 // so it produces no floor candidate.
470 assert!(
471 !rows
472 .iter()
473 .any(|r| matches!(r, Row::Floor { id, .. } if *id == created_id))
474 );
475 }
476
477 /// An object deleted in the checkpoint is recorded at the tombstone
478 /// (lamport) version, not its prior live version.
479 #[tokio::test]
480 async fn process_records_tombstone_for_deleted_object() {
481 let mut builder = TestCheckpointBuilder::new(0)
482 .start_transaction(0)
483 .create_owned_object(0)
484 .finish_transaction();
485 let _cp0 = builder.build_checkpoint();
486 builder = builder
487 .start_transaction(0)
488 .delete_object(0)
489 .finish_transaction();
490 let cp1 = Arc::new(builder.build_checkpoint());
491
492 let deleted_id = TestCheckpointBuilder::derive_object_id(0);
493 let tombstone_version = cp1.transactions[0].effects.lamport_version();
494
495 let rows = ObjectVersionByCheckpoint::default()
496 .process(&cp1)
497 .await
498 .unwrap();
499 let found = rows.iter().find_map(|r| match r {
500 Row::Change {
501 id,
502 checkpoint,
503 version,
504 } if *id == deleted_id => Some((*checkpoint, *version)),
505 _ => None,
506 });
507 assert_eq!(found, Some((1, tombstone_version)));
508 }
509
510 /// An object that existed before the checkpoint and is consumed by
511 /// it produces a floor candidate carrying its incoming version.
512 #[tokio::test]
513 async fn process_emits_floor_candidate_for_input_object() {
514 let mut builder = TestCheckpointBuilder::new(0)
515 .start_transaction(0)
516 .create_owned_object(0)
517 .finish_transaction();
518 let cp0 = builder.build_checkpoint();
519 builder = builder
520 .start_transaction(0)
521 .transfer_object(0, 1)
522 .finish_transaction();
523 let cp1 = Arc::new(builder.build_checkpoint());
524
525 let obj = TestCheckpointBuilder::derive_object_id(0);
526 let incoming = cp0.transactions[0].effects.lamport_version();
527
528 // Within the backfill window: the input object is floored.
529 let rows = ObjectVersionByCheckpoint::with_anchor(Some(100))
530 .process(&cp1)
531 .await
532 .unwrap();
533 let floor = rows.iter().find_map(|r| match r {
534 Row::Floor {
535 id,
536 checkpoint,
537 version,
538 } if *id == obj => Some((*checkpoint, *version)),
539 _ => None,
540 });
541 assert_eq!(floor, Some((1, incoming)), "input object floor candidate");
542 }
543
544 /// Past the restore anchor (tip indexing), the processor produces no
545 /// floor candidates at all, even for input objects.
546 #[tokio::test]
547 async fn process_skips_floor_candidates_past_the_anchor() {
548 let mut builder = TestCheckpointBuilder::new(0)
549 .start_transaction(0)
550 .create_owned_object(0)
551 .finish_transaction();
552 let _cp0 = builder.build_checkpoint();
553 builder = builder
554 .start_transaction(0)
555 .transfer_object(0, 1)
556 .finish_transaction();
557 let cp1 = Arc::new(builder.build_checkpoint());
558
559 // Anchor below this checkpoint (cp 1 > T 0): tip indexing, so no
560 // floor candidates are produced.
561 let rows = ObjectVersionByCheckpoint::with_anchor(Some(0))
562 .process(&cp1)
563 .await
564 .unwrap();
565 assert!(!rows.iter().any(|r| matches!(r, Row::Floor { .. })));
566 // And likewise for a from-genesis build (no anchor).
567 let rows = ObjectVersionByCheckpoint::default()
568 .process(&cp1)
569 .await
570 .unwrap();
571 assert!(!rows.iter().any(|r| matches!(r, Row::Floor { .. })));
572 }
573
574 /// Restore writes a `from_restore` floor row at the anchor, which
575 /// resolves at, above, and (via the fallback) below the anchor.
576 #[test]
577 fn restore_writes_one_row_at_the_anchor() {
578 let (_dir, db, schema) = fresh_db();
579
580 let object =
581 Object::with_id_owner_for_testing(ObjectID::from_single_byte(1), SuiAddress::ZERO);
582
583 let mut batch = db.batch();
584 ObjectVersionByCheckpoint::for_restore(123)
585 .restore(&schema, &object, &mut batch)
586 .unwrap();
587 batch.commit().unwrap();
588
589 for cp in [122, 123, 200] {
590 assert_eq!(
591 schema
592 .get_object_version_at_checkpoint(object.id(), cp)
593 .unwrap(),
594 Some(object.version()),
595 );
596 }
597 }
598
599 #[test]
600 fn restored_anchor_reads_restore_state() {
601 let (_dir, db, _schema) = fresh_db();
602 // No restore row yet.
603 assert_eq!(restored_anchor(&db).unwrap(), None);
604 // A completed restore exposes its anchor.
605 seed_restored_at(&db, 42);
606 assert_eq!(restored_anchor(&db).unwrap(), Some(42));
607 }
608
609 #[test]
610 fn is_first_appearance_true_with_no_prior_row() {
611 let (_dir, _db, schema) = fresh_db();
612 let id = ObjectID::random();
613 assert!(is_first_appearance(&schema.object_version_by_checkpoint, id, 50).unwrap());
614 }
615
616 #[test]
617 fn is_first_appearance_false_when_already_seen() {
618 let (_dir, db, schema) = fresh_db();
619 let id = ObjectID::random();
620 // A prior row exists below the checkpoint.
621 put(&schema, &db, id, 30, 5);
622 assert!(!is_first_appearance(&schema.object_version_by_checkpoint, id, 50).unwrap());
623 }
624
625 #[test]
626 fn is_first_appearance_ignores_rows_at_or_above_the_checkpoint() {
627 let (_dir, db, schema) = fresh_db();
628 let id = ObjectID::random();
629 // The restore floor row sits at the anchor (100), at or above the
630 // queried checkpoint, so it must not count as a prior row.
631 put(&schema, &db, id, 100, 9);
632 assert!(is_first_appearance(&schema.object_version_by_checkpoint, id, 50).unwrap());
633 }
634
635 fn cache(capacity: usize) -> LruCache<ObjectID, ()> {
636 LruCache::new(NonZeroUsize::new(capacity).unwrap())
637 }
638
639 /// Once an object has been resolved as a floor candidate, the next
640 /// resolution short-circuits on the cache: `needs_floor` returns
641 /// `false` without a scan, even though the durable state (no prior
642 /// row) would otherwise report a first appearance.
643 #[test]
644 fn needs_floor_short_circuits_after_first_resolution() {
645 let (_dir, _db, schema) = fresh_db();
646 let cf = &schema.object_version_by_checkpoint;
647 let mut cache = cache(FLOOR_CACHE_CAPACITY);
648 let id = ObjectID::random();
649
650 // First appearance: no prior row, so the floor is needed and the
651 // object is recorded.
652 assert!(needs_floor(&mut cache, cf, id, 50).unwrap());
653 // A later appearance hits the cache and skips the (still empty)
654 // scan, which on its own would report another first appearance.
655 assert!(is_first_appearance(cf, id, 60).unwrap());
656 assert!(!needs_floor(&mut cache, cf, id, 60).unwrap());
657 }
658
659 /// A cold cache (a fresh run) falls back to the durable scan: an
660 /// object whose floor row a prior run persisted is not re-floored.
661 #[test]
662 fn needs_floor_falls_back_to_scan_on_cold_cache() {
663 let (_dir, db, schema) = fresh_db();
664 let cf = &schema.object_version_by_checkpoint;
665 let mut cache = cache(FLOOR_CACHE_CAPACITY);
666 let id = ObjectID::random();
667
668 // A prior run already recorded a row below the checkpoint.
669 put(&schema, &db, id, 30, 5);
670 assert!(!needs_floor(&mut cache, cf, id, 50).unwrap());
671 }
672
673 /// LRU eviction never causes a duplicate floor write: an evicted
674 /// object that reappears falls back to the scan, which finds the
675 /// floor its earlier resolution persisted and reports "not first".
676 #[test]
677 fn needs_floor_survives_eviction() {
678 let (_dir, db, schema) = fresh_db();
679 let cf = &schema.object_version_by_checkpoint;
680 // Capacity one: a second object evicts the first.
681 let mut cache = cache(1);
682 let a = ObjectID::random();
683 let b = ObjectID::random();
684
685 // `a` is a first appearance; mirror the caller by persisting its
686 // synthetic floor at `(a, 0)`.
687 assert!(needs_floor(&mut cache, cf, a, 10).unwrap());
688 put(&schema, &db, a, 0, 5);
689
690 // `b` is a first appearance too, evicting `a` from the cache.
691 assert!(needs_floor(&mut cache, cf, b, 11).unwrap());
692 put(&schema, &db, b, 0, 7);
693
694 // `a` reappears: the cache no longer holds it, so the scan runs
695 // and finds the persisted `(a, 0)` floor -- not a first
696 // appearance, so no rewrite.
697 assert!(!needs_floor(&mut cache, cf, a, 12).unwrap());
698 }
699
700 /// End-to-end shape for a pre-window object that first changes
701 /// inside the window: the rows the restore and backfill would write,
702 /// then the reads they enable. Below the first change, the synthetic
703 /// floor at `(id, 0)` answers with the pre-window version rather than
704 /// the newer restore floor.
705 #[test]
706 fn synthetic_floor_serves_reads_below_first_change() {
707 let (_dir, db, schema) = fresh_db();
708 let id = ObjectID::random();
709 let anchor = 100; // restore tip T
710 let window_entry = SequenceNumber::from_u64(5); // version entering the window
711 let v1 = SequenceNumber::from_u64(6); // after the first in-window change (cp 50)
712
713 // The restore floor at T, and the backfilled change row at the
714 // object's first in-window change (cp 50).
715 let mut batch = db.batch();
716 let (rk, rv) = object_version_by_checkpoint::store_restored(id, anchor, v1);
717 batch
718 .put(&schema.object_version_by_checkpoint, &rk, &rv)
719 .unwrap();
720 let (ck, cv) = object_version_by_checkpoint::store(id, 50, v1);
721 batch
722 .put(&schema.object_version_by_checkpoint, &ck, &cv)
723 .unwrap();
724 batch.commit().unwrap();
725
726 // That change is the object's first appearance in the window, so
727 // the backfill writes a synthetic floor at `(id, 0)`.
728 assert!(is_first_appearance(&schema.object_version_by_checkpoint, id, 50).unwrap());
729 let mut batch = db.batch();
730 let (fk, fv) = object_version_by_checkpoint::store(id, 0, window_entry);
731 batch
732 .put(&schema.object_version_by_checkpoint, &fk, &fv)
733 .unwrap();
734 batch.commit().unwrap();
735
736 // Below the first change, the synthetic floor answers with the
737 // pre-window version (not the restore floor's newer version).
738 assert_eq!(
739 schema.get_object_version_at_checkpoint(id, 30).unwrap(),
740 Some(window_entry),
741 );
742 // At and after the first change, the change row answers.
743 assert_eq!(
744 schema.get_object_version_at_checkpoint(id, 50).unwrap(),
745 Some(v1),
746 );
747 assert_eq!(
748 schema.get_object_version_at_checkpoint(id, 75).unwrap(),
749 Some(v1),
750 );
751 assert_eq!(
752 schema.get_object_version_at_checkpoint(id, 100).unwrap(),
753 Some(v1),
754 );
755 }
756}