diff --git a/.github/workflows/integration-tests.yml b/.github/workflows/integration-tests.yml index 7c1b1666..1a847f95 100644 --- a/.github/workflows/integration-tests.yml +++ b/.github/workflows/integration-tests.yml @@ -585,6 +585,135 @@ 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 \ + --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 + + # 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 \ + --users 2 \ + --uploads-per-user 5 \ + --max-payload-size 1048576 \ + --parallel-uploads + + - 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 +905,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: diff --git a/Cargo.lock b/Cargo.lock index ecae963f..df3a797b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9576,6 +9576,25 @@ 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", + "rand 0.8.5", + "serde", + "serde_json", + "sp-core", + "sp-runtime", + "storage-client", + "subxt-signer", + "tempfile", + "tokio", + "tracing-subscriber 0.3.19", +] + [[package]] name = "storage-client" version = "0.1.0" diff --git a/Cargo.toml b/Cargo.toml index dc71a8ef..3d850622 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] @@ -131,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" @@ -172,6 +176,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 new file mode 100644 index 00000000..434d846c --- /dev/null +++ b/utils/storage-cli/Cargo.toml @@ -0,0 +1,25 @@ +[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 = { workspace = true } +hex = { workspace = true } +rand = { workspace = true } +serde = { workspace = true, features = ["derive", "std"] } +serde_json = { workspace = true, features = ["std"] } + +[dev-dependencies] +tempfile = { workspace = true } 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/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/cli.rs b/utils/storage-cli/src/cli.rs new file mode 100644 index 00000000..bf50770f --- /dev/null +++ b/utils/storage-cli/src/cli.rs @@ -0,0 +1,63 @@ +// SPDX-License-Identifier: Apache-2.0 + +//! 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; + +/// 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)] +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, + + /// 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)] +pub enum Command { + /// Stress-test operations against a provider. + #[command(subcommand)] + StressTest(StressTest), +} diff --git a/utils/storage-cli/src/commands/mod.rs b/utils/storage-cli/src/commands/mod.rs new file mode 100644 index 00000000..906299cd --- /dev/null +++ b/utils/storage-cli/src/commands/mod.rs @@ -0,0 +1,2 @@ +// SPDX-License-Identifier: Apache-2.0 +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..cfbd2bc0 --- /dev/null +++ b/utils/storage-cli/src/commands/stress_test.rs @@ -0,0 +1,357 @@ +// SPDX-License-Identifier: Apache-2.0 + +//! `stress-test` subcommands. + +use std::num::NonZeroUsize; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +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, 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, BucketId}; +use crate::metrics::{summarize, OpOutcome, OpSummary}; + +// === Stress test subcommands === +#[derive(Debug, Subcommand)] +pub enum StressTest { + /// 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 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, + + /// Number of concurrent simulated users, each with its own client (1..N). + #[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 = "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 = "524288")] + pub max_payload_size: NonZeroUsize, + + /// 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 +} + +/// 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, + }; + upload_once(&client, bucket, &random_payload(size)).await +} + +/// 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 + } +} + +/// 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() + } +} + +/// 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")?; + Ok(AccountId32::from(keypair.public_key().0).to_ss58check()) +} + +/// 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 {chain_rpc}"))?; + + let all_buckets_id = admin + .list_my_buckets() + .await + .context("failed to read the account's buckets from chain")?; + + let mut selected = 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_hex)) + { + selected.push(bucket_id); + } + } + + if selected.is_empty() { + bail!( + "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) +} + +/// 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]" + } +} + +/// 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 {}{}", + args.users, + axis(args.parallel_users), + args.uploads_per_user, + axis(args.parallel_uploads), + args.max_payload_size, + bucket_count, + provider_url, + cap, + ); + eprintln!("Running {total_uploads} upload(s)..."); +} + +/// 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()); + + // 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.get(), + args.max_payload_size.get(), + args.parallel_uploads, + sem.clone(), + ) + }; + + if args.parallel_users { + let mut users_set = JoinSet::new(); + for (user_idx, client) in clients.into_iter().enumerate() { + 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), + // 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() { + 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())) +} + +#[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); + } + } + + #[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 new file mode 100644 index 00000000..5b338552 --- /dev/null +++ b/utils/storage-cli/src/common/mod.rs @@ -0,0 +1,100 @@ +// SPDX-License-Identifier: Apache-2.0 + +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) { + (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, + output: crate::metrics::OutputFormat::Text, + } + } + + #[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"); + } +} diff --git a/utils/storage-cli/src/main.rs b/utils/storage-cli/src/main.rs new file mode 100644 index 00000000..8c53dcd3 --- /dev/null +++ b/utils/storage-cli/src/main.rs @@ -0,0 +1,48 @@ +// 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 actions; +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)) => { + 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.labels.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..27fec09e --- /dev/null +++ b/utils/storage-cli/src/metrics.rs @@ -0,0 +1,401 @@ +// 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 via an [`Operation`] implementor — 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, +} + +/// 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, +} + +/// 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 +/// (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 { + /// Display labels for the operation these metrics describe. + pub labels: OpLabels, + /// 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, + /// 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, +} + +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 + } + } +} + +/// 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, + 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 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); + 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() + .filter_map(|o| o.err.clone()) + .take(MAX_SAMPLE_ERRORS) + .collect(); + + OpSummary { + labels: operation.labels(), + total, + ok, + failed: total - ok, + bytes_ok, + elapsed, + lat_min, + lat_avg, + lat_max, + lat_p50, + lat_p95, + lat_p99, + 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, + latency_p50_secs: f64, + latency_p95_secs: f64, + latency_p99_secs: f64, + sample_errors: &'a [String], +} + +impl OpSummary { + fn as_json(&self) -> OpSummaryJson<'_> { + OpSummaryJson { + operation: self.labels.verb, + 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(), + 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, + } + } +} + +/// 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.labels.verb, self.total, self.ok, self.failed + )?; + writeln!( + f, + " data: {:.2} MiB {} in {:.3}s", + mib, self.labels.past_tense, secs + )?; + writeln!( + f, + " throughput: {mib_s:.2} MiB/s, {ops:.1} {}/s", + self.labels.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(), + )?; + 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 { + writeln!(f, " - {e}")?; + } + } + Ok(()) + } +} + +#[cfg(test)] +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![ + 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(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); + 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.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()]); + } + + #[test] + fn summarize_all_failed_has_zero_latency() { + let outcomes = vec![OpOutcome::failure( + 50, + Duration::from_millis(5), + "nope".into(), + )]; + 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); + 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] + 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(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"], "test"); + 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["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"); + } +}