From 2461a24f3794ba839d91a95b5ff8b24dab6b8d7a Mon Sep 17 00:00:00 2001 From: Darwin Subramaniam Date: Wed, 17 Jun 2026 00:28:39 +0800 Subject: [PATCH 01/10] Add storage-cli tooling crate with `stress-test upload` (#175) Add `utils/storage-cli`, a clap-based operator CLI built on the `storage-client` SDK, starting with a `stress-test upload` subcommand. - New workspace member `utils/storage-cli`; promote `clap` to a workspace dependency (version only, features set per-crate). - `stress-test upload` resolves target buckets from chain (MemberBuckets[account] intersected with buckets that have a StorageAgreements[bucket][provider] entry) and uploads generated data over the provider HTTP API. It never creates buckets or agreements and errors clearly when no matching buckets exist. - Reuses AdminClient (chain reads) and StorageUserClient (HTTP upload); no duplicated chain or HTTP logic. --- Cargo.lock | 15 +++ Cargo.toml | 3 + utils/storage-cli/Cargo.toml | 19 +++ utils/storage-cli/README.md | 72 ++++++++++ utils/storage-cli/src/cli.rs | 78 +++++++++++ utils/storage-cli/src/commands/mod.rs | 3 + utils/storage-cli/src/commands/stress_test.rs | 123 ++++++++++++++++++ utils/storage-cli/src/main.rs | 24 ++++ 8 files changed, 337 insertions(+) create mode 100644 utils/storage-cli/Cargo.toml create mode 100644 utils/storage-cli/README.md create mode 100644 utils/storage-cli/src/cli.rs create mode 100644 utils/storage-cli/src/commands/mod.rs create mode 100644 utils/storage-cli/src/commands/stress_test.rs create mode 100644 utils/storage-cli/src/main.rs diff --git a/Cargo.lock b/Cargo.lock index ecae963f..b1e6f56b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9576,6 +9576,21 @@ version = "1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a2eb9349b6444b326872e140eb1cf5e7c522154d69e7a0ffb0fb81c06b37543f" +[[package]] +name = "storage-cli" +version = "0.1.0" +dependencies = [ + "anyhow", + "clap", + "hex", + "sp-core", + "sp-runtime", + "storage-client", + "subxt-signer", + "tokio", + "tracing-subscriber 0.3.19", +] + [[package]] name = "storage-client" version = "0.1.0" diff --git a/Cargo.toml b/Cargo.toml index dc71a8ef..1796042f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -23,6 +23,9 @@ members = [ "storage-interfaces/s3/client", "storage-interfaces/s3/pallet-s3-registry", "storage-interfaces/s3/primitives", + + # Developer tooling + "utils/storage-cli", ] [workspace.package] diff --git a/utils/storage-cli/Cargo.toml b/utils/storage-cli/Cargo.toml new file mode 100644 index 00000000..764d2892 --- /dev/null +++ b/utils/storage-cli/Cargo.toml @@ -0,0 +1,19 @@ +[package] +name = "storage-cli" +version = "0.1.0" +authors.workspace = true +edition.workspace = true +license.workspace = true +repository.workspace = true +description = "Operator CLI for scalable Web3 storage (stress-test and ops tooling)" + +[dependencies] +storage-client = { workspace = true } +clap = { workspace = true, features = ["derive", "env"] } +sp-core = { workspace = true, features = ["std"] } +sp-runtime = { workspace = true, features = ["std"] } +subxt-signer = { workspace = true } +tokio = { workspace = true } +tracing-subscriber = { workspace = true, features = ["env-filter"] } +anyhow = "1" +hex = "0.4" diff --git a/utils/storage-cli/README.md b/utils/storage-cli/README.md new file mode 100644 index 00000000..c2724240 --- /dev/null +++ b/utils/storage-cli/README.md @@ -0,0 +1,72 @@ +# storage-cli + +Storage CLI for [scalable Web3 storage](../../README.md). It consolidates the +on-chain and off-chain storage operations that were previously scattered across +`client/examples/*` and ad-hoc scripts into a single ergonomic binary, built on +top of the [`storage-client`](../../client) SDK. + +## Usage + +```bash +cargo run -p storage-cli -- --help +cargo run -p storage-cli -- stress-test upload --help +``` + +### Global flags + +| Flag | Default | Env | Description | +| -------------------- | ------------------------ | -------------- | -------------------------------------------- | +| `--chain-rpc ` | `ws://127.0.0.1:2222` | `CHAIN_RPC` | Parachain RPC WebSocket endpoint. | +| `--provider-url ` | `http://127.0.0.1:3333` | `PROVIDER_URL` | Provider node HTTP endpoint. | +| `--suri ` | — | | Secret URI for the account, e.g. `//Alice`. | +| `--keyfile ` | — | | File whose contents are the SURI/seed. | + +`--suri` and `--keyfile` are mutually exclusive; exactly one is required. + +## `stress-test upload` + +Uploads generated data to every bucket the account **already** has a storage +agreement with the given provider for. + +```bash +cargo run -p storage-cli -- \ + --suri //Bob \ + stress-test upload --provider --size 4096 +``` + +| Param | Default | Description | +| ---------------------------- | ----------- | ---------------------------------------------------- | +| `--provider ` | required | Provider account (SS58 or `0x`-hex) to target. | +| `--max-buckets-to-write ` | all buckets | Cap the number of buckets written to. | +| `--size ` | `1048576` | Bytes of generated data to upload per bucket. | + +**Behavior** + +1. Derives the account from `--suri`/`--keyfile` and reads its buckets from chain + (`MemberBuckets[account]`). +2. Keeps only buckets that have a `StorageAgreements[bucket][provider]` entry for + the given `--provider`. +3. If none match, it exits with an error — it does **not** create any bucket or + agreement. +4. Uploads generated data to each selected bucket over the provider's HTTP API. + +### Required on-chain setup + +The command only writes to buckets that already have an agreement, so the agreement +must exist first. With a chain and provider running (`just start-chain`, +`just start-provider`), open one — for example via the SDK example: + +```bash +cargo run -p storage-client --example complete_workflow -- \ + ws://127.0.0.1:2222 http://127.0.0.1:3333 //Bob +``` + +That negotiates terms and establishes an agreement, creating a bucket owned by +`//Bob` with a primary agreement to the provider. Run `stress-test upload` as the +same account (`--suri //Bob`) targeting that provider. + +## Limitations + +- **Agreement expiry is not checked.** A bucket is selected based on the presence + of a `StorageAgreements[bucket][provider]` entry; expired-but-not-yet-cleared + agreements are treated as matches. diff --git a/utils/storage-cli/src/cli.rs b/utils/storage-cli/src/cli.rs new file mode 100644 index 00000000..9017b41e --- /dev/null +++ b/utils/storage-cli/src/cli.rs @@ -0,0 +1,78 @@ +//! Command-line argument parsing for the storage operator CLI. + +use clap::{Args, Parser, Subcommand}; +use std::path::PathBuf; + +/// Operator CLI for scalable Web3 storage — drive on-chain and off-chain +/// storage operations from a single tool. +#[derive(Debug, Parser)] +#[command(name = "storage-cli", version, about)] +pub struct Cli { + #[clap(flatten)] + pub global: GlobalArgs, + + #[command(subcommand)] + pub command: Command, +} + +/// Connection and identity flags shared by every subcommand. +#[derive(Debug, Args)] +pub struct GlobalArgs { + /// WebSocket URL for the parachain RPC. + #[arg( + long, + value_name = "URL", + default_value = "ws://127.0.0.1:2222", + env = "CHAIN_RPC" + )] + pub chain_rpc: String, + + /// HTTP URL of the provider node. + #[arg( + long, + value_name = "URL", + default_value = "http://127.0.0.1:3333", + env = "PROVIDER_URL" + )] + pub provider_url: String, + + /// Secret URI (SURI) for the signing/identity account, e.g. "//Alice". + /// Mutually exclusive with `--keyfile`. + #[arg(long, value_name = "SURI", conflicts_with = "keyfile")] + pub suri: Option, + + /// Path to a file whose contents are the SURI/seed for the account. + /// Mutually exclusive with `--suri`. + #[arg(long, value_name = "FILE", conflicts_with = "suri")] + pub keyfile: Option, +} + +#[derive(Debug, Subcommand)] +pub enum Command { + /// Stress-test operations against a provider. + #[command(subcommand)] + StressTest(StressTest), +} + +#[derive(Debug, Subcommand)] +pub enum StressTest { + /// Upload generated data to every bucket the account already has an + /// agreement with the given provider for. + Upload(UploadArgs), +} + +#[derive(Debug, Args)] +pub struct UploadArgs { + /// Provider account (SS58 or 0x-hex) whose agreements select the target + /// buckets. + #[arg(long, value_name = "ACCOUNT")] + pub provider: String, + + /// Cap the number of buckets written to (default: all matching buckets). + #[arg(long, value_name = "N")] + pub max_buckets_to_write: Option, + + /// Bytes of generated data to upload per bucket. + #[arg(long, value_name = "BYTES", default_value_t = 1024 * 1024)] + pub size: usize, +} diff --git a/utils/storage-cli/src/commands/mod.rs b/utils/storage-cli/src/commands/mod.rs new file mode 100644 index 00000000..daf3cc76 --- /dev/null +++ b/utils/storage-cli/src/commands/mod.rs @@ -0,0 +1,3 @@ +//! Subcommand implementations. + +pub mod stress_test; diff --git a/utils/storage-cli/src/commands/stress_test.rs b/utils/storage-cli/src/commands/stress_test.rs new file mode 100644 index 00000000..89e1ea4d --- /dev/null +++ b/utils/storage-cli/src/commands/stress_test.rs @@ -0,0 +1,123 @@ +//! `stress-test` subcommands. + +use anyhow::{anyhow, bail, Context, Result}; +use sp_core::crypto::Ss58Codec; +use sp_runtime::AccountId32; +use storage_client::substrate::SubstrateClient; +use storage_client::{AdminClient, ChunkingStrategy, ClientConfig, StorageUserClient}; +use subxt_signer::{sr25519::Keypair, SecretUri}; + +use crate::cli::{GlobalArgs, UploadArgs}; + +/// Upload generated data to every bucket the account already has an agreement +/// with `--provider` for. +/// +/// This resolves targets from chain (`MemberBuckets[account]` ∩ buckets with a +/// `StorageAgreements[bucket][provider]` entry) and never creates buckets or +/// agreements — if nothing matches, it errors out. +pub async fn upload(global: &GlobalArgs, args: &UploadArgs) -> Result<()> { + // Identity: derive the account whose buckets we look up from the SURI. + let suri = resolve_suri(global)?; + let keypair = Keypair::from_uri(&suri.parse::().context("failed to parse SURI")?) + .context("failed to derive keypair from SURI")?; + let account = AccountId32::from(keypair.public_key().0); + let account_ss58 = account.to_ss58check(); + + // Target provider: parse once, then build the hex needle the SDK reports + // agreements with (`0x` + lowercase hex of the 32 raw account bytes). + // Compare raw bytes, never SS58 strings (prefix differences would mismatch). + let provider = SubstrateClient::parse_account(&args.provider) + .map_err(|e| anyhow!("invalid --provider account: {e}"))?; + let provider_needle = format!("0x{}", hex::encode(provider.as_ref() as &[u8])); + + let config = ClientConfig { + chain_ws_url: global.chain_rpc.clone(), + provider_urls: vec![global.provider_url.clone()], + ..Default::default() + }; + + // Read-only chain access: resolve the buckets that have an agreement with + // the target provider. No signer is set — uploads are off-chain HTTP. + let mut admin = AdminClient::new(config.clone(), account_ss58.clone()) + .context("failed to construct chain client")?; + admin + .connect() + .await + .with_context(|| format!("failed to connect to chain RPC {}", global.chain_rpc))?; + + let all_buckets_id = admin + .list_my_buckets() + .await + .context("failed to read the account's buckets from chain")?; + + let mut selected_buckets_id = Vec::new(); + for bucket_id in all_buckets_id { + let agreements = admin + .list_bucket_agreements(bucket_id) + .await + .with_context(|| format!("failed to read agreements for bucket {bucket_id}"))?; + if agreements + .iter() + .any(|a| a.provider.eq_ignore_ascii_case(&provider_needle)) + { + selected_buckets_id.push(bucket_id); + } + } + + if selected_buckets_id.is_empty() { + bail!( + "account {account_ss58} has no buckets with an agreement to provider {} on {}. \ + Nothing to upload (no bucket or agreement was created).", + args.provider, + global.chain_rpc, + ); + } + + if let Some(max) = args.max_buckets_to_write { + selected_buckets_id.truncate(max); + } + + println!( + "Uploading {} bytes to {} bucket(s) via {}", + args.size, + selected_buckets_id.len(), + global.provider_url, + ); + + // Off-chain HTTP uploads (no chain, no signer). Constant-fill payload,. + let user = StorageUserClient::new(config).context("failed to construct provider client")?; + let payload = vec![0x42; args.size]; + + for bucket in &selected_buckets_id { + let data_root = user + .upload(*bucket, &payload, ChunkingStrategy::default()) + .await + .with_context(|| format!("upload to bucket {bucket} failed"))?; + println!( + " bucket {bucket}: uploaded {} bytes, data_root 0x{}", + payload.len(), + hex::encode(data_root.as_bytes()), + ); + } + + println!("Done: {} bucket(s) written.", selected_buckets_id.len()); + Ok(()) +} + +/// Resolve the SURI from either `--suri` or `--keyfile` (exactly one is required; +/// clap already enforces they are mutually exclusive). +fn resolve_suri(global: &GlobalArgs) -> Result { + match (&global.suri, &global.keyfile) { + (Some(suri), _) => Ok(suri.clone()), + (None, Some(path)) => { + let contents = std::fs::read_to_string(path) + .with_context(|| format!("failed to read keyfile {}", path.display()))?; + let suri = contents.trim().to_string(); + if suri.is_empty() { + bail!("keyfile {} is empty", path.display()); + } + Ok(suri) + } + (None, None) => bail!("one of --suri or --keyfile is required"), + } +} diff --git a/utils/storage-cli/src/main.rs b/utils/storage-cli/src/main.rs new file mode 100644 index 00000000..14d40695 --- /dev/null +++ b/utils/storage-cli/src/main.rs @@ -0,0 +1,24 @@ +//! `storage-cli` — operator CLI for scalable Web3 storage. +//! +//! Drives on-chain and off-chain storage operations through the +//! [`storage-client`](../../client) SDK. See `--help` for the available +//! subcommands. + +mod cli; +mod commands; + +use clap::Parser; + +use crate::cli::{Cli, Command, StressTest}; + +#[tokio::main] +async fn main() -> anyhow::Result<()> { + tracing_subscriber::fmt::init(); + + let cli = Cli::parse(); + match &cli.command { + Command::StressTest(StressTest::Upload(args)) => { + commands::stress_test::upload(&cli.global, args).await + } + } +} From 20757a83aa0fd5e6ef68b7b471c8d7e346d01bf4 Mon Sep 17 00:00:00 2001 From: Darwin Subramaniam Date: Thu, 18 Jun 2026 04:17:27 +0800 Subject: [PATCH 02/10] refactored --- Cargo.lock | 1 + utils/storage-cli/Cargo.toml | 3 + utils/storage-cli/src/{cli.rs => cli_args.rs} | 4 +- utils/storage-cli/src/commands/mod.rs | 3 - utils/storage-cli/src/main.rs | 9 +- utils/storage-cli/src/scenarios/mod.rs | 1 + .../{commands => scenarios}/stress_test.rs | 36 ++----- utils/storage-cli/src/shared.rs | 94 +++++++++++++++++++ 8 files changed, 116 insertions(+), 35 deletions(-) rename utils/storage-cli/src/{cli.rs => cli_args.rs} (93%) delete mode 100644 utils/storage-cli/src/commands/mod.rs create mode 100644 utils/storage-cli/src/scenarios/mod.rs rename utils/storage-cli/src/{commands => scenarios}/stress_test.rs (72%) create mode 100644 utils/storage-cli/src/shared.rs diff --git a/Cargo.lock b/Cargo.lock index b1e6f56b..806f2308 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9587,6 +9587,7 @@ dependencies = [ "sp-runtime", "storage-client", "subxt-signer", + "tempfile", "tokio", "tracing-subscriber 0.3.19", ] diff --git a/utils/storage-cli/Cargo.toml b/utils/storage-cli/Cargo.toml index 764d2892..058a19dc 100644 --- a/utils/storage-cli/Cargo.toml +++ b/utils/storage-cli/Cargo.toml @@ -17,3 +17,6 @@ tokio = { workspace = true } tracing-subscriber = { workspace = true, features = ["env-filter"] } anyhow = "1" hex = "0.4" + +[dev-dependencies] +tempfile = "3" diff --git a/utils/storage-cli/src/cli.rs b/utils/storage-cli/src/cli_args.rs similarity index 93% rename from utils/storage-cli/src/cli.rs rename to utils/storage-cli/src/cli_args.rs index 9017b41e..2d6b8aff 100644 --- a/utils/storage-cli/src/cli.rs +++ b/utils/storage-cli/src/cli_args.rs @@ -1,9 +1,9 @@ -//! Command-line argument parsing for the storage operator CLI. +//! Command-line argument parsing for the storage CLI. use clap::{Args, Parser, Subcommand}; use std::path::PathBuf; -/// Operator CLI for scalable Web3 storage — drive on-chain and off-chain +/// Storage CLI for scalable Web3 storage — drive on-chain and off-chain /// storage operations from a single tool. #[derive(Debug, Parser)] #[command(name = "storage-cli", version, about)] diff --git a/utils/storage-cli/src/commands/mod.rs b/utils/storage-cli/src/commands/mod.rs deleted file mode 100644 index daf3cc76..00000000 --- a/utils/storage-cli/src/commands/mod.rs +++ /dev/null @@ -1,3 +0,0 @@ -//! Subcommand implementations. - -pub mod stress_test; diff --git a/utils/storage-cli/src/main.rs b/utils/storage-cli/src/main.rs index 14d40695..44165ebb 100644 --- a/utils/storage-cli/src/main.rs +++ b/utils/storage-cli/src/main.rs @@ -4,12 +4,13 @@ //! [`storage-client`](../../client) SDK. See `--help` for the available //! subcommands. -mod cli; -mod commands; +mod cli_args; +mod shared; +mod scenarios; use clap::Parser; -use crate::cli::{Cli, Command, StressTest}; +use crate::cli_args::{Cli, Command, StressTest}; #[tokio::main] async fn main() -> anyhow::Result<()> { @@ -18,7 +19,7 @@ async fn main() -> anyhow::Result<()> { let cli = Cli::parse(); match &cli.command { Command::StressTest(StressTest::Upload(args)) => { - commands::stress_test::upload(&cli.global, args).await + scenarios::stress_test::upload(&cli.global, args).await } } } diff --git a/utils/storage-cli/src/scenarios/mod.rs b/utils/storage-cli/src/scenarios/mod.rs new file mode 100644 index 00000000..2b273c0d --- /dev/null +++ b/utils/storage-cli/src/scenarios/mod.rs @@ -0,0 +1 @@ +pub mod stress_test; \ No newline at end of file diff --git a/utils/storage-cli/src/commands/stress_test.rs b/utils/storage-cli/src/scenarios/stress_test.rs similarity index 72% rename from utils/storage-cli/src/commands/stress_test.rs rename to utils/storage-cli/src/scenarios/stress_test.rs index 89e1ea4d..8056484d 100644 --- a/utils/storage-cli/src/commands/stress_test.rs +++ b/utils/storage-cli/src/scenarios/stress_test.rs @@ -7,7 +7,8 @@ use storage_client::substrate::SubstrateClient; use storage_client::{AdminClient, ChunkingStrategy, ClientConfig, StorageUserClient}; use subxt_signer::{sr25519::Keypair, SecretUri}; -use crate::cli::{GlobalArgs, UploadArgs}; +use crate::cli_args::{GlobalArgs, UploadArgs}; +use crate::shared::resolve_suri; /// Upload generated data to every bucket the account already has an agreement /// with `--provider` for. @@ -23,12 +24,13 @@ pub async fn upload(global: &GlobalArgs, args: &UploadArgs) -> Result<()> { let account = AccountId32::from(keypair.public_key().0); let account_ss58 = account.to_ss58check(); - // Target provider: parse once, then build the hex needle the SDK reports - // agreements with (`0x` + lowercase hex of the 32 raw account bytes). - // Compare raw bytes, never SS58 strings (prefix differences would mismatch). - let provider = SubstrateClient::parse_account(&args.provider) - .map_err(|e| anyhow!("invalid --provider account: {e}"))?; - let provider_needle = format!("0x{}", hex::encode(provider.as_ref() as &[u8])); + // Target provider: parse the input and hex-encode it (`0x` + lowercase hex + // of the 32 raw account bytes) for matching against the chain's + // `StorageAgreements`. Match on raw bytes, never SS58 strings — prefix + // differences (`5…` vs `1…`) would make equal accounts compare unequal. + let target_provider_hex = SubstrateClient::parse_account(&args.provider) + .map_err(|e: storage_client::ClientError| anyhow!("invalid --provider account: {e}")) + .map(|ac| format!("0x{}", hex::encode(ac.as_ref() as &[u8])))?; let config = ClientConfig { chain_ws_url: global.chain_rpc.clone(), @@ -58,7 +60,7 @@ pub async fn upload(global: &GlobalArgs, args: &UploadArgs) -> Result<()> { .with_context(|| format!("failed to read agreements for bucket {bucket_id}"))?; if agreements .iter() - .any(|a| a.provider.eq_ignore_ascii_case(&provider_needle)) + .any(|a| a.provider.eq_ignore_ascii_case(&target_provider_hex)) { selected_buckets_id.push(bucket_id); } @@ -103,21 +105,3 @@ pub async fn upload(global: &GlobalArgs, args: &UploadArgs) -> Result<()> { println!("Done: {} bucket(s) written.", selected_buckets_id.len()); Ok(()) } - -/// Resolve the SURI from either `--suri` or `--keyfile` (exactly one is required; -/// clap already enforces they are mutually exclusive). -fn resolve_suri(global: &GlobalArgs) -> Result { - match (&global.suri, &global.keyfile) { - (Some(suri), _) => Ok(suri.clone()), - (None, Some(path)) => { - let contents = std::fs::read_to_string(path) - .with_context(|| format!("failed to read keyfile {}", path.display()))?; - let suri = contents.trim().to_string(); - if suri.is_empty() { - bail!("keyfile {} is empty", path.display()); - } - Ok(suri) - } - (None, None) => bail!("one of --suri or --keyfile is required"), - } -} diff --git a/utils/storage-cli/src/shared.rs b/utils/storage-cli/src/shared.rs new file mode 100644 index 00000000..b99f7c0a --- /dev/null +++ b/utils/storage-cli/src/shared.rs @@ -0,0 +1,94 @@ +use crate::cli_args::GlobalArgs; +use anyhow::{bail, Context, Result}; + +/// Resolve the SURI from either `--suri` or `--keyfile` (exactly one is required; +pub fn resolve_suri(global: &GlobalArgs) -> Result { + match (&global.suri, &global.keyfile) { + (Some(suri), _) => Ok(suri.clone()), + (None, Some(path)) => { + let contents = std::fs::read_to_string(path) + .with_context(|| format!("failed to read keyfile {}", path.display()))?; + let suri = contents.trim().to_string(); + if suri.is_empty() { + bail!("keyfile {} is empty", path.display()); + } + Ok(suri) + } + (None, None) => bail!("one of --suri or --keyfile is required"), + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::io::Write; + use std::path::PathBuf; + + /// Build `GlobalArgs` with placeholder connection flags and the given + /// identity inputs — only `suri`/`keyfile` matter for `resolve_suri`. + fn make_args(suri: Option, keyfile: Option) -> GlobalArgs { + GlobalArgs { + chain_rpc: "ws://127.0.0.1:2222".to_string(), + provider_url: "http://127.0.0.1:3333".to_string(), + suri, + keyfile, + } + } + + #[test] + fn returns_inline_suri() { + let args = make_args(Some("//Alice".to_string()), None); + assert_eq!(resolve_suri(&args).unwrap(), "//Alice"); + } + + #[test] + fn reads_and_trims_keyfile() { + let mut file = tempfile::NamedTempFile::new().unwrap(); + writeln!(file, " //Bob").unwrap(); + let args = make_args(None, Some(file.path().to_path_buf())); + assert_eq!(resolve_suri(&args).unwrap(), "//Bob"); + } + + #[test] + fn rejects_empty_keyfile() { + let mut file = tempfile::NamedTempFile::new().unwrap(); + writeln!(file, " ").unwrap(); + let args = make_args(None, Some(file.path().to_path_buf())); + let err = resolve_suri(&args).unwrap_err().to_string(); + assert!(err.contains("is empty"), "unexpected error: {err}"); + } + + #[test] + fn errors_on_missing_keyfile() { + let path = std::env::temp_dir().join("storage-cli-resolve-suri-does-not-exist"); + let args = make_args(None, Some(path)); + let err = resolve_suri(&args).unwrap_err().to_string(); + assert!( + err.contains("failed to read keyfile"), + "unexpected error: {err}" + ); + } + + #[test] + fn errors_when_neither_provided() { + let args = make_args(None, None); + let err = resolve_suri(&args).unwrap_err().to_string(); + assert!( + err.contains("--suri or --keyfile"), + "unexpected error: {err}" + ); + } + + #[test] + fn inline_suri_wins_over_keyfile() { + // clap's `conflicts_with` blocks this at parse time; this only pins + // the `(Some, _)` arm's precedence within the function itself. + let mut file = tempfile::NamedTempFile::new().unwrap(); + write!(file, "//FromFile").unwrap(); + let args = make_args( + Some("//Inline".to_string()), + Some(file.path().to_path_buf()), + ); + assert_eq!(resolve_suri(&args).unwrap(), "//Inline"); + } +} From 8ffff9ea73b77b6d234c3fbc337a916a3aaa9708 Mon Sep 17 00:00:00 2001 From: Darwin Subramaniam Date: Sat, 20 Jun 2026 00:34:55 +0800 Subject: [PATCH 03/10] refactor(storage-cli): reorganize CLI structure, moved dependency version to workspace and added license header, based on code review --- Cargo.toml | 3 ++ utils/storage-cli/Cargo.toml | 4 +-- utils/storage-cli/src/{cli_args.rs => cli.rs} | 25 ++----------- utils/storage-cli/src/commands/mod.rs | 2 ++ .../{scenarios => commands}/stress_test.rs | 35 +++++++++++++++++-- .../src/{shared.rs => common/mod.rs} | 6 ++-- utils/storage-cli/src/main.rs | 15 ++++---- utils/storage-cli/src/scenarios/mod.rs | 1 - 8 files changed, 56 insertions(+), 35 deletions(-) rename utils/storage-cli/src/{cli_args.rs => cli.rs} (68%) create mode 100644 utils/storage-cli/src/commands/mod.rs rename utils/storage-cli/src/{scenarios => commands}/stress_test.rs (80%) rename utils/storage-cli/src/{shared.rs => common/mod.rs} (97%) delete mode 100644 utils/storage-cli/src/scenarios/mod.rs diff --git a/Cargo.toml b/Cargo.toml index 1796042f..89c2c46c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -147,6 +147,8 @@ serde_with = { version = "3.20.0" } thiserror = "2.0" tracing = { version = "0.1.41", default-features = false } tracing-subscriber = { version = "=0.3.19" } +anyhow = { version = "1.0" } +hex = { version = "0.4" } # Async/HTTP (for provider node) axum = "0.7" @@ -175,6 +177,7 @@ sp-keystore = { version = "0.46.0", default-features = false } tempfile = "3" tokio-test = "0.4" + [profile.release] opt-level = 3 panic = "unwind" diff --git a/utils/storage-cli/Cargo.toml b/utils/storage-cli/Cargo.toml index 058a19dc..6d6056e2 100644 --- a/utils/storage-cli/Cargo.toml +++ b/utils/storage-cli/Cargo.toml @@ -15,8 +15,8 @@ sp-runtime = { workspace = true, features = ["std"] } subxt-signer = { workspace = true } tokio = { workspace = true } tracing-subscriber = { workspace = true, features = ["env-filter"] } -anyhow = "1" -hex = "0.4" +anyhow = { workspace = true } +hex = { workspace = true } [dev-dependencies] tempfile = "3" diff --git a/utils/storage-cli/src/cli_args.rs b/utils/storage-cli/src/cli.rs similarity index 68% rename from utils/storage-cli/src/cli_args.rs rename to utils/storage-cli/src/cli.rs index 2d6b8aff..c952348d 100644 --- a/utils/storage-cli/src/cli_args.rs +++ b/utils/storage-cli/src/cli.rs @@ -1,7 +1,10 @@ +// SPDX-License-Identifier: Apache-2.0 + //! Command-line argument parsing for the storage CLI. use clap::{Args, Parser, Subcommand}; use std::path::PathBuf; +use crate::commands::stress_test::StressTest; /// Storage CLI for scalable Web3 storage — drive on-chain and off-chain /// storage operations from a single tool. @@ -54,25 +57,3 @@ pub enum Command { StressTest(StressTest), } -#[derive(Debug, Subcommand)] -pub enum StressTest { - /// Upload generated data to every bucket the account already has an - /// agreement with the given provider for. - Upload(UploadArgs), -} - -#[derive(Debug, Args)] -pub struct UploadArgs { - /// Provider account (SS58 or 0x-hex) whose agreements select the target - /// buckets. - #[arg(long, value_name = "ACCOUNT")] - pub provider: String, - - /// Cap the number of buckets written to (default: all matching buckets). - #[arg(long, value_name = "N")] - pub max_buckets_to_write: Option, - - /// Bytes of generated data to upload per bucket. - #[arg(long, value_name = "BYTES", default_value_t = 1024 * 1024)] - pub size: usize, -} diff --git a/utils/storage-cli/src/commands/mod.rs b/utils/storage-cli/src/commands/mod.rs new file mode 100644 index 00000000..1cd2e9b8 --- /dev/null +++ b/utils/storage-cli/src/commands/mod.rs @@ -0,0 +1,2 @@ +// SPDX-License-Identifier: Apache-2.0 +pub mod stress_test; \ No newline at end of file diff --git a/utils/storage-cli/src/scenarios/stress_test.rs b/utils/storage-cli/src/commands/stress_test.rs similarity index 80% rename from utils/storage-cli/src/scenarios/stress_test.rs rename to utils/storage-cli/src/commands/stress_test.rs index 8056484d..42fd6d82 100644 --- a/utils/storage-cli/src/scenarios/stress_test.rs +++ b/utils/storage-cli/src/commands/stress_test.rs @@ -1,3 +1,5 @@ +// SPDX-License-Identifier: Apache-2.0 + //! `stress-test` subcommands. use anyhow::{anyhow, bail, Context, Result}; @@ -6,9 +8,38 @@ use sp_runtime::AccountId32; use storage_client::substrate::SubstrateClient; use storage_client::{AdminClient, ChunkingStrategy, ClientConfig, StorageUserClient}; use subxt_signer::{sr25519::Keypair, SecretUri}; +use clap::{Args, Subcommand}; + +use crate::cli::{GlobalArgs}; +use crate::common::resolve_suri; + + +// === Stress test subcommands === +#[derive(Debug, Subcommand)] +pub enum StressTest { + /// Upload generated data to every bucket the account already has an + /// agreement with the given provider for. + #[command(name = "upload")] + ProviderUpload(UploadArgs), +} + +// === `stress-test provider-upload` subcommand === +#[derive(Debug, Args)] +pub struct UploadArgs { + /// Provider account (SS58 or 0x-hex) whose agreements select the target + /// buckets. + #[arg(long, value_name = "ACCOUNT")] + pub provider: String, + + /// Cap the number of buckets written to (default: all matching buckets). + #[arg(long, value_name = "N")] + pub max_buckets_to_write: Option, + + /// Bytes of generated data to upload per bucket. + #[arg(long, value_name = "BYTES", default_value_t = 1024 * 1024)] + pub size: usize, +} -use crate::cli_args::{GlobalArgs, UploadArgs}; -use crate::shared::resolve_suri; /// Upload generated data to every bucket the account already has an agreement /// with `--provider` for. diff --git a/utils/storage-cli/src/shared.rs b/utils/storage-cli/src/common/mod.rs similarity index 97% rename from utils/storage-cli/src/shared.rs rename to utils/storage-cli/src/common/mod.rs index b99f7c0a..a0d13b65 100644 --- a/utils/storage-cli/src/shared.rs +++ b/utils/storage-cli/src/common/mod.rs @@ -1,7 +1,9 @@ -use crate::cli_args::GlobalArgs; +// SPDX-License-Identifier: Apache-2.0 + +use crate::cli::GlobalArgs; use anyhow::{bail, Context, Result}; -/// Resolve the SURI from either `--suri` or `--keyfile` (exactly one is required; +/// Resolve the SURI from either `--suri` or `--keyfile` (exactly one is required) pub fn resolve_suri(global: &GlobalArgs) -> Result { match (&global.suri, &global.keyfile) { (Some(suri), _) => Ok(suri.clone()), diff --git a/utils/storage-cli/src/main.rs b/utils/storage-cli/src/main.rs index 44165ebb..ec5ab565 100644 --- a/utils/storage-cli/src/main.rs +++ b/utils/storage-cli/src/main.rs @@ -1,16 +1,19 @@ +// SPDX-License-Identifier: Apache-2.0 //! `storage-cli` — operator CLI for scalable Web3 storage. //! //! Drives on-chain and off-chain storage operations through the //! [`storage-client`](../../client) SDK. See `--help` for the available //! subcommands. -mod cli_args; -mod shared; -mod scenarios; +mod cli; +mod common; +mod commands; use clap::Parser; -use crate::cli_args::{Cli, Command, StressTest}; +use crate::cli::{Cli, Command}; +use crate::commands::stress_test::StressTest; + #[tokio::main] async fn main() -> anyhow::Result<()> { @@ -18,8 +21,8 @@ async fn main() -> anyhow::Result<()> { let cli = Cli::parse(); match &cli.command { - Command::StressTest(StressTest::Upload(args)) => { - scenarios::stress_test::upload(&cli.global, args).await + Command::StressTest(StressTest::ProviderUpload(args)) => { + commands::stress_test::upload(&cli.global, args).await } } } diff --git a/utils/storage-cli/src/scenarios/mod.rs b/utils/storage-cli/src/scenarios/mod.rs deleted file mode 100644 index 2b273c0d..00000000 --- a/utils/storage-cli/src/scenarios/mod.rs +++ /dev/null @@ -1 +0,0 @@ -pub mod stress_test; \ No newline at end of file From 6f1ed4099ea1b0f8af08591dd810edaac0672e44 Mon Sep 17 00:00:00 2001 From: Darwin Subramaniam Date: Sat, 20 Jun 2026 00:55:23 +0800 Subject: [PATCH 04/10] feat(integration-tests): add utils integration tests for storage-cli stress testing --- .github/workflows/integration-tests.yml | 127 +++++++++++++++++++++++- 1 file changed, 126 insertions(+), 1 deletion(-) diff --git a/.github/workflows/integration-tests.yml b/.github/workflows/integration-tests.yml index 7c1b1666..3fe5b0a3 100644 --- a/.github/workflows/integration-tests.yml +++ b/.github/workflows/integration-tests.yml @@ -585,6 +585,131 @@ jobs: /tmp/omni-node.log /tmp/provider.log + # ── Utils CLI tooling ───────────────────────────────────────────── + utils-integration-tests: + name: Utils Integration Tests + runs-on: parity-large + timeout-minutes: 30 + needs: [set-image, build] + container: + image: ${{ needs.set-image.outputs.CI_IMAGE }} + steps: + - name: Checkout sources + uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6 + with: + persist-credentials: false + + - name: Load common environment variables via env file + run: cat .github/env >> $GITHUB_ENV + + - name: Install just + run: cargo install just --locked || true + + - name: Cache Polkadot SDK binaries + uses: actions/cache@27d5ce7f107fe9357f9df03efb73ab90386fccae # v5 + id: sdk-cache + with: + path: | + .bin/polkadot + .bin/polkadot-execute-worker + .bin/polkadot-prepare-worker + .bin/polkadot-omni-node + .bin/chain-spec-builder + key: polkadot-sdk-${{ env.POLKADOT_SDK_VERSION }} + + - name: Download Polkadot SDK binaries + if: steps.sdk-cache.outputs.cache-hit != 'true' + run: just polkadot_version="$POLKADOT_SDK_VERSION" download-polkadot-sdk-binaries + + - name: Resolve chain-spec script + run: | + RUNTIME=web3-storage-paseo + SCRIPT=$(jq -r --arg r "$RUNTIME" '.[] | select(.name==$r) | .chain_spec_script // empty' scripts/runtimes-matrix.json) + if [ -z "$SCRIPT" ]; then + echo "ERROR: no chain_spec_script for runtime '$RUNTIME' in scripts/runtimes-matrix.json" + exit 1 + fi + echo "Using chain-spec script: $SCRIPT" + echo "CHAIN_SPEC_SCRIPT=$SCRIPT" >> $GITHUB_ENV + + # Before the download so the cached target/ can't clobber it. save-if + # false — only the build job writes this cache. + - name: Rust cache + uses: Swatinem/rust-cache@e18b497796c12c097a38f9edb9d0641fb99eee32 # v2 + with: + shared-key: "integration-tests-build" + save-if: false + + # Extract into target/release so the artifact's contents land back at + # their original paths (chmod, chain_spec_command scripts). + - name: Download build artifacts + uses: actions/download-artifact@3e5f45b2cfb9172054b4087a40e8e0b5a5461e7c # v6 + with: + name: build + path: target/release + + - name: Restore executable bits + run: chmod +x target/release/storage-provider-node + + - name: Start chain (dev mode, 2s blocks) + uses: ./.github/actions/start-e2e-chain + with: + chain-spec-script: ${{ env.CHAIN_SPEC_SCRIPT }} + + - name: Wait for parachain blocks + uses: ./.github/actions/wait-for-parachain + with: + log-file: /tmp/omni-node.log + + - name: Register provider on-chain + run: | + echo "//Alice" > /tmp/alice-key && chmod 600 /tmp/alice-key + cargo run --release -p storage-client --example register_provider \ + ws://127.0.0.1:2222 http://127.0.0.1:3333 /ip4/127.0.0.1/tcp/3333 /tmp/alice-key + + - name: Start provider (inmemory) and wait for health + run: | + nohup ./target/release/storage-provider-node \ + --keyfile /tmp/alice-key --storage-mode inmemory \ + --bind-addr 0.0.0.0:3333 --chain-rpc ws://127.0.0.1:2222 \ + --enable-checkpoint-coordinator > /tmp/provider.log 2>&1 & + + - name: Wait for provider health + uses: ./.github/actions/wait-for-provider-health + + # Seed: //Bob negotiates terms and opens a bucket + primary agreement + # against the //Alice provider, giving the CLI something to upload to. + - name: Seed bucket + agreement (//Bob → //Alice provider) + run: | + cargo run --release -p storage-client --example complete_workflow \ + ws://127.0.0.1:2222 http://127.0.0.1:3333 //Bob + + # Run the CLI's stress-test upload command against the //Alice provider using + # the //Bob key. + - name: storage-cli stress-test upload + run: | + cargo run --release -p storage-cli -- \ + --suri //Bob \ + stress-test upload \ + --provider 5GrwvaEF5zXb26Fz9rcQpDWS57CtERHpNehXCPcNoHGKutQY \ + --size 1048576 + + - name: Stop services + if: always() + run: | + pkill -f "polkadot-omni-node" 2>/dev/null || true + pkill -f "polkadot" 2>/dev/null || true + pkill -f "storage-provider-node" 2>/dev/null || true + + - name: Upload logs (on failure) + if: always() + uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7 + with: + name: utils-integration-test-logs + path: | + /tmp/omni-node.log + /tmp/provider.log + # ── UI Playwright e2e ───────────────────────────────────────────────────── # Drive / console / provider UIs (Vite dev servers) exercised against a # standalone paseo dev chain (omni-node, 2s blocks, no relay chain) + a @@ -776,7 +901,7 @@ jobs: integration-tests-complete: name: Integration Tests - needs: [changes, build, integration-tests, e2e-integration-tests, sc-integration-tests, ui-integration-tests, coverage-gate] + needs: [changes, build, integration-tests, e2e-integration-tests, sc-integration-tests, ui-integration-tests, utils-integration-tests, coverage-gate] if: always() runs-on: ubuntu-latest steps: From ef8421a3b051e13d41319a9dbcae29692d526ea5 Mon Sep 17 00:00:00 2001 From: Darwin Subramaniam Date: Wed, 24 Jun 2026 02:16:51 +0800 Subject: [PATCH 05/10] refactor(cli): reorganize imports and clean up whitespace in storage CLI files --- utils/storage-cli/src/cli.rs | 3 +-- utils/storage-cli/src/commands/mod.rs | 2 +- utils/storage-cli/src/commands/stress_test.rs | 6 ++---- utils/storage-cli/src/main.rs | 3 +-- 4 files changed, 5 insertions(+), 9 deletions(-) diff --git a/utils/storage-cli/src/cli.rs b/utils/storage-cli/src/cli.rs index c952348d..097f6134 100644 --- a/utils/storage-cli/src/cli.rs +++ b/utils/storage-cli/src/cli.rs @@ -2,9 +2,9 @@ //! Command-line argument parsing for the storage CLI. +use crate::commands::stress_test::StressTest; use clap::{Args, Parser, Subcommand}; use std::path::PathBuf; -use crate::commands::stress_test::StressTest; /// Storage CLI for scalable Web3 storage — drive on-chain and off-chain /// storage operations from a single tool. @@ -56,4 +56,3 @@ pub enum Command { #[command(subcommand)] StressTest(StressTest), } - diff --git a/utils/storage-cli/src/commands/mod.rs b/utils/storage-cli/src/commands/mod.rs index 1cd2e9b8..906299cd 100644 --- a/utils/storage-cli/src/commands/mod.rs +++ b/utils/storage-cli/src/commands/mod.rs @@ -1,2 +1,2 @@ // SPDX-License-Identifier: Apache-2.0 -pub mod stress_test; \ No newline at end of file +pub mod stress_test; diff --git a/utils/storage-cli/src/commands/stress_test.rs b/utils/storage-cli/src/commands/stress_test.rs index 42fd6d82..321bec1c 100644 --- a/utils/storage-cli/src/commands/stress_test.rs +++ b/utils/storage-cli/src/commands/stress_test.rs @@ -3,17 +3,16 @@ //! `stress-test` subcommands. use anyhow::{anyhow, bail, Context, Result}; +use clap::{Args, Subcommand}; use sp_core::crypto::Ss58Codec; use sp_runtime::AccountId32; use storage_client::substrate::SubstrateClient; use storage_client::{AdminClient, ChunkingStrategy, ClientConfig, StorageUserClient}; use subxt_signer::{sr25519::Keypair, SecretUri}; -use clap::{Args, Subcommand}; -use crate::cli::{GlobalArgs}; +use crate::cli::GlobalArgs; use crate::common::resolve_suri; - // === Stress test subcommands === #[derive(Debug, Subcommand)] pub enum StressTest { @@ -40,7 +39,6 @@ pub struct UploadArgs { pub size: usize, } - /// Upload generated data to every bucket the account already has an agreement /// with `--provider` for. /// diff --git a/utils/storage-cli/src/main.rs b/utils/storage-cli/src/main.rs index ec5ab565..c59bc254 100644 --- a/utils/storage-cli/src/main.rs +++ b/utils/storage-cli/src/main.rs @@ -6,15 +6,14 @@ //! subcommands. mod cli; -mod common; mod commands; +mod common; use clap::Parser; use crate::cli::{Cli, Command}; use crate::commands::stress_test::StressTest; - #[tokio::main] async fn main() -> anyhow::Result<()> { tracing_subscriber::fmt::init(); From 507ab743245b8e2d8df6011961e759c2ab5b039d Mon Sep 17 00:00:00 2001 From: Darwin Subramaniam Date: Thu, 25 Jun 2026 10:58:50 +0800 Subject: [PATCH 06/10] feat(storage-cli): configurable stress-test scenarios with metrics summary Replace the single constant-fill upload per bucket with load driven entirely by configuration: - --users / --uploads-per-user / --max-payload-size (random payloads) - --parallel-users and --parallel-uploads axes, plus --max-concurrency cap - targets buckets the account already has an agreement with the provider for Add a reusable metrics module (OpOutcome / OpSummary / summarize) tagged by Operation, so a scenario computes and returns aggregate metrics (counts, bytes, throughput, latency) and main owns viewing them, with --output text|json. Wire the new flags into the integration-tests workflow. --- .github/workflows/integration-tests.yml | 5 +- Cargo.lock | 3 + Cargo.toml | 3 +- utils/storage-cli/Cargo.toml | 5 +- utils/storage-cli/src/cli.rs | 5 + utils/storage-cli/src/commands/stress_test.rs | 271 +++++++++++-- utils/storage-cli/src/common/mod.rs | 1 + utils/storage-cli/src/main.rs | 22 +- utils/storage-cli/src/metrics.rs | 367 ++++++++++++++++++ 9 files changed, 648 insertions(+), 34 deletions(-) create mode 100644 utils/storage-cli/src/metrics.rs diff --git a/.github/workflows/integration-tests.yml b/.github/workflows/integration-tests.yml index 3fe5b0a3..23b9e5fa 100644 --- a/.github/workflows/integration-tests.yml +++ b/.github/workflows/integration-tests.yml @@ -692,7 +692,10 @@ jobs: --suri //Bob \ stress-test upload \ --provider 5GrwvaEF5zXb26Fz9rcQpDWS57CtERHpNehXCPcNoHGKutQY \ - --size 1048576 + --users 2 \ + --uploads-per-user 5 \ + --max-payload-size 1048576 \ + --parallel-uploads - name: Stop services if: always() diff --git a/Cargo.lock b/Cargo.lock index 806f2308..df3a797b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9583,6 +9583,9 @@ dependencies = [ "anyhow", "clap", "hex", + "rand 0.8.5", + "serde", + "serde_json", "sp-core", "sp-runtime", "storage-client", diff --git a/Cargo.toml b/Cargo.toml index 89c2c46c..3d850622 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -134,6 +134,7 @@ scale-info = { version = "2.11.6", default-features = false, features = [ ] } # External dependencies +anyhow = { version = "1.0" } async-trait = "0.1" base64 = "0.22" bincode = "1.3" @@ -147,8 +148,6 @@ serde_with = { version = "3.20.0" } thiserror = "2.0" tracing = { version = "0.1.41", default-features = false } tracing-subscriber = { version = "=0.3.19" } -anyhow = { version = "1.0" } -hex = { version = "0.4" } # Async/HTTP (for provider node) axum = "0.7" diff --git a/utils/storage-cli/Cargo.toml b/utils/storage-cli/Cargo.toml index 6d6056e2..434d846c 100644 --- a/utils/storage-cli/Cargo.toml +++ b/utils/storage-cli/Cargo.toml @@ -17,6 +17,9 @@ tokio = { workspace = true } tracing-subscriber = { workspace = true, features = ["env-filter"] } anyhow = { workspace = true } hex = { workspace = true } +rand = { workspace = true } +serde = { workspace = true, features = ["derive", "std"] } +serde_json = { workspace = true, features = ["std"] } [dev-dependencies] -tempfile = "3" +tempfile = { workspace = true } diff --git a/utils/storage-cli/src/cli.rs b/utils/storage-cli/src/cli.rs index 097f6134..bf50770f 100644 --- a/utils/storage-cli/src/cli.rs +++ b/utils/storage-cli/src/cli.rs @@ -3,6 +3,7 @@ //! Command-line argument parsing for the storage CLI. use crate::commands::stress_test::StressTest; +use crate::metrics::OutputFormat; use clap::{Args, Parser, Subcommand}; use std::path::PathBuf; @@ -48,6 +49,10 @@ pub struct GlobalArgs { /// Mutually exclusive with `--suri`. #[arg(long, value_name = "FILE", conflicts_with = "suri")] pub keyfile: Option, + + /// Format for the final metrics summary printed to stdout. + #[arg(long, value_enum, value_name = "FORMAT", default_value_t = OutputFormat::Text)] + pub output: OutputFormat, } #[derive(Debug, Subcommand)] diff --git a/utils/storage-cli/src/commands/stress_test.rs b/utils/storage-cli/src/commands/stress_test.rs index 321bec1c..ec6eda99 100644 --- a/utils/storage-cli/src/commands/stress_test.rs +++ b/utils/storage-cli/src/commands/stress_test.rs @@ -2,6 +2,9 @@ //! `stress-test` subcommands. +use std::sync::Arc; +use std::time::{Duration, Instant}; + use anyhow::{anyhow, bail, Context, Result}; use clap::{Args, Subcommand}; use sp_core::crypto::Ss58Codec; @@ -9,20 +12,28 @@ use sp_runtime::AccountId32; use storage_client::substrate::SubstrateClient; use storage_client::{AdminClient, ChunkingStrategy, ClientConfig, StorageUserClient}; use subxt_signer::{sr25519::Keypair, SecretUri}; +use tokio::sync::Semaphore; +use tokio::task::JoinSet; use crate::cli::GlobalArgs; use crate::common::resolve_suri; +use crate::metrics::{summarize, OpOutcome, OpSummary, Operation}; + +/// Bucket identifier on chain (alias of `storage_primitives::BucketId`, a `u64`). +type BucketId = u64; // === Stress test subcommands === #[derive(Debug, Subcommand)] pub enum StressTest { - /// Upload generated data to every bucket the account already has an + /// Drive configurable upload load against a provider: `users` simulated + /// clients each performing `uploads-per-user` uploads, with either axis run + /// sequentially or in parallel. Targets buckets the account already has an /// agreement with the given provider for. #[command(name = "upload")] ProviderUpload(UploadArgs), } -// === `stress-test provider-upload` subcommand === +// === `stress-test upload` subcommand === #[derive(Debug, Args)] pub struct UploadArgs { /// Provider account (SS58 or 0x-hex) whose agreements select the target @@ -34,18 +45,135 @@ pub struct UploadArgs { #[arg(long, value_name = "N")] pub max_buckets_to_write: Option, - /// Bytes of generated data to upload per bucket. - #[arg(long, value_name = "BYTES", default_value_t = 1024 * 1024)] - pub size: usize, + /// Number of concurrent simulated users, each with its own client (1..N). + #[arg(long, value_name = "N", default_value_t = 1)] + pub users: usize, + + /// Number of uploads each user performs (1..X). + #[arg(long, value_name = "X", default_value_t = 1)] + pub uploads_per_user: usize, + + /// Exact size in bytes of each randomly-generated payload (default 0.5 MiB). + #[arg(long, value_name = "BYTES", default_value_t = 512 * 1024)] + pub max_payload_size: usize, + + /// Run users in parallel (default: sequential). + #[arg(long, default_value_t = false)] + pub parallel_users: bool, + + /// Run each user's uploads in parallel (default: sequential). + #[arg(long, default_value_t = false)] + pub parallel_uploads: bool, + + /// Cap total in-flight uploads across all users (0 = unbounded). + #[arg(long, value_name = "N", default_value_t = 0)] + pub max_concurrency: usize, +} + +/// Pick the target bucket for the `global_idx`-th upload of the whole run, +/// round-robin so load spreads evenly across the selected buckets. +fn bucket_for(global_idx: usize, buckets: &[BucketId]) -> BucketId { + buckets[global_idx % buckets.len()] +} + +/// Generate a payload of exactly `size` random bytes. +fn random_payload(size: usize) -> Vec { + use rand::RngCore; + let mut buf = vec![0u8; size]; + rand::thread_rng().fill_bytes(&mut buf); + buf } -/// Upload generated data to every bucket the account already has an agreement -/// with `--provider` for. +/// Perform one upload, holding a concurrency permit (if any) for its duration. +async fn do_upload( + client: Arc, + bucket: BucketId, + size: usize, + sem: Option>, +) -> OpOutcome { + // Hold the permit until the upload completes; `acquire` only fails if the + // semaphore is closed, which never happens here. + let _permit = match &sem { + Some(s) => s.acquire().await.ok(), + None => None, + }; + let payload = random_payload(size); + let started = Instant::now(); + match client + .upload(bucket, &payload, ChunkingStrategy::default()) + .await + { + Ok(_root) => OpOutcome::success(size, started.elapsed()), + Err(e) => OpOutcome::failure(size, started.elapsed(), e.to_string()), + } +} + +/// Run a single user's `uploads` uploads, either sequentially or in parallel. +async fn run_user( + user_idx: usize, + client: Arc, + buckets: Arc>, + uploads: usize, + size: usize, + parallel: bool, + sem: Option>, +) -> Vec { + // Each user's uploads occupy a contiguous slice of the global index space so + // the round-robin bucket assignment stays even across all users. + let base = user_idx * uploads; + if parallel { + let mut set = JoinSet::new(); + for i in 0..uploads { + let bucket = bucket_for(base + i, &buckets); + let client = client.clone(); + let sem = sem.clone(); + set.spawn(async move { do_upload(client, bucket, size, sem).await }); + } + let mut out = Vec::with_capacity(uploads); + while let Some(res) = set.join_next().await { + match res { + Ok(outcome) => out.push(outcome), + Err(join_err) => out.push(OpOutcome::failure( + size, + Duration::ZERO, + format!("upload task panicked: {join_err}"), + )), + } + } + out + } else { + let mut out = Vec::with_capacity(uploads); + for i in 0..uploads { + let bucket = bucket_for(base + i, &buckets); + out.push(do_upload(client.clone(), bucket, size, sem.clone()).await); + } + out + } +} + +/// Drive configurable upload load against `--provider`. +/// +/// Targets are resolved from chain (`MemberBuckets[account]` ∩ buckets with a +/// `StorageAgreements[bucket][provider]` entry); buckets and agreements are +/// never created — if nothing matches, it errors out. `--users` clients each +/// perform `--uploads-per-user` uploads of `--max-payload-size` random bytes, +/// with users and per-user uploads run sequentially or in parallel per the +/// `--parallel-*` flags, optionally capped by `--max-concurrency`. /// -/// This resolves targets from chain (`MemberBuckets[account]` ∩ buckets with a -/// `StorageAgreements[bucket][provider]` entry) and never creates buckets or -/// agreements — if nothing matches, it errors out. -pub async fn upload(global: &GlobalArgs, args: &UploadArgs) -> Result<()> { +/// Returns the aggregated [`OpSummary`] for the run; the caller (`main`) views +/// them. Per-upload failures are folded into the metrics, so this only returns +/// `Err` for setup failures (bad args, chain connection, no matching buckets). +pub async fn upload(global: &GlobalArgs, args: &UploadArgs) -> Result { + if args.users < 1 { + bail!("--users must be at least 1"); + } + if args.uploads_per_user < 1 { + bail!("--uploads-per-user must be at least 1"); + } + if args.max_payload_size < 1 { + bail!("--max-payload-size must be at least 1 byte"); + } + // Identity: derive the account whose buckets we look up from the SURI. let suri = resolve_suri(global)?; let keypair = Keypair::from_uri(&suri.parse::().context("failed to parse SURI")?) @@ -108,29 +236,114 @@ pub async fn upload(global: &GlobalArgs, args: &UploadArgs) -> Result<()> { selected_buckets_id.truncate(max); } - println!( - "Uploading {} bytes to {} bucket(s) via {}", - args.size, + let total_uploads = args.users.saturating_mul(args.uploads_per_user); + // Progress goes to stderr so stdout carries only the final metrics view + // (keeping `--output json` parseable). + eprintln!( + "Stress test: {} user(s){}, {} upload(s)/user{}, {} bytes each, {} bucket(s) via {}{}", + args.users, + if args.parallel_users { + " [parallel]" + } else { + " [sequential]" + }, + args.uploads_per_user, + if args.parallel_uploads { + " [parallel]" + } else { + " [sequential]" + }, + args.max_payload_size, selected_buckets_id.len(), global.provider_url, + if args.max_concurrency > 0 { + format!(", max in-flight {}", args.max_concurrency) + } else { + String::new() + }, ); + eprintln!("Running {total_uploads} upload(s)..."); - // Off-chain HTTP uploads (no chain, no signer). Constant-fill payload,. - let user = StorageUserClient::new(config).context("failed to construct provider client")?; - let payload = vec![0x42; args.size]; + // Off-chain HTTP uploads (no chain, no signer). One client per user gives + // each simulated user its own connection pool. + let mut clients = Vec::with_capacity(args.users); + for _ in 0..args.users { + clients.push(Arc::new( + StorageUserClient::new(config.clone()) + .context("failed to construct provider client")?, + )); + } - for bucket in &selected_buckets_id { - let data_root = user - .upload(*bucket, &payload, ChunkingStrategy::default()) - .await - .with_context(|| format!("upload to bucket {bucket} failed"))?; - println!( - " bucket {bucket}: uploaded {} bytes, data_root 0x{}", - payload.len(), - hex::encode(data_root.as_bytes()), - ); + let buckets = Arc::new(selected_buckets_id); + let sem = (args.max_concurrency > 0).then(|| Arc::new(Semaphore::new(args.max_concurrency))); + + let started = Instant::now(); + let mut outcomes = Vec::with_capacity(total_uploads); + if args.parallel_users { + let mut users_set = JoinSet::new(); + for (user_idx, client) in clients.into_iter().enumerate() { + let buckets = buckets.clone(); + let sem = sem.clone(); + let uploads = args.uploads_per_user; + let size = args.max_payload_size; + let parallel_uploads = args.parallel_uploads; + users_set.spawn(async move { + run_user( + user_idx, + client, + buckets, + uploads, + size, + parallel_uploads, + sem, + ) + .await + }); + } + while let Some(res) = users_set.join_next().await { + match res { + Ok(user_outcomes) => outcomes.extend(user_outcomes), + Err(join_err) => bail!("user task panicked: {join_err}"), + } + } + } else { + for (user_idx, client) in clients.into_iter().enumerate() { + let user_outcomes = run_user( + user_idx, + client, + buckets.clone(), + args.uploads_per_user, + args.max_payload_size, + args.parallel_uploads, + sem.clone(), + ) + .await; + outcomes.extend(user_outcomes); + } + } + let elapsed = started.elapsed(); + + // Return the aggregated metrics; `main` views them and sets the exit code. + // Individual upload failures live in the metrics, not in this `Result` — + // only setup errors (chain, args, no buckets) above are propagated as `Err`. + Ok(summarize(Operation::Upload, &outcomes, elapsed)) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn random_payload_has_exact_size() { + for size in [1usize, 7, 1024, 512 * 1024] { + assert_eq!(random_payload(size).len(), size); + } } - println!("Done: {} bucket(s) written.", selected_buckets_id.len()); - Ok(()) + #[test] + fn bucket_for_round_robins() { + let buckets = [10u64, 20, 30]; + let picked: Vec = (0..7).map(|i| bucket_for(i, &buckets)).collect(); + assert_eq!(picked, vec![10, 20, 30, 10, 20, 30, 10]); + } } diff --git a/utils/storage-cli/src/common/mod.rs b/utils/storage-cli/src/common/mod.rs index a0d13b65..1740d70a 100644 --- a/utils/storage-cli/src/common/mod.rs +++ b/utils/storage-cli/src/common/mod.rs @@ -34,6 +34,7 @@ mod tests { provider_url: "http://127.0.0.1:3333".to_string(), suri, keyfile, + output: crate::metrics::OutputFormat::Text, } } diff --git a/utils/storage-cli/src/main.rs b/utils/storage-cli/src/main.rs index c59bc254..f68e8180 100644 --- a/utils/storage-cli/src/main.rs +++ b/utils/storage-cli/src/main.rs @@ -8,20 +8,40 @@ mod cli; mod commands; mod common; +mod metrics; +use anyhow::bail; use clap::Parser; use crate::cli::{Cli, Command}; use crate::commands::stress_test::StressTest; +use crate::metrics::{print_text, to_json, OpSummary, OutputFormat}; #[tokio::main] async fn main() -> anyhow::Result<()> { tracing_subscriber::fmt::init(); let cli = Cli::parse(); + + // Scenarios compute and return their metrics; `main` collects every run's + // results and views them here. Adding a `read`/`delete` subcommand means + // pushing another `OpSummary` onto this vec — the viewing below is shared. + let mut all_results: Vec = Vec::new(); match &cli.command { Command::StressTest(StressTest::ProviderUpload(args)) => { - commands::stress_test::upload(&cli.global, args).await + all_results.push(commands::stress_test::upload(&cli.global, args).await?); } } + + match cli.global.output { + OutputFormat::Text => print_text(&all_results), + OutputFormat::Json => println!("{}", to_json(&all_results)?), + } + + // Non-zero exit if any scenario completed with no successful operations. + if let Some(m) = all_results.iter().find(|m| m.total > 0 && m.ok == 0) { + bail!("all {} {} failed", m.total, m.operation.noun_plural()); + } + + Ok(()) } diff --git a/utils/storage-cli/src/metrics.rs b/utils/storage-cli/src/metrics.rs new file mode 100644 index 00000000..ff5e7e7a --- /dev/null +++ b/utils/storage-cli/src/metrics.rs @@ -0,0 +1,367 @@ +// SPDX-License-Identifier: Apache-2.0 + +//! Operation metrics shared across stress-test scenarios. +//! +//! Scenarios record one [`OpOutcome`] per individual operation (an upload +//! today; a read or delete later) and aggregate them with [`summarize`] into an +//! [`OpSummary`], which prints a uniform summary and is returned so callers can +//! consume the numbers programmatically. The aggregation is operation-agnostic: +//! adding a `read`/`delete` scenario means producing `Vec` and +//! tagging the summary with the matching [`Operation`] — no new metrics code. + +use std::fmt; +use std::time::Duration; + +use clap::ValueEnum; +use serde::Serialize; + +/// How to render the collected metrics. +#[derive(Debug, Clone, Copy, PartialEq, Eq, ValueEnum)] +pub enum OutputFormat { + Text, + Json, +} + +/// The kind of operation a stress-test scenario exercises. Extend this as new +/// scenarios are added; [`OpSummary`] formats itself from these labels. +/// +/// `Read`/`Delete` are declared ahead of their scenarios so the metrics +/// abstraction is ready for them; they are exercised by tests until the +/// corresponding `stress-test` subcommands land. +#[allow(dead_code)] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Operation { + /// Off-chain HTTP upload to a provider. + Upload, + /// Off-chain HTTP read/download from a provider. + Read, + /// Delete of stored data. + Delete, +} + +impl Operation { + /// Lowercase verb naming the operation, e.g. `"upload"`. + pub fn as_str(self) -> &'static str { + match self { + Operation::Upload => "upload", + Operation::Read => "read", + Operation::Delete => "delete", + } + } + + /// Plural noun used for counts and rates, e.g. `"uploads"`. + pub fn noun_plural(self) -> &'static str { + match self { + Operation::Upload => "uploads", + Operation::Read => "reads", + Operation::Delete => "deletes", + } + } + + /// Past-tense verb used on the data line, e.g. `"uploaded"`. + pub fn past_tense(self) -> &'static str { + match self { + Operation::Upload => "uploaded", + Operation::Read => "read", + Operation::Delete => "deleted", + } + } +} + +impl fmt::Display for Operation { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(self.as_str()) + } +} + +/// Outcome of a single operation attempt. A stress run records every outcome +/// (success or failure) rather than aborting, so failures surface in the +/// aggregate instead of stopping the run. +#[derive(Debug, Clone)] +pub struct OpOutcome { + /// Whether the operation succeeded. + pub ok: bool, + /// Bytes transferred by this operation (uploaded/read; typically 0 for a + /// delete). + pub bytes: usize, + /// Elapsed time the operation took. + pub elapsed: Duration, + /// Error message if the operation failed. + pub err: Option, +} + +impl OpOutcome { + /// Record a successful operation that transferred `bytes` in `elapsed`. + pub fn success(bytes: usize, elapsed: Duration) -> Self { + Self { + ok: true, + bytes, + elapsed, + err: None, + } + } + + /// Record a failed operation; `bytes` is what it would have transferred. + pub fn failure(bytes: usize, elapsed: Duration, err: String) -> Self { + Self { + ok: false, + bytes, + elapsed, + err: Some(err), + } + } +} + +/// Number of sample error messages retained for the summary. +const MAX_SAMPLE_ERRORS: usize = 5; + +/// Aggregated metrics over every [`OpOutcome`] of a single scenario run. +/// Returned from a scenario so the numbers (counts, bytes, timings) are +/// available to callers in addition to the printed summary. +#[derive(Debug, Clone)] +pub struct OpSummary { + /// Which operation these metrics describe. + pub operation: Operation, + /// Total operations attempted. + pub total: usize, + /// Operations that succeeded. + pub ok: usize, + /// Operations that failed. + pub failed: usize, + /// Bytes transferred across the successful operations. + pub bytes_ok: usize, + /// Total elapsed (wall-clock) duration of the whole run — real time from + /// first to last operation, not the sum of per-operation latencies. + pub elapsed: Duration, + /// Minimum latency over successful operations. + pub lat_min: Duration, + /// Mean latency over successful operations. + pub lat_avg: Duration, + /// Maximum latency over successful operations. + pub lat_max: Duration, + /// Up to [`MAX_SAMPLE_ERRORS`] sample messages from failed operations. + pub sample_errors: Vec, +} + +impl OpSummary { + /// Successful-byte throughput in bytes per second over the elapsed time. + pub fn bytes_per_sec(&self) -> f64 { + let secs = self.elapsed.as_secs_f64(); + if secs > 0.0 { + self.bytes_ok as f64 / secs + } else { + 0.0 + } + } + + /// Successful-operation rate per second over the elapsed time. + pub fn ops_per_sec(&self) -> f64 { + let secs = self.elapsed.as_secs_f64(); + if secs > 0.0 { + self.ok as f64 / secs + } else { + 0.0 + } + } +} + +/// Aggregate per-operation outcomes against the elapsed duration of the run. +pub fn summarize(operation: Operation, outcomes: &[OpOutcome], elapsed: Duration) -> OpSummary { + let total = outcomes.len(); + let ok = outcomes.iter().filter(|o| o.ok).count(); + let bytes_ok = outcomes.iter().filter(|o| o.ok).map(|o| o.bytes).sum(); + + let mut lat_min = Duration::MAX; + let mut lat_max = Duration::ZERO; + let mut lat_sum = Duration::ZERO; + for o in outcomes.iter().filter(|o| o.ok) { + lat_min = lat_min.min(o.elapsed); + lat_max = lat_max.max(o.elapsed); + lat_sum += o.elapsed; + } + let lat_avg = lat_sum.checked_div(ok as u32).unwrap_or(Duration::ZERO); + if ok == 0 { + lat_min = Duration::ZERO; + } + + let sample_errors = outcomes + .iter() + .filter_map(|o| o.err.clone()) + .take(MAX_SAMPLE_ERRORS) + .collect(); + + OpSummary { + operation, + total, + ok, + failed: total - ok, + bytes_ok, + elapsed, + lat_min, + lat_avg, + lat_max, + sample_errors, + } +} + +/// JSON-friendly projection of [`OpSummary`]: operation as a lowercase string, +/// durations as fractional seconds, and the derived throughput rates inlined. +/// Decouples the on-disk/wire shape from the in-memory struct. +#[derive(Debug, Serialize)] +struct OpSummaryJson<'a> { + operation: &'a str, + total: usize, + ok: usize, + failed: usize, + bytes_ok: usize, + elapsed_secs: f64, + throughput_bytes_per_sec: f64, + throughput_ops_per_sec: f64, + latency_min_secs: f64, + latency_avg_secs: f64, + latency_max_secs: f64, + sample_errors: &'a [String], +} + +impl OpSummary { + fn as_json(&self) -> OpSummaryJson<'_> { + OpSummaryJson { + operation: self.operation.as_str(), + total: self.total, + ok: self.ok, + failed: self.failed, + bytes_ok: self.bytes_ok, + elapsed_secs: self.elapsed.as_secs_f64(), + throughput_bytes_per_sec: self.bytes_per_sec(), + throughput_ops_per_sec: self.ops_per_sec(), + latency_min_secs: self.lat_min.as_secs_f64(), + latency_avg_secs: self.lat_avg.as_secs_f64(), + latency_max_secs: self.lat_max.as_secs_f64(), + sample_errors: &self.sample_errors, + } + } +} + +/// Print a human-readable text summary for each scenario's metrics. The text +/// view of the collected results, kept in the CLI front-end so scenarios just +/// compute and return. +pub fn print_text(results: &[OpSummary]) { + for m in results { + print!("{m}"); + } +} + +/// Render all scenario metrics as a pretty-printed JSON array (the `--output +/// json` view). +pub fn to_json(results: &[OpSummary]) -> serde_json::Result { + let view: Vec = results.iter().map(OpSummary::as_json).collect(); + serde_json::to_string_pretty(&view) +} + +impl fmt::Display for OpSummary { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + let secs = self.elapsed.as_secs_f64(); + let mib = self.bytes_ok as f64 / (1024.0 * 1024.0); + let mib_s = self.bytes_per_sec() / (1024.0 * 1024.0); + let ops = self.ops_per_sec(); + writeln!( + f, + "{} results: {} total, {} ok, {} failed", + self.operation, self.total, self.ok, self.failed + )?; + writeln!( + f, + " data: {:.2} MiB {} in {:.3}s", + mib, + self.operation.past_tense(), + secs + )?; + writeln!( + f, + " throughput: {mib_s:.2} MiB/s, {ops:.1} {}/s", + self.operation.noun_plural() + )?; + writeln!( + f, + " latency: min {:.3}s / avg {:.3}s / max {:.3}s", + self.lat_min.as_secs_f64(), + self.lat_avg.as_secs_f64(), + self.lat_max.as_secs_f64(), + )?; + if !self.sample_errors.is_empty() { + writeln!(f, " sample errors:")?; + for e in &self.sample_errors { + writeln!(f, " - {e}")?; + } + } + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn summarize_counts_and_bytes() { + let outcomes = vec![ + OpOutcome::success(100, Duration::from_millis(10)), + OpOutcome::success(100, Duration::from_millis(30)), + OpOutcome::failure(100, Duration::from_millis(5), "boom".into()), + ]; + let m = summarize(Operation::Upload, &outcomes, Duration::from_secs(1)); + assert_eq!(m.operation, Operation::Upload); + assert_eq!(m.total, 3); + assert_eq!(m.ok, 2); + assert_eq!(m.failed, 1); + assert_eq!(m.bytes_ok, 200); + assert_eq!(m.lat_min, Duration::from_millis(10)); + assert_eq!(m.lat_max, Duration::from_millis(30)); + assert_eq!(m.lat_avg, Duration::from_millis(20)); + assert_eq!(m.ops_per_sec(), 2.0); + assert_eq!(m.sample_errors, vec!["boom".to_string()]); + } + + #[test] + fn summarize_all_failed_has_zero_latency() { + let outcomes = vec![OpOutcome::failure( + 50, + Duration::from_millis(5), + "nope".into(), + )]; + let m = summarize(Operation::Read, &outcomes, Duration::from_secs(1)); + assert_eq!(m.ok, 0); + assert_eq!(m.bytes_ok, 0); + assert_eq!(m.lat_min, Duration::ZERO); + assert_eq!(m.lat_avg, Duration::ZERO); + assert_eq!(m.lat_max, Duration::ZERO); + } + + #[test] + fn operation_labels_are_distinct() { + assert_eq!(Operation::Upload.noun_plural(), "uploads"); + assert_eq!(Operation::Read.past_tense(), "read"); + assert_eq!(Operation::Delete.to_string(), "delete"); + } + + #[test] + fn to_json_projects_expected_fields() { + let outcomes = vec![ + OpOutcome::success(1024, Duration::from_millis(10)), + OpOutcome::failure(1024, Duration::from_millis(5), "boom".into()), + ]; + let m = summarize(Operation::Upload, &outcomes, Duration::from_secs(2)); + let json = to_json(&[m]).expect("serializes"); + let v: serde_json::Value = serde_json::from_str(&json).expect("valid json"); + let entry = &v[0]; + assert_eq!(entry["operation"], "upload"); + assert_eq!(entry["total"], 2); + assert_eq!(entry["ok"], 1); + assert_eq!(entry["failed"], 1); + assert_eq!(entry["bytes_ok"], 1024); + assert_eq!(entry["elapsed_secs"], 2.0); + assert_eq!(entry["throughput_ops_per_sec"], 0.5); + assert_eq!(entry["latency_avg_secs"], 0.01); + assert_eq!(entry["sample_errors"][0], "boom"); + } +} From fbc9299b635c7782938a0656f5f812079de1efeb Mon Sep 17 00:00:00 2001 From: Darwin Subramaniam Date: Thu, 25 Jun 2026 11:58:20 +0800 Subject: [PATCH 07/10] refactor(storage-cli): re-architect Operation from enum to trait Model the operation kind as a trait whose implementors supply their display labels, replacing a closed enum with three exhaustive match methods and a pre-declared Read/Delete that required #[allow(dead_code)]. OpSummary now stores the resolved OpLabels and summarize() is generic over the operation, so metrics.rs is fully operation-agnostic: a new operation is a self-contained impl with no central list to extend. The Upload marker lives in the stress-test command for now. Re-architecture: storage-cli-operations --- utils/storage-cli/src/commands/stress_test.rs | 29 ++++- utils/storage-cli/src/main.rs | 2 +- utils/storage-cli/src/metrics.rs | 119 +++++++----------- 3 files changed, 72 insertions(+), 78 deletions(-) diff --git a/utils/storage-cli/src/commands/stress_test.rs b/utils/storage-cli/src/commands/stress_test.rs index ec6eda99..95df689a 100644 --- a/utils/storage-cli/src/commands/stress_test.rs +++ b/utils/storage-cli/src/commands/stress_test.rs @@ -17,11 +17,24 @@ use tokio::task::JoinSet; use crate::cli::GlobalArgs; use crate::common::resolve_suri; -use crate::metrics::{summarize, OpOutcome, OpSummary, Operation}; +use crate::metrics::{summarize, OpLabels, OpOutcome, OpSummary, Operation}; /// Bucket identifier on chain (alias of `storage_primitives::BucketId`, a `u64`). type BucketId = u64; +/// Off-chain HTTP upload to a provider — the operation this scenario exercises. +struct Upload; + +impl Operation for Upload { + fn labels(&self) -> OpLabels { + OpLabels { + verb: "upload", + noun_plural: "uploads", + past_tense: "uploaded", + } + } +} + // === Stress test subcommands === #[derive(Debug, Subcommand)] pub enum StressTest { @@ -321,12 +334,8 @@ pub async fn upload(global: &GlobalArgs, args: &UploadArgs) -> Result outcomes.extend(user_outcomes); } } - let elapsed = started.elapsed(); - // Return the aggregated metrics; `main` views them and sets the exit code. - // Individual upload failures live in the metrics, not in this `Result` — - // only setup errors (chain, args, no buckets) above are propagated as `Err`. - Ok(summarize(Operation::Upload, &outcomes, elapsed)) + Ok(summarize(Upload, &outcomes, started.elapsed())) } #[cfg(test)] @@ -346,4 +355,12 @@ mod tests { let picked: Vec = (0..7).map(|i| bucket_for(i, &buckets)).collect(); assert_eq!(picked, vec![10, 20, 30, 10, 20, 30, 10]); } + + #[test] + fn upload_labels_match() { + let l = Upload.labels(); + assert_eq!(l.verb, "upload"); + assert_eq!(l.noun_plural, "uploads"); + assert_eq!(l.past_tense, "uploaded"); + } } diff --git a/utils/storage-cli/src/main.rs b/utils/storage-cli/src/main.rs index f68e8180..65b77a0d 100644 --- a/utils/storage-cli/src/main.rs +++ b/utils/storage-cli/src/main.rs @@ -40,7 +40,7 @@ async fn main() -> anyhow::Result<()> { // Non-zero exit if any scenario completed with no successful operations. if let Some(m) = all_results.iter().find(|m| m.total > 0 && m.ok == 0) { - bail!("all {} {} failed", m.total, m.operation.noun_plural()); + bail!("all {} {} failed", m.total, m.labels.noun_plural); } Ok(()) diff --git a/utils/storage-cli/src/metrics.rs b/utils/storage-cli/src/metrics.rs index ff5e7e7a..68a12235 100644 --- a/utils/storage-cli/src/metrics.rs +++ b/utils/storage-cli/src/metrics.rs @@ -7,7 +7,7 @@ //! [`OpSummary`], which prints a uniform summary and is returned so callers can //! consume the numbers programmatically. The aggregation is operation-agnostic: //! adding a `read`/`delete` scenario means producing `Vec` and -//! tagging the summary with the matching [`Operation`] — no new metrics code. +//! tagging the summary via an [`Operation`] implementor — no new metrics code. use std::fmt; use std::time::Duration; @@ -22,56 +22,25 @@ pub enum OutputFormat { Json, } -/// The kind of operation a stress-test scenario exercises. Extend this as new -/// scenarios are added; [`OpSummary`] formats itself from these labels. -/// -/// `Read`/`Delete` are declared ahead of their scenarios so the metrics -/// abstraction is ready for them; they are exercised by tests until the -/// corresponding `stress-test` subcommands land. -#[allow(dead_code)] -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum Operation { - /// Off-chain HTTP upload to a provider. - Upload, - /// Off-chain HTTP read/download from a provider. - Read, - /// Delete of stored data. - Delete, +/// Display labels for one kind of operation. Captured into [`OpSummary`] by +/// [`summarize`] so a summary formats itself without knowing the concrete op. +#[derive(Debug, Clone, Copy)] +pub struct OpLabels { + /// Lowercase verb, e.g. `"upload"`. + pub verb: &'static str, + /// Plural noun for counts/rates, e.g. `"uploads"`. + pub noun_plural: &'static str, + /// Past-tense verb for the data line, e.g. `"uploaded"`. + pub past_tense: &'static str, } -impl Operation { - /// Lowercase verb naming the operation, e.g. `"upload"`. - pub fn as_str(self) -> &'static str { - match self { - Operation::Upload => "upload", - Operation::Read => "read", - Operation::Delete => "delete", - } - } - - /// Plural noun used for counts and rates, e.g. `"uploads"`. - pub fn noun_plural(self) -> &'static str { - match self { - Operation::Upload => "uploads", - Operation::Read => "reads", - Operation::Delete => "deletes", - } - } - - /// Past-tense verb used on the data line, e.g. `"uploaded"`. - pub fn past_tense(self) -> &'static str { - match self { - Operation::Upload => "uploaded", - Operation::Read => "read", - Operation::Delete => "deleted", - } - } -} - -impl fmt::Display for Operation { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - f.write_str(self.as_str()) - } +/// One kind of operation a stress-test scenario exercises. Implement it on a +/// marker type in the scenario's own module (e.g. `Upload` in the `stress_test` +/// command): the labels live with the operation, so there is no central list to +/// extend and nothing to keep in sync. +pub trait Operation { + /// Display labels for this operation. + fn labels(&self) -> OpLabels; } /// Outcome of a single operation attempt. A stress run records every outcome @@ -120,8 +89,8 @@ const MAX_SAMPLE_ERRORS: usize = 5; /// available to callers in addition to the printed summary. #[derive(Debug, Clone)] pub struct OpSummary { - /// Which operation these metrics describe. - pub operation: Operation, + /// Display labels for the operation these metrics describe. + pub labels: OpLabels, /// Total operations attempted. pub total: usize, /// Operations that succeeded. @@ -166,7 +135,11 @@ impl OpSummary { } /// Aggregate per-operation outcomes against the elapsed duration of the run. -pub fn summarize(operation: Operation, outcomes: &[OpOutcome], elapsed: Duration) -> OpSummary { +pub fn summarize( + operation: impl Operation, + outcomes: &[OpOutcome], + elapsed: Duration, +) -> OpSummary { let total = outcomes.len(); let ok = outcomes.iter().filter(|o| o.ok).count(); let bytes_ok = outcomes.iter().filter(|o| o.ok).map(|o| o.bytes).sum(); @@ -191,7 +164,7 @@ pub fn summarize(operation: Operation, outcomes: &[OpOutcome], elapsed: Duration .collect(); OpSummary { - operation, + labels: operation.labels(), total, ok, failed: total - ok, @@ -226,7 +199,7 @@ struct OpSummaryJson<'a> { impl OpSummary { fn as_json(&self) -> OpSummaryJson<'_> { OpSummaryJson { - operation: self.operation.as_str(), + operation: self.labels.verb, total: self.total, ok: self.ok, failed: self.failed, @@ -267,19 +240,17 @@ impl fmt::Display for OpSummary { writeln!( f, "{} results: {} total, {} ok, {} failed", - self.operation, self.total, self.ok, self.failed + self.labels.verb, self.total, self.ok, self.failed )?; writeln!( f, " data: {:.2} MiB {} in {:.3}s", - mib, - self.operation.past_tense(), - secs + mib, self.labels.past_tense, secs )?; writeln!( f, " throughput: {mib_s:.2} MiB/s, {ops:.1} {}/s", - self.operation.noun_plural() + self.labels.noun_plural )?; writeln!( f, @@ -302,6 +273,19 @@ impl fmt::Display for OpSummary { mod tests { use super::*; + /// Minimal operation fixture so the generic machinery is testable without + /// depending on any concrete scenario's operation type. + struct TestOp; + impl Operation for TestOp { + fn labels(&self) -> OpLabels { + OpLabels { + verb: "test", + noun_plural: "tests", + past_tense: "tested", + } + } + } + #[test] fn summarize_counts_and_bytes() { let outcomes = vec![ @@ -309,8 +293,8 @@ mod tests { OpOutcome::success(100, Duration::from_millis(30)), OpOutcome::failure(100, Duration::from_millis(5), "boom".into()), ]; - let m = summarize(Operation::Upload, &outcomes, Duration::from_secs(1)); - assert_eq!(m.operation, Operation::Upload); + let m = summarize(TestOp, &outcomes, Duration::from_secs(1)); + assert_eq!(m.labels.verb, "test"); assert_eq!(m.total, 3); assert_eq!(m.ok, 2); assert_eq!(m.failed, 1); @@ -329,7 +313,7 @@ mod tests { Duration::from_millis(5), "nope".into(), )]; - let m = summarize(Operation::Read, &outcomes, Duration::from_secs(1)); + let m = summarize(TestOp, &outcomes, Duration::from_secs(1)); assert_eq!(m.ok, 0); assert_eq!(m.bytes_ok, 0); assert_eq!(m.lat_min, Duration::ZERO); @@ -337,24 +321,17 @@ mod tests { assert_eq!(m.lat_max, Duration::ZERO); } - #[test] - fn operation_labels_are_distinct() { - assert_eq!(Operation::Upload.noun_plural(), "uploads"); - assert_eq!(Operation::Read.past_tense(), "read"); - assert_eq!(Operation::Delete.to_string(), "delete"); - } - #[test] fn to_json_projects_expected_fields() { let outcomes = vec![ OpOutcome::success(1024, Duration::from_millis(10)), OpOutcome::failure(1024, Duration::from_millis(5), "boom".into()), ]; - let m = summarize(Operation::Upload, &outcomes, Duration::from_secs(2)); + let m = summarize(TestOp, &outcomes, Duration::from_secs(2)); let json = to_json(&[m]).expect("serializes"); let v: serde_json::Value = serde_json::from_str(&json).expect("valid json"); let entry = &v[0]; - assert_eq!(entry["operation"], "upload"); + assert_eq!(entry["operation"], "test"); assert_eq!(entry["total"], 2); assert_eq!(entry["ok"], 1); assert_eq!(entry["failed"], 1); From cdb714c28bd150b08f155b9df03810df69f0fee3 Mon Sep 17 00:00:00 2001 From: Darwin Subramaniam Date: Thu, 25 Jun 2026 11:58:50 +0800 Subject: [PATCH 08/10] refactor(storage-cli): re-architect upload into a reusable actions module Split the upload operation out of the stress-test command into a new top-level `actions` module. actions::upload owns the Upload marker and the upload_once(client, bucket, payload) -> OpOutcome primitive; stress_test now only orchestrates (users x uploads, parallelism, concurrency cap) and composes the action. A future read/delete action is a sibling module here, usable by any scenario without touching the metrics layer. Also moves BucketId into `common`, de-duplicates the user dispatch behind a single run_one closure, and switches a panicked user task from aborting the whole run (bail!) to warn-and-continue so partial metrics still reach `main`. Re-architecture: storage-cli-operations --- utils/storage-cli/src/actions/mod.rs | 11 +++ utils/storage-cli/src/actions/upload.rs | 52 +++++++++++ utils/storage-cli/src/commands/stress_test.rs | 91 ++++++------------- utils/storage-cli/src/common/mod.rs | 3 + utils/storage-cli/src/main.rs | 1 + 5 files changed, 93 insertions(+), 65 deletions(-) create mode 100644 utils/storage-cli/src/actions/mod.rs create mode 100644 utils/storage-cli/src/actions/upload.rs diff --git a/utils/storage-cli/src/actions/mod.rs b/utils/storage-cli/src/actions/mod.rs new file mode 100644 index 00000000..5f48412d --- /dev/null +++ b/utils/storage-cli/src/actions/mod.rs @@ -0,0 +1,11 @@ +// SPDX-License-Identifier: Apache-2.0 + +//! Reusable storage actions that scenarios compose. +//! +//! Each action owns its `Operation` marker (from [`crate::metrics`]) and a +//! primitive that performs the operation once and returns a measured +//! [`crate::metrics::OpOutcome`]. Scenarios such as `stress-test` drive these +//! primitives under load; a future `read`/`delete` action is a sibling module +//! here, usable by any scenario without touching the metrics layer. + +pub mod upload; diff --git a/utils/storage-cli/src/actions/upload.rs b/utils/storage-cli/src/actions/upload.rs new file mode 100644 index 00000000..58176dd2 --- /dev/null +++ b/utils/storage-cli/src/actions/upload.rs @@ -0,0 +1,52 @@ +// SPDX-License-Identifier: Apache-2.0 + +//! Upload action: perform a single off-chain HTTP upload and measure it. + +use std::time::Instant; + +use storage_client::{ChunkingStrategy, StorageUserClient}; + +use crate::common::BucketId; +use crate::metrics::{OpLabels, OpOutcome, Operation}; + +/// Off-chain HTTP upload to a provider. +pub struct Upload; + +impl Operation for Upload { + fn labels(&self) -> OpLabels { + OpLabels { + verb: "upload", + noun_plural: "uploads", + past_tense: "uploaded", + } + } +} + +/// Perform one upload of `payload` to `bucket`, returning the measured outcome. +pub async fn upload_once( + client: &StorageUserClient, + bucket: BucketId, + payload: &[u8], +) -> OpOutcome { + let started = Instant::now(); + match client + .upload(bucket, payload, ChunkingStrategy::default()) + .await + { + Ok(_root) => OpOutcome::success(payload.len(), started.elapsed()), + Err(e) => OpOutcome::failure(payload.len(), started.elapsed(), e.to_string()), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn upload_labels_match() { + let l = Upload.labels(); + assert_eq!(l.verb, "upload"); + assert_eq!(l.noun_plural, "uploads"); + assert_eq!(l.past_tense, "uploaded"); + } +} diff --git a/utils/storage-cli/src/commands/stress_test.rs b/utils/storage-cli/src/commands/stress_test.rs index 95df689a..c65993b0 100644 --- a/utils/storage-cli/src/commands/stress_test.rs +++ b/utils/storage-cli/src/commands/stress_test.rs @@ -10,30 +10,15 @@ use clap::{Args, Subcommand}; use sp_core::crypto::Ss58Codec; use sp_runtime::AccountId32; use storage_client::substrate::SubstrateClient; -use storage_client::{AdminClient, ChunkingStrategy, ClientConfig, StorageUserClient}; +use storage_client::{AdminClient, ClientConfig, StorageUserClient}; use subxt_signer::{sr25519::Keypair, SecretUri}; use tokio::sync::Semaphore; use tokio::task::JoinSet; +use crate::actions::upload::{upload_once, Upload}; use crate::cli::GlobalArgs; -use crate::common::resolve_suri; -use crate::metrics::{summarize, OpLabels, OpOutcome, OpSummary, Operation}; - -/// Bucket identifier on chain (alias of `storage_primitives::BucketId`, a `u64`). -type BucketId = u64; - -/// Off-chain HTTP upload to a provider — the operation this scenario exercises. -struct Upload; - -impl Operation for Upload { - fn labels(&self) -> OpLabels { - OpLabels { - verb: "upload", - noun_plural: "uploads", - past_tense: "uploaded", - } - } -} +use crate::common::{resolve_suri, BucketId}; +use crate::metrics::{summarize, OpOutcome, OpSummary}; // === Stress test subcommands === #[derive(Debug, Subcommand)] @@ -110,15 +95,7 @@ async fn do_upload( Some(s) => s.acquire().await.ok(), None => None, }; - let payload = random_payload(size); - let started = Instant::now(); - match client - .upload(bucket, &payload, ChunkingStrategy::default()) - .await - { - Ok(_root) => OpOutcome::success(size, started.elapsed()), - Err(e) => OpOutcome::failure(size, started.elapsed(), e.to_string()), - } + upload_once(&client, bucket, &random_payload(size)).await } /// Run a single user's `uploads` uploads, either sequentially or in parallel. @@ -292,46 +269,38 @@ pub async fn upload(global: &GlobalArgs, args: &UploadArgs) -> Result let started = Instant::now(); let mut outcomes = Vec::with_capacity(total_uploads); + // Build each user's future once; spawn it for parallelism or await it in + // sequence. Passing the `Copy` config values positionally keeps the spawned + // future `'static` (it owns its `usize`/`bool`/`Arc`s), so the closure only + // borrows `buckets`/`sem`/`args` locally. + let run_one = |user_idx: usize, client: Arc| { + run_user( + user_idx, + client, + buckets.clone(), + args.uploads_per_user, + args.max_payload_size, + args.parallel_uploads, + sem.clone(), + ) + }; + if args.parallel_users { let mut users_set = JoinSet::new(); for (user_idx, client) in clients.into_iter().enumerate() { - let buckets = buckets.clone(); - let sem = sem.clone(); - let uploads = args.uploads_per_user; - let size = args.max_payload_size; - let parallel_uploads = args.parallel_uploads; - users_set.spawn(async move { - run_user( - user_idx, - client, - buckets, - uploads, - size, - parallel_uploads, - sem, - ) - .await - }); + users_set.spawn(run_one(user_idx, client)); } while let Some(res) = users_set.join_next().await { match res { Ok(user_outcomes) => outcomes.extend(user_outcomes), - Err(join_err) => bail!("user task panicked: {join_err}"), + // A panicked user task is a bug, not load — warn and keep the + // partial results rather than discarding the whole run. + Err(join_err) => eprintln!("warning: a user task failed: {join_err}"), } } } else { for (user_idx, client) in clients.into_iter().enumerate() { - let user_outcomes = run_user( - user_idx, - client, - buckets.clone(), - args.uploads_per_user, - args.max_payload_size, - args.parallel_uploads, - sem.clone(), - ) - .await; - outcomes.extend(user_outcomes); + outcomes.extend(run_one(user_idx, client).await); } } @@ -355,12 +324,4 @@ mod tests { let picked: Vec = (0..7).map(|i| bucket_for(i, &buckets)).collect(); assert_eq!(picked, vec![10, 20, 30, 10, 20, 30, 10]); } - - #[test] - fn upload_labels_match() { - let l = Upload.labels(); - assert_eq!(l.verb, "upload"); - assert_eq!(l.noun_plural, "uploads"); - assert_eq!(l.past_tense, "uploaded"); - } } diff --git a/utils/storage-cli/src/common/mod.rs b/utils/storage-cli/src/common/mod.rs index 1740d70a..5b338552 100644 --- a/utils/storage-cli/src/common/mod.rs +++ b/utils/storage-cli/src/common/mod.rs @@ -3,6 +3,9 @@ use crate::cli::GlobalArgs; use anyhow::{bail, Context, Result}; +/// Bucket identifier on chain (alias of `storage_primitives::BucketId`, a `u64`). +pub type BucketId = u64; + /// Resolve the SURI from either `--suri` or `--keyfile` (exactly one is required) pub fn resolve_suri(global: &GlobalArgs) -> Result { match (&global.suri, &global.keyfile) { diff --git a/utils/storage-cli/src/main.rs b/utils/storage-cli/src/main.rs index 65b77a0d..8c53dcd3 100644 --- a/utils/storage-cli/src/main.rs +++ b/utils/storage-cli/src/main.rs @@ -5,6 +5,7 @@ //! [`storage-client`](../../client) SDK. See `--help` for the available //! subcommands. +mod actions; mod cli; mod commands; mod common; From c788a9a790458e5ec1bbdb13eba195cc2b75a0e4 Mon Sep 17 00:00:00 2001 From: Darwin Subramaniam Date: Mon, 29 Jun 2026 19:39:59 +0800 Subject: [PATCH 09/10] refactor: update upload arguments to use NonZeroUsize and improve error handling --- utils/storage-cli/src/commands/stress_test.rs | 218 ++++++++++-------- 1 file changed, 124 insertions(+), 94 deletions(-) diff --git a/utils/storage-cli/src/commands/stress_test.rs b/utils/storage-cli/src/commands/stress_test.rs index c65993b0..cfbd2bc0 100644 --- a/utils/storage-cli/src/commands/stress_test.rs +++ b/utils/storage-cli/src/commands/stress_test.rs @@ -2,6 +2,7 @@ //! `stress-test` subcommands. +use std::num::NonZeroUsize; use std::sync::Arc; use std::time::{Duration, Instant}; @@ -44,16 +45,16 @@ pub struct UploadArgs { pub max_buckets_to_write: Option, /// Number of concurrent simulated users, each with its own client (1..N). - #[arg(long, value_name = "N", default_value_t = 1)] - pub users: usize, + #[arg(long, value_name = "N", default_value = "1")] + pub users: NonZeroUsize, /// Number of uploads each user performs (1..X). - #[arg(long, value_name = "X", default_value_t = 1)] - pub uploads_per_user: usize, + #[arg(long, value_name = "X", default_value = "1")] + pub uploads_per_user: NonZeroUsize, /// Exact size in bytes of each randomly-generated payload (default 0.5 MiB). - #[arg(long, value_name = "BYTES", default_value_t = 512 * 1024)] - pub max_payload_size: usize, + #[arg(long, value_name = "BYTES", default_value = "524288")] + pub max_payload_size: NonZeroUsize, /// Run users in parallel (default: sequential). #[arg(long, default_value_t = false)] @@ -141,65 +142,46 @@ async fn run_user( } } -/// Drive configurable upload load against `--provider`. -/// -/// Targets are resolved from chain (`MemberBuckets[account]` ∩ buckets with a -/// `StorageAgreements[bucket][provider]` entry); buckets and agreements are -/// never created — if nothing matches, it errors out. `--users` clients each -/// perform `--uploads-per-user` uploads of `--max-payload-size` random bytes, -/// with users and per-user uploads run sequentially or in parallel per the -/// `--parallel-*` flags, optionally capped by `--max-concurrency`. -/// -/// Returns the aggregated [`OpSummary`] for the run; the caller (`main`) views -/// them. Per-upload failures are folded into the metrics, so this only returns -/// `Err` for setup failures (bad args, chain connection, no matching buckets). -pub async fn upload(global: &GlobalArgs, args: &UploadArgs) -> Result { - if args.users < 1 { - bail!("--users must be at least 1"); - } - if args.uploads_per_user < 1 { - bail!("--uploads-per-user must be at least 1"); - } - if args.max_payload_size < 1 { - bail!("--max-payload-size must be at least 1 byte"); +/// Build the client config (chain RPC + provider URL) from the global flags. +fn build_config(global: &GlobalArgs) -> ClientConfig { + ClientConfig { + chain_ws_url: global.chain_rpc.clone(), + provider_urls: vec![global.provider_url.clone()], + ..Default::default() } +} - // Identity: derive the account whose buckets we look up from the SURI. +/// Derive the SS58 account whose buckets we upload to, from `--suri`/`--keyfile`. +fn resolve_account(global: &GlobalArgs) -> Result { let suri = resolve_suri(global)?; let keypair = Keypair::from_uri(&suri.parse::().context("failed to parse SURI")?) .context("failed to derive keypair from SURI")?; - let account = AccountId32::from(keypair.public_key().0); - let account_ss58 = account.to_ss58check(); - - // Target provider: parse the input and hex-encode it (`0x` + lowercase hex - // of the 32 raw account bytes) for matching against the chain's - // `StorageAgreements`. Match on raw bytes, never SS58 strings — prefix - // differences (`5…` vs `1…`) would make equal accounts compare unequal. - let target_provider_hex = SubstrateClient::parse_account(&args.provider) - .map_err(|e: storage_client::ClientError| anyhow!("invalid --provider account: {e}")) - .map(|ac| format!("0x{}", hex::encode(ac.as_ref() as &[u8])))?; - - let config = ClientConfig { - chain_ws_url: global.chain_rpc.clone(), - provider_urls: vec![global.provider_url.clone()], - ..Default::default() - }; + Ok(AccountId32::from(keypair.public_key().0).to_ss58check()) +} - // Read-only chain access: resolve the buckets that have an agreement with - // the target provider. No signer is set — uploads are off-chain HTTP. - let mut admin = AdminClient::new(config.clone(), account_ss58.clone()) +/// Resolve the buckets the account can upload to via `provider_hex`: those with +/// a `StorageAgreements[bucket][provider]` entry. Read-only chain access (no +/// signer); buckets and agreements are never created. Errors if none match. +async fn discover_target_buckets( + config: &ClientConfig, + account_ss58: &str, + provider_hex: &str, + provider_display: &str, + chain_rpc: &str, +) -> Result> { + let mut admin = AdminClient::new(config.clone(), account_ss58.to_string()) .context("failed to construct chain client")?; admin .connect() .await - .with_context(|| format!("failed to connect to chain RPC {}", global.chain_rpc))?; + .with_context(|| format!("failed to connect to chain RPC {chain_rpc}"))?; let all_buckets_id = admin .list_my_buckets() .await .context("failed to read the account's buckets from chain")?; - let mut selected_buckets_id = Vec::new(); + let mut selected = Vec::new(); for bucket_id in all_buckets_id { let agreements = admin .list_bucket_agreements(bucket_id) @@ -207,68 +189,75 @@ pub async fn upload(global: &GlobalArgs, args: &UploadArgs) -> Result .with_context(|| format!("failed to read agreements for bucket {bucket_id}"))?; if agreements .iter() - .any(|a| a.provider.eq_ignore_ascii_case(&target_provider_hex)) + .any(|a| a.provider.eq_ignore_ascii_case(provider_hex)) { - selected_buckets_id.push(bucket_id); + selected.push(bucket_id); } } - if selected_buckets_id.is_empty() { + if selected.is_empty() { bail!( - "account {account_ss58} has no buckets with an agreement to provider {} on {}. \ - Nothing to upload (no bucket or agreement was created).", - args.provider, - global.chain_rpc, + "account {account_ss58} has no buckets with an agreement to provider \ + {provider_display} on {chain_rpc}. Nothing to upload (no bucket or \ + agreement was created)." ); } + Ok(selected) +} - if let Some(max) = args.max_buckets_to_write { - selected_buckets_id.truncate(max); +/// Construct `count` provider clients — one per simulated user, so each gets its +/// own connection pool. Off-chain HTTP only (no chain, no signer). +fn build_clients(config: &ClientConfig, count: usize) -> Result>> { + (0..count) + .map(|_| { + StorageUserClient::new(config.clone()) + .map(Arc::new) + .context("failed to construct provider client") + }) + .collect() +} + +/// Label for a parallelism axis in the progress banner. +fn axis(parallel: bool) -> &'static str { + if parallel { + "[parallel]" + } else { + "[sequential]" } +} - let total_uploads = args.users.saturating_mul(args.uploads_per_user); - // Progress goes to stderr so stdout carries only the final metrics view - // (keeping `--output json` parseable). +/// Print the run configuration to stderr (stdout carries only the final metrics +/// view, keeping `--output json` parseable). +fn print_banner(args: &UploadArgs, total_uploads: usize, bucket_count: usize, provider_url: &str) { + let cap = if args.max_concurrency > 0 { + format!(", max in-flight {}", args.max_concurrency) + } else { + String::new() + }; eprintln!( - "Stress test: {} user(s){}, {} upload(s)/user{}, {} bytes each, {} bucket(s) via {}{}", + "Stress test: {} user(s) {}, {} upload(s)/user {}, {} bytes each, {} bucket(s) via {}{}", args.users, - if args.parallel_users { - " [parallel]" - } else { - " [sequential]" - }, + axis(args.parallel_users), args.uploads_per_user, - if args.parallel_uploads { - " [parallel]" - } else { - " [sequential]" - }, + axis(args.parallel_uploads), args.max_payload_size, - selected_buckets_id.len(), - global.provider_url, - if args.max_concurrency > 0 { - format!(", max in-flight {}", args.max_concurrency) - } else { - String::new() - }, + bucket_count, + provider_url, + cap, ); eprintln!("Running {total_uploads} upload(s)..."); +} - // Off-chain HTTP uploads (no chain, no signer). One client per user gives - // each simulated user its own connection pool. - let mut clients = Vec::with_capacity(args.users); - for _ in 0..args.users { - clients.push(Arc::new( - StorageUserClient::new(config.clone()) - .context("failed to construct provider client")?, - )); - } - - let buckets = Arc::new(selected_buckets_id); +/// Drive the configured upload load: one future per user, spawned for +/// parallelism or awaited in sequence, collecting every [`OpOutcome`]. +async fn run_load( + clients: Vec>, + buckets: Arc>, + args: &UploadArgs, +) -> Vec { let sem = (args.max_concurrency > 0).then(|| Arc::new(Semaphore::new(args.max_concurrency))); + let mut outcomes = Vec::with_capacity(clients.len() * args.uploads_per_user.get()); - let started = Instant::now(); - let mut outcomes = Vec::with_capacity(total_uploads); // Build each user's future once; spawn it for parallelism or await it in // sequence. Passing the `Copy` config values positionally keeps the spawned // future `'static` (it owns its `usize`/`bool`/`Arc`s), so the closure only @@ -278,8 +267,8 @@ pub async fn upload(global: &GlobalArgs, args: &UploadArgs) -> Result user_idx, client, buckets.clone(), - args.uploads_per_user, - args.max_payload_size, + args.uploads_per_user.get(), + args.max_payload_size.get(), args.parallel_uploads, sem.clone(), ) @@ -303,7 +292,48 @@ pub async fn upload(global: &GlobalArgs, args: &UploadArgs) -> Result outcomes.extend(run_one(user_idx, client).await); } } + outcomes +} +/// Drive configurable upload load against `--provider`. +/// +/// Targets are resolved from chain (`MemberBuckets[account]` ∩ buckets with a +/// `StorageAgreements[bucket][provider]` entry); buckets and agreements are +/// never created — if nothing matches, it errors out. `--users` clients each +/// perform `--uploads-per-user` uploads of `--max-payload-size` random bytes, +/// with users and per-user uploads run sequentially or in parallel per the +/// `--parallel-*` flags, optionally capped by `--max-concurrency`. +/// +/// Returns the aggregated [`OpSummary`] for the run; the caller (`main`) views +/// them. Per-upload failures are folded into the metrics, so this only returns +/// `Err` for setup failures (bad provider, chain connection, no matching buckets). +pub async fn upload(global: &GlobalArgs, args: &UploadArgs) -> Result { + let account_ss58 = resolve_account(global)?; + + let provider_hex = SubstrateClient::parse_account(&args.provider) + .map_err(|e: storage_client::ClientError| anyhow!("invalid --provider account: {e}")) + .map(|ac| format!("0x{}", hex::encode(ac.as_ref() as &[u8])))?; + + let config = build_config(global); + + let mut buckets = discover_target_buckets( + &config, + &account_ss58, + &provider_hex, + &args.provider, + &global.chain_rpc, + ) + .await?; + if let Some(max) = args.max_buckets_to_write { + buckets.truncate(max); + } + + let total_uploads = args.users.get().saturating_mul(args.uploads_per_user.get()); + print_banner(args, total_uploads, buckets.len(), &global.provider_url); + + let clients = build_clients(&config, args.users.get())?; + let started = Instant::now(); + let outcomes = run_load(clients, Arc::new(buckets), args).await; Ok(summarize(Upload, &outcomes, started.elapsed())) } From 9cdb7ab0acd825cd256491a1be9ef8bd843621ec Mon Sep 17 00:00:00 2001 From: Darwin Subramaniam Date: Tue, 7 Jul 2026 18:04:51 +0800 Subject: [PATCH 10/10] feat(metrics): use the disable-auth on provider node and added percentile on the metric --- .github/workflows/integration-tests.yml | 3 +- utils/storage-cli/src/metrics.rs | 79 +++++++++++++++++++++---- 2 files changed, 70 insertions(+), 12 deletions(-) diff --git a/.github/workflows/integration-tests.yml b/.github/workflows/integration-tests.yml index 23b9e5fa..1a847f95 100644 --- a/.github/workflows/integration-tests.yml +++ b/.github/workflows/integration-tests.yml @@ -672,7 +672,8 @@ jobs: nohup ./target/release/storage-provider-node \ --keyfile /tmp/alice-key --storage-mode inmemory \ --bind-addr 0.0.0.0:3333 --chain-rpc ws://127.0.0.1:2222 \ - --enable-checkpoint-coordinator > /tmp/provider.log 2>&1 & + --enable-checkpoint-coordinator \ + --disable-auth-i-know-what-i-am-doing > /tmp/provider.log 2>&1 & - name: Wait for provider health uses: ./.github/actions/wait-for-provider-health diff --git a/utils/storage-cli/src/metrics.rs b/utils/storage-cli/src/metrics.rs index 68a12235..27fec09e 100644 --- a/utils/storage-cli/src/metrics.rs +++ b/utils/storage-cli/src/metrics.rs @@ -108,6 +108,12 @@ pub struct OpSummary { pub lat_avg: Duration, /// Maximum latency over successful operations. pub lat_max: Duration, + /// 50th-percentile (median) latency over successful operations. + pub lat_p50: Duration, + /// 95th-percentile latency over successful operations. + pub lat_p95: Duration, + /// 99th-percentile latency over successful operations. + pub lat_p99: Duration, /// Up to [`MAX_SAMPLE_ERRORS`] sample messages from failed operations. pub sample_errors: Vec, } @@ -134,6 +140,17 @@ impl OpSummary { } } +/// Nearest-rank percentile (`p` in `[0, 100]`) over an already-sorted slice of +/// durations. `Duration::ZERO` on an empty slice. +fn percentile(sorted: &[Duration], p: f64) -> Duration { + if sorted.is_empty() { + return Duration::ZERO; + } + let rank = ((p / 100.0) * sorted.len() as f64).ceil() as usize; + let idx = rank.saturating_sub(1).min(sorted.len() - 1); + sorted[idx] +} + /// Aggregate per-operation outcomes against the elapsed duration of the run. pub fn summarize( operation: impl Operation, @@ -144,18 +161,20 @@ pub fn summarize( let ok = outcomes.iter().filter(|o| o.ok).count(); let bytes_ok = outcomes.iter().filter(|o| o.ok).map(|o| o.bytes).sum(); - let mut lat_min = Duration::MAX; - let mut lat_max = Duration::ZERO; - let mut lat_sum = Duration::ZERO; - for o in outcomes.iter().filter(|o| o.ok) { - lat_min = lat_min.min(o.elapsed); - lat_max = lat_max.max(o.elapsed); - lat_sum += o.elapsed; - } + let mut ok_latencies: Vec = outcomes + .iter() + .filter(|o| o.ok) + .map(|o| o.elapsed) + .collect(); + ok_latencies.sort_unstable(); + + let lat_min = ok_latencies.first().copied().unwrap_or(Duration::ZERO); + let lat_max = ok_latencies.last().copied().unwrap_or(Duration::ZERO); + let lat_sum: Duration = ok_latencies.iter().sum(); let lat_avg = lat_sum.checked_div(ok as u32).unwrap_or(Duration::ZERO); - if ok == 0 { - lat_min = Duration::ZERO; - } + let lat_p50 = percentile(&ok_latencies, 50.0); + let lat_p95 = percentile(&ok_latencies, 95.0); + let lat_p99 = percentile(&ok_latencies, 99.0); let sample_errors = outcomes .iter() @@ -173,6 +192,9 @@ pub fn summarize( lat_min, lat_avg, lat_max, + lat_p50, + lat_p95, + lat_p99, sample_errors, } } @@ -193,6 +215,9 @@ struct OpSummaryJson<'a> { latency_min_secs: f64, latency_avg_secs: f64, latency_max_secs: f64, + latency_p50_secs: f64, + latency_p95_secs: f64, + latency_p99_secs: f64, sample_errors: &'a [String], } @@ -210,6 +235,9 @@ impl OpSummary { latency_min_secs: self.lat_min.as_secs_f64(), latency_avg_secs: self.lat_avg.as_secs_f64(), latency_max_secs: self.lat_max.as_secs_f64(), + latency_p50_secs: self.lat_p50.as_secs_f64(), + latency_p95_secs: self.lat_p95.as_secs_f64(), + latency_p99_secs: self.lat_p99.as_secs_f64(), sample_errors: &self.sample_errors, } } @@ -259,6 +287,13 @@ impl fmt::Display for OpSummary { self.lat_avg.as_secs_f64(), self.lat_max.as_secs_f64(), )?; + writeln!( + f, + " percentile: p50 {:.3}s / p95 {:.3}s / p99 {:.3}s", + self.lat_p50.as_secs_f64(), + self.lat_p95.as_secs_f64(), + self.lat_p99.as_secs_f64(), + )?; if !self.sample_errors.is_empty() { writeln!(f, " sample errors:")?; for e in &self.sample_errors { @@ -302,6 +337,9 @@ mod tests { assert_eq!(m.lat_min, Duration::from_millis(10)); assert_eq!(m.lat_max, Duration::from_millis(30)); assert_eq!(m.lat_avg, Duration::from_millis(20)); + assert_eq!(m.lat_p50, Duration::from_millis(10)); + assert_eq!(m.lat_p95, Duration::from_millis(30)); + assert_eq!(m.lat_p99, Duration::from_millis(30)); assert_eq!(m.ops_per_sec(), 2.0); assert_eq!(m.sample_errors, vec!["boom".to_string()]); } @@ -319,6 +357,22 @@ mod tests { assert_eq!(m.lat_min, Duration::ZERO); assert_eq!(m.lat_avg, Duration::ZERO); assert_eq!(m.lat_max, Duration::ZERO); + assert_eq!(m.lat_p50, Duration::ZERO); + assert_eq!(m.lat_p95, Duration::ZERO); + assert_eq!(m.lat_p99, Duration::ZERO); + } + + #[test] + fn percentile_uses_nearest_rank() { + let durations: Vec = (1..=100).map(Duration::from_millis).collect(); + assert_eq!(percentile(&durations, 50.0), Duration::from_millis(50)); + assert_eq!(percentile(&durations, 95.0), Duration::from_millis(95)); + assert_eq!(percentile(&durations, 99.0), Duration::from_millis(99)); + } + + #[test] + fn percentile_of_empty_is_zero() { + assert_eq!(percentile(&[], 50.0), Duration::ZERO); } #[test] @@ -339,6 +393,9 @@ mod tests { assert_eq!(entry["elapsed_secs"], 2.0); assert_eq!(entry["throughput_ops_per_sec"], 0.5); assert_eq!(entry["latency_avg_secs"], 0.01); + assert_eq!(entry["latency_p50_secs"], 0.01); + assert_eq!(entry["latency_p95_secs"], 0.01); + assert_eq!(entry["latency_p99_secs"], 0.01); assert_eq!(entry["sample_errors"][0], "boom"); } }