sui_rpc_store/reader/
mod.rs

1// Copyright (c) Mysten Labs, Inc.
2// SPDX-License-Identifier: Apache-2.0
3
4//! Adapter that exposes [`RpcStoreSchema`] through the trait stack
5//! `sui-rpc-api` consumes — [`ObjectStore`], [`ReadStore`],
6//! [`RuntimeObjectResolver`], [`RpcStateReader`], and [`RpcIndexes`].
7//!
8//! The adapter type, [`RpcStoreReader`], is generic over a
9//! [`Reader`] so a single struct serves both tip reads (`R = Db`)
10//! and point-in-time reads bound to a snapshot (`R = Snapshot`).
11//! Callers requesting "give me the latest" hold the tip reader;
12//! callers requesting "show me state at checkpoint X" hold the
13//! snapshot-bound one. The choice of consistency context is made
14//! once, at the point [`RpcStoreReader::at_snapshot`] is called.
15//!
16//! [`ObjectStore`]: sui_types::storage::ObjectStore
17//! [`ReadStore`]: sui_types::storage::ReadStore
18//! [`RuntimeObjectResolver`]: sui_types::storage::RuntimeObjectResolver
19//! [`RpcStateReader`]: sui_types::storage::RpcStateReader
20//! [`RpcIndexes`]: sui_types::storage::RpcIndexes
21//! [`Reader`]: sui_consistent_store::reader::Reader
22
23mod child_resolver;
24mod indexes;
25#[cfg(test)]
26mod integration_test;
27mod layout;
28mod object_store;
29mod read_store;
30mod state_reader;
31
32use std::sync::Arc;
33
34use sui_consistent_store::Db;
35use sui_consistent_store::SchemaAtSnapshot;
36use sui_consistent_store::Snapshot;
37use sui_consistent_store::reader::Reader;
38
39use crate::RpcStoreSchema;
40
41/// Adapter exposing [`RpcStoreSchema`] through the
42/// `sui-rpc-api` reader-trait stack.
43///
44/// Construct one of two ways:
45///
46/// - [`RpcStoreReader::new`] binds to tip reads (`R = Db`). Use
47///   this for callers that want the latest committed state.
48/// - [`RpcStoreReader::at_snapshot`] takes a captured
49///   [`Snapshot`] and returns a [`RpcStoreReader<Snapshot>`] whose
50///   every read returns the state at that snapshot. Use this for
51///   "show me state at checkpoint X" requests.
52///
53/// `RpcStoreReader` holds an `Arc<RpcStoreSchema<R>>` so trait
54/// impls can hand a `&self` to any of the inherent read helpers
55/// already defined on the schema. The wrapper itself is `Clone`
56/// (cheap, `Arc`-backed) so it can be handed to
57/// `sui-rpc-api::StateReader::new(Arc::new(reader))`.
58pub struct RpcStoreReader<R: Reader = Db> {
59    /// The `Db` handle. Held separately from `schema` so trait
60    /// impls that need framework-level access (chain id, pipeline
61    /// watermarks) don't have to walk through a typed CF.
62    db: Db,
63
64    /// Typed handles to every CF the read paths exercise.
65    schema: Arc<RpcStoreSchema<R>>,
66
67    /// The pipelines registered on this deployment, whose watermarks
68    /// bound the reported indexed tip
69    /// (`get_highest_indexed_checkpoint_seq_number`). Kept explicit
70    /// rather than derived from the watermark CF's rows so a stale
71    /// row left behind by a pipeline that is no longer registered
72    /// cannot pin the reported tip, and so a registered pipeline
73    /// that has not committed yet correctly reads as "nothing fully
74    /// indexed".
75    pipelines: Arc<[&'static str]>,
76}
77
78impl<R: Reader> RpcStoreReader<R> {
79    /// Bind the adapter to an existing [`RpcStoreSchema`].
80    ///
81    /// `db` must be the same [`Db`] the schema was opened against.
82    /// Holding both separately is cheap (each is `Arc`-backed) and
83    /// avoids a `schema.epochs.reader().db()` style detour inside
84    /// hot read paths.
85    ///
86    /// The registered pipeline set defaults to the embedded
87    /// fullnode's cohorts ([`LIVE_COHORT`] + [`HISTORY_COHORT`]) —
88    /// the only in-tree deployment. A deployment that registers a
89    /// different set (e.g. a standalone node running the raw
90    /// chain-data pipelines) must override it via
91    /// [`Self::with_pipelines`] so the indexed-tip bound covers
92    /// exactly what it serves.
93    ///
94    /// [`LIVE_COHORT`]: crate::LIVE_COHORT
95    /// [`HISTORY_COHORT`]: crate::HISTORY_COHORT
96    pub fn new(db: Db, schema: Arc<RpcStoreSchema<R>>) -> Self {
97        let pipelines = crate::LIVE_COHORT
98            .iter()
99            .chain(crate::HISTORY_COHORT)
100            .copied()
101            .collect();
102        Self {
103            db,
104            schema,
105            pipelines,
106        }
107    }
108
109    /// Override the registered pipeline set whose watermarks bound
110    /// the reported indexed tip. See [`Self::new`] for the default.
111    pub fn with_pipelines(mut self, pipelines: impl IntoIterator<Item = &'static str>) -> Self {
112        self.pipelines = pipelines.into_iter().collect();
113        self
114    }
115
116    /// Borrow the underlying [`Db`] handle. Used by trait impls
117    /// that read directly from the framework CFs (chain id,
118    /// pipeline watermarks) rather than going through the typed
119    /// user schema.
120    pub fn db(&self) -> &Db {
121        &self.db
122    }
123
124    /// Borrow the typed schema this adapter is bound to. Trait
125    /// impls reach through this to call any of the inherent read
126    /// helpers `RpcStoreSchema` exposes per-CF.
127    pub fn schema(&self) -> &RpcStoreSchema<R> {
128        &self.schema
129    }
130
131    /// The highest checkpoint the live-object cohort (owned objects, types,
132    /// balances) has committed -- `min(checkpoint_hi_inclusive)` across its
133    /// pipelines, or `None` if any has no watermark yet.
134    ///
135    /// The embedded indexer follows the tip asynchronously, so this lags the
136    /// executed tip and bounds the checkpoint at which the live-object index
137    /// surface is readable. The history cohort backfills independently and is
138    /// deliberately excluded here -- its availability is exposed separately.
139    pub fn highest_live_committed_checkpoint(
140        &self,
141    ) -> sui_types::storage::error::Result<
142        Option<sui_types::messages_checkpoint::CheckpointSequenceNumber>,
143    > {
144        self.min_committed(crate::LIVE_COHORT.iter().copied())
145    }
146
147    /// The highest checkpoint every pipeline in `pipelines` has
148    /// committed (`min(checkpoint_hi_inclusive)`), or `None` if any of
149    /// them has no watermark yet. Private, but reachable from the
150    /// sibling trait-impl modules (children of this module).
151    fn min_committed(
152        &self,
153        pipelines: impl IntoIterator<Item = &'static str>,
154    ) -> sui_types::storage::error::Result<
155        Option<sui_types::messages_checkpoint::CheckpointSequenceNumber>,
156    > {
157        let framework = self.db().framework();
158        let mut min_hi: Option<u64> = None;
159        for name in pipelines {
160            let key = sui_consistent_store::PipelineTaskKey::new(name);
161            let Some(watermark) = framework
162                .watermarks
163                .get(&key)
164                .map_err(sui_types::storage::error::Error::custom)?
165            else {
166                return Ok(None);
167            };
168            let hi = watermark.checkpoint_hi_inclusive;
169            min_hi = Some(min_hi.map_or(hi, |m| m.min(hi)));
170        }
171        Ok(min_hi)
172    }
173}
174
175impl RpcStoreReader<Db> {
176    /// Re-project this reader against a captured [`Snapshot`].
177    ///
178    /// The returned [`RpcStoreReader<Snapshot>`] reads every CF
179    /// through the snapshot's `ReadOptions`, so reads are
180    /// consistent with the point in time at which the snapshot
181    /// was taken (rather than the tip). The original tip reader
182    /// is unaffected.
183    pub fn at_snapshot(&self, snap: &Snapshot) -> RpcStoreReader<Snapshot> {
184        RpcStoreReader {
185            db: self.db.clone(),
186            schema: Arc::new(self.schema.at(snap)),
187            pipelines: self.pipelines.clone(),
188        }
189    }
190}
191
192impl<R: Reader> Clone for RpcStoreReader<R> {
193    fn clone(&self) -> Self {
194        Self {
195            db: self.db.clone(),
196            schema: self.schema.clone(),
197            pipelines: self.pipelines.clone(),
198        }
199    }
200}
201
202#[cfg(test)]
203mod tests {
204    use sui_consistent_store::DbOptions;
205
206    use super::*;
207
208    #[test]
209    fn new_binds_db_and_schema() {
210        let dir = tempfile::tempdir().unwrap();
211        let (db, schema) = Db::open::<RpcStoreSchema>(dir.path(), DbOptions::default()).unwrap();
212        let reader = RpcStoreReader::new(db.clone(), Arc::new(schema));
213        // Smoke check: both handles are reachable and cloneable.
214        let _ = reader.clone();
215        assert!(reader.schema().get_pruning_watermarks().unwrap().is_none());
216    }
217
218    #[test]
219    fn at_snapshot_returns_snapshot_bound_reader() {
220        let dir = tempfile::tempdir().unwrap();
221        let (db, schema) = Db::open::<RpcStoreSchema>(dir.path(), DbOptions::default()).unwrap();
222        let reader = RpcStoreReader::new(db.clone(), Arc::new(schema));
223
224        db.take_snapshot(sui_consistent_store::Watermark::for_checkpoint(0));
225        let snap = db.at_snapshot(0).expect("snapshot retained");
226        let snap_reader = reader.at_snapshot(&snap);
227        // Both readers see the same (empty) state on a fresh DB.
228        assert!(reader.schema().get_pruning_watermarks().unwrap().is_none());
229        assert!(
230            snap_reader
231                .schema()
232                .get_pruning_watermarks()
233                .unwrap()
234                .is_none()
235        );
236    }
237}
sui_rpc_store/reader/mod.rs

sui_rpc_store/reader/
mod.rs
