iota_core/checkpoints/
checkpoint_output.rs

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

iota_core/checkpoints/
checkpoint_output.rs
