linera_service/cli/validator_benchmark/
mod.rs

1// Copyright (c) Zefchain Labs, Inc.
2// SPDX-License-Identifier: Apache-2.0
3
4//! Multi-layer pre-onboarding benchmark for a candidate validator.
5//!
6//! Tracking: linera-io/linera-infra#1198.
7
8mod bulk_download;
9mod config;
10mod latency;
11mod partial_sync;
12mod preflight;
13mod progress;
14mod read_latency;
15mod report;
16mod rpc;
17mod tip_lag;
18
19use std::{io::IsTerminal as _, time::Duration};
20
21use anyhow::Result;
22use chrono::Utc;
23pub use config::Benchmark;
24use linera_base::identifiers::ChainId;
25use linera_client::client_context::ClientContext;
26use linera_core::{
27    data_types::ChainInfoQuery,
28    node::{ValidatorNode, ValidatorNodeProvider as _},
29};
30
31use self::{
32    progress::Progress,
33    report::{Candidate, Layers, Metadata, Observer, OutputSpec, Report, Writer},
34};
35
36impl Benchmark {
37    /// Runs the pre-onboarding benchmark against the candidate validator.
38    pub async fn run(
39        &self,
40        context: &mut ClientContext<
41            impl linera_core::Environment<ValidatorNode = linera_rpc::Client>,
42        >,
43    ) -> Result<()> {
44        // Validate output specs up front so a typo fails fast, before any work.
45        let output_specs = OutputSpec::parse_all(&self.output)?;
46
47        let progress = Progress::new(!self.no_progress && std::io::stderr().is_terminal());
48        let rpc_timeout = Duration::from_secs(self.rpc_timeout_secs);
49
50        let started_at = Utc::now();
51        let node = context.make_node_provider().make_node(&self.address)?;
52        let writer = Writer::new(output_specs);
53
54        // Build the report up front and flush file targets after each layer, so
55        // an interrupted run still leaves the completed layers on disk.
56        let mut report = Report {
57            metadata: Metadata {
58                tool_version: env!("CARGO_PKG_VERSION").to_string(),
59                candidate: Candidate {
60                    address: self.address.clone(),
61                    public_key: self.public_key.map(|k| k.to_string()),
62                    version_info: None,
63                    network_description: None,
64                },
65                observer: Observer {
66                    location: self.observer_location.clone(),
67                    hostname: std::env::var("HOSTNAME").unwrap_or_default(),
68                    started_at: started_at.to_rfc3339(),
69                    ended_at: None,
70                    duration_secs: None,
71                },
72                config: serde_json::to_value(self)?,
73                chains_tested: self.chain.iter().map(|c| c.to_string()).collect(),
74                complete: false,
75            },
76            layers: Layers::default(),
77        };
78
79        // L1 preflight also yields version/network info for the report metadata.
80        // When skipped, fetch those two cheap fields best-effort anyway.
81        if !self.skip_preflight {
82            let outcome = preflight::run(&node, rpc_timeout, &progress).await;
83            if self.abort_on_preflight_fail
84                && outcome.report.status == report::PreflightStatus::Fail
85            {
86                progress.clear();
87                anyhow::bail!(
88                    "preflight failed for {}: {:?}",
89                    self.address,
90                    outcome.report.errors
91                );
92            }
93            report.metadata.candidate.version_info = outcome.version_info;
94            report.metadata.candidate.network_description = outcome.network_description;
95            report.layers.preflight = Some(outcome.report);
96        } else {
97            report.metadata.candidate.version_info =
98                rpc::timed(rpc_timeout, node.get_version_info())
99                    .await
100                    .ok()
101                    .map(|v| format!("{v:?}"));
102            report.metadata.candidate.network_description =
103                rpc::timed(rpc_timeout, node.get_network_description())
104                    .await
105                    .ok()
106                    .and_then(|nd| serde_json::to_value(nd).ok());
107        }
108        writer.write_files(&report)?;
109
110        // A not-yet-committee candidate may hold few or no blocks. Seed first when
111        // --deep so the read layers below exercise a candidate that actually has
112        // the data, and warn about any chain it does not hold and will not seed.
113        let deep_chain = self.deep.then(|| self.deep_chain.unwrap_or(self.chain[0]));
114        warn_unheld_chains(&node, &self.chain, deep_chain, rpc_timeout).await;
115        if let Some(deep_chain) = deep_chain {
116            report.layers.partial_sync = Some(
117                Box::pin(partial_sync::run(
118                    &node,
119                    context,
120                    deep_chain,
121                    self.deep_blocks,
122                    rpc_timeout,
123                    &progress,
124                ))
125                .await?,
126            );
127            writer.write_files(&report)?;
128        }
129
130        // Layer futures are large; box them at the await site (clippy::large_futures).
131        if !self.skip_read_baseline {
132            report.layers.read_baseline = Some(
133                Box::pin(read_latency::run_baseline(
134                    &node,
135                    &self.chain,
136                    self.baseline_requests,
137                    rpc_timeout,
138                    &progress,
139                ))
140                .await,
141            );
142            writer.write_files(&report)?;
143        }
144
145        if !self.skip_read_stress {
146            report.layers.read_stress = Some(
147                Box::pin(read_latency::run_stress(
148                    &node,
149                    &self.chain,
150                    &self.stress_levels,
151                    Duration::from_secs(self.stress_duration_secs),
152                    rpc_timeout,
153                    &progress,
154                ))
155                .await,
156            );
157            writer.write_files(&report)?;
158        }
159
160        if !self.skip_bulk_download {
161            report.layers.bulk_download = Some(
162                Box::pin(bulk_download::run(
163                    &node,
164                    &self.chain,
165                    self.bulk_batch_size,
166                    &self.bulk_concurrency,
167                    &self.bulk_height_range,
168                    rpc_timeout,
169                    &progress,
170                ))
171                .await?,
172            );
173            writer.write_files(&report)?;
174        }
175
176        if !self.skip_tip_lag {
177            report.layers.tip_lag = Some(
178                Box::pin(tip_lag::run(
179                    &node,
180                    context,
181                    &self.chain,
182                    self.tip_lag_samples,
183                    Duration::from_secs(self.tip_lag_interval_secs),
184                    rpc_timeout,
185                    &progress,
186                ))
187                .await?,
188            );
189            writer.write_files(&report)?;
190        }
191
192        let ended_at = Utc::now();
193        report.metadata.observer.ended_at = Some(ended_at.to_rfc3339());
194        report.metadata.observer.duration_secs =
195            Some(u64::try_from((ended_at - started_at).num_seconds()).unwrap_or(0));
196        report.metadata.complete = true;
197        progress.clear();
198        writer.emit(&report)?;
199        Ok(())
200    }
201}
202
203/// Warn about chains the candidate does not hold (read layers would be shallow),
204/// excluding one that `--deep` is about to seed. A chain with tip 0 or an
205/// unreachable lookup is treated as not held.
206async fn warn_unheld_chains(
207    node: &impl ValidatorNode,
208    chains: &[ChainId],
209    seeded: Option<ChainId>,
210    rpc_timeout: Duration,
211) {
212    let mut unheld = Vec::new();
213    for &chain in chains {
214        if Some(chain) == seeded {
215            continue;
216        }
217        let held = rpc::timed(
218            rpc_timeout,
219            node.handle_chain_info_query(ChainInfoQuery::new(chain)),
220        )
221        .await
222        .is_ok_and(|response| response.info.next_block_height.0 > 0);
223        if !held {
224            unheld.push(chain.to_string());
225        }
226    }
227    if !unheld.is_empty() {
228        tracing::warn!(
229            "candidate does not hold chain(s) [{}]; read layers (L3-L5) will be shallow. \
230             Pre-sync them (`linera validator sync`) or pass `--deep` to seed blocks first.",
231            unheld.join(", ")
232        );
233    }
234}
linera_service/cli/validator_benchmark/mod.rs

linera_service/cli/validator_benchmark/
mod.rs
