sui_rpc_loadgen/payload/
get_checkpoints.rs

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
// Copyright (c) Mysten Labs, Inc.
// SPDX-License-Identifier: Apache-2.0

use crate::payload::validation::check_transactions;
use crate::payload::{GetCheckpoints, ProcessPayload, RpcCommandProcessor, SignerInfo};
use anyhow::Result;
use async_trait::async_trait;
use dashmap::DashSet;
use futures::future::join_all;
use itertools::Itertools;
use std::sync::Arc;

use crate::payload::checkpoint_utils::get_latest_checkpoint_stats;
use sui_json_rpc_types::CheckpointId;
use sui_types::base_types::TransactionDigest;
use tokio::sync::Mutex;
use tracing::log::warn;
use tracing::{debug, error, info};

#[async_trait]
impl<'a> ProcessPayload<'a, &'a GetCheckpoints> for RpcCommandProcessor {
    async fn process(
        &'a self,
        op: &'a GetCheckpoints,
        _signer_info: &Option<SignerInfo>,
    ) -> Result<()> {
        let clients = self.get_clients().await?;

        let checkpoint_stats = get_latest_checkpoint_stats(&clients, op.end).await;
        let max_checkpoint = checkpoint_stats.max_latest_checkpoint();
        debug!("GetCheckpoints({}, {:?})", op.start, max_checkpoint,);

        // TODO(chris): read `cross_validate` from config
        let cross_validate = true;

        for seq in op.start..=max_checkpoint {
            let transaction_digests: Arc<Mutex<DashSet<TransactionDigest>>> =
                Arc::new(Mutex::new(DashSet::new()));
            let checkpoints = join_all(clients.iter().enumerate().map(|(i, client)| {
                let transaction_digests = transaction_digests.clone();
                let end_checkpoint_for_clients = checkpoint_stats.latest_checkpoints.clone();
                async move {
                    if end_checkpoint_for_clients[i] < seq {
                        // TODO(chris) log actual url
                        warn!(
                            "The RPC server corresponding to the {i}th url has a outdated checkpoint number {}.\
                            The latest checkpoint number is {seq}",
                            end_checkpoint_for_clients[i]
                        );
                        return None;
                    }

                    match client
                        .read_api()
                        .get_checkpoint(CheckpointId::SequenceNumber(seq))
                        .await {
                        Ok(t) => {
                            if t.sequence_number != seq {
                                error!("The RPC server corresponding to the {i}th url has unexpected checkpoint sequence number {}, expected {seq}", t.sequence_number,);
                            }
                            for digest in t.transactions.iter() {
                                transaction_digests.lock().await.insert(*digest);
                            }
                            Some(t)
                        },
                        Err(err) => {
                            error!("Failed to fetch checkpoint {seq} on the {i}th url: {err}");
                            None
                        }
                    }
                }
            }))
                .await;

            let transaction_digests = transaction_digests
                .lock()
                .await
                .iter()
                .map(|digest| *digest)
                .collect::<Vec<_>>();

            if op.verify_transactions {
                let transaction_responses = check_transactions(
                    &clients,
                    &transaction_digests,
                    cross_validate,
                    op.verify_objects,
                )
                .await
                .into_iter()
                .concat();

                if op.record {
                    debug!("adding addresses and object ids from response");
                    self.add_addresses_from_response(&transaction_responses);
                    self.add_object_ids_from_response(&transaction_responses);
                };
            }

            if op.record {
                debug!("adding transaction digests from response");
                self.add_transaction_digests(transaction_digests);
            };

            if cross_validate {
                let valid_checkpoint = checkpoints.iter().enumerate().find_map(|(i, x)| {
                    if x.is_some() {
                        Some((i, x.clone().unwrap()))
                    } else {
                        None
                    }
                });

                if valid_checkpoint.is_none() {
                    error!("none of the urls are returning valid checkpoint for seq {seq}");
                    continue;
                }
                // safe to unwrap because we check some above
                let (valid_checkpoint_idx, valid_checkpoint) = valid_checkpoint.unwrap();
                for (i, x) in checkpoints.iter().enumerate() {
                    if i == valid_checkpoint_idx {
                        continue;
                    }
                    // ignore the None value because it's warned above
                    let eq = x.is_none() || x.as_ref().unwrap() == &valid_checkpoint;
                    if !eq {
                        error!("getCheckpoint {seq} has a different result between the {valid_checkpoint_idx}th and {i}th URL {:?} {:?}", x, checkpoints[valid_checkpoint_idx])
                    }
                }
            }

            if seq % 10000 == 0 {
                info!("Finished processing checkpoint {seq}");
            }
        }

        Ok(())
    }
}