sui_core/checkpoints/
checkpoint_output.rs

1// Copyright (c) Mysten Labs, Inc.
2// SPDX-License-Identifier: Apache-2.0
3
4use crate::authority::StableSyncAuthoritySigner;
5use crate::authority::authority_per_epoch_store::AuthorityPerEpochStore;
6use crate::consensus_adapter::SubmitToConsensus;
7use async_trait::async_trait;
8use std::sync::Arc;
9use sui_types::base_types::AuthorityName;
10use sui_types::error::SuiResult;
11use sui_types::message_envelope::Message;
12use sui_types::messages_checkpoint::{
13    CertifiedCheckpointSummary, CheckpointContents, CheckpointSignatureMessage, CheckpointSummary,
14    SignedCheckpointSummary, VerifiedCheckpoint,
15};
16use sui_types::messages_consensus::ConsensusTransaction;
17use tracing::{debug, info, instrument, trace};
18
19use super::{CheckpointMetrics, CheckpointStore};
20
21#[async_trait]
22pub trait CheckpointOutput: Sync + Send + 'static {
23    async fn checkpoint_created(
24        &self,
25        summary: &CheckpointSummary,
26        contents: &CheckpointContents,
27        epoch_store: &Arc<AuthorityPerEpochStore>,
28        checkpoint_store: &Arc<CheckpointStore>,
29    ) -> SuiResult;
30}
31
32#[async_trait]
33pub trait CertifiedCheckpointOutput: Sync + Send + 'static {
34    async fn certified_checkpoint_created(&self, summary: &CertifiedCheckpointSummary)
35    -> SuiResult;
36}
37
38pub struct SubmitCheckpointToConsensus<T> {
39    sender: T,
40    signer: StableSyncAuthoritySigner,
41    authority: AuthorityName,
42    log_checkpoint_output: LogCheckpointOutput,
43}
44
45impl<T> SubmitCheckpointToConsensus<T> {
46    pub fn new(
47        sender: T,
48        signer: StableSyncAuthoritySigner,
49        authority: AuthorityName,
50        metrics: Arc<CheckpointMetrics>,
51    ) -> Self {
52        Self {
53            sender,
54            signer,
55            authority,
56            log_checkpoint_output: LogCheckpointOutput::new(metrics),
57        }
58    }
59}
60
61pub struct LogCheckpointOutput {
62    pub metrics: Arc<CheckpointMetrics>,
63}
64
65impl LogCheckpointOutput {
66    pub fn new(metrics: Arc<CheckpointMetrics>) -> Self {
67        Self { metrics }
68    }
69}
70
71#[async_trait]
72impl<T: SubmitToConsensus> CheckpointOutput for SubmitCheckpointToConsensus<T> {
73    #[instrument(level = "debug", skip_all)]
74    async fn checkpoint_created(
75        &self,
76        summary: &CheckpointSummary,
77        contents: &CheckpointContents,
78        epoch_store: &Arc<AuthorityPerEpochStore>,
79        checkpoint_store: &Arc<CheckpointStore>,
80    ) -> SuiResult {
81        self.log_checkpoint_output
82            .checkpoint_created(summary, contents, epoch_store, checkpoint_store)
83            .await?;
84
85        let checkpoint_timestamp = summary.timestamp_ms;
86        let checkpoint_seq = summary.sequence_number;
87
88        let highest_verified_checkpoint = checkpoint_store
89            .get_highest_verified_checkpoint()?
90            .map(|x| *x.sequence_number());
91
92        if Some(checkpoint_seq) > highest_verified_checkpoint {
93            debug!(
94                "Sending checkpoint signature at sequence {checkpoint_seq} to consensus, timestamp {checkpoint_timestamp}",
95            );
96
97            let summary = SignedCheckpointSummary::new(
98                epoch_store.epoch(),
99                summary.clone(),
100                &*self.signer,
101                self.authority,
102            );
103
104            let message = CheckpointSignatureMessage { summary };
105            assert!(
106                epoch_store
107                    .protocol_config()
108                    .consensus_checkpoint_signature_key_includes_digest()
109            );
110            let transaction = ConsensusTransaction::new_checkpoint_signature_message_v2(message);
111            self.sender
112                .submit_to_consensus(&[transaction], epoch_store)?;
113            self.log_checkpoint_output
114                .metrics
115                .last_sent_checkpoint_signature
116                .set(checkpoint_seq as i64);
117        } else {
118            debug!(
119                "Checkpoint at sequence {checkpoint_seq} is already certified, skipping signature submission to consensus",
120            );
121            self.log_checkpoint_output
122                .metrics
123                .last_skipped_checkpoint_signature_submission
124                .set(checkpoint_seq as i64);
125        }
126
127        Ok(())
128    }
129}
130
131#[async_trait]
132impl CheckpointOutput for LogCheckpointOutput {
133    async fn checkpoint_created(
134        &self,
135        summary: &CheckpointSummary,
136        contents: &CheckpointContents,
137        _epoch_store: &Arc<AuthorityPerEpochStore>,
138        _checkpoint_store: &Arc<CheckpointStore>,
139    ) -> SuiResult {
140        self.metrics.checkpoint_creation_latency.observe(
141            summary
142                .timestamp()
143                .elapsed()
144                .unwrap_or_default()
145                .as_secs_f64(),
146        );
147        self.metrics.checkpoint_creation_latency_ms.observe(
148            summary
149                .timestamp()
150                .elapsed()
151                .unwrap_or_default()
152                .as_millis() as u64,
153        );
154
155        trace!(
156            "Including following transactions in checkpoint {}: {:?}",
157            summary.sequence_number, contents
158        );
159        debug!(
160            "Creating checkpoint {:?} at epoch {}, sequence {}, previous digest {:?}, transactions count {}, content digest {:?}, end_of_epoch_data {:?}",
161            summary.digest(),
162            summary.epoch,
163            summary.sequence_number,
164            summary.previous_digest,
165            contents.size(),
166            summary.content_digest,
167            summary.end_of_epoch_data,
168        );
169
170        Ok(())
171    }
172}
173
174#[async_trait]
175impl CertifiedCheckpointOutput for LogCheckpointOutput {
176    async fn certified_checkpoint_created(
177        &self,
178        summary: &CertifiedCheckpointSummary,
179    ) -> SuiResult {
180        info!(
181            "Certified checkpoint with sequence {} and digest {}",
182            summary.sequence_number,
183            summary.digest()
184        );
185        Ok(())
186    }
187}
188
189pub struct SendCheckpointToStateSync {
190    handle: sui_network::state_sync::Handle,
191}
192
193impl SendCheckpointToStateSync {
194    pub fn new(handle: sui_network::state_sync::Handle) -> Self {
195        Self { handle }
196    }
197}
198
199#[async_trait]
200impl CertifiedCheckpointOutput for SendCheckpointToStateSync {
201    #[instrument(level = "debug", skip_all)]
202    async fn certified_checkpoint_created(
203        &self,
204        summary: &CertifiedCheckpointSummary,
205    ) -> SuiResult {
206        info!(
207            "Certified checkpoint with sequence {} and digest {}",
208            summary.sequence_number,
209            summary.digest()
210        );
211        self.handle
212            .send_checkpoint(VerifiedCheckpoint::new_unchecked(summary.to_owned()))
213            .await;
214
215        Ok(())
216    }
217}
sui_core/checkpoints/checkpoint_output.rs

sui_core/checkpoints/
checkpoint_output.rs
