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.observe(
85            summary
86                .timestamp()
87                .elapsed()
88                .unwrap_or_default()
89                .as_secs_f64(),
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            self.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.metrics
122                .last_skipped_checkpoint_signature_submission
123                .set(checkpoint_seq as i64);
124        }
125
126        if checkpoint_timestamp >= self.next_reconfiguration_timestamp_ms {
127            // close_epoch is ok if called multiple times
128            self.sender.close_epoch(epoch_store);
129        }
130        Ok(())
131    }
132}
133
134#[async_trait]
135impl CheckpointOutput for LogCheckpointOutput {
136    async fn checkpoint_created(
137        &self,
138        summary: &CheckpointSummary,
139        contents: &CheckpointContents,
140        _epoch_store: &Arc<AuthorityPerEpochStore>,
141        _checkpoint_store: &Arc<CheckpointStore>,
142    ) -> IotaResult {
143        trace!(
144            "Including following transactions in checkpoint {}: {:?}",
145            summary.sequence_number, contents
146        );
147        info!(
148            "Creating checkpoint {:?} at epoch {}, sequence {}, previous digest {:?}, transactions count {}, content digest {:?}, end_of_epoch_data {:?}",
149            summary.digest(),
150            summary.epoch,
151            summary.sequence_number,
152            summary.previous_digest,
153            contents.size(),
154            summary.content_digest,
155            summary.end_of_epoch_data,
156        );
157
158        Ok(())
159    }
160}
161
162#[async_trait]
163impl CertifiedCheckpointOutput for LogCheckpointOutput {
164    async fn certified_checkpoint_created(
165        &self,
166        summary: &CertifiedCheckpointSummary,
167    ) -> IotaResult {
168        debug!(
169            "Certified checkpoint with sequence {} and digest {}",
170            summary.sequence_number,
171            summary.digest()
172        );
173        Ok(())
174    }
175}
176
177pub struct SendCheckpointToStateSync {
178    handle: iota_network::state_sync::Handle,
179}
180
181impl SendCheckpointToStateSync {
182    pub fn new(handle: iota_network::state_sync::Handle) -> Self {
183        Self { handle }
184    }
185}
186
187#[async_trait]
188impl CertifiedCheckpointOutput for SendCheckpointToStateSync {
189    #[instrument(level = "debug", skip_all)]
190    async fn certified_checkpoint_created(
191        &self,
192        summary: &CertifiedCheckpointSummary,
193    ) -> IotaResult {
194        debug!(
195            "Certified checkpoint with sequence {} and digest {}",
196            summary.sequence_number,
197            summary.digest()
198        );
199        self.handle
200            .send_checkpoint(VerifiedCheckpoint::new_unchecked(summary.to_owned()))
201            .await;
202
203        Ok(())
204    }
205}
iota_core/checkpoints/checkpoint_output.rs

iota_core/checkpoints/
checkpoint_output.rs
