sui_core/
rpc_store_ingestion_client.rs

1// Copyright (c) Mysten Labs, Inc.
2// SPDX-License-Identifier: Apache-2.0
3
4//! [`IngestionClientTrait`] backed by a fullnode's local checkpoint and
5//! perpetual stores.
6//!
7//! The embedded `sui-rpc-store` indexer fetches historical checkpoints by
8//! sequence number -- to backfill the history cohort and to fill gaps the live
9//! broadcast stream drops. On a fullnode every executed checkpoint's data
10//! already lives in the local stores, so this client assembles a [`Checkpoint`]
11//! from them via [`ReadStore::get_checkpoint_data`] rather than fetching from a
12//! remote object store or gRPC endpoint.
13//!
14//! [`Checkpoint`]: sui_types::full_checkpoint_content::Checkpoint
15
16use async_trait::async_trait;
17use sui_indexer_alt_framework::ingestion::ingestion_client::CheckpointError;
18use sui_indexer_alt_framework::ingestion::ingestion_client::CheckpointResult;
19use sui_indexer_alt_framework::ingestion::ingestion_client::IngestionClientTrait;
20use sui_types::digests::ChainIdentifier;
21use sui_types::storage::ReadStore;
22
23/// An [`IngestionClientTrait`] over any [`ReadStore`]. In production `R` is the
24/// fullnode's [`RocksDbStore`], which reads the local checkpoint and perpetual
25/// stores; the generic bound keeps it unit-testable against an in-memory store.
26///
27/// `chain_id` is captured at construction rather than derived from the genesis
28/// checkpoint, because genesis may have been pruned away on a long-running
29/// fullnode.
30///
31/// [`RocksDbStore`]: crate::storage::RocksDbStore
32pub struct PerpetualStoreIngestionClient<R> {
33    store: R,
34    chain_id: ChainIdentifier,
35}
36
37impl<R> PerpetualStoreIngestionClient<R> {
38    pub fn new(store: R, chain_id: ChainIdentifier) -> Self {
39        Self { store, chain_id }
40    }
41}
42
43#[async_trait]
44impl<R> IngestionClientTrait for PerpetualStoreIngestionClient<R>
45where
46    R: ReadStore + Send + Sync + 'static,
47{
48    async fn chain_id(&self) -> anyhow::Result<ChainIdentifier> {
49        Ok(self.chain_id)
50    }
51
52    async fn checkpoint(&self, checkpoint: u64) -> CheckpointResult {
53        // A checkpoint below the perpetual store's pruning watermark, or above
54        // the highest executed one, is simply absent. Report it as NotFound so
55        // the framework's `wait_for` can retry checkpoints that have not been
56        // executed yet, while pruned checkpoints stay permanently unavailable.
57        let lowest = self
58            .store
59            .get_lowest_available_checkpoint()
60            .map_err(|e| CheckpointError::Fetch(e.into()))?;
61        if checkpoint < lowest {
62            // The pruning gate holds the perpetual floor at or below every
63            // embedded pipeline's committed watermark, so this firing means a
64            // checkpoint was pruned out from under a pipeline that still
65            // needs it -- a permanently unserveable request the framework
66            // will retry forever, stalling that pipeline. This log line is
67            // the stall's primary signal; keep it specific.
68            tracing::error!(
69                checkpoint,
70                lowest_available = lowest,
71                "the embedded rpc-store indexer requested a checkpoint below \
72                 the perpetual store's pruning floor; the requesting pipeline \
73                 cannot make progress",
74            );
75        }
76
77        let summary = self
78            .store
79            .get_checkpoint_by_sequence_number(checkpoint)
80            .ok_or(CheckpointError::NotFound)?;
81        let contents = self
82            .store
83            .get_checkpoint_contents_by_digest(&summary.content_digest)
84            .ok_or(CheckpointError::NotFound)?;
85        self.store
86            .get_checkpoint_data(summary, contents)
87            .map_err(|e| CheckpointError::Fetch(anyhow::Error::from(e)))
88    }
89
90    async fn latest_checkpoint_number(&self) -> anyhow::Result<u64> {
91        Ok(self.store.get_latest_checkpoint_sequence_number()?)
92    }
93}
94
95#[cfg(test)]
96mod tests {
97    use super::*;
98    use crate::rpc_store_test_utils::store_with;
99    use crate::rpc_store_test_utils::test_chain_id;
100
101    #[tokio::test]
102    async fn chain_id_returns_configured_value() {
103        let client = PerpetualStoreIngestionClient::new(store_with([]), test_chain_id());
104        assert_eq!(client.chain_id().await.unwrap(), test_chain_id());
105    }
106
107    #[tokio::test]
108    async fn checkpoint_missing_is_not_found() {
109        let client = PerpetualStoreIngestionClient::new(store_with([1, 2]), test_chain_id());
110        assert!(matches!(
111            client.checkpoint(5).await,
112            Err(CheckpointError::NotFound)
113        ));
114    }
115
116    #[tokio::test]
117    async fn checkpoint_present_round_trips() {
118        let client = PerpetualStoreIngestionClient::new(store_with([0, 1, 2]), test_chain_id());
119        let cp = client.checkpoint(1).await.unwrap();
120        assert_eq!(*cp.summary.sequence_number(), 1);
121    }
122
123    #[tokio::test]
124    async fn checkpoint_summary_without_contents_is_not_found() {
125        let mut store = store_with([3]);
126        store.drop_contents_for = Some(3);
127        let client = PerpetualStoreIngestionClient::new(store, test_chain_id());
128        assert!(matches!(
129            client.checkpoint(3).await,
130            Err(CheckpointError::NotFound)
131        ));
132    }
133
134    #[tokio::test]
135    async fn latest_checkpoint_number_is_highest_executed() {
136        let client = PerpetualStoreIngestionClient::new(store_with([0, 4, 9]), test_chain_id());
137        assert_eq!(client.latest_checkpoint_number().await.unwrap(), 9);
138    }
139}
sui_core/rpc_store_ingestion_client.rs

sui_core/
rpc_store_ingestion_client.rs
