Skip to main content

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}