diff --git a/Cargo.lock b/Cargo.lock index 31cd6a97..b97ff55f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3167,6 +3167,26 @@ dependencies = [ "tokio", ] +[[package]] +name = "guardian-multisig-e2e-benchmark" +version = "0.1.0" +dependencies = [ + "anyhow", + "chrono", + "clap", + "guardian-client", + "hex", + "miden-multisig-client", + "miden-protocol", + "rand 0.9.2", + "rand_chacha 0.9.0", + "serde", + "serde_json", + "tempfile", + "tokio", + "toml 0.8.23", +] + [[package]] name = "guardian-prod-benchmarks" version = "0.1.0" diff --git a/Cargo.toml b/Cargo.toml index 73816df9..bc940c5e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,5 +1,6 @@ [workspace] members = [ + "benchmarks/multisig-e2e", "benchmarks/prod-server", "crates/shared", "crates/client", diff --git a/benchmarks/multisig-e2e/.gitignore b/benchmarks/multisig-e2e/.gitignore new file mode 100644 index 00000000..344c9846 --- /dev/null +++ b/benchmarks/multisig-e2e/.gitignore @@ -0,0 +1,2 @@ +reports/*.json +reports/*.jsonl diff --git a/benchmarks/multisig-e2e/Cargo.toml b/benchmarks/multisig-e2e/Cargo.toml new file mode 100644 index 00000000..8616be78 --- /dev/null +++ b/benchmarks/multisig-e2e/Cargo.toml @@ -0,0 +1,21 @@ +[package] +name = "guardian-multisig-e2e-benchmark" +version = "0.1.0" +edition = "2024" +publish = false + +[dependencies] +anyhow = { workspace = true } +chrono = { workspace = true, features = ["serde"] } +clap = { version = "4.5", features = ["derive"] } +guardian-client = { path = "../../crates/client" } +hex = { workspace = true } +miden-multisig-client = { path = "../../crates/miden-multisig-client" } +miden-protocol = { workspace = true } +rand = { workspace = true } +rand_chacha = { workspace = true } +serde = { workspace = true, features = ["derive"] } +serde_json = { workspace = true } +tempfile = { workspace = true } +tokio = { workspace = true, features = ["fs", "macros", "rt-multi-thread", "sync", "time"] } +toml = "0.8" diff --git a/benchmarks/multisig-e2e/README.md b/benchmarks/multisig-e2e/README.md new file mode 100644 index 00000000..1f456f3b --- /dev/null +++ b/benchmarks/multisig-e2e/README.md @@ -0,0 +1,99 @@ +# Guardian multisig end-to-end benchmark + +This benchmark measures a real multisig proposal lifecycle. Two persistent 1-of-1 accounts +alternate P2ID transfers through Guardian while the Miden transactions are proved and submitted. +A deterministic fraction of received notes is consumed. + +It is intentionally separate from `benchmarks/prod-server`: that harness measures Guardian API +capacity with synthetic deltas, while this one measures the full Rust SDK, Guardian, prover, and +Miden network path. + +## Local Guardian workflow + +Start a local Guardian server configured for the same Miden network as the benchmark. Create the +two persistent accounts: + +```bash +cargo run -p guardian-multisig-e2e-benchmark -- prepare \ + --miden-endpoint https://rpc.testnet.miden.io +``` + +`prepare` registers Alice and Bob with Guardian and writes their account IDs and Falcon secret keys +to `.guardian/bench/multisig-e2e-accounts.json`. The file is mode `0600` on Unix and is ignored by +git. It is not overwritten automatically because the account and secret-key binding must remain +stable. + +Each generated account is persisted before it is registered with Guardian. If preparation stops +partway through, the partial fixture retains every key that may have been registered. Preserve or +move that file before deciding whether to reprovision. + +Fund both printed account IDs from the faucet selected for the run. Update `testnet.local.toml` +with the faucet's hex or bech32 address, then turn the funding notes into spendable vault assets: + +```bash +cargo run -p guardian-multisig-e2e-benchmark -- bootstrap \ + --config benchmarks/multisig-e2e/testnet.local.toml +``` + +Check balances and connectivity without mutating either account: + +```bash +cargo run -p guardian-multisig-e2e-benchmark -- preflight \ + --config benchmarks/multisig-e2e/testnet.local.toml +``` + +Run the benchmark: + +```bash +cargo run --release -p guardian-multisig-e2e-benchmark -- run \ + --config benchmarks/multisig-e2e/testnet.local.toml +``` + +## Measurements and artifacts + +Each completed send is flushed as one JSONL record under `reports/`. A companion manifest captures +the endpoints, public account IDs, workload parameters, and random seed without copying account +secrets. If a run stops early, a failure sidecar captures the operation and full error chain. + +The primary timings are: + +- `send_proposal_ms`: end-to-end proposal time, including retry attempts and retry sleep. +- `send_proposal_retry_wait_ms`: time spent sleeping between proposal attempts. +- active proposal time: derived as proposal time minus retry sleep. +- `send_execution_ms`: send execution, proving, and submission. +- `note_visibility_ms`: time until the receiver can discover the new note. +- `total_ms`: the complete operation, including optional note consumption. + +The summary compares both end-to-end and active proposal medians in the first and last workload +quintiles. The active comparison separates Guardian/client work from configured retry sleep. + +Canonicalization observations are queued during the run and collected after foreground account +operations stop. The separate `canonicalization.json` artifact uses Guardian's `canonical_at` +timestamp. A timestamp at or before the local observation start is recorded as zero rather than +falling back to the deferred wall clock. All observations share one `timeout_seconds` drain +deadline, so final collection is bounded independently of observation count. + +If Guardian reports that a prior delta is still pending, the benchmark waits +`proposal_retry_interval_ms`, then reruns the proposal workflow against the latest account state. +Transient Guardian connection failures use the same retry deadline for proposal creation and +execution. Miden's transient `block_to`-ahead-of-chain-tip sync race is also retried regardless of +which sync endpoint reports it. Other errors fail immediately. + +`max_duration_seconds` stops scheduling new operations at an operation boundary. The shared +canonicalization drain runs afterward before the final summary is written. + +Summarize a completed or manually interrupted JSONL report: + +```bash +cargo run -p guardian-multisig-e2e-benchmark -- summarize \ + --report benchmarks/multisig-e2e/reports/multisig-e2e-.jsonl +``` + +## Funding calculation + +The runner requires a conservative starting vault balance of +`amount * ceil(operations / 2)` for each account. This guarantees that deterministic consume +choices cannot make a run fail for lack of spendable funds. Funding notes do not count until +`bootstrap` has consumed them. + +For a quick smoke, lower `operations` before attempting a long run. diff --git a/benchmarks/multisig-e2e/src/canonicalization.rs b/benchmarks/multisig-e2e/src/canonicalization.rs new file mode 100644 index 00000000..ba5fd241 --- /dev/null +++ b/benchmarks/multisig-e2e/src/canonicalization.rs @@ -0,0 +1,285 @@ +use std::cell::RefCell; +use std::fs; +use std::path::Path; +use std::time::{Duration, Instant}; + +use anyhow::{Context, Result, anyhow}; +use chrono::{DateTime, Utc}; +use guardian_client::GuardianClient; +use miden_multisig_client::AccountId; +use serde::Serialize; + +use crate::fixture::Fixture; +use crate::runtime::load_observer; + +#[derive(Debug, Clone, Copy, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum ProposalKind { + Send, + Consume, +} + +#[derive(Debug)] +struct Observation { + operation: u64, + kind: ProposalKind, + nonce: u64, + started_at: DateTime, + started: Instant, +} + +#[derive(Debug)] +struct QueuedObservation { + account: String, + observation: Observation, +} + +#[derive(Debug, Clone, Copy, Serialize)] +#[serde(rename_all = "snake_case")] +enum ObservationStatus { + Canonical, + Discarded, + TimedOut, + ObservationFailed, +} + +#[derive(Debug, Serialize)] +struct CanonicalizationRecord { + operation: u64, + account: String, + account_id: String, + proposal_kind: ProposalKind, + nonce: u64, + started_at: DateTime, + observed_at: DateTime, + elapsed_ms: u64, + polls: u64, + status: ObservationStatus, + canonical_at: Option, + error: Option, +} + +pub struct CanonicalizationTracker { + fixture: Fixture, + observations: RefCell>, + poll_interval: Duration, + timeout: Duration, +} + +impl CanonicalizationTracker { + pub async fn start( + fixture: &Fixture, + poll_interval: Duration, + timeout: Duration, + ) -> Result { + for account in &fixture.accounts { + AccountId::from_hex(&account.account_id) + .with_context(|| format!("invalid account ID for {}", account.label))?; + } + + Ok(Self { + fixture: fixture.clone(), + observations: RefCell::new(Vec::new()), + poll_interval, + timeout, + }) + } + + pub fn observe( + &self, + account: &str, + operation: u64, + kind: ProposalKind, + nonce: u64, + ) -> Result<()> { + if !self + .fixture + .accounts + .iter() + .any(|fixture_account| fixture_account.label == account) + { + return Err(anyhow!( + "no canonicalization observer for account {account}" + )); + } + self.observations.borrow_mut().push(QueuedObservation { + account: account.to_string(), + observation: Observation { + operation, + kind, + nonce, + started_at: Utc::now(), + started: Instant::now(), + }, + }); + Ok(()) + } + + pub async fn finish(self, path: &Path) -> Result<()> { + let mut records = Vec::new(); + let observations = self.observations.into_inner(); + let deadline = Instant::now() + self.timeout; + println!( + "collecting {} canonicalization observations with a shared {}s deadline", + observations.len(), + self.timeout.as_secs() + ); + for account in &self.fixture.accounts { + let mut observer = load_observer(&self.fixture, account).await?; + let account_id = AccountId::from_hex(&account.account_id) + .with_context(|| format!("invalid account ID for {}", account.label))?; + for queued in observations + .iter() + .filter(|queued| queued.account == account.label) + { + records.push( + observe_delta( + &mut observer, + &account.label, + account_id, + &queued.observation, + self.poll_interval, + deadline, + ) + .await, + ); + } + } + records.sort_by_key(|record| (record.operation, kind_order(record.proposal_kind))); + fs::write(path, serde_json::to_vec_pretty(&records)?) + .with_context(|| format!("failed to write {}", path.display())) + } +} + +async fn observe_delta( + observer: &mut GuardianClient, + label: &str, + account_id: AccountId, + observation: &Observation, + poll_interval: Duration, + deadline: Instant, +) -> CanonicalizationRecord { + let mut polls = 0; + + loop { + polls += 1; + let poll_error = match observer.get_delta(&account_id, observation.nonce).await { + Ok(response) => { + if let Some(delta) = response.delta { + if let Some(canonical_at) = delta.canonical_at { + return record( + label, + account_id, + observation, + polls, + ObservationStatus::Canonical, + Some(canonical_at), + None, + ); + } + if delta.discarded_at.is_some() { + return record( + label, + account_id, + observation, + polls, + ObservationStatus::Discarded, + None, + None, + ); + } + } + None + } + Err(error) => Some(error.to_string()), + }; + + let now = Instant::now(); + if now >= deadline { + let status = if poll_error.is_some() { + ObservationStatus::ObservationFailed + } else { + ObservationStatus::TimedOut + }; + return record( + label, + account_id, + observation, + polls, + status, + None, + poll_error, + ); + } + tokio::time::sleep(poll_interval.min(deadline - now)).await; + } +} + +fn record( + label: &str, + account_id: AccountId, + observation: &Observation, + polls: u64, + status: ObservationStatus, + canonical_at: Option, + error: Option, +) -> CanonicalizationRecord { + let elapsed_ms = canonical_at + .as_deref() + .and_then(|value| canonical_elapsed_ms(observation.started_at, value)) + .unwrap_or_else(|| wall_elapsed_ms(observation.started)); + CanonicalizationRecord { + operation: observation.operation, + account: label.to_string(), + account_id: account_id.to_string(), + proposal_kind: observation.kind, + nonce: observation.nonce, + started_at: observation.started_at, + observed_at: Utc::now(), + elapsed_ms, + polls, + status, + canonical_at, + error, + } +} + +fn kind_order(kind: ProposalKind) -> u8 { + match kind { + ProposalKind::Send => 0, + ProposalKind::Consume => 1, + } +} + +fn canonical_elapsed_ms(started_at: DateTime, canonical_at: &str) -> Option { + let canonical_at = DateTime::parse_from_rfc3339(canonical_at) + .ok()? + .with_timezone(&Utc); + Some(u64::try_from((canonical_at - started_at).num_milliseconds()).unwrap_or(0)) +} + +fn wall_elapsed_ms(started: Instant) -> u64 { + u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn canonical_elapsed_clamps_timestamp_before_observation_to_zero() { + let started_at = DateTime::parse_from_rfc3339("2026-07-23T08:00:01Z") + .unwrap() + .with_timezone(&Utc); + + assert_eq!( + canonical_elapsed_ms(started_at, "2026-07-23T08:00:00Z"), + Some(0) + ); + } + + #[test] + fn canonical_elapsed_rejects_invalid_timestamp() { + assert_eq!(canonical_elapsed_ms(Utc::now(), "not-a-timestamp"), None); + } +} diff --git a/benchmarks/multisig-e2e/src/config.rs b/benchmarks/multisig-e2e/src/config.rs new file mode 100644 index 00000000..548291cc --- /dev/null +++ b/benchmarks/multisig-e2e/src/config.rs @@ -0,0 +1,152 @@ +use std::fs; +use std::path::{Path, PathBuf}; + +use anyhow::{Context, Result, anyhow, bail}; +use miden_multisig_client::Endpoint; +use serde::Deserialize; + +#[derive(Debug, Clone, Deserialize)] +pub struct RunConfig { + pub accounts_file: PathBuf, + pub faucet_id: String, + #[serde(default = "default_operations")] + pub operations: u64, + #[serde(default = "default_amount")] + pub amount: u64, + #[serde(default = "default_consume_probability")] + pub consume_probability: f64, + #[serde(default = "default_seed")] + pub seed: u64, + #[serde(default = "default_poll_interval_ms")] + pub poll_interval_ms: u64, + #[serde(default = "default_timeout_seconds")] + pub timeout_seconds: u64, + #[serde(default = "default_proposal_retry_interval_ms")] + pub proposal_retry_interval_ms: u64, + #[serde(default = "default_proposal_retry_timeout_seconds")] + pub proposal_retry_timeout_seconds: u64, + #[serde(default)] + pub max_duration_seconds: Option, + #[serde(default = "default_artifacts_dir")] + pub artifacts_dir: PathBuf, +} + +impl RunConfig { + pub fn load(path: &Path) -> Result { + let contents = fs::read_to_string(path) + .with_context(|| format!("failed to read benchmark config {}", path.display()))?; + let config: Self = toml::from_str(&contents) + .with_context(|| format!("failed to parse benchmark config {}", path.display()))?; + config.validate()?; + Ok(config) + } + + fn validate(&self) -> Result<()> { + if self.operations == 0 { + bail!("operations must be greater than zero"); + } + if self.amount == 0 { + bail!("amount must be greater than zero"); + } + if !(0.0..=1.0).contains(&self.consume_probability) { + bail!("consume_probability must be between 0 and 1"); + } + if self.poll_interval_ms == 0 || self.timeout_seconds == 0 { + bail!("poll_interval_ms and timeout_seconds must be greater than zero"); + } + if self.proposal_retry_interval_ms == 0 || self.proposal_retry_timeout_seconds == 0 { + bail!( + "proposal_retry_interval_ms and proposal_retry_timeout_seconds must be greater than zero" + ); + } + if self.max_duration_seconds == Some(0) { + bail!("max_duration_seconds must be greater than zero when set"); + } + Ok(()) + } +} + +pub fn parse_miden_endpoint(input: &str) -> Result { + let (protocol, authority) = input + .split_once("://") + .ok_or_else(|| anyhow!("Miden endpoint must start with http:// or https://"))?; + if protocol != "http" && protocol != "https" { + bail!("unsupported Miden endpoint protocol '{protocol}'"); + } + if authority.is_empty() || authority.contains('/') { + bail!("Miden endpoint must contain only a host and optional port"); + } + + let (host, port) = match authority.rsplit_once(':') { + Some((host, port)) if !host.is_empty() => { + let port = port + .parse::() + .with_context(|| format!("invalid Miden endpoint port '{port}'"))?; + (host.to_string(), Some(port)) + } + _ => (authority.to_string(), None), + }; + Ok(Endpoint::new(protocol.to_string(), host, port)) +} + +fn default_operations() -> u64 { + 300 +} + +fn default_amount() -> u64 { + 1 +} + +fn default_consume_probability() -> f64 { + 0.5 +} + +fn default_seed() -> u64 { + 42 +} + +fn default_poll_interval_ms() -> u64 { + 1_000 +} + +fn default_timeout_seconds() -> u64 { + 180 +} + +fn default_proposal_retry_interval_ms() -> u64 { + 1_000 +} + +fn default_proposal_retry_timeout_seconds() -> u64 { + 180 +} + +fn default_artifacts_dir() -> PathBuf { + PathBuf::from("benchmarks/multisig-e2e/reports") +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn parses_endpoint_with_port() { + let endpoint = parse_miden_endpoint("http://localhost:57291").unwrap(); + assert_eq!(endpoint.protocol(), "http"); + assert_eq!(endpoint.host(), "localhost"); + assert_eq!(endpoint.port(), Some(57291)); + } + + #[test] + fn parses_endpoint_without_port() { + let endpoint = parse_miden_endpoint("https://rpc.devnet.miden.io").unwrap(); + assert_eq!(endpoint.protocol(), "https"); + assert_eq!(endpoint.host(), "rpc.devnet.miden.io"); + assert_eq!(endpoint.port(), None); + } + + #[test] + fn rejects_endpoint_path() { + assert!(parse_miden_endpoint("https://rpc.devnet.miden.io/path").is_err()); + } +} diff --git a/benchmarks/multisig-e2e/src/fixture.rs b/benchmarks/multisig-e2e/src/fixture.rs new file mode 100644 index 00000000..8d650c53 --- /dev/null +++ b/benchmarks/multisig-e2e/src/fixture.rs @@ -0,0 +1,175 @@ +use std::fs; +use std::io::Write; +use std::path::Path; + +use anyhow::{Context, Result, bail}; +use miden_multisig_client::{MultisigClient, SecretKey}; +use miden_protocol::utils::serde::Serializable; +use serde::{Deserialize, Serialize}; +use tempfile::{NamedTempFile, TempDir}; + +use crate::config::parse_miden_endpoint; + +const FIXTURE_VERSION: u32 = 1; + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct AccountFixture { + pub label: String, + pub account_id: String, + pub secret_key_hex: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct Fixture { + pub version: u32, + pub guardian_endpoint: String, + pub miden_endpoint: String, + pub accounts: Vec, +} + +impl Fixture { + pub fn load(path: &Path) -> Result { + let contents = fs::read_to_string(path) + .with_context(|| format!("failed to read account fixture {}", path.display()))?; + let fixture: Self = serde_json::from_str(&contents) + .with_context(|| format!("failed to parse account fixture {}", path.display()))?; + if fixture.version != FIXTURE_VERSION { + bail!( + "unsupported account fixture version {}; expected {}", + fixture.version, + FIXTURE_VERSION + ); + } + if fixture.accounts.len() != 2 { + bail!("account fixture must contain exactly two accounts"); + } + Ok(fixture) + } +} + +pub async fn prepare( + guardian_endpoint: String, + miden_endpoint: String, + output: &Path, +) -> Result { + if output.exists() { + bail!( + "refusing to overwrite existing account fixture {}; move it explicitly to reprovision", + output.display() + ); + } + let endpoint = parse_miden_endpoint(&miden_endpoint)?; + let mut fixture = Fixture { + version: FIXTURE_VERSION, + guardian_endpoint, + miden_endpoint, + accounts: Vec::with_capacity(2), + }; + + for label in ["alice", "bob"] { + let secret_key = SecretKey::new(); + let secret_key_hex = hex::encode(secret_key.to_bytes()); + let data_dir = + TempDir::new().context("failed to create temporary Miden client directory")?; + let mut client = MultisigClient::builder() + .miden_endpoint(endpoint.clone()) + .guardian_endpoint(fixture.guardian_endpoint.clone()) + .account_dir(data_dir.path()) + .with_secret_key(secret_key) + .build() + .await + .with_context(|| format!("failed to build {label} client"))?; + let commitment = client.user_commitment(); + let account_id = client + .create_account(1, vec![commitment]) + .await + .with_context(|| format!("failed to create {label} account"))? + .id(); + fixture.accounts.push(AccountFixture { + label: label.to_string(), + account_id: account_id.to_string(), + secret_key_hex, + }); + persist_fixture(output, &fixture)?; + client + .push_account() + .await + .with_context(|| format!("failed to register {label} account with Guardian"))?; + } + + Ok(fixture) +} + +fn persist_fixture(output: &Path, fixture: &Fixture) -> Result<()> { + let parent = output + .parent() + .filter(|parent| !parent.as_os_str().is_empty()); + if let Some(parent) = parent { + fs::create_dir_all(parent) + .with_context(|| format!("failed to create fixture directory {}", parent.display()))?; + } + let directory = parent.unwrap_or_else(|| Path::new(".")); + let mut temporary = NamedTempFile::new_in(directory) + .with_context(|| format!("failed to create fixture file in {}", directory.display()))?; + serde_json::to_writer_pretty(temporary.as_file_mut(), fixture)?; + temporary.as_file_mut().write_all(b"\n")?; + temporary.as_file_mut().sync_all()?; + restrict_permissions(temporary.path())?; + temporary + .persist(output) + .map_err(|error| error.error) + .with_context(|| format!("failed to write account fixture {}", output.display()))?; + Ok(()) +} + +#[cfg(unix)] +fn restrict_permissions(path: &Path) -> Result<()> { + use std::os::unix::fs::PermissionsExt; + + fs::set_permissions(path, fs::Permissions::from_mode(0o600)) + .with_context(|| format!("failed to restrict permissions on {}", path.display())) +} + +#[cfg(not(unix))] +fn restrict_permissions(path: &Path) -> Result<()> { + bail!( + "cannot restrict permissions on {} on this platform; refusing to persist secret keys unprotected", + path.display() + ) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn persists_partial_fixture_for_account_recovery() { + let directory = TempDir::new().unwrap(); + let path = directory.path().join("accounts.json"); + let mut fixture = Fixture { + version: FIXTURE_VERSION, + guardian_endpoint: "http://localhost:50051".to_string(), + miden_endpoint: "https://rpc.testnet.miden.io".to_string(), + accounts: vec![AccountFixture { + label: "alice".to_string(), + account_id: "0xalice".to_string(), + secret_key_hex: "secret".to_string(), + }], + }; + + persist_fixture(&path, &fixture).unwrap(); + + let persisted: Fixture = serde_json::from_str(&fs::read_to_string(&path).unwrap()).unwrap(); + assert_eq!(persisted.accounts[0].secret_key_hex, "secret"); + + fixture.accounts.push(AccountFixture { + label: "bob".to_string(), + account_id: "0xbob".to_string(), + secret_key_hex: "another-secret".to_string(), + }); + persist_fixture(&path, &fixture).unwrap(); + + let persisted: Fixture = serde_json::from_str(&fs::read_to_string(path).unwrap()).unwrap(); + assert_eq!(persisted.accounts.len(), 2); + } +} diff --git a/benchmarks/multisig-e2e/src/main.rs b/benchmarks/multisig-e2e/src/main.rs new file mode 100644 index 00000000..6a895804 --- /dev/null +++ b/benchmarks/multisig-e2e/src/main.rs @@ -0,0 +1,88 @@ +mod canonicalization; +mod config; +mod fixture; +mod runner; +mod runtime; + +use std::path::PathBuf; + +use anyhow::Result; +use clap::{Parser, Subcommand}; +use miden_multisig_client::AccountId; +use miden_protocol::address::NetworkId; + +use config::RunConfig; + +#[derive(Debug, Parser)] +#[command(about = "Real multisig proposal benchmark for Guardian")] +struct Cli { + #[command(subcommand)] + command: Command, +} + +#[derive(Debug, Subcommand)] +enum Command { + Prepare { + #[arg(long, default_value = "http://localhost:50051")] + guardian_endpoint: String, + #[arg(long, default_value = "https://rpc.devnet.miden.io")] + miden_endpoint: String, + #[arg(long, default_value = ".guardian/bench/multisig-e2e-accounts.json")] + accounts_file: PathBuf, + }, + Preflight { + #[arg(long)] + config: PathBuf, + }, + Bootstrap { + #[arg(long)] + config: PathBuf, + }, + Run { + #[arg(long)] + config: PathBuf, + }, + Summarize { + #[arg(long)] + report: PathBuf, + }, +} + +#[tokio::main] +async fn main() -> Result<()> { + match Cli::parse().command { + Command::Prepare { + guardian_endpoint, + miden_endpoint, + accounts_file, + } => { + let fixture = + fixture::prepare(guardian_endpoint, miden_endpoint, &accounts_file).await?; + println!("wrote {}", accounts_file.display()); + let network_id = if fixture.miden_endpoint.contains("testnet") { + NetworkId::Testnet + } else { + NetworkId::Devnet + }; + for account in fixture.accounts { + let account_id = AccountId::from_hex(&account.account_id)?; + println!( + "{}: {} ({})", + account.label, + account.account_id, + account_id.to_bech32(network_id.clone()) + ); + } + } + Command::Preflight { config } => runner::preflight(&RunConfig::load(&config)?).await?, + Command::Bootstrap { config } => runner::bootstrap(&RunConfig::load(&config)?).await?, + Command::Run { config } => { + let path = runner::run(&RunConfig::load(&config)?).await?; + println!("wrote {}", path.display()); + } + Command::Summarize { report } => { + runner::summarize_report(&report)?; + } + } + Ok(()) +} diff --git a/benchmarks/multisig-e2e/src/runner.rs b/benchmarks/multisig-e2e/src/runner.rs new file mode 100644 index 00000000..86f5b1c4 --- /dev/null +++ b/benchmarks/multisig-e2e/src/runner.rs @@ -0,0 +1,912 @@ +use std::collections::HashSet; +use std::fs::{self, File}; +use std::io::{BufRead, BufReader, BufWriter, Write}; +use std::path::{Path, PathBuf}; +use std::time::{Duration, Instant}; + +use anyhow::{Context, Result, anyhow, bail}; +use chrono::{DateTime, Utc}; +use guardian_client::GuardianClient; +use miden_multisig_client::{ + AccountId, ConsumableNote, MultisigError, NoteFilter, Proposal, TransactionType, +}; +use rand::{Rng, SeedableRng}; +use rand_chacha::ChaCha8Rng; +use serde::{Deserialize, Serialize}; + +use crate::canonicalization::{CanonicalizationTracker, ProposalKind}; +use crate::config::RunConfig; +use crate::fixture::Fixture; +use crate::runtime::{BenchClient, is_miden_sync_tip_ahead, load_clients, load_observer}; + +#[derive(Debug, Deserialize, Serialize)] +pub struct OperationRecord { + pub operation: u64, + pub started_at: DateTime, + pub sender: String, + pub receiver: String, + pub amount: u64, + pub consumed: bool, + pub send_proposal_id: String, + pub send_nonce: u64, + #[serde(default)] + pub send_proposal_retries: u64, + #[serde(default)] + pub send_proposal_retry_wait_ms: u64, + pub send_proposal_ms: u64, + pub send_execution_ms: u64, + pub send_canonicalization_ms: Option, + pub note_visibility_ms: u64, + pub note_id: String, + pub consume_proposal_id: Option, + pub consume_nonce: Option, + #[serde(default)] + pub consume_proposal_retries: Option, + #[serde(default)] + pub consume_proposal_retry_wait_ms: Option, + pub consume_proposal_ms: Option, + pub consume_execution_ms: Option, + pub consume_canonicalization_ms: Option, + pub total_ms: u64, +} + +#[derive(Debug, Serialize)] +struct RunManifest { + schema_version: u32, + started_at: DateTime, + guardian_endpoint: String, + miden_endpoint: String, + account_ids: Vec, + faucet_id: String, + operations: u64, + amount: u64, + consume_probability: f64, + seed: u64, + poll_interval_ms: u64, + timeout_seconds: u64, + proposal_retry_interval_ms: u64, + proposal_retry_timeout_seconds: u64, + max_duration_seconds: Option, + records_file: String, + canonicalization_file: String, +} + +#[derive(Debug, Serialize)] +struct FailureRecord { + failed_at: DateTime, + operation: u64, + sender: String, + receiver: String, + error: String, +} + +#[derive(Debug, Clone, Copy)] +struct OperationSpec { + index: u64, + faucet_id: AccountId, + amount: u64, + consumed: bool, +} + +struct ProposalAttempt { + proposal: Proposal, + retries: u64, + retry_wait_ms: u64, +} + +pub async fn preflight(config: &RunConfig) -> Result<()> { + let fixture = Fixture::load(&config.accounts_file)?; + let faucet_id = parse_faucet_id(config)?; + let mut clients = load_clients(&fixture, config).await?; + + for client in &mut clients { + sync_network_with_retry(client, config) + .await + .with_context(|| { + format!( + "failed to refresh notes for {} during preflight", + client.label + ) + })?; + let notes = client + .client + .list_consumable_notes_filtered(NoteFilter::by_faucet(faucet_id)) + .await?; + let note_balance: u64 = notes + .iter() + .map(|note| note.amount_for_faucet(faucet_id)) + .sum(); + let nonce = client + .client + .account() + .ok_or_else(|| anyhow!("{} account is not loaded", client.label))? + .nonce(); + println!( + "{}: account={} nonce={} vault_balance={} consumable_note_balance={} notes={}", + client.label, + client.account_id, + nonce, + client.balance(faucet_id), + note_balance, + notes.len() + ); + } + + let required = required_balance(config); + println!( + "worst-case starting vault balance per account for this profile: {}", + required + ); + Ok(()) +} + +pub async fn bootstrap(config: &RunConfig) -> Result<()> { + let fixture = Fixture::load(&config.accounts_file)?; + let faucet_id = parse_faucet_id(config)?; + let mut clients = load_clients(&fixture, config).await?; + + for (client, fixture_account) in clients.iter_mut().zip(&fixture.accounts) { + let mut observer = load_observer(&fixture, fixture_account).await?; + sync_network_with_retry(client, config).await?; + let notes = client + .client + .list_consumable_notes_filtered(NoteFilter::by_faucet(faucet_id)) + .await?; + if notes.is_empty() { + println!("{}: no consumable faucet notes", client.label); + continue; + } + for note in notes { + let amount = note.amount_for_faucet(faucet_id); + let proposal = propose_with_retry( + client, + TransactionType::consume_notes(vec![note.id]), + config, + ) + .await + .with_context(|| format!("failed to create {} bootstrap proposal", client.label))? + .proposal; + ensure_ready(&proposal.status, &proposal.id)?; + execute_with_retry(client, &proposal.id, config) + .await + .with_context(|| { + format!("failed to execute {} bootstrap proposal", client.label) + })?; + await_canonical( + &mut observer, + &client.label, + client.account_id, + proposal.nonce, + config, + ) + .await?; + println!( + "{}: consumed note {} ({} units), nonce {}", + client.label, note.id, amount, proposal.nonce + ); + } + } + Ok(()) +} + +pub async fn run(config: &RunConfig) -> Result { + let fixture = Fixture::load(&config.accounts_file)?; + let faucet_id = parse_faucet_id(config)?; + let mut clients = load_clients(&fixture, config).await?; + ensure_starting_balances(&clients, faucet_id, config)?; + fs::create_dir_all(&config.artifacts_dir).with_context(|| { + format!( + "failed to create artifact directory {}", + config.artifacts_dir.display() + ) + })?; + let started_at = Utc::now(); + let run_name = format!("multisig-e2e-{}", started_at.format("%Y%m%dT%H%M%S%.3fZ")); + let path = config.artifacts_dir.join(format!("{run_name}.jsonl")); + let canonicalization_path = config + .artifacts_dir + .join(format!("{run_name}.canonicalization.json")); + write_manifest( + config, + &fixture, + started_at, + path.file_name() + .and_then(|name| name.to_str()) + .ok_or_else(|| anyhow!("artifact path is not valid UTF-8"))?, + canonicalization_path + .file_name() + .and_then(|name| name.to_str()) + .ok_or_else(|| anyhow!("canonicalization path is not valid UTF-8"))?, + config + .artifacts_dir + .join(format!("{run_name}.manifest.json")), + )?; + let file = File::create(&path) + .with_context(|| format!("failed to create artifact {}", path.display()))?; + let mut writer = BufWriter::new(file); + let mut rng = ChaCha8Rng::seed_from_u64(config.seed); + let mut records = Vec::with_capacity(config.operations as usize); + let run_started = Instant::now(); + let tracker = CanonicalizationTracker::start( + &fixture, + Duration::from_millis(config.poll_interval_ms), + Duration::from_secs(config.timeout_seconds), + ) + .await?; + + for operation in 0..config.operations { + if duration_limit_reached(run_started, config.max_duration_seconds) { + println!( + "duration limit reached after {} completed operations", + records.len() + ); + break; + } + let consumed = rng.random_bool(config.consume_probability); + let (sender, receiver) = client_pair(&mut clients, operation); + let spec = OperationSpec { + index: operation, + faucet_id, + amount: config.amount, + consumed, + }; + let result = execute_operation(spec, sender, receiver, config, &tracker).await; + let record = match result { + Ok(record) => record, + Err(error) => { + writer.flush()?; + let failure = FailureRecord { + failed_at: Utc::now(), + operation: operation + 1, + sender: sender.label.clone(), + receiver: receiver.label.clone(), + error: format!("{error:#}"), + }; + let failure_path = config + .artifacts_dir + .join(format!("{run_name}.failure.json")); + fs::write(&failure_path, serde_json::to_vec_pretty(&failure)?)?; + tracker.finish(&canonicalization_path).await?; + return Err(error).with_context(|| { + format!( + "operation {} failed; details written to {}", + operation + 1, + failure_path.display() + ) + }); + } + }; + serde_json::to_writer(&mut writer, &record)?; + writer.write_all(b"\n")?; + writer.flush()?; + println!( + "{}/{} {} -> {} proposal={}ms retries={} retry_wait={}ms execute={}ms canonical=deferred note={}ms consumed={}", + operation + 1, + config.operations, + record.sender, + record.receiver, + record.send_proposal_ms, + record.send_proposal_retries, + record.send_proposal_retry_wait_ms, + record.send_execution_ms, + record.note_visibility_ms, + record.consumed + ); + records.push(record); + } + + writer.flush()?; + drop(writer); + tracker.finish(&canonicalization_path).await?; + write_and_print_summary(&path)?; + Ok(path) +} + +fn write_manifest( + config: &RunConfig, + fixture: &Fixture, + started_at: DateTime, + records_file: &str, + canonicalization_file: &str, + path: PathBuf, +) -> Result<()> { + let manifest = RunManifest { + schema_version: 4, + started_at, + guardian_endpoint: fixture.guardian_endpoint.clone(), + miden_endpoint: fixture.miden_endpoint.clone(), + account_ids: fixture + .accounts + .iter() + .map(|account| account.account_id.clone()) + .collect(), + faucet_id: config.faucet_id.clone(), + operations: config.operations, + amount: config.amount, + consume_probability: config.consume_probability, + seed: config.seed, + poll_interval_ms: config.poll_interval_ms, + timeout_seconds: config.timeout_seconds, + proposal_retry_interval_ms: config.proposal_retry_interval_ms, + proposal_retry_timeout_seconds: config.proposal_retry_timeout_seconds, + max_duration_seconds: config.max_duration_seconds, + records_file: records_file.to_string(), + canonicalization_file: canonicalization_file.to_string(), + }; + fs::write(&path, serde_json::to_vec_pretty(&manifest)?) + .with_context(|| format!("failed to write run manifest {}", path.display())) +} + +async fn execute_operation( + spec: OperationSpec, + sender: &mut BenchClient, + receiver: &mut BenchClient, + config: &RunConfig, + tracker: &CanonicalizationTracker, +) -> Result { + let started_at = Utc::now(); + let started = Instant::now(); + let existing_notes = consumable_note_ids(receiver, spec.faucet_id, config).await?; + let proposal_started = Instant::now(); + let send_attempt = propose_with_retry( + sender, + TransactionType::transfer(receiver.account_id, spec.faucet_id, spec.amount), + config, + ) + .await?; + let send_proposal_ms = elapsed_ms(proposal_started); + let send = send_attempt.proposal; + ensure_ready(&send.status, &send.id)?; + + let execution_started = Instant::now(); + execute_with_retry(sender, &send.id, config).await?; + let send_execution_ms = elapsed_ms(execution_started); + tracker.observe( + &sender.label, + spec.index + 1, + ProposalKind::Send, + send.nonce, + )?; + let note_started = Instant::now(); + let note = await_new_note( + receiver, + spec.faucet_id, + spec.amount, + &existing_notes, + config, + ) + .await?; + let note_visibility_ms = elapsed_ms(note_started); + + let mut consume_proposal_id = None; + let mut consume_nonce = None; + let mut consume_proposal_retries = None; + let mut consume_proposal_retry_wait_ms = None; + let mut consume_proposal_ms = None; + let mut consume_execution_ms = None; + if spec.consumed { + let proposal_started = Instant::now(); + let consume_attempt = propose_with_retry( + receiver, + TransactionType::consume_notes(vec![note.id]), + config, + ) + .await?; + consume_proposal_ms = Some(elapsed_ms(proposal_started)); + consume_proposal_retries = Some(consume_attempt.retries); + consume_proposal_retry_wait_ms = Some(consume_attempt.retry_wait_ms); + let consume = consume_attempt.proposal; + ensure_ready(&consume.status, &consume.id)?; + let execution_started = Instant::now(); + execute_with_retry(receiver, &consume.id, config).await?; + consume_execution_ms = Some(elapsed_ms(execution_started)); + tracker.observe( + &receiver.label, + spec.index + 1, + ProposalKind::Consume, + consume.nonce, + )?; + consume_proposal_id = Some(consume.id); + consume_nonce = Some(consume.nonce); + } + + Ok(OperationRecord { + operation: spec.index + 1, + started_at, + sender: sender.label.clone(), + receiver: receiver.label.clone(), + amount: spec.amount, + consumed: spec.consumed, + send_proposal_id: send.id, + send_nonce: send.nonce, + send_proposal_retries: send_attempt.retries, + send_proposal_retry_wait_ms: send_attempt.retry_wait_ms, + send_proposal_ms, + send_execution_ms, + send_canonicalization_ms: None, + note_visibility_ms, + note_id: note.id.to_string(), + consume_proposal_id, + consume_nonce, + consume_proposal_retries, + consume_proposal_retry_wait_ms, + consume_proposal_ms, + consume_execution_ms, + consume_canonicalization_ms: None, + total_ms: elapsed_ms(started), + }) +} + +async fn propose_with_retry( + client: &mut BenchClient, + transaction_type: TransactionType, + config: &RunConfig, +) -> Result { + let deadline = Instant::now() + Duration::from_secs(config.proposal_retry_timeout_seconds); + let retry_interval = Duration::from_millis(config.proposal_retry_interval_ms); + let mut retries = 0; + let mut retry_wait_ms = 0; + + loop { + match client + .client + .propose_transaction(transaction_type.clone()) + .await + { + Ok(proposal) => { + return Ok(ProposalAttempt { + proposal, + retries, + retry_wait_ms, + }); + } + Err(error) if is_retryable_proposal_error(&error) => { + let now = Instant::now(); + if now >= deadline { + bail!( + "timed out retrying proposal for {} after transient error: {}", + client.label, + error + ); + } + retries += 1; + let wait_started = Instant::now(); + tokio::time::sleep(retry_interval.min(deadline - now)).await; + retry_wait_ms += elapsed_ms(wait_started); + } + Err(error) => return Err(error.into()), + } + } +} + +async fn execute_with_retry( + client: &mut BenchClient, + proposal_id: &str, + config: &RunConfig, +) -> Result<()> { + let deadline = Instant::now() + Duration::from_secs(config.proposal_retry_timeout_seconds); + let retry_interval = Duration::from_millis(config.proposal_retry_interval_ms); + loop { + match client.client.execute_proposal(proposal_id).await { + Ok(()) => return Ok(()), + Err(error) if is_retryable_execution_error(&error) => { + let now = Instant::now(); + if now >= deadline { + return Err(error.into()); + } + tokio::time::sleep(retry_interval.min(deadline - now)).await; + } + Err(error) => return Err(error.into()), + } + } +} + +fn is_retryable_proposal_error(error: &MultisigError) -> bool { + match error { + MultisigError::GuardianConnection(_) => true, + MultisigError::GuardianServer(message) => { + message.contains("conflict_pending_delta") + || message.contains("There's already a pending change for this account") + } + MultisigError::MidenClient(_) => is_miden_sync_tip_ahead(error), + _ => false, + } +} + +fn is_retryable_execution_error(error: &MultisigError) -> bool { + matches!(error, MultisigError::GuardianConnection(_)) || is_miden_sync_tip_ahead(error) +} + +fn client_pair( + clients: &mut [BenchClient], + operation: u64, +) -> (&mut BenchClient, &mut BenchClient) { + let (alice, bob) = clients.split_at_mut(1); + if operation.is_multiple_of(2) { + (&mut alice[0], &mut bob[0]) + } else { + (&mut bob[0], &mut alice[0]) + } +} + +async fn consumable_note_ids( + client: &mut BenchClient, + faucet_id: AccountId, + config: &RunConfig, +) -> Result> { + sync_network_with_retry(client, config).await?; + Ok(client + .client + .list_consumable_notes_filtered(NoteFilter::by_faucet(faucet_id)) + .await? + .into_iter() + .map(|note| note.id.to_string()) + .collect()) +} + +async fn await_new_note( + client: &mut BenchClient, + faucet_id: AccountId, + amount: u64, + existing_notes: &HashSet, + config: &RunConfig, +) -> Result { + let deadline = Instant::now() + Duration::from_secs(config.timeout_seconds); + let retry_interval = Duration::from_millis(config.proposal_retry_interval_ms); + loop { + sync_network_until(client, deadline, retry_interval).await?; + let notes = client + .client + .list_consumable_notes_filtered(NoteFilter::by_faucet(faucet_id)) + .await?; + if let Some(note) = notes.into_iter().find(|note| { + !existing_notes.contains(¬e.id.to_string()) + && note.amount_for_faucet(faucet_id) == amount + }) { + return Ok(note); + } + let now = Instant::now(); + if now >= deadline { + bail!("timed out waiting for the P2ID note to become consumable"); + } + tokio::time::sleep(Duration::from_millis(config.poll_interval_ms).min(deadline - now)) + .await; + } +} + +async fn sync_network_with_retry(client: &mut BenchClient, config: &RunConfig) -> Result<()> { + let deadline = Instant::now() + Duration::from_secs(config.proposal_retry_timeout_seconds); + let retry_interval = Duration::from_millis(config.proposal_retry_interval_ms); + sync_network_until(client, deadline, retry_interval).await +} + +async fn sync_network_until( + client: &mut BenchClient, + deadline: Instant, + retry_interval: Duration, +) -> Result<()> { + loop { + match client.client.sync_network_only().await { + Ok(()) => return Ok(()), + Err(error) if is_miden_sync_tip_ahead(&error) => { + let now = Instant::now(); + if now >= deadline { + return Err(error.into()); + } + tokio::time::sleep(retry_interval.min(deadline - now)).await; + } + Err(error) => return Err(error.into()), + } + } +} + +async fn await_canonical( + observer: &mut GuardianClient, + label: &str, + account_id: AccountId, + nonce: u64, + config: &RunConfig, +) -> Result { + let started = Instant::now(); + let deadline = started + Duration::from_secs(config.timeout_seconds); + loop { + let poll_error = match observer.get_delta(&account_id, nonce).await { + Ok(response) => { + if let Some(delta) = response.delta { + if delta.canonical_at.is_some() { + return Ok(elapsed_ms(started)); + } + if delta.discarded_at.is_some() { + bail!("Guardian discarded {} nonce {}", label, nonce); + } + } + None + } + Err(error) => Some(error.to_string()), + }; + if Instant::now() >= deadline { + if let Some(error) = poll_error { + bail!( + "timed out observing {} nonce {} after polling error: {}", + label, + nonce, + error + ); + } + bail!( + "timed out waiting for Guardian to canonicalize {} nonce {}", + label, + nonce + ); + } + tokio::time::sleep(Duration::from_millis(config.poll_interval_ms)).await; + } +} + +fn required_balance(config: &RunConfig) -> u64 { + config.amount.saturating_mul(config.operations.div_ceil(2)) +} + +fn ensure_starting_balances( + clients: &[BenchClient], + faucet_id: AccountId, + config: &RunConfig, +) -> Result<()> { + let required = required_balance(config); + for client in clients { + let balance = client.balance(faucet_id); + if balance < required { + bail!( + "{} vault balance is {}; profile requires {} in the worst case. Fund the account and run bootstrap", + client.label, + balance, + required + ); + } + } + Ok(()) +} + +fn ensure_ready(status: &miden_multisig_client::ProposalStatus, id: &str) -> Result<()> { + if !status.is_ready() { + bail!("1-of-1 proposal {id} was not ready after creation"); + } + Ok(()) +} + +fn parse_faucet_id(config: &RunConfig) -> Result { + AccountId::parse(&config.faucet_id) + .map(|(account_id, _network_id)| account_id) + .context("faucet_id is not a valid Miden hex or bech32 account ID") +} + +fn elapsed_ms(started: Instant) -> u64 { + u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX) +} + +fn duration_limit_reached(started: Instant, max_duration_seconds: Option) -> bool { + max_duration_seconds + .map(|seconds| started.elapsed() >= Duration::from_secs(seconds)) + .unwrap_or(false) +} + +#[derive(Debug, Serialize)] +struct LatencyStats { + samples: usize, + min_ms: u64, + p50_ms: u64, + p95_ms: u64, + max_ms: u64, + mean_ms: f64, +} + +#[derive(Debug, Serialize)] +struct RunSummary { + completed_operations: usize, + consumed_operations: usize, + unconsumed_operations: usize, + measured_runtime_ms: u64, + send_proposal_retries: u64, + send_proposal_retry_wait: LatencyStats, + send_proposal: LatencyStats, + send_execution: LatencyStats, + operation_total: LatencyStats, + first_quintile_send_proposal: LatencyStats, + last_quintile_send_proposal: LatencyStats, + send_proposal_p50_growth_percent: f64, + send_proposal_active: LatencyStats, + first_quintile_send_proposal_active: LatencyStats, + last_quintile_send_proposal_active: LatencyStats, + send_proposal_active_p50_growth_percent: f64, +} + +pub fn summarize_report(path: &Path) -> Result { + write_and_print_summary(path) +} + +fn write_and_print_summary(path: &Path) -> Result { + let records = read_records(path)?; + let summary = RunSummary::from_records(&records)?; + let summary_path = path.with_extension("summary.json"); + fs::write(&summary_path, serde_json::to_vec_pretty(&summary)?) + .with_context(|| format!("failed to write {}", summary_path.display()))?; + println!( + "completed={} consumed={} runtime={:.1}min proposal_p50={}ms proposal_p95={}ms retries={} first_q_p50={}ms last_q_p50={}ms growth={:+.1}% active_first_q_p50={}ms active_last_q_p50={}ms active_growth={:+.1}%", + summary.completed_operations, + summary.consumed_operations, + summary.measured_runtime_ms as f64 / 60_000.0, + summary.send_proposal.p50_ms, + summary.send_proposal.p95_ms, + summary.send_proposal_retries, + summary.first_quintile_send_proposal.p50_ms, + summary.last_quintile_send_proposal.p50_ms, + summary.send_proposal_p50_growth_percent, + summary.first_quintile_send_proposal_active.p50_ms, + summary.last_quintile_send_proposal_active.p50_ms, + summary.send_proposal_active_p50_growth_percent + ); + println!("wrote {}", summary_path.display()); + Ok(summary_path) +} + +impl RunSummary { + fn from_records(records: &[OperationRecord]) -> Result { + if records.is_empty() { + bail!("report contains no completed operations"); + } + let quintile_size = records.len().div_ceil(5).max(1); + let first = &records[..quintile_size]; + let last = &records[records.len() - quintile_size..]; + let first_stats = latency_stats(first.iter().map(|record| record.send_proposal_ms))?; + let last_stats = latency_stats(last.iter().map(|record| record.send_proposal_ms))?; + let growth = percent_growth(first_stats.p50_ms, last_stats.p50_ms); + let active_ms = |record: &OperationRecord| { + active_proposal_ms(record.send_proposal_ms, record.send_proposal_retry_wait_ms) + }; + let first_active_stats = latency_stats(first.iter().map(active_ms))?; + let last_active_stats = latency_stats(last.iter().map(active_ms))?; + let active_growth = percent_growth(first_active_stats.p50_ms, last_active_stats.p50_ms); + + Ok(Self { + completed_operations: records.len(), + consumed_operations: records.iter().filter(|record| record.consumed).count(), + unconsumed_operations: records.iter().filter(|record| !record.consumed).count(), + measured_runtime_ms: records.iter().map(|record| record.total_ms).sum(), + send_proposal_retries: records + .iter() + .map(|record| record.send_proposal_retries) + .sum(), + send_proposal_retry_wait: latency_stats( + records + .iter() + .map(|record| record.send_proposal_retry_wait_ms), + )?, + send_proposal: latency_stats(records.iter().map(|record| record.send_proposal_ms))?, + send_execution: latency_stats(records.iter().map(|record| record.send_execution_ms))?, + operation_total: latency_stats(records.iter().map(|record| record.total_ms))?, + first_quintile_send_proposal: first_stats, + last_quintile_send_proposal: last_stats, + send_proposal_p50_growth_percent: growth, + send_proposal_active: latency_stats(records.iter().map(active_ms))?, + first_quintile_send_proposal_active: first_active_stats, + last_quintile_send_proposal_active: last_active_stats, + send_proposal_active_p50_growth_percent: active_growth, + }) + } +} + +fn read_records(path: &Path) -> Result> { + let file = File::open(path) + .with_context(|| format!("failed to open benchmark report {}", path.display()))?; + BufReader::new(file) + .lines() + .enumerate() + .filter_map(|(index, line)| match line { + Ok(line) if line.trim().is_empty() => None, + result => Some((index, result)), + }) + .map(|(index, line)| { + let line = line.with_context(|| format!("failed to read line {}", index + 1))?; + serde_json::from_str(&line) + .with_context(|| format!("failed to parse report line {}", index + 1)) + }) + .collect() +} + +fn latency_stats(values: impl IntoIterator) -> Result { + let mut values: Vec = values.into_iter().collect(); + if values.is_empty() { + bail!("cannot summarize an empty latency series"); + } + values.sort_unstable(); + let sum: u128 = values.iter().map(|value| u128::from(*value)).sum(); + let p50_index = (values.len() - 1) / 2; + let p95_index = (values.len() * 95).div_ceil(100).saturating_sub(1); + Ok(LatencyStats { + samples: values.len(), + min_ms: values[0], + p50_ms: values[p50_index], + p95_ms: values[p95_index], + max_ms: values[values.len() - 1], + mean_ms: sum as f64 / values.len() as f64, + }) +} + +fn percent_growth(first: u64, last: u64) -> f64 { + if first == 0 { + return 0.0; + } + (last as f64 - first as f64) * 100.0 / first as f64 +} + +fn active_proposal_ms(total_ms: u64, retry_wait_ms: u64) -> u64 { + total_ms.saturating_sub(retry_wait_ms) +} + +#[cfg(test)] +mod tests { + use super::*; + use miden_protocol::address::NetworkId; + + #[test] + fn latency_stats_uses_nearest_rank_percentiles() { + let stats = latency_stats([30, 10, 20]).unwrap(); + assert_eq!(stats.p50_ms, 20); + assert_eq!(stats.p95_ms, 30); + } + + #[test] + fn parses_testnet_faucet_address() { + let account_id = AccountId::from_hex("0x023e462be6c55661144b5440f07c2c").unwrap(); + let config = RunConfig { + accounts_file: PathBuf::new(), + faucet_id: account_id.to_bech32(NetworkId::Testnet), + operations: 1, + amount: 1, + consume_probability: 0.5, + seed: 42, + poll_interval_ms: 1_000, + timeout_seconds: 180, + proposal_retry_interval_ms: 1_000, + proposal_retry_timeout_seconds: 180, + max_duration_seconds: None, + artifacts_dir: PathBuf::new(), + }; + + assert_eq!(parse_faucet_id(&config).unwrap(), account_id); + } + + #[test] + fn recognizes_only_known_transient_proposal_errors() { + let pending = MultisigError::GuardianServer( + "There's already a pending change for this account. Finish it first.".to_string(), + ); + let unrelated = MultisigError::GuardianServer("invalid signature".to_string()); + + assert!(is_retryable_proposal_error(&pending)); + assert!(!is_retryable_proposal_error(&unrelated)); + } + + #[test] + fn retries_guardian_connection_errors_for_proposal_and_execution() { + let connection = MultisigError::GuardianConnection("temporarily unavailable".to_string()); + + assert!(is_retryable_proposal_error(&connection)); + assert!(is_retryable_execution_error(&connection)); + } + + #[test] + fn recognizes_sync_tip_race_from_nullifier_endpoint() { + let error = MultisigError::MidenClient( + "endpoint: SyncNullifiers, message: block_to (857206) is greater than chain tip (857205)" + .to_string(), + ); + + assert!(is_miden_sync_tip_ahead(&error)); + } + + #[test] + fn active_proposal_time_excludes_retry_sleep() { + assert_eq!(active_proposal_ms(4_500, 3_000), 1_500); + } +} diff --git a/benchmarks/multisig-e2e/src/runtime.rs b/benchmarks/multisig-e2e/src/runtime.rs new file mode 100644 index 00000000..711ce41b --- /dev/null +++ b/benchmarks/multisig-e2e/src/runtime.rs @@ -0,0 +1,124 @@ +use anyhow::{Context, Result}; +use guardian_client::{Auth, FalconRpoSigner, GuardianClient}; +use miden_multisig_client::{AccountId, MultisigClient, MultisigError, SecretKey}; +use miden_protocol::asset::Asset; +use miden_protocol::utils::serde::Deserializable; +use tempfile::TempDir; + +use crate::config::RunConfig; +use crate::config::parse_miden_endpoint; +use crate::fixture::{AccountFixture, Fixture}; + +pub struct BenchClient { + pub label: String, + pub account_id: AccountId, + pub client: MultisigClient, + _data_dir: TempDir, +} + +impl BenchClient { + pub fn balance(&self, faucet_id: AccountId) -> u64 { + self.client + .account() + .into_iter() + .flat_map(|account| account.inner().vault().assets()) + .filter_map(|asset| match asset { + Asset::Fungible(asset) if asset.faucet_id() == faucet_id => { + Some(asset.amount().as_u64()) + } + _ => None, + }) + .sum() + } +} + +pub async fn load_clients(fixture: &Fixture, config: &RunConfig) -> Result> { + let mut clients = Vec::with_capacity(fixture.accounts.len()); + for account in &fixture.accounts { + clients.push(load_client(fixture, account, config).await?); + } + Ok(clients) +} + +async fn load_client( + fixture: &Fixture, + account: &AccountFixture, + config: &RunConfig, +) -> Result { + let endpoint = parse_miden_endpoint(&fixture.miden_endpoint)?; + let account_id = AccountId::from_hex(&account.account_id) + .with_context(|| format!("invalid account ID for {}", account.label))?; + let data_dir = TempDir::new().context("failed to create temporary Miden client directory")?; + let mut client = MultisigClient::builder() + .miden_endpoint(endpoint) + .guardian_endpoint(fixture.guardian_endpoint.clone()) + .account_dir(data_dir.path()) + .with_secret_key(parse_secret_key(account)?) + .build() + .await + .with_context(|| format!("failed to build {} client", account.label))?; + client + .pull_account(account_id) + .await + .with_context(|| format!("failed to pull {} account from Guardian", account.label))?; + sync_with_retry(&mut client, config) + .await + .with_context(|| format!("failed to sync {} account", account.label))?; + + Ok(BenchClient { + label: account.label.clone(), + account_id, + client, + _data_dir: data_dir, + }) +} + +async fn sync_with_retry(client: &mut MultisigClient, config: &RunConfig) -> Result<()> { + let deadline = std::time::Instant::now() + + std::time::Duration::from_secs(config.proposal_retry_timeout_seconds); + let retry_interval = std::time::Duration::from_millis(config.proposal_retry_interval_ms); + loop { + match client.sync().await { + Ok(()) => return Ok(()), + Err(error) if is_miden_sync_tip_ahead(&error) => { + let now = std::time::Instant::now(); + if now >= deadline { + return Err(error.into()); + } + tokio::time::sleep(retry_interval.min(deadline - now)).await; + } + Err(error) => return Err(error.into()), + } + } +} + +pub(crate) fn is_miden_sync_tip_ahead(error: &MultisigError) -> bool { + matches!( + error, + MultisigError::MidenClient(message) + if message.contains("block_to (") + && message.contains("is greater than chain tip (") + ) +} + +pub async fn load_observer(fixture: &Fixture, account: &AccountFixture) -> Result { + let observer = GuardianClient::connect(fixture.guardian_endpoint.clone()) + .await + .with_context(|| { + format!( + "failed to connect {} canonicalization observer", + account.label + ) + })? + .with_auth(Auth::FalconRpoSigner(FalconRpoSigner::new( + parse_secret_key(account)?, + ))); + Ok(observer) +} + +fn parse_secret_key(account: &AccountFixture) -> Result { + let bytes = hex::decode(&account.secret_key_hex) + .with_context(|| format!("invalid secret key hex for {}", account.label))?; + SecretKey::read_from_bytes(&bytes) + .with_context(|| format!("invalid Falcon secret key for {}", account.label)) +} diff --git a/benchmarks/multisig-e2e/testnet.local.toml b/benchmarks/multisig-e2e/testnet.local.toml new file mode 100644 index 00000000..52583ce4 --- /dev/null +++ b/benchmarks/multisig-e2e/testnet.local.toml @@ -0,0 +1,12 @@ +accounts_file = ".guardian/bench/multisig-e2e-accounts.json" +faucet_id = "mtst1aqj93e2yvy5wdv2skadca0vuuypfnp80" +operations = 50 +amount = 1 +consume_probability = 0.5 +seed = 42 +poll_interval_ms = 1000 +timeout_seconds = 180 +proposal_retry_interval_ms = 1000 +proposal_retry_timeout_seconds = 180 +max_duration_seconds = 1800 +artifacts_dir = "benchmarks/multisig-e2e/reports"