sui_rpc_store/indexer/
checkpoint_seq_by_digest.rs

1// Copyright (c) Mysten Labs, Inc.
2// SPDX-License-Identifier: Apache-2.0
3
4//! Sequential pipeline that populates the
5//! [`schema::checkpoint_seq_by_digest`](crate::schema::checkpoint_seq_by_digest)
6//! CF: one `CheckpointDigest → checkpoint_seq` row per checkpoint.
7
8use std::sync::Arc;
9
10use async_trait::async_trait;
11use sui_indexer_alt_framework::pipeline::Processor;
12use sui_indexer_alt_framework::pipeline::sequential;
13use sui_types::digests::CheckpointDigest;
14use sui_types::full_checkpoint_content::Checkpoint;
15use sui_types::message_envelope::Message;
16use sui_types::messages_checkpoint::CheckpointSummary;
17
18use crate::indexer::Schema;
19use crate::indexer::Store;
20use crate::schema::checkpoint_seq_by_digest;
21use crate::schema::primitives::U64Varint;
22
23/// Pipeline marker for `checkpoint_seq_by_digest`.
24pub struct CheckpointSeqByDigest;
25
26pub struct Row {
27    pub digest: CheckpointDigest,
28    pub seq: u64,
29}
30
31#[async_trait]
32impl Processor for CheckpointSeqByDigest {
33    const NAME: &'static str = "checkpoint_seq_by_digest";
34    type Value = Row;
35
36    async fn process(&self, checkpoint: &Arc<Checkpoint>) -> anyhow::Result<Vec<Row>> {
37        let summary: &CheckpointSummary = checkpoint.summary.data();
38        Ok(vec![Row {
39            digest: summary.digest(),
40            seq: summary.sequence_number,
41        }])
42    }
43}
44
45#[async_trait]
46impl sequential::Handler for CheckpointSeqByDigest {
47    type Store = Store;
48    type Batch = Vec<Row>;
49
50    fn batch(&self, batch: &mut Self::Batch, values: std::vec::IntoIter<Row>) {
51        batch.extend(values);
52    }
53
54    async fn commit<'a>(
55        &self,
56        batch: &Self::Batch,
57        conn: &mut sui_consistent_store::Connection<'a, Schema>,
58    ) -> anyhow::Result<usize> {
59        let cf = &conn.store.schema().checkpoint_seq_by_digest;
60        for row in batch {
61            conn.batch.put(
62                cf,
63                &checkpoint_seq_by_digest::Key(row.digest),
64                &U64Varint(row.seq),
65            )?;
66        }
67        Ok(batch.len())
68    }
69}
70
71#[cfg(test)]
72mod tests {
73    use std::sync::Arc;
74
75    use sui_types::test_checkpoint_data_builder::TestCheckpointBuilder;
76
77    use super::*;
78
79    #[tokio::test]
80    async fn process_emits_one_row_per_checkpoint() {
81        let checkpoint = Arc::new(TestCheckpointBuilder::new(11).build_checkpoint());
82        let rows = CheckpointSeqByDigest.process(&checkpoint).await.unwrap();
83        assert_eq!(rows.len(), 1);
84        assert_eq!(rows[0].seq, 11);
85    }
86}
sui_rpc_store/indexer/checkpoint_seq_by_digest.rs

sui_rpc_store/indexer/
checkpoint_seq_by_digest.rs
