sui_tool/

commands.rs

// Copyright (c) Mysten Labs, Inc.
// SPDX-License-Identifier: Apache-2.0

use crate::db_tool::{DbToolCommand, execute_db_tool_command, print_db_all_tables};
use crate::{
    ConciseObjectOutput, GroupedObjectOutput, SnapshotVerifyMode, VerboseObjectOutput,
    check_completed_snapshot, download_formal_snapshot, get_latest_available_epoch, get_object,
    get_transaction_block, make_clients, restore_from_db_checkpoint,
};
use anyhow::Result;
use consensus_core::storage::{Store, rocksdb_store::RocksDBStore};
use consensus_core::{BlockAPI, CommitAPI, CommitRange};
use futures::TryStreamExt;
use futures::future::join_all;
use std::path::PathBuf;
use std::{collections::BTreeMap, env, sync::Arc};
use sui_config::genesis::Genesis;
use sui_core::authority_client::AuthorityAPI;
use sui_protocol_config::Chain;
use sui_rpc_api::Client;
use sui_types::gas_coin::GasCoin;
use sui_types::messages_consensus::ConsensusTransaction;
use sui_types::transaction::Transaction;
use telemetry_subscribers::TracingHandle;

use sui_types::{
    base_types::*, crypto::AuthorityPublicKeyBytes, messages_grpc::TransactionInfoRequest,
};

use clap::*;
use sui_config::Config;
use sui_config::object_storage_config::{ObjectStoreConfig, ObjectStoreType};
use sui_types::messages_checkpoint::{
    CheckpointRequest, CheckpointResponse, CheckpointSequenceNumber,
};

#[derive(Parser, Clone, ValueEnum)]
pub enum Verbosity {
    Grouped,
    Concise,
    Verbose,
}

#[derive(Parser)]
pub enum ToolCommand {
    #[command(name = "scan-consensus-commits")]
    ScanConsensusCommits {
        #[arg(long = "db-path")]
        db_path: String,
        #[arg(long = "start-commit")]
        start_commit: Option<u32>,
        #[arg(long = "end-commit")]
        end_commit: Option<u32>,
    },

    /// Inspect if a specific object is or all gas objects owned by an address are locked by validators
    #[command(name = "locked-object")]
    LockedObject {
        /// Either id or address must be provided
        /// The object to check
        #[arg(long, help = "The object ID to fetch")]
        id: Option<ObjectID>,
        /// Either id or address must be provided
        /// If provided, check all gas objects owned by this account
        #[arg(long = "address")]
        address: Option<SuiAddress>,
        /// RPC address to provide the up-to-date committee info
        #[arg(long = "fullnode-rpc-url")]
        fullnode_rpc_url: String,
        /// Should attempt to rescue the object if it's locked but not fully locked
        #[arg(long = "rescue")]
        rescue: bool,
    },

    /// Fetch the same object from all validators
    #[command(name = "fetch-object")]
    FetchObject {
        #[arg(long, help = "The object ID to fetch")]
        id: ObjectID,

        #[arg(long, help = "Fetch object at a specific sequence")]
        version: Option<u64>,

        #[arg(
            long,
            help = "Validator to fetch from - if not specified, all validators are queried"
        )]
        validator: Option<AuthorityName>,

        // RPC address to provide the up-to-date committee info
        #[arg(long = "fullnode-rpc-url")]
        fullnode_rpc_url: String,

        /// Concise mode groups responses by results.
        /// prints tabular output suitable for processing with unix tools. For
        /// instance, to quickly check that all validators agree on the history of an object:
        /// ```text
        /// $ sui-tool fetch-object --id 0x260efde76ebccf57f4c5e951157f5c361cde822c \
        ///      --genesis $HOME/.sui/sui_config/genesis.blob \
        ///      --verbosity concise --concise-no-header
        /// ```
        #[arg(
            value_enum,
            long = "verbosity",
            default_value = "grouped",
            ignore_case = true
        )]
        verbosity: Verbosity,

        #[arg(
            long = "concise-no-header",
            help = "don't show header in concise output"
        )]
        concise_no_header: bool,
    },

    /// Fetch the effects association with transaction `digest`
    #[command(name = "fetch-transaction")]
    FetchTransaction {
        // RPC address to provide the up-to-date committee info
        #[arg(long = "fullnode-rpc-url")]
        fullnode_rpc_url: String,

        #[arg(long, help = "The transaction ID to fetch")]
        digest: TransactionDigest,

        /// If true, show the input transaction as well as the effects
        #[arg(long = "show-tx")]
        show_input_tx: bool,
    },

    /// Tool to read validator & node db.
    #[command(name = "db-tool")]
    DbTool {
        /// Path of the DB to read
        #[arg(long = "db-path")]
        db_path: String,
        #[command(subcommand)]
        cmd: Option<DbToolCommand>,
    },
    /// Download all packages to the local filesystem from a GraphQL service. Each package gets its
    /// own sub-directory, named for its ID on chain and version containing two metadata files
    /// (linkage.json and origins.json), a file containing the overall object and a file for every
    /// module it contains. Each module file is named for its module name, with a .mv suffix, and
    /// contains Move bytecode (suitable for passing into a disassembler).
    #[command(name = "dump-packages")]
    DumpPackages {
        /// Connection information for a GraphQL service.
        #[clap(long, short)]
        rpc_url: String,

        /// Path to a non-existent directory that can be created and filled with package information.
        #[clap(long, short)]
        output_dir: PathBuf,

        /// Only fetch packages that were created before this checkpoint (given by its sequence
        /// number).
        #[clap(long)]
        before_checkpoint: Option<u64>,

        /// If false (default), log level will be overridden to "off", and output will be reduced to
        /// necessary status information.
        #[clap(short, long = "verbose")]
        verbose: bool,
    },

    #[command(name = "dump-validators")]
    DumpValidators {
        #[arg(long = "genesis")]
        genesis: PathBuf,

        #[arg(
            long = "concise",
            help = "show concise output - name, protocol key and network address"
        )]
        concise: bool,
    },

    #[command(name = "dump-genesis")]
    DumpGenesis {
        #[arg(long = "genesis")]
        genesis: PathBuf,
    },

    /// Fetch authenticated checkpoint information at a specific sequence number.
    /// If sequence number is not specified, get the latest authenticated checkpoint.
    #[command(name = "fetch-checkpoint")]
    FetchCheckpoint {
        // RPC address to provide the up-to-date committee info
        #[arg(long = "fullnode-rpc-url")]
        fullnode_rpc_url: String,

        #[arg(long, help = "Fetch checkpoint at a specific sequence number")]
        sequence_number: Option<CheckpointSequenceNumber>,
    },

    #[command(name = "anemo")]
    Anemo {
        #[command(next_help_heading = "foo", flatten)]
        args: anemo_cli::Args,
    },

    #[command(name = "restore-db")]
    RestoreFromDBCheckpoint {
        #[arg(long = "config-path")]
        config_path: PathBuf,
        #[arg(long = "db-checkpoint-path")]
        db_checkpoint_path: PathBuf,
    },

    // Restore from formal (slim, DB agnostic) snapshot. Note that this is only supported
    /// for protocol versions supporting `commit_root_state_digest`. For mainnet, this is
    /// epoch 20+, and for testnet this is epoch 12+
    #[clap(
        name = "download-formal-snapshot",
        about = "Downloads formal database snapshot via cloud object store, outputs to local disk"
    )]
    DownloadFormalSnapshot {
        #[clap(long = "epoch", conflicts_with = "latest")]
        epoch: Option<u64>,
        #[clap(long = "genesis")]
        genesis: PathBuf,
        #[clap(long = "path")]
        path: PathBuf,
        /// Number of parallel downloads to perform. Defaults to 50, max 200.
        #[clap(long = "num-parallel-downloads")]
        num_parallel_downloads: Option<usize>,
        /// Number of parallel chunks for object insertion. Defaults to 8.
        #[clap(long = "num-parallel-chunks", default_value = "8")]
        num_parallel_chunks: usize,
        /// Verification mode to employ.
        #[clap(long = "verify", default_value = "normal")]
        verify: Option<SnapshotVerifyMode>,
        /// Network to download snapshot for. Defaults to "mainnet".
        /// If `--snapshot-bucket` or `--archive-bucket` is not specified,
        /// the value of this flag is used to construct default bucket names.
        #[clap(long = "network", default_value = "mainnet")]
        network: Chain,
        /// Snapshot bucket name. If not specified, defaults are
        /// based on value of `--network` flag.
        #[clap(long = "snapshot-bucket", conflicts_with = "no_sign_request")]
        snapshot_bucket: Option<String>,
        /// Snapshot bucket type
        #[clap(
            long = "snapshot-bucket-type",
            conflicts_with = "no_sign_request",
            help = "Required if --no-sign-request is not set"
        )]
        snapshot_bucket_type: Option<ObjectStoreType>,
        /// Path to snapshot directory on local filesystem.
        /// Only applicable if `--snapshot-bucket-type` is "file".
        #[clap(long = "snapshot-path")]
        snapshot_path: Option<PathBuf>,
        /// If true, no authentication is needed for snapshot restores
        #[clap(
            long = "no-sign-request",
            conflicts_with_all = &["snapshot_bucket", "snapshot_bucket_type"],
            help = "if set, no authentication is needed for snapshot restore"
        )]
        no_sign_request: bool,
        /// Download snapshot of the latest available epoch.
        /// If `--epoch` is specified, then this flag gets ignored.
        #[clap(
            long = "latest",
            conflicts_with = "epoch",
            help = "defaults to latest available snapshot in chosen bucket"
        )]
        latest: bool,
        /// If false (default), log level will be overridden to "off",
        /// and output will be reduced to necessary status information.
        #[clap(long = "verbose")]
        verbose: bool,

        /// Number of retries for failed HTTP requests when downloading snapshot files.
        /// Set to 0 to disable retries.
        #[clap(long = "max-retries", default_value = "10")]
        max_retries: usize,
        /// Port for the Prometheus metrics server. Defaults to 9185.
        #[clap(long = "metrics-port", default_value = "9185")]
        metrics_port: u16,
    },

    /// Interactive shell for navigating the validator database.
    #[command(name = "db-shell")]
    DbShell(crate::db_shell::DbShellArgs),

    /// Interactive Rhai shell for inspecting a TideHunter database.
    #[cfg(all(feature = "tideconsole", not(windows)))]
    #[command(name = "tideconsole")]
    TideConsole {
        /// Path to a TideHunter database directory to open on startup (bound to variable 'db').
        #[arg(short, long)]
        db: Option<PathBuf>,
        /// Rhai snippet to evaluate non-interactively, then exit.
        #[arg(short, long)]
        exec: Option<String>,
        /// Path to a Rhai script file to evaluate non-interactively, then exit.
        #[arg(short, long)]
        script: Option<PathBuf>,
    },
}

async fn check_locked_object(
    sui_client: &Client,
    committee: Arc<BTreeMap<AuthorityPublicKeyBytes, u64>>,
    id: ObjectID,
    rescue: bool,
) -> anyhow::Result<()> {
    let clients = Arc::new(make_clients(sui_client).await?);
    let output = get_object(id, None, None, clients.clone()).await?;
    let output = GroupedObjectOutput::new(output, committee);
    if output.fully_locked {
        println!("Object {} is fully locked.", id);
        return Ok(());
    }
    let top_record = output.voting_power.first().unwrap();
    let top_record_stake = top_record.1;
    let top_record = top_record.0.clone().unwrap();
    if top_record.4.is_none() {
        println!(
            "Object {} does not seem to be locked by majority of validators (unlocked stake: {})",
            id, top_record_stake
        );
        return Ok(());
    }

    let tx_digest = top_record.2;
    if !rescue {
        println!("Object {} is rescueable, top tx: {:?}", id, tx_digest);
        return Ok(());
    }
    println!("Object {} is rescueable, trying tx {}", id, tx_digest);
    let validator = output
        .grouped_results
        .get(&Some(top_record))
        .unwrap()
        .first()
        .unwrap();
    let client = &clients.get(validator).unwrap().1;
    let tx = client
        .handle_transaction_info_request(TransactionInfoRequest {
            transaction_digest: tx_digest,
        })
        .await?
        .transaction;
    let tx = Transaction::new(tx);
    let res = sui_client.clone().execute_transaction(&tx).await;
    match res {
        Ok(_) => {
            println!("Transaction executed successfully ({:?})", tx_digest);
        }
        Err(e) => {
            println!("Failed to execute transaction ({:?}): {:?}", tx_digest, e);
        }
    }
    Ok(())
}

impl ToolCommand {
    #[allow(clippy::format_in_format_args)]
    pub async fn execute(self, tracing_handle: TracingHandle) -> Result<(), anyhow::Error> {
        match self {
            ToolCommand::ScanConsensusCommits {
                db_path,
                start_commit,
                end_commit,
            } => {
                let rocks_db_store = RocksDBStore::new(&db_path);

                let start_commit = start_commit.unwrap_or(0);
                let end_commit = end_commit.unwrap_or(u32::MAX);

                let commits = rocks_db_store
                    .scan_commits(CommitRange::new(start_commit..=end_commit))
                    .unwrap();
                println!("found {} consensus commits", commits.len());

                for commit in commits {
                    let inner = &*commit;
                    let block_refs = inner.blocks();
                    let blocks = rocks_db_store.read_blocks(block_refs).unwrap();

                    for block in blocks.iter().flatten() {
                        let data = block.transactions_data();
                        println!(
                            "\"index\": \"{}\", \"leader\": \"{}\", \"blocks\": \"{:#?}\", {} txs",
                            inner.index(),
                            inner.leader(),
                            inner.blocks(),
                            data.len()
                        );
                        for txns in &data {
                            let tx: ConsensusTransaction = bcs::from_bytes(txns).unwrap();
                            println!("\t{:?}", tx.key());
                        }
                    }
                }
            }
            ToolCommand::LockedObject {
                id,
                fullnode_rpc_url,
                rescue,
                address,
            } => {
                let sui_client = Client::new(fullnode_rpc_url)?;
                let committee = Arc::new(
                    sui_client
                        .get_committee(None)
                        .await?
                        .voting_rights
                        .into_iter()
                        .collect::<BTreeMap<_, _>>(),
                );
                let object_ids = match id {
                    Some(id) => vec![id],
                    None => {
                        let address = address.expect("Either id or address must be provided");
                        sui_client
                            .list_owned_objects(address, Some(GasCoin::type_()))
                            .map_ok(|o| o.id())
                            .try_collect()
                            .await?
                    }
                };
                for ids in object_ids.chunks(30) {
                    let mut tasks = vec![];
                    for id in ids {
                        tasks.push(check_locked_object(
                            &sui_client,
                            committee.clone(),
                            *id,
                            rescue,
                        ))
                    }
                    join_all(tasks)
                        .await
                        .into_iter()
                        .collect::<Result<Vec<_>, _>>()?;
                }
            }
            ToolCommand::FetchObject {
                id,
                validator,
                version,
                fullnode_rpc_url,
                verbosity,
                concise_no_header,
            } => {
                let sui_client = Client::new(fullnode_rpc_url)?;
                let clients = Arc::new(make_clients(&sui_client).await?);
                let output = get_object(id, version, validator, clients).await?;

                match verbosity {
                    Verbosity::Grouped => {
                        let committee = Arc::new(
                            sui_client
                                .get_committee(None)
                                .await?
                                .voting_rights
                                .into_iter()
                                .collect::<BTreeMap<_, _>>(),
                        );
                        println!("{}", GroupedObjectOutput::new(output, committee));
                    }
                    Verbosity::Verbose => {
                        println!("{}", VerboseObjectOutput(output));
                    }
                    Verbosity::Concise => {
                        if !concise_no_header {
                            println!("{}", ConciseObjectOutput::header());
                        }
                        println!("{}", ConciseObjectOutput(output));
                    }
                }
            }
            ToolCommand::FetchTransaction {
                digest,
                show_input_tx,
                fullnode_rpc_url,
            } => {
                print!(
                    "{}",
                    get_transaction_block(digest, show_input_tx, fullnode_rpc_url).await?
                );
            }
            ToolCommand::DbTool { db_path, cmd } => {
                let path = PathBuf::from(db_path);
                match cmd {
                    Some(c) => execute_db_tool_command(path, c).await?,
                    None => print_db_all_tables(path)?,
                }
            }
            ToolCommand::DumpPackages {
                rpc_url,
                output_dir,
                before_checkpoint,
                verbose,
            } => {
                if !verbose {
                    tracing_handle
                        .update_log("off")
                        .expect("Failed to update log level");
                }

                sui_package_dump::dump(rpc_url, output_dir, before_checkpoint).await?;
            }
            ToolCommand::DumpValidators { genesis, concise } => {
                let genesis = Genesis::load(genesis).unwrap();
                if !concise {
                    println!("{:#?}", genesis.validator_set_for_tooling());
                } else {
                    for (i, val_info) in genesis.validator_set_for_tooling().iter().enumerate() {
                        let metadata = val_info.verified_metadata();
                        println!(
                            "#{:<2} {:<20} {:?} {:?} {}",
                            i,
                            metadata.name,
                            metadata.sui_pubkey_bytes().concise(),
                            metadata.net_address,
                            anemo::PeerId(metadata.network_pubkey.0.to_bytes()),
                        )
                    }
                }
            }
            ToolCommand::DumpGenesis { genesis } => {
                let genesis = Genesis::load(genesis)?;
                println!("{:#?}", genesis);
            }
            ToolCommand::FetchCheckpoint {
                sequence_number,
                fullnode_rpc_url,
            } => {
                let sui_client = Client::new(fullnode_rpc_url)?;
                let clients = make_clients(&sui_client).await?;

                for (name, (_, client)) in clients {
                    let resp = client
                        .handle_checkpoint(CheckpointRequest {
                            sequence_number,
                            request_content: true,
                        })
                        .await
                        .unwrap();
                    let CheckpointResponse {
                        checkpoint,
                        contents,
                    } = resp;

                    let summary = checkpoint.clone().unwrap().data().clone();
                    // write summary to file
                    let mut file = std::fs::File::create("/tmp/ckpt_summary")
                        .expect("Failed to create /tmp/summary");
                    let bytes =
                        bcs::to_bytes(&summary).expect("Failed to serialize summary to BCS");
                    use std::io::Write;
                    file.write_all(&bytes)
                        .expect("Failed to write summary to /tmp/ckpt_summary");

                    println!("Validator: {:?}\n", name.concise());
                    println!("Checkpoint: {:?}\n", checkpoint);
                    println!("Content: {:?}\n", contents);
                }
            }
            ToolCommand::Anemo { args } => {
                let config = crate::make_anemo_config();
                anemo_cli::run(config, args).await
            }
            ToolCommand::RestoreFromDBCheckpoint {
                config_path,
                db_checkpoint_path,
            } => {
                let config = sui_config::NodeConfig::load(config_path)?;
                restore_from_db_checkpoint(&config, &db_checkpoint_path).await?;
            }
            ToolCommand::DownloadFormalSnapshot {
                epoch,
                genesis,
                path,
                num_parallel_downloads,
                num_parallel_chunks,
                verify,
                network,
                snapshot_bucket,
                snapshot_bucket_type,
                snapshot_path,
                no_sign_request,
                latest,
                verbose,
                max_retries,
                metrics_port,
            } => {
                if !verbose {
                    tracing_handle
                        .update_log("off")
                        .expect("Failed to update log level");
                }
                let num_parallel_downloads = num_parallel_downloads.unwrap_or(50).min(200);
                let snapshot_bucket =
                    snapshot_bucket.or_else(|| match (network, no_sign_request) {
                        (Chain::Mainnet, false) => Some(
                            env::var("MAINNET_FORMAL_SIGNED_BUCKET")
                                .unwrap_or("mysten-mainnet-formal".to_string()),
                        ),
                        (Chain::Mainnet, true) => env::var("MAINNET_FORMAL_UNSIGNED_BUCKET").ok(),
                        (Chain::Testnet, true) => env::var("TESTNET_FORMAL_UNSIGNED_BUCKET").ok(),
                        (Chain::Testnet, _) => Some(
                            env::var("TESTNET_FORMAL_SIGNED_BUCKET")
                                .unwrap_or("mysten-testnet-formal".to_string()),
                        ),
                        (Chain::Unknown, _) => {
                            panic!("Cannot generate default snapshot bucket for unknown network");
                        }
                    });

                let aws_endpoint = env::var("AWS_SNAPSHOT_ENDPOINT").ok().or_else(|| {
                    if no_sign_request {
                        if network == Chain::Mainnet {
                            Some("https://formal-snapshot.mainnet.sui.io".to_string())
                        } else if network == Chain::Testnet {
                            Some("https://formal-snapshot.testnet.sui.io".to_string())
                        } else {
                            None
                        }
                    } else {
                        None
                    }
                });

                let snapshot_bucket_type = if no_sign_request {
                    ObjectStoreType::S3
                } else {
                    snapshot_bucket_type
                        .expect("You must set either --snapshot-bucket-type or --no-sign-request")
                };
                let snapshot_store_config = match snapshot_bucket_type {
                    ObjectStoreType::S3 => ObjectStoreConfig {
                        object_store: Some(ObjectStoreType::S3),
                        bucket: snapshot_bucket.filter(|s| !s.is_empty()),
                        aws_access_key_id: env::var("AWS_SNAPSHOT_ACCESS_KEY_ID").ok(),
                        aws_secret_access_key: env::var("AWS_SNAPSHOT_SECRET_ACCESS_KEY").ok(),
                        aws_region: env::var("AWS_SNAPSHOT_REGION").ok(),
                        aws_endpoint: aws_endpoint.filter(|s| !s.is_empty()),
                        aws_virtual_hosted_style_request: env::var(
                            "AWS_SNAPSHOT_VIRTUAL_HOSTED_REQUESTS",
                        )
                        .ok()
                        .and_then(|b| b.parse().ok())
                        .unwrap_or(no_sign_request),
                        object_store_connection_limit: 200,
                        no_sign_request,
                        ..Default::default()
                    },
                    ObjectStoreType::GCS => ObjectStoreConfig {
                        object_store: Some(ObjectStoreType::GCS),
                        bucket: snapshot_bucket,
                        google_service_account: env::var("GCS_SNAPSHOT_SERVICE_ACCOUNT_FILE_PATH")
                            .ok(),
                        object_store_connection_limit: 200,
                        no_sign_request,
                        ..Default::default()
                    },
                    ObjectStoreType::Azure => ObjectStoreConfig {
                        object_store: Some(ObjectStoreType::Azure),
                        bucket: snapshot_bucket,
                        azure_storage_account: env::var("AZURE_SNAPSHOT_STORAGE_ACCOUNT").ok(),
                        azure_storage_access_key: env::var("AZURE_SNAPSHOT_STORAGE_ACCESS_KEY")
                            .ok(),
                        object_store_connection_limit: 200,
                        no_sign_request,
                        ..Default::default()
                    },
                    ObjectStoreType::File => {
                        if snapshot_path.is_some() {
                            ObjectStoreConfig {
                                object_store: Some(ObjectStoreType::File),
                                directory: snapshot_path,
                                ..Default::default()
                            }
                        } else {
                            panic!(
                                "--snapshot-path must be specified for --snapshot-bucket-type=file"
                            );
                        }
                    }
                };

                let ingestion_url = match network {
                    Chain::Mainnet => "https://checkpoints.mainnet.sui.io",
                    Chain::Testnet => "https://checkpoints.testnet.sui.io",
                    _ => panic!("Cannot generate default ingestion url for unknown network"),
                };

                let latest_available_epoch =
                    latest.then_some(get_latest_available_epoch(&snapshot_store_config).await?);
                let epoch_to_download = epoch.or(latest_available_epoch).expect(
                    "Either pass epoch with --epoch <epoch_num> or use latest with --latest",
                );

                if let Err(e) =
                    check_completed_snapshot(&snapshot_store_config, epoch_to_download).await
                {
                    panic!(
                        "Aborting snapshot restore: {}, snapshot may not be uploaded yet",
                        e
                    );
                }

                let verify = verify.unwrap_or_default();
                download_formal_snapshot(
                    &path,
                    epoch_to_download,
                    &genesis,
                    snapshot_store_config,
                    ingestion_url,
                    num_parallel_downloads,
                    num_parallel_chunks,
                    network,
                    verify,
                    max_retries,
                    metrics_port,
                )
                .await?;
            }
            ToolCommand::DbShell(args) => {
                tokio::task::spawn_blocking(move || crate::db_shell::run(args)).await??;
            }
            #[cfg(all(feature = "tideconsole", not(windows)))]
            ToolCommand::TideConsole { db, exec, script } => {
                tokio::task::spawn_blocking(move || crate::tideconsole_cmd::run(db, exec, script))
                    .await??;
            }
        };
        Ok(())
    }
}