diff --git a/.github/workflows/bench.yml b/.github/workflows/bench.yml new file mode 100644 index 00000000..9ca6660b --- /dev/null +++ b/.github/workflows/bench.yml @@ -0,0 +1,131 @@ +name: Bench + +# Real-world benchmark suite: Driven's engine vs rclone (bench/README.md). +# +# COST POLICY: this workflow uploads REAL bytes to a REAL Google account and +# takes minutes to hours. It therefore NEVER runs on `pull_request` and NEVER on +# a plain push - only: +# +# workflow_dispatch - on demand, with a chosen scale and tool list. +# v* tag pushes - at the SMOKE scale only, as a release-time check that +# the suite still works and nothing has fallen off a +# cliff. Same gating shape as chaos.yml's real-drive job. +# +# Like `chaos-real-drive`, the job degrades to a clean SKIP (never red) when the +# credentials are absent, so a fork or a rotated-away secret does not fail a +# release. Every job is time-boxed so a hung upload cannot burn hours of runner +# budget. + +on: + workflow_dispatch: + inputs: + scale: + description: "Fixture scale" + type: choice + default: smoke + options: [smoke, small, medium, full] + tools: + description: "Comma-separated tools to measure" + type: string + default: "driven,rclone" + push: + tags: ["v*"] + +concurrency: + # One benchmark at a time: two concurrent runs would contend for the same + # uplink and produce numbers that mean nothing. Never cancel a running one - + # a cancelled run leaves its uploaded folder behind. + group: ${{ github.workflow }} + cancel-in-progress: false + +permissions: + contents: read + +env: + CARGO_TERM_COLOR: always + RUST_BACKTRACE: 1 + # The harness boots the headless core, whose state layer uses sqlx + # compile-time-checked queries; CI has no live DB (same as ci.yml). + SQLX_OFFLINE: "true" + CARGO_PROFILE_DEV_DEBUG: "0" + +jobs: + bench: + name: benchmark (${{ inputs.scale || 'smoke' }}) + runs-on: ubuntu-latest + # A `full` run is meant to take hours; everything else finishes long before + # this. The cap exists so a hung upload cannot run until the 6h default. + timeout-minutes: 180 + env: + DRIVEN_E2E_REFRESH_TOKEN: ${{ secrets.DRIVEN_E2E_REFRESH_TOKEN }} + DRIVEN_E2E_DEST_FOLDER_ID: ${{ secrets.DRIVEN_E2E_DEST_FOLDER_ID }} + DRIVEN_OAUTH_CLIENT_ID: ${{ secrets.DRIVEN_OAUTH_CLIENT_ID }} + DRIVEN_OAUTH_CLIENT_SECRET: ${{ secrets.DRIVEN_OAUTH_CLIENT_SECRET }} + # A tag push has no inputs; the release-time check is deliberately small. + BENCH_SCALE: ${{ inputs.scale || 'smoke' }} + BENCH_TOOLS: ${{ inputs.tools || 'driven,rclone' }} + steps: + - uses: actions/checkout@v7 + + - name: Check for credentials + id: creds + # Absent secrets are a clean skip, not a failure: forks and rotated + # secrets must not turn a release tag red. + run: | + if [ -n "$DRIVEN_E2E_REFRESH_TOKEN" ] && [ -n "$DRIVEN_E2E_DEST_FOLDER_ID" ] \ + && [ -n "$DRIVEN_OAUTH_CLIENT_SECRET" ]; then + echo "present=true" >> "$GITHUB_OUTPUT" + else + echo "present=false" >> "$GITHUB_OUTPUT" + echo "Bench credentials are not available; skipping the benchmark run." + fi + + - uses: dtolnay/rust-toolchain@stable + if: steps.creds.outputs.present == 'true' + + - name: Install build deps + if: steps.creds.outputs.present == 'true' + run: | + sudo apt-get update + sudo apt-get install -y libssl-dev + + - name: Install rclone + if: steps.creds.outputs.present == 'true' + run: | + sudo apt-get install -y rclone + rclone version + + - uses: Swatinem/rust-cache@v2 + if: steps.creds.outputs.present == 'true' + with: + # Builds a subset of the workspace, so it restores the same per-OS + # cache ci.yml warms. This workflow never runs on main, so it is + # purely restore-only. + shared-key: "workspace" + save-if: false + + - name: Run the benchmark + if: steps.creds.outputs.present == 'true' + run: | + # The upload cap is deliberate; the larger scales opt out of it + # explicitly rather than the harness silently ignoring it. + full="" + case "$BENCH_SCALE" in + medium|full) full="--full" ;; + esac + cargo run --release -p driven-bench -- run \ + --scale "$BENCH_SCALE" \ + --tools "$BENCH_TOOLS" \ + $full + + - name: Publish the report to the run summary + # `always()` so a failed benchmark still shows its table - the numbers + # are the point, and a partial run is usually the interesting one. + if: always() && steps.creds.outputs.present == 'true' + run: | + latest=$(ls -1t bench/results/*.md 2>/dev/null | head -n 1 || true) + if [ -n "$latest" ]; then + cat "$latest" >> "$GITHUB_STEP_SUMMARY" + else + echo "No benchmark report was produced." >> "$GITHUB_STEP_SUMMARY" + fi diff --git a/.github/workflows/coverage.yml b/.github/workflows/coverage.yml index 2f947948..ff492d16 100644 --- a/.github/workflows/coverage.yml +++ b/.github/workflows/coverage.yml @@ -12,10 +12,13 @@ name: Coverage # there is no baseline, so the gate is informational (fail-open) for that one # run, then enforces from the next PR onward. # -# Scope: the library crates (`--exclude src-tauri --exclude driven-chaos`). -# src-tauri is a thin IPC layer over driven-core; driven-chaos is the stress -# harness. Both are excluded from the measured/report set (their tests are not -# run for coverage), but `--workspace --exclude` still auto-includes any NEW +# Scope: the library crates (`--exclude src-tauri --exclude driven-chaos +# --exclude driven-bench`). src-tauri is a thin IPC layer over driven-core; +# driven-chaos is the stress harness; driven-bench is the benchmark harness, +# which mostly spawns child processes and talks to a real Google account and so +# is largely unreachable without credentials. All three are excluded from the +# measured/report set (their tests are not run for coverage), but +# `--workspace --exclude` still auto-includes any NEW # crate in the gate, which a hand-maintained `-p` list would silently miss. The # Vue/TS app (`ui/`) is measured in full. (telemetry-worker is its own toolchain # and is out of scope for this gate.) @@ -88,6 +91,7 @@ jobs: - name: Rust coverage (library crates) run: | cargo llvm-cov --workspace --exclude src-tauri --exclude driven-chaos \ + --exclude driven-bench \ --summary-only --json --output-path coverage-rust.json RUST_PCT=$(jq '.data[0].totals.lines.percent' coverage-rust.json) echo "HEAD_RUST=$RUST_PCT" >> "$GITHUB_ENV" diff --git a/Cargo.lock b/Cargo.lock index df96cebe..7407babe 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1569,6 +1569,31 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "driven-bench" +version = "2.3.0" +dependencies = [ + "anyhow", + "async-trait", + "bytes", + "clap", + "driven-core", + "driven-diskstat", + "driven-drive", + "driven-power", + "driven-test-fixtures", + "futures", + "libc", + "serde", + "serde_json", + "tempfile", + "tokio", + "tracing", + "tracing-subscriber", + "uuid", + "windows 0.62.2", +] + [[package]] name = "driven-chaos" version = "2.3.0" diff --git a/Cargo.toml b/Cargo.toml index 4efcd449..fc08bc49 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -13,6 +13,7 @@ members = [ "crates/driven-cli", "crates/driven-test-fixtures", "crates/driven-chaos", + "crates/driven-bench", "src-tauri", ] # M9b (SPEC s16): the telemetry Cloudflare Worker has its own (TypeScript/wrangler) diff --git a/bench/README.md b/bench/README.md new file mode 100644 index 00000000..411815f7 --- /dev/null +++ b/bench/README.md @@ -0,0 +1,214 @@ +# Driven benchmark suite + +Measures Driven's real backup engine against [rclone](https://rclone.org/) on the +two workloads that dominate real backup sets, and writes a report you can put in +a release note. + +The suite exists to answer one question honestly: **is Driven constrained by the +hardware (CPU, disk, network, the Drive API) or by its own algorithms?** A +competitor that is doing roughly the same work is the cheapest way to tell those +apart, and it catches the regressions a synthetic microbenchmark never would. + +- Harness crate: [`crates/driven-bench`](../crates/driven-bench) +- Reports land in [`bench/results/`](results/) +- CI workflow: [`.github/workflows/bench.yml`](../.github/workflows/bench.yml) + +## What it measures + +Two fixture shapes: + +| Shape | What it is | What it stresses | +| --- | --- | --- | +| `huge` | A few very large files, flat | Raw upload throughput, chunking, the resumable path | +| `tiny-deep` | Up to a million small files, nested 8 directories deep | Walking, hashing, per-file bookkeeping, request round-trips | + +Two phases per shape, in order, against the same destination: + +1. **cold** - the tree has never been uploaded. Everything transfers. +2. **incremental** - 0.1% of the files are rewritten deterministically, then the + same command runs again. Almost nothing should transfer; what is being + measured is how fast each tool can work out that nothing changed. + +Per phase the report records wall-clock time, throughput in MiB/s and files/s, +bytes and files transferred, child-process CPU time, peak working set, the +concurrency the tool ran at, and - for Driven only - the number of Drive API +requests. + +## Prerequisites + +- **rclone on `PATH`** (or `--rclone `). Windows: `choco install rclone`, or + unzip the official build from . Linux: + `sudo apt-get install -y rclone`. Any recent version works; the report records + which one ran. +- **Credentials** for the dedicated automation Google account, in the + environment: + + | Variable | Purpose | + | --- | --- | + | `DRIVEN_E2E_REFRESH_TOKEN` | Google refresh token with `drive` scope | + | `DRIVEN_E2E_DEST_FOLDER_ID` | The folder every run writes beneath | + | `DRIVEN_OAUTH_CLIENT_ID` | The OAuth client the token was minted with | + | `DRIVEN_OAUTH_CLIENT_SECRET` | Its secret | + + Locally these live in the gitignored `.env.test` at the repo root, which the + harness loads automatically - you do not need to source it yourself. Existing + environment variables always win, so a CI secret is never shadowed by a stale + local file. In CI they are repository secrets. These are the same four + variables the real-Drive e2e suite uses (`design/E2E_REAL.md`). + +- **Disk**: fixtures are cached under `target/bench-fixtures/`. The `full` scale + needs roughly 10 GB there. `just bench-fixture-clean` reclaims it. + +## Running it + +```powershell +# The usual local run: ~1.2 GB uploaded, both tools, both shapes. +just bench + +# Prove the pipeline works without waiting: a few hundred MB. +just bench smoke + +# One shape only, and keep the uploaded folder for inspection. +cargo run -p driven-bench -- run --scale small --shape tiny-deep --keep-remote + +# The shapes the suite is really about. Needs --full to clear the upload cap. +cargo run -p driven-bench -- run --scale full --full +``` + +`bench/run.ps1` is the same thing with a friendlier front end +(`.\bench\run.ps1 -Scale smoke`). + +### Scales + +| Scale | `huge` | `tiny-deep` | Uploaded per tool | Rough duration (both tools) | +| --- | --- | --- | --- | --- | +| `smoke` | 2 x 8 MiB | 300 files | ~17 MiB | a few minutes | +| `small` (default) | 4 x 128 MiB | 50,000 files | ~610 MiB | 20-60 minutes | +| `medium` | 4 x 512 MiB | 200,000 files | ~2.4 GiB | a few hours | +| `full` | 4 x 2 GiB | 1,000,000 files | ~10 GiB | most of a day | + +Durations depend almost entirely on your uplink and on Drive's per-file rate +limits; the tiny-files shapes are bound by request rate, not bandwidth, so they +take far longer than their size suggests. + +**Cost.** Everything uploaded is trashed at the end of the run, so the storage +cost is transient, but the bytes still cross your connection twice per tool +(once per tool, cold) and count against the account's Drive API quota. Do not +run `full` on a metered connection. + +## Safety rails + +The suite writes to a real Drive account, so: + +- The destination folder id must be given explicitly, by `--dest` or + `DRIVEN_E2E_DEST_FOLDER_ID`. There is no default and no discovery step; the + run aborts before generating a single byte if it is missing. +- Every remote write happens under **one** freshly created `driven-bench-` + folder inside that destination, with a subfolder per scenario (tool x fixture). + The two phases of a scenario deliberately share that subfolder - and, for + Driven, one state database - because an incremental run has to see what the + cold run left behind. A fresh folder or a fresh database per phase would make + the "incremental" numbers a second cold upload wearing the wrong label. +- Cleanup trashes exactly that run folder, **by the id it was created with**. The + suite never lists the destination folder and never matches anything by name, + so it cannot touch data it did not create. If cleanup fails it prints the + folder id to trash by hand rather than retrying blindly. +- Total upload bytes are capped at 2 GiB by default. Exceeding it is an error + that tells you to pass `--full` or lower `--scale`, rather than silently + uploading ten times what you expected. +- Credentials are read from the environment only - never from the OS keychain - + and are never printed. The rclone config containing the token is written to a + temporary directory that is deleted when the run ends. + +## Interpreting the numbers + +### What is and is not apples-to-apples + +The two tools are given identical input, identical destinations and identical +network conditions. They do **not** do identical work, and pretending otherwise +would make the comparison useless rather than fair: + +| | Driven | rclone | +| --- | --- | --- | +| Change detection | Hashes content; maintains a local SQLite state database | Compares size + modification time; no database | +| Cold phase cost | Pays for hashing and state writes on top of the upload | Pays for the upload | +| Incremental phase | Knows exactly what changed, from state | Re-lists the remote and compares every file | +| Crash recovery | Reconciles from `pending_ops` on restart | Re-runs from scratch | +| Concurrency | `min(cores * 2, 16)` by default | 4 transfers by default | +| API requests | Instrumented and reported | Not exposed; the column is blank, which means "not measurable", not zero | + +Two consequences worth stating plainly: + +- **Driven should be expected to lose, or roughly tie, on the cold `huge` + phase.** It is doing strictly more work per byte (hashing, state) for benefits + that only pay off later. If it loses *badly* there, that is a real finding. +- **The incremental phase is where the state database is supposed to earn its + keep**, especially on `tiny-deep`, where rclone has to re-list a million remote + objects and Driven does not. If Driven is not clearly ahead there, that is also + a real finding. + +The concurrency defaults differ because each tool runs at its **stock +settings** - that is what a user actually gets. To isolate the algorithms +instead, equalise them with `--rclone-transfers 16` (or whatever +`min(cores * 2, 16)` is on your machine, which the report prints in the `Conc` +column). + +### Other caveats + +- **Fixture content is incompressible** (a seeded SplitMix64 stream). That is + deliberate: on zero-filled or text-like data, any tool that compresses on the + wire - including Driven's own small-file bundling - would post numbers it could + never reach on the photos, videos and archives that dominate real backup sets. + It does mean these numbers say nothing about how much compression helps on a + compressible corpus. +- **The mutation step changes content only** - no creates, no deletes. That keeps + `rclone copy` a fair match for Driven; a deletion would require `rclone sync` + to be comparable with Driven's trash pass, and `rclone sync` deletes on the + remote, which is a materially riskier command to point at a benchmark folder. +- **Driven's numbers come from a real engine run**, not from `driven-cli sync`. + That subcommand is a debug driver that walks only the top level of a folder and + keeps no state, so it would upload zero files from the `tiny-deep` fixture. The + harness assembles the same `SqliteStateRepo` -> `DefaultExecutor` -> + `SyncOrchestrator` stack `src-tauri/src/assembly.rs` does. Encryption, VSS, + hooks and network probing are off - they are opt-in features with no rclone + equivalent, and the probe traffic would pollute the API-call count. +- **Cross-host comparisons are meaningless.** The report always records the OS, + architecture and CPU count; only compare runs from the same machine and + connection. +- Both tools run as **child processes**, so CPU time and peak memory come from + the same OS accounting for both. On Unix, peak RSS is a high-water mark across + reaped children, so it is reported only when the child raised it; on Windows it + is exact per process. + +### Why rclone, and why not restic + +rclone is the right first comparison: it mirrors a local tree to Drive, which is +what Driven does, so the numbers mean the same thing on both sides. + +`restic` was considered and deliberately left out. It stores a content-addressed, +chunked, deduplicating repository rather than a mirror of your files, so its +"upload" is a different operation with different outputs - and on a re-run its +deduplication would flatter it on exactly the workload this suite measures. Adding +it would produce a bigger table, not a more honest one. If a chunked-repo +comparison is wanted later it deserves its own scenario and its own caveats +section, not a third column here. + +## When it runs + +Never on pull requests, and never on a push to `main` - real uploads on every PR +would be slow and expensive. Only: + +- **manually**, via `workflow_dispatch` (inputs: `scale`, `tools`), and +- **on `v*` tag pushes**, at the `smoke` scale, as a release-time regression + check. + +The workflow is time-boxed so a hung run cannot burn hours, and it skips cleanly +when the credentials are absent (a fork, or a missing secret) rather than failing +red - the same policy `chaos.yml` uses for its real-Drive job. + +## Trending over time + +Each run writes `bench/results/.md` and `.json`. The JSON +keeps every field the table omits, so two runs can be diffed without re-running +anything. Results are committed only when you want a durable record; the +directory is otherwise a scratch area. diff --git a/bench/results/.gitignore b/bench/results/.gitignore new file mode 100644 index 00000000..7c24c91f --- /dev/null +++ b/bench/results/.gitignore @@ -0,0 +1,8 @@ +# Benchmark reports are generated artefacts of a local run, and a run happens +# every time someone types `just bench` - so they are ignored by default rather +# than swept into an unrelated commit. +# +# To keep one deliberately as a durable record to compare against later: +# git add -f bench/results/.md bench/results/.json +*.md +*.json diff --git a/bench/results/.gitkeep b/bench/results/.gitkeep new file mode 100644 index 00000000..bcf4a05b --- /dev/null +++ b/bench/results/.gitkeep @@ -0,0 +1,2 @@ +# Benchmark reports land here as .md and .json. +# Commit one only when you want a durable record to compare against later. diff --git a/bench/run.ps1 b/bench/run.ps1 new file mode 100644 index 00000000..cc124d5e --- /dev/null +++ b/bench/run.ps1 @@ -0,0 +1,113 @@ +<# +.SYNOPSIS + Runs the Driven vs rclone benchmark suite locally on Windows. + +.DESCRIPTION + A thin front end over `cargo run -p driven-bench -- run`. It checks the two + things that most often go wrong before a long run starts - a missing rclone + binary and missing credentials - so you find out in seconds rather than + after the fixtures have been generated. + + Credentials come from the gitignored .env.test at the repo root, which the + harness loads itself; this script only reports whether it is there. + + See bench/README.md for scales, costs and how to read the results. + +.PARAMETER Scale + Fixture size: smoke, small (default), medium or full. + +.PARAMETER Tools + Comma-separated tools to measure. Defaults to "driven,rclone". + +.PARAMETER Shape + Restrict to one fixture shape: huge or tiny-deep. + +.PARAMETER Rclone + Path to the rclone binary, when it is not on PATH. + +.PARAMETER Full + Lift the 2 GiB upload cap. Required for -Scale full. + +.PARAMETER KeepRemote + Leave the uploaded run folder in Drive instead of trashing it. + +.EXAMPLE + .\bench\run.ps1 -Scale smoke + +.EXAMPLE + .\bench\run.ps1 -Scale full -Full +#> +[CmdletBinding()] +param( + [ValidateSet("smoke", "small", "medium", "full")] + [string]$Scale = "small", + + [string]$Tools = "driven,rclone", + + [ValidateSet("huge", "tiny-deep")] + [string]$Shape, + + [string]$Rclone, + + [switch]$Full, + + [switch]$KeepRemote +) + +$ErrorActionPreference = "Stop" + +$repoRoot = Split-Path -Parent $PSScriptRoot + +# --- preflight ------------------------------------------------------------ +# Both checks are advisory: the harness enforces them properly. They exist so a +# long run fails in the first second rather than after fixture generation. + +if ($Tools -split "," -contains "rclone") { + $rclonePath = if ($Rclone) { $Rclone } else { (Get-Command rclone -ErrorAction SilentlyContinue).Source } + if (-not $rclonePath) { + Write-Error @" +rclone was not found on PATH. +Install it with 'choco install rclone', or unzip the official build from +https://rclone.org/downloads/ and pass -Rclone . +"@ + } + Write-Host "rclone: $rclonePath" +} + +$envFile = Join-Path $repoRoot ".env.test" +if (Test-Path $envFile) { + Write-Host "credentials: $envFile" +} elseif ($env:DRIVEN_E2E_REFRESH_TOKEN) { + Write-Host "credentials: from the environment" +} else { + Write-Error @" +No credentials found: neither $envFile nor DRIVEN_E2E_REFRESH_TOKEN is present. +See bench/README.md, 'Prerequisites'. +"@ +} + +if ($Scale -eq "full" -and -not $Full) { + Write-Warning "-Scale full uploads ~10 GB per tool and needs -Full to clear the upload cap." +} + +# --- run ------------------------------------------------------------------ + +$benchArgs = @("run", "--scale", $Scale, "--tools", $Tools) +if ($Shape) { $benchArgs += @("--shape", $Shape) } +if ($Rclone) { $benchArgs += @("--rclone", $Rclone) } +if ($Full) { $benchArgs += "--full" } +if ($KeepRemote) { $benchArgs += "--keep-remote" } + +Write-Host "scale: $Scale" +Write-Host "tools: $Tools" +Write-Host "" + +Push-Location $repoRoot +try { + # --release matters: a dev-profile build spends its time in Driven's hashing + # and encryption paths rather than measuring them. + & cargo run --release -p driven-bench -- @benchArgs + exit $LASTEXITCODE +} finally { + Pop-Location +} diff --git a/crates/driven-bench/Cargo.toml b/crates/driven-bench/Cargo.toml new file mode 100644 index 00000000..551bb99b --- /dev/null +++ b/crates/driven-bench/Cargo.toml @@ -0,0 +1,58 @@ +[package] +name = "driven-bench" +version.workspace = true +edition.workspace = true +rust-version.workspace = true +license.workspace = true +repository.workspace = true +description = "Driven real-world benchmark harness; see bench/README.md" +publish = false + +[[bin]] +name = "driven-bench" +path = "src/main.rs" + +[dependencies] +# The harness drives the REAL engine (the same scan -> plan -> execute pipeline +# the desktop app runs), so it boots driven-core directly against a live +# `GoogleDriveStore` - exactly the assembly `src-tauri/src/assembly.rs` performs, +# minus the GUI/VSS/crypto/hook layers. It does NOT depend on src-tauri. +driven-core = { path = "../driven-core" } +driven-drive = { path = "../driven-drive" } +driven-power = { path = "../driven-power" } +# The adaptive upload-parallelism controller's per-OS disk-busy reader, so the +# benchmarked engine behaves like a production build (adaptive is default-ON). +driven-diskstat = { path = "../driven-diskstat" } +# Deterministic on-AC power for the run (a laptop on battery would otherwise +# have the orchestrator's power gate decide the result). +driven-test-fixtures = { path = "../driven-test-fixtures" } + +anyhow.workspace = true +async-trait.workspace = true +clap.workspace = true +serde.workspace = true +serde_json.workspace = true +tokio.workspace = true +tracing.workspace = true +tracing-subscriber.workspace = true +uuid.workspace = true +bytes.workspace = true +futures.workspace = true +# Per-scenario state databases and rclone configs live in a throwaway directory +# that is removed when the run ends. +tempfile = "3" + +# Per-child CPU time + peak working set, so `driven` and `rclone` are measured +# by the SAME OS accounting rather than one in-process and one out. +[target.'cfg(windows)'.dependencies] +windows = { version = "0.62", features = [ + "Win32_Foundation", + "Win32_System_Threading", + "Win32_System_ProcessStatus", +] } + +[target.'cfg(unix)'.dependencies] +libc.workspace = true + +[dev-dependencies] +tempfile = "3" diff --git a/crates/driven-bench/src/agent.rs b/crates/driven-bench/src/agent.rs new file mode 100644 index 00000000..23c3e75e --- /dev/null +++ b/crates/driven-bench/src/agent.rs @@ -0,0 +1,429 @@ +//! The Driven side of the benchmark: one real backup cycle, in a child process. +//! +//! This is NOT a simplified upload loop. It assembles the same engine the +//! desktop app assembles in `src-tauri/src/assembly.rs` - `SqliteStateRepo` -> +//! `DefaultExecutor` (with the adaptive upload pool and the AIMD pacer) -> +//! `SyncOrchestrator` - and runs `run_cycle`, so what gets measured is the real +//! scan -> plan -> execute -> verify pipeline against a live `GoogleDriveStore`. +//! Benchmarking `driven-cli sync` instead would have measured a debug driver +//! that walks only the top level of the source folder and keeps no state. +//! +//! What is deliberately NOT wired, and why: +//! +//! - **VSS / crypto / hooks**: off. Encryption and shadow copies are opt-in +//! features; including them would measure a configuration most users do not +//! run, and rclone has no equivalent to compare against. +//! - **Network probing**: replaced with an always-online probe. The real prober +//! issues its own HTTP requests, which would pollute the API-call count +//! without telling us anything about backup throughput. +//! - **Power gating**: a fixed on-AC state, and both `skip_on_battery` and +//! `skip_on_metered` are off, so an unplugged laptop cannot silently turn a +//! benchmark into a no-op that looks blazingly fast. +//! +//! The process prints exactly one machine-readable line, prefixed with +//! [`METRICS_PREFIX`], for the parent harness to parse. + +use std::path::PathBuf; +use std::sync::{Arc, Mutex}; +use std::time::Instant; + +use anyhow::{Context, Result}; +use serde::{Deserialize, Serialize}; +use tokio::sync::broadcast::error::RecvError; + +use driven_core::executor::{DefaultExecutor, ExecutorDeps}; +use driven_core::network::{NetworkProbe, NetworkState, ServiceHealth, ServiceName}; +use driven_core::orchestrator::{OrchestratorConfig, SyncOrchestrator, TickSource}; +use driven_core::pacer::AimdPacer; +use driven_core::state::{AccountRow, SourceRow, SqliteStateRepo, StateRepo}; +use driven_core::time::{Clock, SystemClock}; +use driven_core::types::{AccountId, AccountState, OrchestratorEvent, SourceId}; +use driven_drive::remote_store::RemoteStore; +use driven_power::{PowerSource, PowerState}; +use driven_test_fixtures::power::FakePowerSource; + +use crate::counting_store::{ApiCounts, CountingStore}; +use crate::creds::BenchCreds; + +/// The stdout marker the parent harness looks for. +pub const METRICS_PREFIX: &str = "DRIVEN_BENCH_METRICS "; + +/// What one engine cycle reported about itself. +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +// Missing fields decode as zero, so a metrics line written by an older build +// still parses instead of failing the whole phase. +#[serde(default)] +pub struct AgentMetrics { + /// Time inside `run_cycle`, excluding process startup and Drive auth. + pub engine_ms: u64, + /// Time from the start of the cycle until the planner ran, i.e. how long + /// walking and hashing the tree took. + /// + /// This is the number that says WHERE a slow run went. On the million-tiny- + /// files shape a cold pass can be bound by the local scan or by Drive + /// round-trips, and the totals alone cannot tell those apart. `None` when + /// the cycle ended before the planner reported (nothing to do, or an error). + pub scan_ms: Option, + /// Time from the planner finishing to the end of the cycle, i.e. the + /// upload half. `engine_ms - scan_ms` net of the planning step itself. + pub upload_ms: Option, + /// Files the executor finished, from the progress stream. + pub files_done: u64, + /// Bytes the executor moved, from the progress stream. + pub bytes_done: u64, + /// Files the planner decided to upload this cycle. On an incremental run + /// this is the change-detection result: it should equal the mutated file + /// count, not the whole tree. + pub planned_uploads: u64, + /// Bytes the planner decided to upload this cycle. + pub planned_bytes: u64, + /// Errors the executor reported. + pub errors: u64, + /// `upload_done` rows written during the cycle (the durable count). + pub logged_files_uploaded: u64, + /// Summed `upload_done` bytes during the cycle (the durable count). + pub logged_bytes_uploaded: u64, + /// Drive requests, counted at the store seam. + pub api: ApiCounts, +} + +/// Arguments for the in-child engine run. +#[derive(Debug, Clone, clap::Args)] +pub struct AgentArgs { + /// The local folder to back up. + #[arg(long)] + pub source: PathBuf, + /// The Drive folder id to upload into. + #[arg(long)] + pub dest_folder_id: String, + /// The state database for this scenario. Reused across the cold and + /// incremental phases so the incremental phase actually has prior state to + /// compare against. + #[arg(long)] + pub state_db: PathBuf, +} + +/// A network probe that always answers "online". +struct AlwaysOnline; + +#[async_trait::async_trait] +impl NetworkProbe for AlwaysOnline { + async fn probe(&self) -> NetworkState { + NetworkState::Online + } + fn service_health(&self, _service: ServiceName) -> ServiceHealth { + ServiceHealth::Closed + } + fn note_outcome(&self, _service: ServiceName, _ok: bool) {} +} + +/// Runs one backup cycle and prints the metrics line. +pub async fn run(args: AgentArgs) -> Result<()> { + let creds = BenchCreds::from_env()?; + let clock: Arc = Arc::new(SystemClock); + + let state: Arc = Arc::new( + SqliteStateRepo::open(&args.state_db) + .await + .with_context(|| format!("opening state db {}", args.state_db.display()))?, + ); + + // One account, reused across phases so the second cycle sees the first + // cycle's `file_state` rows. + let account_id = match state.list_accounts().await?.into_iter().next() { + Some(existing) => existing.id, + None => { + let id = AccountId::new_v4(); + state + .upsert_account(&AccountRow { + id, + email: "bench@driven.invalid".into(), + display_name: Some("driven-bench".into()), + state: AccountState::Ok, + encryption_master_key_id: None, + created_at: clock.now_ms(), + last_synced_at: None, + }) + .await?; + id + } + }; + + // Same source row across phases, keyed by the local path. A fresh SourceId + // per phase would orphan every `file_state` row and turn the incremental + // phase into a second cold upload - the classic way to report a great + // change-detection number that means nothing. + let local_path = args.source.to_string_lossy().into_owned(); + let source = match state + .list_sources() + .await? + .into_iter() + .find(|s| s.local_path == local_path) + { + Some(existing) => existing, + None => { + let row = new_source( + account_id, + &local_path, + &args.dest_folder_id, + clock.now_ms(), + ); + state.upsert_source(&row).await?; + row + } + }; + + let real: Arc = Arc::new(creds.build_store()?); + let (remote, counters) = CountingStore::new(real); + + let pacer = Arc::new(AimdPacer::new(clock.clone(), None)); + let power: Arc = Arc::new(FakePowerSource::new(PowerState { + ac_connected: true, + battery_percent: None, + on_metered_network: false, + network_reachable: true, + })); + let network: Arc = Arc::new(AlwaysOnline); + + // Mirror the app's adaptive upload parallelism (DESIGN s11.4.7): the pool + // must be built here and injected, because `DefaultExecutor` otherwise + // constructs its own and any configured concurrency is silently ignored. + let upload_pool = + driven_core::adaptive::UploadPool::new(driven_core::adaptive::default_pool_size()); + let throughput = driven_core::adaptive::ThroughputProbe::new(); + + let executor = Arc::new( + DefaultExecutor::with_clock( + ExecutorDeps { + remote: remote.clone(), + state: state.clone(), + pacer: pacer.clone(), + crypto: None, + vss: None, + network: None, + }, + clock.clone(), + ) + .with_upload_pool(upload_pool.clone()) + .with_throughput_probe(throughput.clone()), + ); + + let config = OrchestratorConfig { + // A laptop on battery, or a runner whose connection looks metered, must + // not turn the benchmark into a paused no-op. + skip_on_battery: false, + skip_on_metered: false, + ..Default::default() + }; + + let mut orchestrator = SyncOrchestrator::new( + account_id, + state.clone(), + executor, + power, + network, + clock.clone(), + config, + ); + orchestrator = orchestrator.with_pacer(pacer.clone()); + let disk: Arc = Arc::new( + driven_diskstat::RealDiskBusyProbe::new(PathBuf::from(&source.local_path)), + ); + orchestrator = orchestrator.with_adaptive_controller(Arc::new( + driven_core::adaptive::AdaptiveController::new( + upload_pool, + throughput, + disk, + pacer, + clock.clone(), + ), + )); + let orchestrator = Arc::new(orchestrator); + + let events = orchestrator.subscribe(); + let window_start = clock.now_ms(); + let started = Instant::now(); + + // Consume the event stream WHILE the cycle runs, not afterwards. Draining it + // at the end would lose the phase boundaries entirely (a timestamp cannot be + // recovered from a buffered event) and, on a large run, would also lose + // events to broadcast lag before they were ever read. + let observed = Arc::new(Mutex::new(Observed::default())); + let observer = tokio::spawn(observe(events, observed.clone(), started)); + + orchestrator + .run_cycle(TickSource::Manual) + .await + .context("running the backup cycle")?; + let engine_ms = started.elapsed().as_millis() as u64; + let window_end = clock.now_ms(); + + // The orchestrator owns the broadcast sender, so the observer's `recv` never + // returns `Closed` on its own; the cycle is over, so stop it. + observer.abort(); + let observed = observed.lock().expect("observer mutex").clone(); + + let metrics = AgentMetrics { + engine_ms, + scan_ms: observed.scan_ms, + upload_ms: observed.scan_ms.map(|scan| engine_ms.saturating_sub(scan)), + files_done: observed.files_done, + bytes_done: observed.bytes_done, + planned_uploads: observed.planned_uploads, + planned_bytes: observed.planned_bytes, + errors: observed.errors, + api: counters.snapshot(), + ..Default::default() + }; + let mut metrics = metrics; + + // The durable counterpart to the progress stream: the broadcast channel can + // lag on a large run, the activity rows cannot. + let telemetry = state + .telemetry_events_since(window_start, window_end.max(window_start + 1)) + .await + .context("reading the run's activity rows")?; + metrics.logged_files_uploaded = telemetry.files_uploaded; + metrics.logged_bytes_uploaded = telemetry.bytes_uploaded; + + println!("{METRICS_PREFIX}{}", serde_json::to_string(&metrics)?); + Ok(()) +} + +/// What watching the orchestrator's event stream revealed about one cycle. +#[derive(Debug, Clone, Default)] +struct Observed { + /// Elapsed time when the planner first reported - the end of the scan. + scan_ms: Option, + files_done: u64, + bytes_done: u64, + planned_uploads: u64, + planned_bytes: u64, + errors: u64, +} + +/// Folds orchestrator events into `observed` as they arrive. +/// +/// The executor emits cumulative progress snapshots and the orchestrator +/// forwards a closing one whose per-counter values may be lower, so each counter +/// takes its maximum rather than its last value. Runs until aborted. +async fn observe( + mut events: tokio::sync::broadcast::Receiver, + observed: Arc>, + started: Instant, +) { + loop { + match events.recv().await { + Ok(OrchestratorEvent::Progress { progress, .. }) => { + let mut o = observed.lock().expect("observer mutex"); + o.files_done = o.files_done.max(progress.files_done); + o.bytes_done = o.bytes_done.max(progress.bytes_done); + o.errors = o.errors.max(progress.errors); + } + Ok(OrchestratorEvent::StateChanged { + state: driven_core::types::OrchestratorState::Planning { plan }, + }) => { + let mut o = observed.lock().expect("observer mutex"); + // The FIRST planning event ends the scan; a later one (a second + // source, say) must not overwrite that boundary. + o.scan_ms + .get_or_insert_with(|| started.elapsed().as_millis() as u64); + o.planned_uploads = o.planned_uploads.max(plan.uploads as u64); + o.planned_bytes = o.planned_bytes.max(plan.bytes); + } + Ok(_) => {} + // A lagged receiver has dropped events; the durable activity rows + // are the authority for totals, so keep reading what is left. + Err(RecvError::Lagged(_)) => {} + Err(RecvError::Closed) => return, + } + } +} + +/// Builds the benchmark's backup source row. +fn new_source(account_id: AccountId, local_path: &str, folder_id: &str, now: i64) -> SourceRow { + SourceRow { + id: SourceId::new_v4(), + account_id, + display_name: "driven-bench".into(), + enabled: true, + local_path: local_path.to_string(), + drive_folder_id: folder_id.to_string(), + drive_id: None, + drive_folder_path: "/driven-bench".into(), + encryption_enabled: false, + wrapped_source_key: None, + respect_gitignore: false, + include_patterns: vec![], + exclude_patterns: vec![], + placeholder_policy: Default::default(), + schedule_json_v2_reserved: None, + // A week, so the deep-verify pass never fires mid-benchmark and turns + // one run's numbers into an outlier. + deep_verify_interval_secs: 604_800, + last_full_scan_at: None, + last_deep_verify_at: Some(now), + mtime_granularity_ns: None, + created_at: now, + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn metrics_round_trip_through_the_marker_line() { + let metrics = AgentMetrics { + engine_ms: 1234, + scan_ms: Some(400), + upload_ms: Some(834), + files_done: 7, + bytes_done: 42, + planned_uploads: 7, + ..Default::default() + }; + let line = format!( + "{METRICS_PREFIX}{}", + serde_json::to_string(&metrics).unwrap() + ); + let parsed = crate::tools::parse_agent_metrics(&line).expect("parses"); + assert_eq!(parsed.engine_ms, 1234); + assert_eq!(parsed.scan_ms, Some(400)); + assert_eq!(parsed.upload_ms, Some(834)); + assert_eq!(parsed.files_done, 7); + assert_eq!(parsed.planned_uploads, 7); + } + + #[test] + fn a_cycle_with_no_planning_event_reports_no_scan_time() { + // A cycle that ends before the planner reports (nothing to do, or an + // error) must leave the column empty rather than claim a zero-second + // scan, which would read as "the walk was instant". + let observed = Observed::default(); + assert_eq!(observed.scan_ms, None); + let upload_ms = observed.scan_ms.map(|s| 50u64.saturating_sub(s)); + assert!(upload_ms.is_none()); + } + + #[test] + fn the_first_planning_event_fixes_the_scan_boundary() { + // A later planning event (a second source, say) must not move it. + let mut observed = Observed::default(); + observed.scan_ms.get_or_insert(300); + observed.scan_ms.get_or_insert(900); + assert_eq!(observed.scan_ms, Some(300)); + } + + #[test] + fn the_bench_source_never_enables_encryption_or_gitignore_rules() { + // Both would change what is uploaded and make the comparison with rclone + // meaningless, so they are pinned here rather than left to a default. + let row = new_source(AccountId::new_v4(), "/tmp/x", "folder", 0); + assert!(!row.encryption_enabled); + assert!(!row.respect_gitignore); + assert!(row.include_patterns.is_empty()); + assert!(row.exclude_patterns.is_empty()); + assert!(row.enabled); + } +} diff --git a/crates/driven-bench/src/counting_store.rs b/crates/driven-bench/src/counting_store.rs new file mode 100644 index 00000000..d2f46a71 --- /dev/null +++ b/crates/driven-bench/src/counting_store.rs @@ -0,0 +1,294 @@ +//! A [`RemoteStore`] decorator that counts the Drive requests underneath it. +//! +//! Neither `driven-core` nor `driven-drive` keeps a request counter, and "how +//! many API calls did that cost" is one of the more interesting numbers a backup +//! benchmark can report - on the million-tiny-files shape it is usually the +//! binding constraint, not bandwidth. Wrapping the store is the same seam the +//! executor already uses for `BreakerReportingStore`, so it needs no core change +//! and adds one relaxed atomic increment per call. +//! +//! One caveat the report repeats: `resume_chunk` is counted per CHUNK, which is +//! the honest unit (each chunk is its own HTTP PUT), so a resumable upload of a +//! large file contributes many calls. + +use std::collections::{HashMap, HashSet}; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::Arc; + +use async_trait::async_trait; +use bytes::Bytes; +use serde::{Deserialize, Serialize}; + +use driven_drive::remote_store::{ + AboutInfo, DownloadStream, DriveContext, RemoteEntry, RemoteStore, ResumableKind, + ResumableSession, ResumeProgress, SharedDrive, UploadBody, +}; + +/// Per-operation request counts collected during one run. +#[derive(Debug, Default, Clone, Serialize, Deserialize)] +pub struct ApiCounts { + pub ensure_folder: u64, + pub list_folder: u64, + pub list_shared_drives: u64, + pub create: u64, + pub update: u64, + pub resumable_session: u64, + pub resume_chunk: u64, + pub trash: u64, + pub delete_permanent: u64, + pub metadata: u64, + pub download: u64, + pub find_by_op_uuid: u64, + pub list_source_object_ids: u64, + pub about: u64, + /// The sum of every field above. + pub total: u64, +} + +/// The live counters behind a [`CountingStore`]. +#[derive(Debug, Default)] +pub struct Counters { + ensure_folder: AtomicU64, + list_folder: AtomicU64, + list_shared_drives: AtomicU64, + create: AtomicU64, + update: AtomicU64, + resumable_session: AtomicU64, + resume_chunk: AtomicU64, + trash: AtomicU64, + delete_permanent: AtomicU64, + metadata: AtomicU64, + download: AtomicU64, + find_by_op_uuid: AtomicU64, + list_source_object_ids: AtomicU64, + about: AtomicU64, +} + +impl Counters { + /// Takes a snapshot of every counter. + pub fn snapshot(&self) -> ApiCounts { + let load = |c: &AtomicU64| c.load(Ordering::Relaxed); + let mut counts = ApiCounts { + ensure_folder: load(&self.ensure_folder), + list_folder: load(&self.list_folder), + list_shared_drives: load(&self.list_shared_drives), + create: load(&self.create), + update: load(&self.update), + resumable_session: load(&self.resumable_session), + resume_chunk: load(&self.resume_chunk), + trash: load(&self.trash), + delete_permanent: load(&self.delete_permanent), + metadata: load(&self.metadata), + download: load(&self.download), + find_by_op_uuid: load(&self.find_by_op_uuid), + list_source_object_ids: load(&self.list_source_object_ids), + about: load(&self.about), + total: 0, + }; + counts.total = counts.ensure_folder + + counts.list_folder + + counts.list_shared_drives + + counts.create + + counts.update + + counts.resumable_session + + counts.resume_chunk + + counts.trash + + counts.delete_permanent + + counts.metadata + + counts.download + + counts.find_by_op_uuid + + counts.list_source_object_ids + + counts.about; + counts + } +} + +/// Wraps a [`RemoteStore`], counting every call before delegating. +pub struct CountingStore { + inner: Arc, + counters: Arc, +} + +impl CountingStore { + /// Wraps `inner`, returning the store and the counters to read afterwards. + pub fn new(inner: Arc) -> (Arc, Arc) { + let counters = Arc::new(Counters::default()); + let store = Arc::new(Self { + inner, + counters: counters.clone(), + }); + (store, counters) + } +} + +/// Increments one counter. +fn bump(counter: &AtomicU64) { + counter.fetch_add(1, Ordering::Relaxed); +} + +#[async_trait] +impl RemoteStore for CountingStore { + async fn ensure_folder( + &self, + parent_id: &str, + name: &str, + drive_context: &DriveContext, + ) -> anyhow::Result { + bump(&self.counters.ensure_folder); + self.inner + .ensure_folder(parent_id, name, drive_context) + .await + } + + async fn list_folder( + &self, + folder_id: &str, + drive_context: &DriveContext, + ) -> anyhow::Result> { + bump(&self.counters.list_folder); + self.inner.list_folder(folder_id, drive_context).await + } + + async fn list_shared_drives(&self) -> anyhow::Result> { + bump(&self.counters.list_shared_drives); + self.inner.list_shared_drives().await + } + + async fn create( + &self, + parent_id: &str, + name: &str, + mime: &str, + body: UploadBody, + app_properties: HashMap, + ) -> anyhow::Result { + bump(&self.counters.create); + self.inner + .create(parent_id, name, mime, body, app_properties) + .await + } + + async fn update( + &self, + file_id: &str, + body: UploadBody, + app_properties_patch: HashMap, + ) -> anyhow::Result { + bump(&self.counters.update); + self.inner.update(file_id, body, app_properties_patch).await + } + + async fn resumable_session( + &self, + kind: ResumableKind, + mime: &str, + size: u64, + ) -> anyhow::Result { + bump(&self.counters.resumable_session); + self.inner.resumable_session(kind, mime, size).await + } + + async fn resume_chunk( + &self, + session: &ResumableSession, + offset: u64, + chunk: Bytes, + ) -> anyhow::Result { + bump(&self.counters.resume_chunk); + self.inner.resume_chunk(session, offset, chunk).await + } + + async fn trash(&self, file_id: &str) -> anyhow::Result<()> { + bump(&self.counters.trash); + self.inner.trash(file_id).await + } + + async fn delete_permanent(&self, file_id: &str) -> anyhow::Result<()> { + bump(&self.counters.delete_permanent); + self.inner.delete_permanent(file_id).await + } + + async fn metadata(&self, file_id: &str) -> anyhow::Result { + bump(&self.counters.metadata); + self.inner.metadata(file_id).await + } + + async fn download(&self, file_id: &str) -> anyhow::Result { + bump(&self.counters.download); + self.inner.download(file_id).await + } + + async fn find_by_op_uuid( + &self, + parent_id: &str, + op_uuid: &str, + drive_context: &DriveContext, + ) -> anyhow::Result> { + bump(&self.counters.find_by_op_uuid); + self.inner + .find_by_op_uuid(parent_id, op_uuid, drive_context) + .await + } + + async fn list_source_object_ids( + &self, + source_id: &str, + drive_context: &DriveContext, + ) -> anyhow::Result> { + bump(&self.counters.list_source_object_ids); + self.inner + .list_source_object_ids(source_id, drive_context) + .await + } + + async fn about(&self) -> anyhow::Result { + bump(&self.counters.about); + self.inner.about().await + } +} + +#[cfg(test)] +mod tests { + use super::*; + use driven_drive::fake::InMemoryRemoteStore; + + #[tokio::test] + async fn counts_each_delegated_call_and_totals_them() { + let fake = Arc::new(InMemoryRemoteStore::new()); + let root = fake.root_id().to_string(); + let (store, counters) = CountingStore::new(fake); + + assert_eq!(counters.snapshot().total, 0); + + let ctx = DriveContext::MyDrive; + let folder = store.ensure_folder(&root, "a", &ctx).await.unwrap(); + store + .create( + &folder.id, + "f.bin", + "application/octet-stream", + UploadBody::Bytes(vec![1, 2, 3].into()), + HashMap::new(), + ) + .await + .unwrap(); + store.list_folder(&folder.id, &ctx).await.unwrap(); + store.list_folder(&folder.id, &ctx).await.unwrap(); + + let counts = counters.snapshot(); + assert_eq!(counts.ensure_folder, 1); + assert_eq!(counts.create, 1); + assert_eq!(counts.list_folder, 2); + assert_eq!(counts.total, 4, "total must be the sum of every counter"); + } + + #[tokio::test] + async fn a_failing_call_is_still_counted() { + let fake = Arc::new(InMemoryRemoteStore::new()); + let (store, counters) = CountingStore::new(fake); + // An id that was never created: the call fails, but it still cost a + // request, so the benchmark must count it. + assert!(store.metadata("no-such-id").await.is_err()); + assert_eq!(counters.snapshot().metadata, 1); + } +} diff --git a/crates/driven-bench/src/creds.rs b/crates/driven-bench/src/creds.rs new file mode 100644 index 00000000..7eb35808 --- /dev/null +++ b/crates/driven-bench/src/creds.rs @@ -0,0 +1,255 @@ +//! Credential loading and the destination safety rails. +//! +//! The suite talks to a REAL Google Drive account, so the rules here are +//! deliberately strict: +//! +//! - Credentials come from the environment only. Nothing is read from the OS +//! keychain (so a bench run can never pick up the maintainer's personal +//! account by accident) and nothing is ever printed. +//! - The destination folder must be named explicitly - by `--dest` or by +//! `DRIVEN_E2E_DEST_FOLDER_ID`. There is no default and no discovery step. +//! - Every remote write goes under one freshly created run folder inside that +//! destination, and cleanup trashes exactly that folder by the id it was +//! created with. The suite never lists the destination and never matches by +//! name, so it cannot delete anything it did not create. + +use std::path::Path; +use std::sync::Arc; + +use anyhow::{Context, Result}; + +use driven_drive::google::token_store::RefreshingTokenSource; +use driven_drive::google::GoogleDriveStore; +use driven_drive::remote_store::{DriveContext, RemoteStore}; +use driven_drive::{CustomCaConfig, ProxyConfig}; + +/// Environment variable names, shared with the real-Drive e2e suite so one set +/// of secrets serves both (design/E2E_REAL.md). +pub const ENV_REFRESH_TOKEN: &str = "DRIVEN_E2E_REFRESH_TOKEN"; +pub const ENV_DEST_FOLDER_ID: &str = "DRIVEN_E2E_DEST_FOLDER_ID"; +pub const ENV_CLIENT_ID: &str = "DRIVEN_OAUTH_CLIENT_ID"; +pub const ENV_CLIENT_SECRET: &str = "DRIVEN_OAUTH_CLIENT_SECRET"; + +/// The resolved credentials for a bench run. Deliberately has no `Debug` impl - +/// a stray `{:?}` is the classic way a token ends up in a log. +pub struct BenchCreds { + pub client_id: String, + pub client_secret: String, + pub refresh_token: String, +} + +impl BenchCreds { + /// Reads the credentials from the environment. + pub fn from_env() -> Result { + Ok(Self { + client_id: required(ENV_CLIENT_ID)?, + client_secret: required(ENV_CLIENT_SECRET)?, + refresh_token: required(ENV_REFRESH_TOKEN)?, + }) + } + + /// Builds a live Drive store from the refresh token. + /// + /// Unlike the desktop app this never touches the keychain, so a rotated + /// refresh token is not persisted anywhere - which is what we want for a + /// throwaway benchmark identity. + pub fn build_store(&self) -> Result { + let ca = CustomCaConfig::none(); + let proxy = ProxyConfig::system(); + let tokens = RefreshingTokenSource::from_stored_refresh_token( + self.refresh_token.clone(), + self.client_id.clone(), + self.client_secret.clone(), + &ca, + &proxy, + ) + .context("building the Drive token source from the refresh token")?; + GoogleDriveStore::with_default_clients(tokens, &ca, &proxy) + .context("building the Drive store") + } +} + +/// Reads one required environment variable, with an error that says how to fix +/// it rather than just naming the variable. +fn required(key: &str) -> Result { + match std::env::var(key) { + Ok(v) if !v.trim().is_empty() => Ok(v), + _ => anyhow::bail!( + "{key} is not set. Source the gitignored .env.test at the repo root \ + (or set the four DRIVEN_* bench variables by hand); see bench/README.md." + ), + } +} + +/// Resolves the destination folder id from an explicit flag or the environment. +/// +/// This is a hard precondition, checked before a single byte is generated: a +/// benchmark that guessed its destination could write into a real backup. +pub fn resolve_dest_folder_id(explicit: Option<&str>) -> Result { + if let Some(id) = explicit { + let id = id.trim(); + if !id.is_empty() { + return Ok(id.to_string()); + } + } + required(ENV_DEST_FOLDER_ID).context( + "the benchmark refuses to run without an explicit destination: pass --dest ", + ) +} + +/// Loads `KEY=VALUE` pairs from a dotenv-style file, if it exists. +/// +/// Existing environment variables always win, so CI secrets are never shadowed +/// by a stale local file. Values are not logged. +pub fn load_dotenv(path: &Path) -> Result { + let Ok(contents) = std::fs::read_to_string(path) else { + return Ok(false); + }; + for line in contents.lines() { + let line = line.trim(); + if line.is_empty() || line.starts_with('#') { + continue; + } + let Some((key, value)) = line.split_once('=') else { + continue; + }; + let key = key.trim(); + let value = value.trim().trim_matches('"').trim_matches('\''); + if key.is_empty() || std::env::var_os(key).is_some() { + continue; + } + // SAFETY-ish: this runs during single-threaded startup, before the + // tokio runtime or any worker thread reads the environment. + std::env::set_var(key, value); + } + Ok(true) +} + +/// A remote folder created by the harness, which trashes itself on request. +/// +/// Holding the id from creation is the entire safety story: cleanup never +/// enumerates, never searches by name, and therefore cannot touch anything that +/// was already in the destination. +pub struct RunFolder { + store: Arc, + pub id: String, + pub name: String, +} + +impl RunFolder { + /// Creates `name` under `parent_id`. + pub async fn create( + store: Arc, + parent_id: &str, + name: String, + ) -> Result { + let entry = store + .ensure_folder(parent_id, &name, &DriveContext::MyDrive) + .await + .with_context(|| format!("creating the run folder '{name}' under {parent_id}"))?; + Ok(Self { + store, + id: entry.id, + name, + }) + } + + /// Creates a child folder under this one, for one scenario's uploads. + pub async fn child(&self, name: &str) -> Result { + let entry = self + .store + .ensure_folder(&self.id, name, &DriveContext::MyDrive) + .await + .with_context(|| format!("creating scenario folder '{name}'"))?; + Ok(entry.id) + } + + /// Trashes this folder - and only this folder - by the id it was created + /// with. Trashing a folder trashes its subtree, so one call cleans up every + /// scenario beneath it. Already-gone is success (`trash` treats a 404 as + /// the desired state). + pub async fn cleanup(&self) -> Result<()> { + self.store + .trash(&self.id) + .await + .with_context(|| format!("trashing the run folder {} ({})", self.name, self.id)) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::io::Write; + + #[test] + fn resolve_dest_prefers_the_explicit_flag() { + assert_eq!( + resolve_dest_folder_id(Some("folder-abc")).unwrap(), + "folder-abc" + ); + } + + #[test] + fn resolve_dest_errors_when_nothing_is_set() { + // The variable is intentionally absent in the unit-test environment. + if std::env::var_os(ENV_DEST_FOLDER_ID).is_some() { + return; + } + let err = resolve_dest_folder_id(None).unwrap_err().to_string(); + assert!( + err.contains("--dest"), + "the error must tell the user how to fix it, got: {err}" + ); + } + + #[test] + fn resolve_dest_treats_blank_as_unset() { + if std::env::var_os(ENV_DEST_FOLDER_ID).is_some() { + return; + } + assert!(resolve_dest_folder_id(Some(" ")).is_err()); + } + + #[test] + fn dotenv_parses_pairs_and_skips_comments() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join(".env.test"); + let mut f = std::fs::File::create(&path).unwrap(); + writeln!(f, "# a comment").unwrap(); + writeln!(f).unwrap(); + writeln!(f, "DRIVEN_BENCH_DOTENV_PROBE=hello").unwrap(); + writeln!(f, "DRIVEN_BENCH_DOTENV_QUOTED=\"quoted value\"").unwrap(); + drop(f); + + std::env::remove_var("DRIVEN_BENCH_DOTENV_PROBE"); + std::env::remove_var("DRIVEN_BENCH_DOTENV_QUOTED"); + assert!(load_dotenv(&path).unwrap()); + assert_eq!(std::env::var("DRIVEN_BENCH_DOTENV_PROBE").unwrap(), "hello"); + assert_eq!( + std::env::var("DRIVEN_BENCH_DOTENV_QUOTED").unwrap(), + "quoted value" + ); + std::env::remove_var("DRIVEN_BENCH_DOTENV_PROBE"); + std::env::remove_var("DRIVEN_BENCH_DOTENV_QUOTED"); + } + + #[test] + fn dotenv_never_overrides_an_existing_variable() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join(".env.test"); + std::fs::write(&path, "DRIVEN_BENCH_DOTENV_WINS=from-file\n").unwrap(); + std::env::set_var("DRIVEN_BENCH_DOTENV_WINS", "from-env"); + load_dotenv(&path).unwrap(); + assert_eq!( + std::env::var("DRIVEN_BENCH_DOTENV_WINS").unwrap(), + "from-env", + "a CI secret must never be shadowed by a stale local file" + ); + std::env::remove_var("DRIVEN_BENCH_DOTENV_WINS"); + } + + #[test] + fn dotenv_missing_file_is_not_an_error() { + assert!(!load_dotenv(Path::new("no/such/.env.test")).unwrap()); + } +} diff --git a/crates/driven-bench/src/fixture.rs b/crates/driven-bench/src/fixture.rs new file mode 100644 index 00000000..abfe950b --- /dev/null +++ b/crates/driven-bench/src/fixture.rs @@ -0,0 +1,651 @@ +//! Deterministic fixture generation for the benchmark suite. +//! +//! Two shapes cover the cases the suite is meant to answer (bench/README.md): +//! +//! - [`Shape::Huge`] - a handful of multi-hundred-megabyte files. Exercises raw +//! upload throughput, chunking and the resumable path; almost no per-file +//! overhead. +//! - [`Shape::TinyDeep`] - up to a million small files spread over a deeply +//! nested tree. Exercises walking, hashing, per-file bookkeeping and request +//! round-trips; almost no raw byte throughput. +//! +//! Everything is a pure function of `(shape, scale, seed)`: the same triple +//! always produces byte-identical trees, on any machine, so two tools are +//! compared against the same input and a re-run compares against the same input +//! as last week. +//! +//! # Why the content is pseudo-random +//! +//! File bodies are filled from a seeded SplitMix64 stream, which is effectively +//! incompressible. That is a deliberate choice: a tool that compresses on the +//! wire (or, in Driven's case, packs cold small files into a `.tar.gz` bundle) +//! would otherwise score wildly better on zero-filled or text-like fixtures than +//! it ever would on the photos, videos and archives that dominate a real backup +//! set. Incompressible content measures the transport, not the entropy of the +//! test data. The trade-off is documented in bench/README.md so nobody reads +//! these numbers as "compression does not help" - on a compressible corpus it +//! very much does. + +use std::fs; +use std::io::Write; +use std::path::{Path, PathBuf}; + +use anyhow::{Context, Result}; +use serde::{Deserialize, Serialize}; + +/// Bytes written per `write_all` call while filling a file body. +const WRITE_CHUNK: usize = 1 << 20; + +/// Small files per leaf directory in the [`Shape::TinyDeep`] tree. Keeping this +/// low forces a wide, genuinely deep tree rather than a few fat directories. +const FILES_PER_LEAF: usize = 4; + +/// Smallest / largest body size for a [`Shape::TinyDeep`] file. The spread is +/// what makes the shape realistic - a fixed size would let a tool tune to it. +const TINY_MIN_BYTES: u64 = 64; +const TINY_MAX_BYTES: u64 = 4096; + +/// The file name of the manifest describing a materialised fixture. It lives +/// BESIDE the tree (not inside it) so it is never itself uploaded. +const MANIFEST_NAME: &str = "manifest.json"; + +/// The subdirectory holding the actual files. The tool under test is pointed at +/// this path, never at the fixture root. +const TREE_DIR: &str = "tree"; + +/// The two fixture shapes. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, clap::ValueEnum)] +#[serde(rename_all = "kebab-case")] +pub enum Shape { + /// A few very large files in one flat directory. + Huge, + /// Very many very small files in a deeply nested tree. + TinyDeep, +} + +impl Shape { + /// The stable slug used in fixture directory names and report tables. + pub fn slug(self) -> &'static str { + match self { + Shape::Huge => "huge", + Shape::TinyDeep => "tiny-deep", + } + } +} + +impl std::fmt::Display for Shape { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(self.slug()) + } +} + +/// A fully-resolved description of one fixture tree. +/// +/// This is the single source of truth for "what does this fixture contain": +/// path layout, per-file sizes and file contents are all derived from it. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct FixtureSpec { + /// Which shape to build. + pub shape: Shape, + /// How many files the tree holds. + pub files: usize, + /// For [`Shape::Huge`], the exact size of every file. Ignored for + /// [`Shape::TinyDeep`], whose sizes are drawn per file from the seed. + pub huge_file_bytes: u64, + /// Directory nesting depth for [`Shape::TinyDeep`]. Ignored for + /// [`Shape::Huge`], which is flat. + pub depth: usize, + /// The PRNG seed. Same seed, same bytes, forever. + pub seed: u64, +} + +impl FixtureSpec { + /// The directory name this spec materialises into, unique per spec so two + /// scales can coexist in the fixture cache. + pub fn dir_name(&self) -> String { + match self.shape { + Shape::Huge => format!( + "huge-{}x{}-s{}", + self.files, self.huge_file_bytes, self.seed + ), + Shape::TinyDeep => format!("tiny-deep-{}-d{}-s{}", self.files, self.depth, self.seed), + } + } + + /// Total byte size of the tree, computed without touching the disk. + pub fn total_bytes(&self) -> u64 { + match self.shape { + Shape::Huge => self.huge_file_bytes * self.files as u64, + Shape::TinyDeep => (0..self.files).map(|i| self.file_size(i)).sum(), + } + } + + /// The size of file `index`, derived from the seed. + pub fn file_size(&self, index: usize) -> u64 { + match self.shape { + Shape::Huge => self.huge_file_bytes, + Shape::TinyDeep => { + let r = + splitmix64(self.seed ^ mix(index as u64).wrapping_mul(0x9E37_79B9_7F4A_7C15)); + TINY_MIN_BYTES + (r % (TINY_MAX_BYTES - TINY_MIN_BYTES + 1)) + } + } + } + + /// The tree-relative path of file `index`. + /// + /// For [`Shape::TinyDeep`] the leaf directory is the base-`fanout` + /// representation of the file's leaf index, padded to exactly `depth` + /// components - so every file really does sit `depth` levels down, and no + /// empty directories are ever created. + pub fn file_path(&self, index: usize) -> PathBuf { + match self.shape { + Shape::Huge => PathBuf::from(format!("file_{index:04}.bin")), + Shape::TinyDeep => { + let leaf = index / FILES_PER_LEAF; + let fanout = self.fanout(); + let mut path = PathBuf::new(); + let mut rest = leaf; + // Least-significant digit first is fine: the mapping only has to + // be a stable bijection, not sorted. + for _ in 0..self.depth { + path.push(format!("d{:02}", rest % fanout)); + rest /= fanout; + } + path.push(format!("f{index:07}.bin")); + path + } + } + } + + /// The directory branching factor: the smallest `f >= 2` with + /// `f^depth >= leaves`, so the tree is exactly `depth` deep and no wider + /// than it needs to be. + fn fanout(&self) -> usize { + let leaves = self.files.div_ceil(FILES_PER_LEAF).max(1); + let mut f = 2usize; + loop { + // Saturating power: a big fanout with a big depth overflows long + // before it stops satisfying the bound. + let mut acc: u128 = 1; + for _ in 0..self.depth { + acc = acc.saturating_mul(f as u128); + } + if acc >= leaves as u128 || f >= 64 { + return f; + } + f += 1; + } + } + + /// The PRNG state that seeds file `index`'s body. `generation` is bumped by + /// [`Fixture::mutate`] so a mutated file gets genuinely different bytes. + fn content_seed(&self, index: usize, generation: u64) -> u64 { + splitmix64(self.seed ^ mix(index as u64) ^ generation.wrapping_mul(0xD1B5_4A32_D192_ED03)) + } +} + +/// The on-disk record of a materialised fixture, stored beside the tree. +#[derive(Debug, Clone, Serialize, Deserialize)] +struct Manifest { + spec: FixtureSpec, + /// Indices whose bodies are currently at generation 1 (mutated) rather than + /// generation 0 (pristine). Persisted so an interrupted run leaves enough + /// information to restore the tree instead of rebuilding it. + mutated: Vec, +} + +/// A materialised fixture on disk. +pub struct Fixture { + root: PathBuf, + spec: FixtureSpec, + mutated: Vec, +} + +impl Fixture { + /// The directory to point a tool at. + pub fn tree(&self) -> PathBuf { + self.root.join(TREE_DIR) + } + + /// Materialises `spec` under `cache_root`, reusing an existing tree when one + /// matches. + /// + /// A reused tree is always returned PRISTINE: if a previous run mutated + /// files and did not restore them (or crashed midway), the recorded indices + /// are rewritten back to their generation-0 bodies. That matters because the + /// cold-upload phase of the next tool must see byte-identical input to the + /// one the previous tool saw. + pub fn build(cache_root: &Path, spec: &FixtureSpec) -> Result { + let root = cache_root.join(spec.dir_name()); + let manifest_path = root.join(MANIFEST_NAME); + + if let Ok(bytes) = fs::read(&manifest_path) { + if let Ok(manifest) = serde_json::from_slice::(&bytes) { + if manifest.spec == *spec { + let mut fixture = Fixture { + root, + spec: spec.clone(), + mutated: manifest.mutated, + }; + if !fixture.mutated.is_empty() { + eprintln!( + "fixture {}: restoring {} previously-mutated file(s)", + spec.dir_name(), + fixture.mutated.len() + ); + fixture.restore()?; + } + return Ok(fixture); + } + } + } + + // No usable cache: rebuild from scratch. + if root.exists() { + fs::remove_dir_all(&root) + .with_context(|| format!("clearing stale fixture {}", root.display()))?; + } + fs::create_dir_all(root.join(TREE_DIR)) + .with_context(|| format!("creating fixture root {}", root.display()))?; + + let fixture = Fixture { + root, + spec: spec.clone(), + mutated: Vec::new(), + }; + fixture.write_range(0..spec.files, 0)?; + fixture.save_manifest()?; + Ok(fixture) + } + + /// Rewrites a deterministic `fraction` of the tree's files with fresh + /// content, modelling the "small changes since the last backup" case. + /// + /// Selection is a fixed stride derived from the seed, so the same fixture + /// always mutates the same files - both tools see the identical delta. + /// Returns the indices touched. + /// + /// Content changes but the file set does not: no creates, no deletes. That + /// keeps `rclone copy` and Driven comparable (see bench/README.md - a delete + /// would need `rclone sync` to be a fair match for Driven's trash pass). + pub fn mutate(&mut self, fraction: f64) -> Result> { + let count = ((self.spec.files as f64 * fraction).round() as usize) + .clamp(1, self.spec.files) + .min(self.spec.files); + let stride = (self.spec.files / count).max(1); + let offset = (splitmix64(self.spec.seed ^ 0xBEEF) as usize) % stride; + + let indices: Vec = (0..count) + .map(|n| (offset + n * stride) % self.spec.files) + .collect(); + + for &index in &indices { + self.write_one(index, 1)?; + } + + self.mutated = indices.clone(); + self.save_manifest()?; + Ok(indices) + } + + /// Rewrites every mutated file back to its pristine body. + pub fn restore(&mut self) -> Result<()> { + let indices = std::mem::take(&mut self.mutated); + for index in indices { + self.write_one(index, 0)?; + } + self.save_manifest() + } + + /// Writes files `range` at content `generation`, spreading the work over the + /// available cores (a million small files is a thread-bound workload, and a + /// single-threaded generator would dominate the time spent benchmarking). + fn write_range(&self, range: std::ops::Range, generation: u64) -> Result<()> { + let workers = std::thread::available_parallelism() + .map(|n| n.get()) + .unwrap_or(4) + .clamp(1, 16); + let total = range.len(); + let chunk = total.div_ceil(workers).max(1); + + eprintln!( + "fixture {}: writing {} file(s), {} across {} worker(s)", + self.spec.dir_name(), + total, + human_bytes(self.spec.total_bytes()), + workers + ); + + let results: Vec> = std::thread::scope(|scope| { + let handles: Vec<_> = (0..workers) + .map(|w| { + let start = range.start + w * chunk; + let end = (start + chunk).min(range.end); + scope.spawn(move || { + for index in start..end { + self.write_one(index, generation)?; + } + Ok(()) + }) + }) + .collect(); + handles + .into_iter() + .map(|h| { + h.join() + .unwrap_or_else(|_| Err(anyhow::anyhow!("fixture worker panicked"))) + }) + .collect() + }); + for r in results { + r?; + } + Ok(()) + } + + /// Writes exactly one file's body at `generation`, creating parent + /// directories as needed. + fn write_one(&self, index: usize, generation: u64) -> Result<()> { + let rel = self.spec.file_path(index); + let path = self.tree().join(&rel); + if let Some(parent) = path.parent() { + fs::create_dir_all(parent).with_context(|| format!("creating {}", parent.display()))?; + } + let size = self.spec.file_size(index); + let mut state = self.spec.content_seed(index, generation); + let mut file = fs::File::create(&path) + .with_context(|| format!("creating fixture file {}", path.display()))?; + + let mut written = 0u64; + let mut buf = vec![0u8; WRITE_CHUNK.min(size.max(1) as usize)]; + while written < size { + let want = ((size - written) as usize).min(buf.len()); + fill_random(&mut buf[..want], &mut state); + file.write_all(&buf[..want]) + .with_context(|| format!("writing fixture file {}", path.display()))?; + written += want as u64; + } + file.flush()?; + Ok(()) + } + + fn save_manifest(&self) -> Result<()> { + let manifest = Manifest { + spec: self.spec.clone(), + mutated: self.mutated.clone(), + }; + let bytes = serde_json::to_vec_pretty(&manifest)?; + fs::write(self.root.join(MANIFEST_NAME), bytes) + .with_context(|| format!("writing manifest under {}", self.root.display()))?; + Ok(()) + } +} + +/// Deletes every cached fixture under `cache_root`. +pub fn clean(cache_root: &Path) -> Result<()> { + if cache_root.exists() { + fs::remove_dir_all(cache_root) + .with_context(|| format!("removing fixture cache {}", cache_root.display()))?; + } + Ok(()) +} + +/// Fills `buf` with a SplitMix64 stream, advancing `state`. +fn fill_random(buf: &mut [u8], state: &mut u64) { + for word in buf.chunks_mut(8) { + *state = state.wrapping_add(0x9E37_79B9_7F4A_7C15); + let value = splitmix64(*state); + let bytes = value.to_le_bytes(); + word.copy_from_slice(&bytes[..word.len()]); + } +} + +/// The SplitMix64 finaliser - a fast, well-distributed 64-bit mixer. +fn splitmix64(mut z: u64) -> u64 { + z = (z ^ (z >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9); + z = (z ^ (z >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB); + z ^ (z >> 31) +} + +/// Mixes an index into a well-spread 64-bit value. +fn mix(x: u64) -> u64 { + splitmix64(x.wrapping_add(0x2545_F491_4F6C_DD1D)) +} + +/// Formats a byte count for human-facing log lines. +pub fn human_bytes(bytes: u64) -> String { + const UNITS: [&str; 5] = ["B", "KiB", "MiB", "GiB", "TiB"]; + let mut value = bytes as f64; + let mut unit = 0; + while value >= 1024.0 && unit < UNITS.len() - 1 { + value /= 1024.0; + unit += 1; + } + if unit == 0 { + format!("{bytes} B") + } else { + format!("{value:.1} {}", UNITS[unit]) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn tiny_spec() -> FixtureSpec { + FixtureSpec { + shape: Shape::TinyDeep, + files: 40, + huge_file_bytes: 0, + depth: 4, + seed: 7, + } + } + + /// The files `spec` says must exist under `root`, as + /// `(tree-relative slash path, contents)`, sorted. + /// + /// Reads by the path the spec PREDICTS rather than by walking the tree. + /// That is the stronger assertion: the fixture's entire contract is that + /// its layout is a pure function of the spec, so a helper that discovered + /// paths from disk would happily pass even if `file_path` and the writer + /// had drifted together. A missing file fails loudly here. Use + /// [`count_files`] alongside it to catch a stray EXTRA file. + fn read_expected(root: &Path, spec: &FixtureSpec) -> Vec<(String, Vec)> { + let mut out: Vec<(String, Vec)> = (0..spec.files) + .map(|index| { + let rel = spec.file_path(index); + let bytes = fs::read(root.join(&rel)) + .unwrap_or_else(|e| panic!("fixture file {} missing: {e}", rel.display())); + (rel.to_string_lossy().replace('\\', "/"), bytes) + }) + .collect(); + out.sort(); + out + } + + /// How many files exist under `root`, so a test can catch a leftover file + /// from an earlier spec that [`read_expected`] would never look at. + fn count_files(root: &Path) -> usize { + let mut count = 0; + let mut stack = vec![root.to_path_buf()]; + while let Some(dir) = stack.pop() { + for entry in fs::read_dir(&dir).unwrap() { + let entry = entry.unwrap(); + // Ask the directory entry rather than re-statting its path: one + // fewer syscall, and nothing to get wrong about the path itself. + if entry.file_type().unwrap().is_dir() { + stack.push(entry.path()); + } else { + count += 1; + } + } + } + count + } + + #[test] + fn same_seed_produces_byte_identical_trees() { + let a = tempdir(); + let b = tempdir(); + let spec = tiny_spec(); + let fa = Fixture::build(a.path(), &spec).unwrap(); + let fb = Fixture::build(b.path(), &spec).unwrap(); + assert_eq!( + read_expected(&fa.tree(), &spec), + read_expected(&fb.tree(), &spec) + ); + } + + #[test] + fn different_seeds_produce_different_content() { + let a = tempdir(); + let b = tempdir(); + let spec = tiny_spec(); + let fa = Fixture::build(a.path(), &spec).unwrap(); + let mut other = spec.clone(); + other.seed = 8; + let fb = Fixture::build(b.path(), &other).unwrap(); + assert_ne!( + read_expected(&fa.tree(), &spec), + read_expected(&fb.tree(), &other) + ); + } + + #[test] + fn tree_is_exactly_depth_deep_and_has_the_requested_file_count() { + let dir = tempdir(); + let spec = tiny_spec(); + let f = Fixture::build(dir.path(), &spec).unwrap(); + let files = read_expected(&f.tree(), &spec); + assert_eq!(files.len(), spec.files); + assert_eq!( + count_files(&f.tree()), + spec.files, + "the tree must hold exactly the spec's files and nothing else" + ); + for (rel, _) in &files { + // depth directory components plus the file name. + assert_eq!( + rel.split('/').count(), + spec.depth + 1, + "{rel} is not {} levels deep", + spec.depth + ); + } + } + + #[test] + fn total_bytes_matches_what_was_written() { + let dir = tempdir(); + let spec = tiny_spec(); + let f = Fixture::build(dir.path(), &spec).unwrap(); + let on_disk: u64 = read_expected(&f.tree(), &spec) + .iter() + .map(|(_, b)| b.len() as u64) + .sum(); + assert_eq!(on_disk, spec.total_bytes()); + } + + #[test] + fn huge_shape_is_flat_and_exact_size() { + let dir = tempdir(); + let spec = FixtureSpec { + shape: Shape::Huge, + files: 3, + huge_file_bytes: 4096, + depth: 0, + seed: 1, + }; + let f = Fixture::build(dir.path(), &spec).unwrap(); + let files = read_expected(&f.tree(), &spec); + assert_eq!(files.len(), 3); + assert_eq!(count_files(&f.tree()), 3); + for (rel, bytes) in files { + assert!(!rel.contains('/'), "huge shape must be flat, got {rel}"); + assert_eq!(bytes.len(), 4096); + } + } + + #[test] + fn mutate_changes_only_the_selected_files_and_restore_undoes_it() { + let dir = tempdir(); + let spec = tiny_spec(); + let mut f = Fixture::build(dir.path(), &spec).unwrap(); + let before = read_expected(&f.tree(), &spec); + + let touched = f.mutate(0.1).unwrap(); + assert_eq!(touched.len(), 4, "10% of 40 files"); + let after = read_expected(&f.tree(), &spec); + assert_eq!( + count_files(&f.tree()), + spec.files, + "mutate must not add or remove files" + ); + let changed = before + .iter() + .zip(after.iter()) + .filter(|(a, b)| a.1 != b.1) + .count(); + assert_eq!(changed, touched.len()); + + f.restore().unwrap(); + assert_eq!( + read_expected(&f.tree(), &spec), + before, + "restore must be exact" + ); + } + + #[test] + fn mutation_selection_is_deterministic() { + let a = tempdir(); + let b = tempdir(); + let mut fa = Fixture::build(a.path(), &tiny_spec()).unwrap(); + let mut fb = Fixture::build(b.path(), &tiny_spec()).unwrap(); + assert_eq!(fa.mutate(0.1).unwrap(), fb.mutate(0.1).unwrap()); + } + + #[test] + fn rebuild_restores_a_fixture_left_mutated_by_a_crashed_run() { + let dir = tempdir(); + let spec = tiny_spec(); + let mut f = Fixture::build(dir.path(), &spec).unwrap(); + let pristine = read_expected(&f.tree(), &spec); + f.mutate(0.1).unwrap(); + assert_ne!(read_expected(&f.tree(), &spec), pristine); + drop(f); + + // A fresh build over the same cache must hand back a pristine tree. + let reused = Fixture::build(dir.path(), &spec).unwrap(); + assert_eq!(read_expected(&reused.tree(), &spec), pristine); + } + + #[test] + fn a_changed_spec_rebuilds_rather_than_reusing() { + let dir = tempdir(); + let spec = tiny_spec(); + let f = Fixture::build(dir.path(), &spec).unwrap(); + assert_eq!(count_files(&f.tree()), spec.files); + + let mut smaller = spec.clone(); + smaller.files = 12; + let f2 = Fixture::build(dir.path(), &smaller).unwrap(); + // Exactly 12 - the 28 files the previous spec wrote must be GONE, not + // left behind to be uploaded as part of the next benchmark. + assert_eq!(count_files(&f2.tree()), 12); + assert_eq!(read_expected(&f2.tree(), &smaller).len(), 12); + } + + #[test] + fn human_bytes_is_readable() { + assert_eq!(human_bytes(512), "512 B"); + assert_eq!(human_bytes(1536), "1.5 KiB"); + assert_eq!(human_bytes(1 << 30), "1.0 GiB"); + } + + fn tempdir() -> tempfile::TempDir { + tempfile::tempdir().expect("temp dir") + } +} diff --git a/crates/driven-bench/src/main.rs b/crates/driven-bench/src/main.rs new file mode 100644 index 00000000..a3aecadc --- /dev/null +++ b/crates/driven-bench/src/main.rs @@ -0,0 +1,520 @@ +//! `driven-bench` - the real-world benchmark suite (bench/README.md). +//! +//! Compares Driven's actual backup engine against `rclone` on the two shapes +//! that dominate real backup sets - a few very large files, and very many very +//! small ones in a deep tree - across a cold upload and an incremental re-run +//! after a small change. +//! +//! It is deliberately NOT part of any normal build or test path: it uploads real +//! bytes to a real Google account and costs real time. It runs on demand +//! (`just bench`, `bench/run.ps1`) or from the tag-gated `bench.yml` workflow. + +use std::path::{Path, PathBuf}; +use std::sync::Arc; + +use anyhow::{Context, Result}; +use clap::{Parser, Subcommand}; + +mod agent; +mod counting_store; +mod creds; +mod fixture; +mod procstat; +mod report; +mod tools; + +use crate::fixture::{human_bytes, Fixture, FixtureSpec, Shape}; +use crate::report::{RunReport, ScenarioReport}; +use crate::tools::{Phase, PhaseResult, Tool}; + +/// Default ceiling on the bytes one invocation may upload, summed over every +/// tool and fixture. A benchmark that quietly pushes tens of gigabytes to a +/// personal Drive is a bad benchmark; `--full` lifts it deliberately. +const DEFAULT_MAX_UPLOAD_BYTES: u64 = 2 * 1024 * 1024 * 1024; + +/// Fraction of the tree rewritten before the incremental phase. +const MUTATION_FRACTION: f64 = 0.001; + +#[derive(Debug, Parser)] +#[command( + name = "driven-bench", + version, + about = "Driven vs rclone benchmark suite" +)] +struct Cli { + #[command(subcommand)] + command: Command, +} + +#[derive(Debug, Subcommand)] +enum Command { + /// Run the benchmark matrix and write a report. + Run(RunArgs), + /// Build, mutate or delete fixture trees without benchmarking anything. + #[command(subcommand)] + Fixture(FixtureCommand), + /// Internal: run ONE Driven backup cycle and print its metrics. + /// + /// The harness re-invokes itself with this so the engine is measured as a + /// child process, exactly like rclone is. + #[command(hide = true)] + AgentSync(agent::AgentArgs), +} + +/// The fixture sizes a run can be invoked at. +#[derive(Debug, Clone, Copy, PartialEq, Eq, clap::ValueEnum)] +enum Scale { + /// Minutes, a few hundred megabytes. For proving the pipeline works. + Smoke, + /// The default: enough data for the numbers to mean something. + Small, + /// A serious local run. + Medium, + /// The shapes the suite is really about: multi-gigabyte files and a + /// million small ones. Needs `--full` to clear the upload cap. + Full, +} + +impl Scale { + fn slug(self) -> &'static str { + match self { + Scale::Smoke => "smoke", + Scale::Small => "small", + Scale::Medium => "medium", + Scale::Full => "full", + } + } + + /// The fixtures this scale runs, in report order. + fn specs(self, seed: u64) -> Vec { + let mib = 1024 * 1024; + let (huge_files, huge_bytes, tiny_files, depth) = match self { + Scale::Smoke => (2, 8 * mib, 300, 5), + Scale::Small => (4, 128 * mib, 50_000, 8), + Scale::Medium => (4, 512 * mib, 200_000, 8), + Scale::Full => (4, 2048 * mib, 1_000_000, 8), + }; + vec![ + FixtureSpec { + shape: Shape::Huge, + files: huge_files, + huge_file_bytes: huge_bytes, + depth: 0, + seed, + }, + FixtureSpec { + shape: Shape::TinyDeep, + files: tiny_files, + huge_file_bytes: 0, + depth, + seed, + }, + ] + } +} + +#[derive(Debug, clap::Args)] +struct RunArgs { + /// Fixture size. + #[arg(long, value_enum, default_value = "small")] + scale: Scale, + /// Which tools to measure. + #[arg(long, value_delimiter = ',', default_values = ["driven", "rclone"])] + tools: Vec, + /// Only run one fixture shape instead of both. + #[arg(long, value_enum)] + shape: Option, + /// The destination Drive folder id. Required, by flag or by + /// `DRIVEN_E2E_DEST_FOLDER_ID` - there is no default. + #[arg(long)] + dest: Option, + /// Fixture PRNG seed. The same seed always produces the same trees. + #[arg(long, default_value_t = 1)] + seed: u64, + /// Lift the default upload cap (needed for `--scale full`). + #[arg(long)] + full: bool, + /// Override the upload cap, in bytes. + #[arg(long)] + max_upload_bytes: Option, + /// Path to the rclone binary. Defaults to `rclone` on `PATH`. + #[arg(long)] + rclone: Option, + /// rclone's parallel transfer count. Defaults to rclone's own default (4), + /// i.e. each tool runs at its stock settings; pass Driven's pool size here + /// to compare the algorithms at equal concurrency instead. + #[arg(long, default_value_t = 4)] + rclone_transfers: u64, + /// Where to cache generated fixtures. + #[arg(long)] + fixture_root: Option, + /// Where to write the report. + #[arg(long)] + results_dir: Option, + /// Leave the uploaded run folder in Drive instead of trashing it. + #[arg(long)] + keep_remote: bool, +} + +#[derive(Debug, Subcommand)] +enum FixtureCommand { + /// Materialise a fixture tree without uploading anything. + Build { + #[arg(long, value_enum)] + shape: Shape, + #[arg(long, value_enum, default_value = "small")] + scale: Scale, + #[arg(long, default_value_t = 1)] + seed: u64, + #[arg(long)] + fixture_root: Option, + }, + /// Delete every cached fixture. + Clean { + #[arg(long)] + fixture_root: Option, + }, +} + +#[tokio::main] +async fn main() -> Result<()> { + tracing_subscriber::fmt() + .with_env_filter( + tracing_subscriber::EnvFilter::try_from_default_env() + .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("warn")), + ) + .with_writer(std::io::stderr) + .init(); + + // A gitignored .env.test at the repo root is the local credential source; + // in CI the same names arrive as secrets and always win. + let _ = creds::load_dotenv(&repo_root().join(".env.test")); + + match Cli::parse().command { + Command::Run(args) => run(args).await, + Command::Fixture(cmd) => run_fixture(cmd), + Command::AgentSync(args) => agent::run(args).await, + } +} + +/// The repo root, derived from this crate's manifest directory. +fn repo_root() -> PathBuf { + Path::new(env!("CARGO_MANIFEST_DIR")) + .parent() + .and_then(Path::parent) + .map(Path::to_path_buf) + .unwrap_or_else(|| PathBuf::from(".")) +} + +/// Removes repeated tools while preserving the order they were given in. +fn dedupe_tools(tools: &[Tool]) -> Vec { + let mut unique: Vec = Vec::with_capacity(tools.len()); + for &tool in tools { + if !unique.contains(&tool) { + unique.push(tool); + } + } + unique +} + +fn fixture_root(explicit: Option) -> PathBuf { + explicit.unwrap_or_else(|| repo_root().join("target").join("bench-fixtures")) +} + +fn run_fixture(cmd: FixtureCommand) -> Result<()> { + match cmd { + FixtureCommand::Build { + shape, + scale, + seed, + fixture_root: root, + } => { + let root = fixture_root(root); + let spec = scale + .specs(seed) + .into_iter() + .find(|s| s.shape == shape) + .expect("every scale defines both shapes"); + let fixture = Fixture::build(&root, &spec)?; + println!("{}", fixture.tree().display()); + Ok(()) + } + FixtureCommand::Clean { fixture_root: root } => { + let root = fixture_root(root); + fixture::clean(&root)?; + println!("removed {}", root.display()); + Ok(()) + } + } +} + +async fn run(mut args: RunArgs) -> Result<()> { + if args.tools.is_empty() { + anyhow::bail!("--tools must name at least one tool"); + } + // A repeated tool would reuse the same scenario folder AND the same state + // database, so its second "cold" phase would silently be an incremental one + // reported as cold. Collapse duplicates rather than measure a lie. + args.tools = dedupe_tools(&args.tools); + + // --- preconditions, all checked before a single byte is generated ------- + let dest_folder_id = creds::resolve_dest_folder_id(args.dest.as_deref())?; + let creds = creds::BenchCreds::from_env()?; + + let specs: Vec = args + .scale + .specs(args.seed) + .into_iter() + .filter(|s| args.shape.is_none_or(|shape| shape == s.shape)) + .collect(); + + let cap = args.max_upload_bytes.unwrap_or(DEFAULT_MAX_UPLOAD_BYTES); + let planned: u64 = specs.iter().map(|s| s.total_bytes()).sum::() * args.tools.len() as u64; + if !args.full && planned > cap { + anyhow::bail!( + "this run would upload {} ({} per tool x {} tool(s)), over the {} cap.\n\ + Pass --full to lift the cap, --max-upload-bytes to raise it, or use a smaller --scale.", + human_bytes(planned), + human_bytes(planned / args.tools.len() as u64), + args.tools.len(), + human_bytes(cap) + ); + } + + let rclone_binary = if args.tools.contains(&Tool::Rclone) { + let found = tools::find_rclone(args.rclone.as_deref()).context( + "rclone was requested but no rclone binary was found: install it, put it on PATH, \ + or pass --rclone (see bench/README.md)", + )?; + Some(found) + } else { + None + }; + + // --- one run folder, created up front, trashed at the end --------------- + let store: Arc = Arc::new(creds.build_store()?); + let run_name = format!("driven-bench-{}", uuid::Uuid::new_v4()); + let run_folder = creds::RunFolder::create(store, &dest_folder_id, run_name.clone()) + .await + .context("creating the run folder - check the destination folder id and the credentials")?; + println!("run folder: {run_name} ({})", run_folder.id); + println!( + "plan: {} fixture(s) x {} tool(s), up to {} uploaded", + specs.len(), + args.tools.len(), + human_bytes(planned) + ); + + // Everything after this point must reach the cleanup below, so the body's + // error is captured rather than returned. + let outcome = run_scenarios(&args, &specs, &creds, &run_folder, rclone_binary.as_deref()).await; + + if args.keep_remote { + println!( + "--keep-remote: leaving {run_name} ({}) in Drive", + run_folder.id + ); + } else if let Err(err) = run_folder.cleanup().await { + eprintln!("WARNING: failed to trash the run folder {run_name}: {err:#}"); + eprintln!(" trash it by hand: folder id {}", run_folder.id); + } else { + println!("cleaned up run folder {run_name}"); + } + + let scenarios = outcome?; + + let report = RunReport { + started_at: report::utc_timestamp(), + scale: args.scale.slug().to_string(), + seed: args.seed, + host: format!("{}/{}", std::env::consts::OS, std::env::consts::ARCH), + cpus: std::thread::available_parallelism() + .map(|n| n.get()) + .unwrap_or(0), + driven_version: env!("CARGO_PKG_VERSION").to_string(), + rclone_version: rclone_binary.as_deref().and_then(tools::rclone_version), + tools: args.tools.clone(), + scenarios, + }; + + let results_dir = args + .results_dir + .clone() + .unwrap_or_else(|| repo_root().join("bench").join("results")); + let (md, json) = report.write_to(&results_dir)?; + + println!("\n{}", report.to_markdown()); + println!("report: {}", md.display()); + println!("raw: {}", json.display()); + + if !report.all_ok() { + anyhow::bail!("at least one benchmark phase failed - see the table above"); + } + Ok(()) +} + +/// Runs every (fixture x tool) scenario, each in its own remote subfolder and +/// its own state database. +async fn run_scenarios( + args: &RunArgs, + specs: &[FixtureSpec], + creds: &creds::BenchCreds, + run_folder: &creds::RunFolder, + rclone_binary: Option<&Path>, +) -> Result> { + let fixture_cache = fixture_root(args.fixture_root.clone()); + let work = tempfile::tempdir().context("creating the harness work directory")?; + + let mut scenarios = Vec::new(); + for spec in specs { + // Built once and shared by every tool, so they upload identical bytes. + let mut fixture = Fixture::build(&fixture_cache, spec)?; + let mut results = Vec::new(); + + for &tool in &args.tools { + let scenario = format!("{}-{}", tool.slug(), spec.shape.slug()); + println!("\n=== {scenario} ==="); + let folder_id = run_folder.child(&scenario).await?; + + // The cold and incremental phases share one destination folder and, + // for Driven, one state database. A fresh state db per phase would + // make the incremental phase a second cold upload. + let state_db = work.path().join(&scenario).join("state.db"); + let rclone_config = work.path().join(format!("{scenario}.conf")); + std::fs::create_dir_all(state_db.parent().expect("state db has a parent"))?; + if tool == Tool::Rclone { + tools::write_rclone_config(&rclone_config, creds, &folder_id)?; + } + + // Resolved before the closure so mutating the fixture between + // phases does not collide with a live borrow of it. + let tree = fixture.tree(); + let phase = |phase: Phase| -> Result { + let result = match tool { + Tool::Driven => tools::run_driven(phase, &tree, &folder_id, &state_db)?, + Tool::Rclone => tools::run_rclone( + phase, + rclone_binary.expect("checked when rclone was requested"), + &rclone_config, + &tree, + args.rclone_transfers, + )?, + }; + println!( + " {phase:<12} {:>8.1}s {:>6} files {:>10}{}", + result.wall_secs, + result + .files_transferred + .map(|f| f.to_string()) + .unwrap_or_else(|| "?".into()), + result + .bytes_transferred + .map(human_bytes) + .unwrap_or_else(|| "?".into()), + result + .detail + .as_deref() + .map(|d| format!(" [{d}]")) + .unwrap_or_default(), + ); + Ok(result) + }; + + results.push(phase(Phase::Cold)?); + + let touched = fixture.mutate(MUTATION_FRACTION)?; + println!(" mutated {} of {} file(s)", touched.len(), spec.files); + + results.push(phase(Phase::Incremental)?); + + // Hand the next tool a pristine tree - it must see exactly what this + // one saw on its cold phase. + fixture.restore()?; + } + + scenarios.push(ScenarioReport { + spec: spec.clone(), + results, + }); + } + Ok(scenarios) +} + +#[cfg(test)] +mod tests { + use super::*; + use clap::CommandFactory; + + #[test] + fn cli_definition_is_valid() { + Cli::command().debug_assert(); + } + + #[test] + fn every_scale_defines_both_shapes() { + for scale in [Scale::Smoke, Scale::Small, Scale::Medium, Scale::Full] { + let specs = scale.specs(1); + assert_eq!(specs.len(), 2, "{} must define both shapes", scale.slug()); + assert!(specs.iter().any(|s| s.shape == Shape::Huge)); + assert!(specs.iter().any(|s| s.shape == Shape::TinyDeep)); + } + } + + #[test] + fn scales_increase_monotonically() { + let total = |scale: Scale| -> u64 { scale.specs(1).iter().map(|s| s.total_bytes()).sum() }; + assert!(total(Scale::Smoke) < total(Scale::Small)); + assert!(total(Scale::Small) < total(Scale::Medium)); + assert!(total(Scale::Medium) < total(Scale::Full)); + } + + #[test] + fn the_smoke_scale_fits_under_the_default_cap_for_both_tools() { + let planned: u64 = Scale::Smoke + .specs(1) + .iter() + .map(|s| s.total_bytes()) + .sum::() + * 2; + assert!( + planned < DEFAULT_MAX_UPLOAD_BYTES, + "the smoke scale must never trip the upload cap" + ); + } + + #[test] + fn the_full_scale_exceeds_the_default_cap_so_it_needs_an_explicit_opt_in() { + let planned: u64 = Scale::Full.specs(1).iter().map(|s| s.total_bytes()).sum(); + assert!( + planned > DEFAULT_MAX_UPLOAD_BYTES, + "the full scale must require --full rather than running by accident" + ); + } + + #[test] + fn the_full_scale_is_the_shape_the_suite_is_about() { + let specs = Scale::Full.specs(1); + let tiny = specs.iter().find(|s| s.shape == Shape::TinyDeep).unwrap(); + assert_eq!(tiny.files, 1_000_000); + let huge = specs.iter().find(|s| s.shape == Shape::Huge).unwrap(); + assert_eq!(huge.huge_file_bytes, 2 * 1024 * 1024 * 1024); + } + + #[test] + fn dedupe_tools_collapses_repeats_and_keeps_order() { + assert_eq!( + dedupe_tools(&[Tool::Rclone, Tool::Driven, Tool::Rclone]), + vec![Tool::Rclone, Tool::Driven] + ); + assert_eq!(dedupe_tools(&[Tool::Driven]), vec![Tool::Driven]); + assert!(dedupe_tools(&[]).is_empty()); + } + + #[test] + fn repo_root_contains_the_workspace_manifest() { + assert!( + repo_root().join("Cargo.toml").is_file(), + "repo_root() must resolve to the workspace root" + ); + } +} diff --git a/crates/driven-bench/src/procstat.rs b/crates/driven-bench/src/procstat.rs new file mode 100644 index 00000000..744e4791 --- /dev/null +++ b/crates/driven-bench/src/procstat.rs @@ -0,0 +1,245 @@ +//! Runs a child process and measures it with OS accounting. +//! +//! Every benchmarked tool - including Driven itself - runs as a CHILD of the +//! harness, so all of them are measured the same way. That symmetry is the whole +//! point of this module: if Driven's engine ran in-process, its "CPU time" would +//! silently include fixture generation, report rendering and the harness's own +//! Drive calls, and would not be comparable to rclone's. +//! +//! Wall time is measured by the harness around spawn/wait. CPU time and peak +//! memory come from the OS: +//! +//! - Windows (the primary platform) reports both exactly, per process, via +//! `GetProcessTimes` and `GetProcessMemoryInfo` on the child handle - which +//! stay valid after exit for as long as the handle is open. +//! - Unix reads `getrusage(RUSAGE_CHILDREN)` around the child. CPU time is an +//! exact delta because the harness never runs two children at once. Peak RSS +//! is a high-water mark across ALL reaped children, so it is reported only +//! when this child raised it; otherwise it is `None` rather than a number that +//! belongs to an earlier child. + +use std::io::Read; +use std::process::{Command, Stdio}; +use std::time::{Duration, Instant}; + +use anyhow::{Context, Result}; + +/// What one measured child run produced. +#[derive(Debug, Clone)] +pub struct ProcMetrics { + /// Wall-clock time from spawn to exit. + pub wall: Duration, + /// User + system CPU time, when the OS could attribute it. + pub cpu: Option, + /// Peak resident set / working set in bytes, when the OS could attribute it. + pub peak_rss_bytes: Option, + /// The process exit code, when it exited normally. + pub exit_code: Option, + /// Captured stdout. + pub stdout: String, + /// Captured stderr. + pub stderr: String, +} + +impl ProcMetrics { + /// Whether the child exited zero. + pub fn success(&self) -> bool { + self.exit_code == Some(0) + } +} + +/// Spawns `cmd`, captures its output, waits for it, and returns timings. +/// +/// Output is drained on background threads so a chatty child can never deadlock +/// on a full pipe buffer. +pub fn run_measured(cmd: &mut Command) -> Result { + cmd.stdout(Stdio::piped()).stderr(Stdio::piped()); + + let before = rusage_children(); + let started = Instant::now(); + let mut child = cmd + .spawn() + .with_context(|| format!("spawning {:?}", cmd.get_program()))?; + + let mut stdout_pipe = child.stdout.take().expect("piped stdout"); + let mut stderr_pipe = child.stderr.take().expect("piped stderr"); + + let (stdout, stderr) = std::thread::scope(|scope| { + let out = scope.spawn(move || { + let mut buf = String::new(); + let _ = stdout_pipe.read_to_string(&mut buf); + buf + }); + let err = scope.spawn(move || { + let mut buf = String::new(); + let _ = stderr_pipe.read_to_string(&mut buf); + buf + }); + ( + out.join().unwrap_or_default(), + err.join().unwrap_or_default(), + ) + }); + + let status = child.wait().context("waiting for child")?; + let wall = started.elapsed(); + + // Query the handle BEFORE `child` drops (Windows closes it on drop). + let (cpu, peak_rss_bytes) = child_resource_usage(&child, before); + + Ok(ProcMetrics { + wall, + cpu, + peak_rss_bytes, + exit_code: status.code(), + stdout, + stderr, + }) +} + +#[cfg(windows)] +mod platform { + use std::os::windows::io::AsRawHandle; + use std::time::Duration; + + use windows::Win32::Foundation::{FILETIME, HANDLE}; + use windows::Win32::System::ProcessStatus::{GetProcessMemoryInfo, PROCESS_MEMORY_COUNTERS}; + use windows::Win32::System::Threading::GetProcessTimes; + + /// No baseline is needed on Windows: every counter is per-process. A unit + /// STRUCT rather than `()` so the shared call site stays lint-clean. + pub struct Baseline; + + pub fn baseline() -> Baseline { + Baseline + } + + /// Converts a `FILETIME` (100-nanosecond ticks) to a `Duration`. + fn filetime_to_duration(ft: FILETIME) -> Duration { + let ticks = ((ft.dwHighDateTime as u64) << 32) | ft.dwLowDateTime as u64; + Duration::from_nanos(ticks.saturating_mul(100)) + } + + pub fn usage( + child: &std::process::Child, + _before: Baseline, + ) -> (Option, Option) { + let handle = HANDLE(child.as_raw_handle()); + + let mut creation = FILETIME::default(); + let mut exit = FILETIME::default(); + let mut kernel = FILETIME::default(); + let mut user = FILETIME::default(); + let cpu = + unsafe { GetProcessTimes(handle, &mut creation, &mut exit, &mut kernel, &mut user) } + .ok() + .map(|()| filetime_to_duration(kernel) + filetime_to_duration(user)); + + let mut counters = PROCESS_MEMORY_COUNTERS::default(); + let size = std::mem::size_of::() as u32; + let peak = unsafe { GetProcessMemoryInfo(handle, &mut counters, size) } + .ok() + .map(|()| counters.PeakWorkingSetSize as u64); + + (cpu, peak) + } +} + +#[cfg(unix)] +mod platform { + use std::time::Duration; + + /// The `(cpu, peak_rss)` reading taken before the child started. + pub type Baseline = Option<(Duration, u64)>; + + fn read() -> Baseline { + let mut usage: libc::rusage = unsafe { std::mem::zeroed() }; + // SAFETY: `usage` is a valid, fully-initialised rusage for the kernel to + // write into; RUSAGE_CHILDREN is a documented constant. + if unsafe { libc::getrusage(libc::RUSAGE_CHILDREN, &mut usage) } != 0 { + return None; + } + let to_duration = |tv: libc::timeval| { + Duration::new(tv.tv_sec as u64, (tv.tv_usec as u32).saturating_mul(1000)) + }; + // Linux reports ru_maxrss in kilobytes, macOS in bytes. + let scale: u64 = if cfg!(target_os = "macos") { 1 } else { 1024 }; + Some(( + to_duration(usage.ru_utime) + to_duration(usage.ru_stime), + (usage.ru_maxrss.max(0) as u64).saturating_mul(scale), + )) + } + + pub fn baseline() -> Baseline { + read() + } + + pub fn usage( + _child: &std::process::Child, + before: Baseline, + ) -> (Option, Option) { + let (Some((cpu_before, rss_before)), Some((cpu_after, rss_after))) = (before, read()) + else { + return (None, None); + }; + // Children run one at a time, so the CPU delta is exactly this child's. + let cpu = cpu_after.checked_sub(cpu_before); + // ru_maxrss is a high-water mark over every child reaped so far; it only + // tells us about THIS child when this child raised it. + let peak = (rss_after > rss_before).then_some(rss_after); + (cpu, peak) + } +} + +fn rusage_children() -> platform::Baseline { + platform::baseline() +} + +fn child_resource_usage( + child: &std::process::Child, + before: platform::Baseline, +) -> (Option, Option) { + platform::usage(child, before) +} + +#[cfg(test)] +mod tests { + use super::*; + + /// Spawns the test binary itself in a mode that just exits, which is + /// portable in a way that `sleep` / `timeout` are not. + fn trivial_child() -> Command { + let mut cmd = Command::new(std::env::current_exe().unwrap()); + // A filter that matches no test: the harness starts, reports 0 tests and + // exits 0 - cheap, and available on every platform. + cmd.arg("--exact").arg("driven_bench_no_such_test"); + cmd + } + + #[test] + fn measures_a_successful_child() { + let m = run_measured(&mut trivial_child()).unwrap(); + assert!(m.success(), "child failed: {}", m.stderr); + assert!(m.wall > Duration::ZERO); + // stdout of a libtest run always mentions how many tests ran. + assert!( + m.stdout.contains("test result") || m.stdout.contains("running"), + "unexpected child stdout: {}", + m.stdout + ); + } + + #[test] + fn reports_a_failing_child_without_erroring() { + let mut cmd = Command::new(std::env::current_exe().unwrap()); + cmd.arg("--this-flag-does-not-exist"); + let m = run_measured(&mut cmd).unwrap(); + assert!(!m.success(), "expected a non-zero exit"); + } + + #[test] + fn missing_binary_is_an_error_not_a_panic() { + let mut cmd = Command::new("driven-bench-definitely-not-a-real-binary"); + assert!(run_measured(&mut cmd).is_err()); + } +} diff --git a/crates/driven-bench/src/report.rs b/crates/driven-bench/src/report.rs new file mode 100644 index 00000000..3a861e2f --- /dev/null +++ b/crates/driven-bench/src/report.rs @@ -0,0 +1,331 @@ +//! Report rendering: a markdown table for humans, raw JSON for trending. +//! +//! Both files are written per run under `bench/results/`, named by UTC +//! timestamp. The markdown is what goes in a PR or a release note; the JSON +//! keeps every field (including the ones the table omits for width) so a later +//! run can be diffed against an earlier one without re-running anything. + +use std::path::{Path, PathBuf}; +use std::time::{SystemTime, UNIX_EPOCH}; + +use anyhow::{Context, Result}; +use serde::Serialize; + +use crate::fixture::{human_bytes, FixtureSpec}; +use crate::tools::{PhaseResult, Tool}; + +/// One fixture's worth of measurements. +#[derive(Debug, Clone, Serialize)] +pub struct ScenarioReport { + pub spec: FixtureSpec, + pub results: Vec, +} + +/// The whole run. +#[derive(Debug, Clone, Serialize)] +pub struct RunReport { + /// UTC timestamp, `YYYY-MM-DDTHH:MM:SSZ`. + pub started_at: String, + /// The scale name the run was invoked with. + pub scale: String, + /// The fixture seed, so a re-run can reproduce the same trees. + pub seed: u64, + /// OS + architecture the numbers were produced on. Comparing across hosts + /// is meaningless, so the report always says which host it was. + pub host: String, + /// Logical CPU count, for context on the concurrency columns. + pub cpus: usize, + /// The Driven version under test. + pub driven_version: String, + /// The rclone build, when rclone took part. + pub rclone_version: Option, + /// Which tools ran. + pub tools: Vec, + pub scenarios: Vec, +} + +impl RunReport { + /// Whether every measured phase succeeded. + pub fn all_ok(&self) -> bool { + self.scenarios + .iter() + .flat_map(|s| s.results.iter()) + .all(|r| r.ok) + } + + /// Renders the human-facing markdown report. + pub fn to_markdown(&self) -> String { + let mut out = String::new(); + out.push_str("# Driven benchmark run\n\n"); + out.push_str(&format!("- **When (UTC):** {}\n", self.started_at)); + out.push_str(&format!( + "- **Scale:** {} (seed {})\n", + self.scale, self.seed + )); + out.push_str(&format!( + "- **Host:** {} ({} logical CPUs)\n", + self.host, self.cpus + )); + out.push_str(&format!("- **Driven:** {}\n", self.driven_version)); + if let Some(version) = &self.rclone_version { + out.push_str(&format!("- **rclone:** {version}\n")); + } + out.push('\n'); + + for scenario in &self.scenarios { + let spec = &scenario.spec; + out.push_str(&format!("## {} fixture\n\n", spec.shape)); + out.push_str(&format!( + "{} files, {} total{}\n\n", + spec.files, + human_bytes(spec.total_bytes()), + match spec.shape { + crate::fixture::Shape::TinyDeep => + format!(", nested {} directories deep", spec.depth), + crate::fixture::Shape::Huge => String::new(), + } + )); + out.push_str( + "| Tool | Phase | Wall s | Scan s | MiB/s | files/s | Files | Bytes | API calls | CPU s | Peak RSS | Conc | Notes |\n", + ); + out.push_str( + "| --- | --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | --- |\n", + ); + for r in &scenario.results { + out.push_str(&format!( + "| {} | {} | {:.1} | {} | {} | {} | {} | {} | {} | {} | {} | {} | {} |\n", + r.tool, + r.phase, + r.wall_secs, + opt_f(r.scan_secs, 1), + // Two places: on the tiny-files shape the byte rate is a + // small fraction of a MiB/s and one place rounds it to 0.0. + opt_f(r.mib_per_sec(), 2), + opt_f(r.files_per_sec(), 1), + opt_u(r.files_transferred), + r.bytes_transferred.map(human_bytes).unwrap_or_else(dash), + opt_u(r.api_calls), + opt_f(r.cpu_secs, 1), + r.peak_rss_bytes.map(human_bytes).unwrap_or_else(dash), + opt_u(r.concurrency), + r.detail.clone().unwrap_or_else(|| { + if r.ok { + String::new() + } else { + "FAILED".to_string() + } + }), + )); + } + out.push('\n'); + } + + out.push_str("## Reading these numbers\n\n"); + out.push_str( + "The two tools do different amounts of work, on purpose - see `bench/README.md`, \ + \"What is and is not apples-to-apples\". In short: Driven maintains a local state \ + database and hashes file content, which costs it time on the cold phase and buys it \ + precision on the incremental phase; rclone compares size and modification time and \ + keeps no database. `API calls` is instrumented inside Driven's Drive client and has \ + no rclone equivalent, so a blank cell there means \"not measurable\", not zero.\n\n\ + `Scan s` is the time Driven spent walking and hashing before the first upload \ + started, so `Wall s - Scan s` is the upload half. That split is what says WHERE a \ + slow run went: a large scan share means the local walk is the constraint, a small \ + one means Drive round-trips are. rclone interleaves listing with transferring and \ + exposes no such boundary, so its cell is blank.\n", + ); + out + } + + /// Writes `/.md` and `/.json`, returning + /// both paths. + pub fn write_to(&self, dir: &Path) -> Result<(PathBuf, PathBuf)> { + std::fs::create_dir_all(dir) + .with_context(|| format!("creating the results directory {}", dir.display()))?; + let stem = self.started_at.replace([':', '-'], ""); + let md = dir.join(format!("{stem}.md")); + let json = dir.join(format!("{stem}.json")); + std::fs::write(&md, self.to_markdown()) + .with_context(|| format!("writing {}", md.display()))?; + std::fs::write(&json, serde_json::to_string_pretty(self)?) + .with_context(|| format!("writing {}", json.display()))?; + Ok((md, json)) + } +} + +fn dash() -> String { + "-".to_string() +} + +fn opt_f(value: Option, places: usize) -> String { + value.map_or_else(dash, |v| format!("{v:.places$}")) +} + +fn opt_u(value: Option) -> String { + value.map_or_else(dash, |v| v.to_string()) +} + +/// Formats "now" as `YYYY-MM-DDTHH:MM:SSZ`. +/// +/// Hand-rolled rather than pulling in a date library: the harness needs exactly +/// one timestamp and the crate is deliberately dependency-light. +pub fn utc_timestamp() -> String { + let secs = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_secs()) + .unwrap_or(0); + let (days, rem) = ((secs / 86_400) as i64, secs % 86_400); + let (year, month, day) = civil_from_days(days); + format!( + "{year:04}-{month:02}-{day:02}T{:02}:{:02}:{:02}Z", + rem / 3600, + (rem % 3600) / 60, + rem % 60 + ) +} + +/// Howard Hinnant's `civil_from_days`: days since the Unix epoch to (y, m, d). +fn civil_from_days(days: i64) -> (i64, u32, u32) { + let z = days + 719_468; + let era = if z >= 0 { z } else { z - 146_096 } / 146_097; + let doe = (z - era * 146_097) as u64; + let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365; + let y = yoe as i64 + era * 400; + let doy = doe - (365 * yoe + yoe / 4 - yoe / 100); + let mp = (5 * doy + 2) / 153; + let d = (doy - (153 * mp + 2) / 5 + 1) as u32; + let m = if mp < 10 { mp + 3 } else { mp - 9 } as u32; + (if m <= 2 { y + 1 } else { y }, m, d) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::fixture::Shape; + use crate::tools::Phase; + + fn spec() -> FixtureSpec { + FixtureSpec { + shape: Shape::TinyDeep, + files: 100, + huge_file_bytes: 0, + depth: 4, + seed: 3, + } + } + + fn result(tool: Tool, ok: bool) -> PhaseResult { + PhaseResult { + tool, + phase: Phase::Cold, + wall_secs: 10.0, + cpu_secs: Some(2.5), + peak_rss_bytes: Some(1 << 20), + files_transferred: Some(100), + bytes_transferred: Some(10 * 1_048_576), + api_calls: (tool == Tool::Driven).then_some(120), + concurrency: Some(8), + scan_secs: (tool == Tool::Driven).then_some(3.0), + ok, + detail: None, + } + } + + fn report(results: Vec) -> RunReport { + RunReport { + started_at: "2026-07-25T16:00:00Z".into(), + scale: "smoke".into(), + seed: 3, + host: "windows/x86_64".into(), + cpus: 8, + driven_version: "2.3.0".into(), + rclone_version: Some("rclone v1.74.4".into()), + tools: vec![Tool::Driven, Tool::Rclone], + scenarios: vec![ScenarioReport { + spec: spec(), + results, + }], + } + } + + #[test] + fn markdown_has_a_row_per_result_and_names_both_tools() { + let md = report(vec![result(Tool::Driven, true), result(Tool::Rclone, true)]).to_markdown(); + assert!(md.contains("| driven | cold |")); + assert!(md.contains("| rclone | cold |")); + assert!(md.contains("rclone v1.74.4")); + assert!(md.contains("tiny-deep fixture")); + } + + #[test] + fn an_unmeasurable_cell_renders_as_a_dash_not_a_zero() { + let md = report(vec![result(Tool::Rclone, true)]).to_markdown(); + let row = md + .lines() + .find(|l| l.starts_with("| rclone |")) + .expect("rclone row"); + // rclone has no request counter; the API column must be a dash so the + // table never claims it made zero requests. + assert!(row.contains(" - |"), "expected a dash cell in: {row}"); + } + + #[test] + fn the_scan_column_shows_drivens_split_and_stays_blank_for_rclone() { + let md = report(vec![result(Tool::Driven, true), result(Tool::Rclone, true)]).to_markdown(); + assert!(md.contains("| Scan s |"), "the table must carry the column"); + let driven = md + .lines() + .find(|l| l.starts_with("| driven |")) + .expect("driven row"); + assert!( + driven.contains("| 3.0 |"), + "driven must report its scan time, got: {driven}" + ); + let rclone = md + .lines() + .find(|l| l.starts_with("| rclone |")) + .expect("rclone row"); + // rclone interleaves listing with transferring; the cell must be a dash, + // never a zero that would read as "no scan needed". + assert!( + !rclone.contains("| 0.0 |"), + "rclone must not claim a zero scan, got: {rclone}" + ); + } + + #[test] + fn all_ok_is_false_when_any_phase_failed() { + assert!(report(vec![result(Tool::Driven, true)]).all_ok()); + assert!(!report(vec![result(Tool::Driven, false)]).all_ok()); + } + + #[test] + fn writes_both_a_markdown_and_a_json_file() { + let dir = tempfile::tempdir().unwrap(); + let (md, json) = report(vec![result(Tool::Driven, true)]) + .write_to(dir.path()) + .unwrap(); + assert!(md.exists() && json.exists()); + let parsed: serde_json::Value = + serde_json::from_str(&std::fs::read_to_string(&json).unwrap()).unwrap(); + assert_eq!(parsed["scale"], "smoke"); + assert_eq!(parsed["scenarios"][0]["results"][0]["tool"], "driven"); + } + + #[test] + fn timestamp_is_iso_8601_utc() { + let ts = utc_timestamp(); + assert_eq!(ts.len(), 20, "unexpected timestamp {ts}"); + assert!(ts.ends_with('Z')); + assert_eq!(&ts[4..5], "-"); + } + + #[test] + fn civil_from_days_matches_known_dates() { + assert_eq!(civil_from_days(0), (1970, 1, 1)); + assert_eq!(civil_from_days(19_000), (2022, 1, 8)); + // A leap day and the day after it, the classic off-by-one. + assert_eq!(civil_from_days(18_321), (2020, 2, 29)); + assert_eq!(civil_from_days(18_322), (2020, 3, 1)); + } +} diff --git a/crates/driven-bench/src/tools.rs b/crates/driven-bench/src/tools.rs new file mode 100644 index 00000000..5eed52fe --- /dev/null +++ b/crates/driven-bench/src/tools.rs @@ -0,0 +1,462 @@ +//! The tools under test, and how each one is driven and measured. +//! +//! Both tools run as child processes measured by [`crate::procstat`], upload +//! into the same Drive folder tree under the same account, and are given the +//! same source directory. What differs is unavoidable and is reported rather +//! than hidden - see bench/README.md, "What is and is not apples-to-apples". + +use std::path::Path; +use std::process::Command; +use std::time::Duration; + +use anyhow::{Context, Result}; +use serde::Serialize; + +use crate::agent::{AgentMetrics, METRICS_PREFIX}; +use crate::creds::BenchCreds; +use crate::procstat::run_measured; + +/// A tool the suite can benchmark. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, clap::ValueEnum)] +#[serde(rename_all = "kebab-case")] +pub enum Tool { + /// Driven's real engine, run headlessly in a child process. + Driven, + /// `rclone copy` against a Drive remote built from the same credentials. + Rclone, +} + +impl Tool { + /// The stable slug used in folder names, JSON and report tables. + pub fn slug(self) -> &'static str { + match self { + Tool::Driven => "driven", + Tool::Rclone => "rclone", + } + } +} + +impl std::fmt::Display for Tool { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(self.slug()) + } +} + +/// Which half of a scenario a measurement belongs to. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[serde(rename_all = "kebab-case")] +pub enum Phase { + /// First upload of the tree into an empty destination. + Cold, + /// Re-run after a small deterministic change to the same tree. + Incremental, +} + +impl Phase { + pub fn slug(self) -> &'static str { + match self { + Phase::Cold => "cold", + Phase::Incremental => "incremental", + } + } +} + +impl std::fmt::Display for Phase { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(self.slug()) + } +} + +/// One measured tool run. +#[derive(Debug, Clone, Serialize)] +pub struct PhaseResult { + pub tool: Tool, + pub phase: Phase, + /// Wall-clock seconds for the whole child process. + pub wall_secs: f64, + /// CPU seconds, when the OS attributed them. + pub cpu_secs: Option, + /// Peak working set in bytes, when the OS attributed it. + pub peak_rss_bytes: Option, + /// Files the tool reported transferring. + pub files_transferred: Option, + /// Bytes the tool reported transferring. + pub bytes_transferred: Option, + /// Drive requests, when the tool can be instrumented. Only Driven can be: + /// rclone exposes no request counter, so this stays `None` for it rather + /// than being guessed from transfer counts. + pub api_calls: Option, + /// Upload concurrency the tool ran at, for the report to show alongside the + /// timings. + pub concurrency: Option, + /// Seconds spent walking and hashing before any upload began, when the tool + /// can report it. Only Driven can: it is the number that says whether a slow + /// run was bound by the local scan or by Drive round-trips. rclone + /// interleaves listing with transferring and exposes no such boundary, so + /// this stays `None` for it rather than being invented. + pub scan_secs: Option, + /// Whether the child exited zero. + pub ok: bool, + /// A short human-facing note - the failure reason when `ok` is false. + pub detail: Option, +} + +impl PhaseResult { + /// Throughput in mebibytes per second, or `None` when nothing moved. + pub fn mib_per_sec(&self) -> Option { + let bytes = self.bytes_transferred?; + if bytes == 0 || self.wall_secs <= 0.0 { + return None; + } + Some(bytes as f64 / 1_048_576.0 / self.wall_secs) + } + + /// Files per second, or `None` when nothing moved. + pub fn files_per_sec(&self) -> Option { + let files = self.files_transferred?; + if files == 0 || self.wall_secs <= 0.0 { + return None; + } + Some(files as f64 / self.wall_secs) + } +} + +/// Builds a failed result carrying the reason, so one broken tool degrades to a +/// row that says why instead of aborting the whole suite. +fn failed(tool: Tool, phase: Phase, wall: Duration, detail: String) -> PhaseResult { + PhaseResult { + tool, + phase, + wall_secs: wall.as_secs_f64(), + cpu_secs: None, + peak_rss_bytes: None, + files_transferred: None, + bytes_transferred: None, + api_calls: None, + concurrency: None, + scan_secs: None, + ok: false, + detail: Some(detail), + } +} + +/// Extracts the agent's metrics line from a child's stdout. +pub fn parse_agent_metrics(stdout: &str) -> Option { + stdout + .lines() + .filter_map(|line| line.trim().strip_prefix(METRICS_PREFIX)) + .next_back() + .and_then(|json| serde_json::from_str(json).ok()) +} + +/// Runs one Driven phase by re-invoking this binary's hidden `agent-sync` +/// subcommand, so the engine is measured as a child exactly like rclone is. +pub fn run_driven( + phase: Phase, + source: &Path, + dest_folder_id: &str, + state_db: &Path, +) -> Result { + let exe = std::env::current_exe().context("locating the bench binary")?; + let mut cmd = Command::new(exe); + cmd.arg("agent-sync") + .arg("--source") + .arg(source) + .arg("--dest-folder-id") + .arg(dest_folder_id) + .arg("--state-db") + .arg(state_db) + // Keep the child's log volume bounded: at info level a million-file run + // would spend real time formatting lines nobody reads. + .env("RUST_LOG", "warn"); + + let m = run_measured(&mut cmd)?; + if !m.success() { + return Ok(failed( + Tool::Driven, + phase, + m.wall, + format!( + "driven agent exited {:?}: {}", + m.exit_code, + last_lines(&m.stderr, 3) + ), + )); + } + + let Some(agent) = parse_agent_metrics(&m.stdout) else { + return Ok(failed( + Tool::Driven, + phase, + m.wall, + "driven agent printed no metrics line".to_string(), + )); + }; + + // Prefer the durable activity-row counts; fall back to the progress stream, + // which can lag on a large run. + let files = if agent.logged_files_uploaded > 0 { + agent.logged_files_uploaded + } else { + agent.files_done + }; + let bytes = if agent.logged_bytes_uploaded > 0 { + agent.logged_bytes_uploaded + } else { + agent.bytes_done + }; + + Ok(PhaseResult { + tool: Tool::Driven, + phase, + wall_secs: m.wall.as_secs_f64(), + cpu_secs: m.cpu.map(|c| c.as_secs_f64()), + peak_rss_bytes: m.peak_rss_bytes, + files_transferred: Some(files), + bytes_transferred: Some(bytes), + api_calls: Some(agent.api.total), + concurrency: Some(driven_core::adaptive::default_pool_size() as u64), + scan_secs: agent.scan_ms.map(|ms| ms as f64 / 1000.0), + ok: agent.errors == 0, + detail: (agent.errors > 0).then(|| format!("{} executor error(s)", agent.errors)), + }) +} + +/// Writes an rclone config for a Drive remote rooted at `folder_id`. +/// +/// rclone will not accept a token whose `access_token` is empty (it treats the +/// whole token as unparseable and reports "there's no refresh token"), but it is +/// perfectly happy to refresh a NON-empty token that has already expired. So the +/// config carries a placeholder access token with an expiry in the past, and +/// rclone mints a real one from the refresh token on first use - no separate +/// token-minting request, and the same credential Driven uses. +/// +/// The file lands in a caller-owned temp directory (rclone rewrites it with the +/// refreshed token) and is never logged. +pub fn write_rclone_config(path: &Path, creds: &BenchCreds, folder_id: &str) -> Result<()> { + let token = format!( + r#"{{"access_token":"driven-bench-placeholder","token_type":"Bearer","refresh_token":"{}","expiry":"2000-01-01T00:00:00Z"}}"#, + creds.refresh_token + ); + let config = format!( + "[bench]\n\ + type = drive\n\ + client_id = {}\n\ + client_secret = {}\n\ + scope = drive\n\ + root_folder_id = {}\n\ + token = {}\n", + creds.client_id, creds.client_secret, folder_id, token + ); + std::fs::write(path, config) + .with_context(|| format!("writing the rclone config to {}", path.display()))?; + Ok(()) +} + +/// Parses the `stats` object out of rclone's JSON log. +/// +/// rclone prints one final stats record at the end of a run; with +/// `--stats-log-level NOTICE` it is emitted even without `-v`, which matters +/// because the per-file `-v` lines would be a million lines long on the tiny +/// files shape. Returns `(bytes, transfers, errors)`. +pub fn parse_rclone_stats(stderr: &str) -> Option<(u64, u64, u64)> { + stderr + .lines() + .filter_map(|line| serde_json::from_str::(line.trim()).ok()) + .filter_map(|value| value.get("stats").cloned()) + .rfind(|stats| stats.is_object()) + .map(|stats| { + let get = |key: &str| stats.get(key).and_then(|v| v.as_u64()).unwrap_or(0); + (get("bytes"), get("transfers"), get("errors")) + }) +} + +/// Runs one rclone phase. +pub fn run_rclone( + phase: Phase, + binary: &Path, + config: &Path, + source: &Path, + transfers: u64, +) -> Result { + let mut cmd = Command::new(binary); + cmd.arg("--config") + .arg(config) + .arg("copy") + .arg(source) + .arg("bench:") + .arg("--use-json-log") + // A single final stats record instead of a per-file log. + .arg("--stats") + .arg("100000h") + .arg("--stats-log-level") + .arg("NOTICE") + .arg("--transfers") + .arg(transfers.to_string()) + .arg("--checkers") + .arg(transfers.to_string()); + + let m = run_measured(&mut cmd)?; + let stats = parse_rclone_stats(&m.stderr); + if !m.success() { + return Ok(failed( + Tool::Rclone, + phase, + m.wall, + format!( + "rclone exited {:?}: {}", + m.exit_code, + last_lines(&m.stderr, 3) + ), + )); + } + + let (bytes, transfers_done, errors) = stats.unwrap_or((0, 0, 0)); + Ok(PhaseResult { + tool: Tool::Rclone, + phase, + wall_secs: m.wall.as_secs_f64(), + cpu_secs: m.cpu.map(|c| c.as_secs_f64()), + peak_rss_bytes: m.peak_rss_bytes, + files_transferred: Some(transfers_done), + bytes_transferred: Some(bytes), + // rclone exposes no request counter. + api_calls: None, + concurrency: Some(transfers), + // rclone interleaves listing and transferring; there is no scan phase + // to report, so the column stays empty instead of guessing. + scan_secs: None, + ok: errors == 0, + detail: (errors > 0).then(|| format!("{errors} rclone error(s)")), + }) +} + +/// Reports the rclone version string, for the report header. +pub fn rclone_version(binary: &Path) -> Option { + let out = Command::new(binary).arg("version").output().ok()?; + let text = String::from_utf8_lossy(&out.stdout); + text.lines().next().map(|l| l.trim().to_string()) +} + +/// Finds the rclone binary: an explicit path, then `PATH`. +pub fn find_rclone(explicit: Option<&Path>) -> Option { + if let Some(path) = explicit { + return path.exists().then(|| path.to_path_buf()); + } + let name = if cfg!(windows) { + "rclone.exe" + } else { + "rclone" + }; + std::env::var_os("PATH").and_then(|paths| { + std::env::split_paths(&paths) + .map(|dir| dir.join(name)) + .find(|candidate| candidate.is_file()) + }) +} + +/// The last `n` non-empty lines of a captured stream, for error messages. +fn last_lines(text: &str, n: usize) -> String { + let lines: Vec<&str> = text.lines().filter(|l| !l.trim().is_empty()).collect(); + let start = lines.len().saturating_sub(n); + lines[start..].join(" | ") +} + +#[cfg(test)] +mod tests { + use super::*; + + /// A real final stats line captured from `rclone v1.74.4`. + const RCLONE_JSON_LOG: &str = r#"{"time":"2026-07-25T16:34:13-05:00","level":"notice","msg":"there was a message"} +{"time":"2026-07-25T16:34:13-05:00","level":"info","msg":"\nTransferred: 292.969 KiB\n","stats":{"bytes":300000,"checks":0,"deletes":0,"elapsedTime":0.008,"errors":0,"listed":5,"speed":0,"totalBytes":300000,"totalTransfers":3,"transfers":3},"source":"accounting/stats.go:551"}"#; + + #[test] + fn parses_rclone_stats_from_the_json_log() { + let (bytes, transfers, errors) = parse_rclone_stats(RCLONE_JSON_LOG).expect("stats"); + assert_eq!(bytes, 300_000); + assert_eq!(transfers, 3); + assert_eq!(errors, 0); + } + + #[test] + fn rclone_stats_are_none_when_no_stats_record_was_printed() { + assert!(parse_rclone_stats("not json at all\n{\"level\":\"info\"}").is_none()); + } + + #[test] + fn rclone_errors_are_read_from_the_stats_record() { + let log = r#"{"msg":"x","stats":{"bytes":1,"transfers":0,"errors":2}}"#; + assert_eq!(parse_rclone_stats(log), Some((1, 0, 2))); + } + + #[test] + fn agent_metrics_are_taken_from_the_last_marker_line() { + let stdout = format!( + "some log noise\n{METRICS_PREFIX}{{\"engine_ms\":1}}\n{METRICS_PREFIX}{{\"engine_ms\":2}}\n" + ); + assert_eq!(parse_agent_metrics(&stdout).unwrap().engine_ms, 2); + } + + #[test] + fn agent_metrics_are_none_without_a_marker() { + assert!(parse_agent_metrics("nothing here").is_none()); + } + + #[test] + fn rclone_config_embeds_an_expired_placeholder_so_rclone_refreshes() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("rclone.conf"); + let creds = BenchCreds { + client_id: "cid".into(), + client_secret: "csec".into(), + refresh_token: "rt-value".into(), + }; + write_rclone_config(&path, &creds, "folder-1").unwrap(); + let text = std::fs::read_to_string(&path).unwrap(); + assert!(text.contains("root_folder_id = folder-1")); + assert!(text.contains("\"refresh_token\":\"rt-value\"")); + assert!( + text.contains("\"expiry\":\"2000-01-01T00:00:00Z\""), + "the placeholder must already be expired so rclone refreshes it" + ); + assert!( + !text.contains("\"access_token\":\"\""), + "rclone rejects an empty access_token outright" + ); + } + + #[test] + fn throughput_helpers_handle_the_nothing_moved_case() { + let mut r = PhaseResult { + tool: Tool::Driven, + phase: Phase::Incremental, + wall_secs: 2.0, + cpu_secs: None, + peak_rss_bytes: None, + files_transferred: Some(0), + bytes_transferred: Some(0), + api_calls: None, + concurrency: None, + scan_secs: None, + ok: true, + detail: None, + }; + assert!(r.mib_per_sec().is_none()); + assert!(r.files_per_sec().is_none()); + + r.bytes_transferred = Some(2 * 1_048_576); + r.files_transferred = Some(4); + assert_eq!(r.mib_per_sec().unwrap(), 1.0); + assert_eq!(r.files_per_sec().unwrap(), 2.0); + } + + #[test] + fn last_lines_trims_to_the_tail() { + assert_eq!(last_lines("a\n\nb\nc\n", 2), "b | c"); + } + + #[test] + fn find_rclone_rejects_a_missing_explicit_path() { + assert!(find_rclone(Some(Path::new("no/such/rclone"))).is_none()); + } +} diff --git a/justfile b/justfile index c7b50e3b..367bc35f 100644 --- a/justfile +++ b/justfile @@ -62,6 +62,26 @@ chaos-soak args="--duration 30m": $env:DRIVEN_CHAOS_SOAK="1"; cargo run -p driven-chaos -- run-all --hermetic cargo run -p driven-chaos -- fuzz {{args}} +# --- benchmark suite (bench/README.md) --- + +# Compare Driven's real engine against rclone on a live Drive account. Uploads +# REAL bytes and takes real time - see bench/README.md for scales and costs. +# `just bench smoke` proves the pipeline in a few minutes; the default `small` +# scale uploads ~610 MiB per tool. Needs rclone on PATH and the DRIVEN_E2E_* +# credentials (the gitignored .env.test at the repo root is loaded for you). +bench scale="small" args="": + cargo run --release -p driven-bench -- run --scale {{scale}} {{args}} + +# Materialise a benchmark fixture without uploading anything, e.g. +# `just bench-fixture tiny-deep small`. +bench-fixture shape scale="small": + cargo run --release -p driven-bench -- fixture build --shape {{shape}} --scale {{scale}} + +# Delete every cached benchmark fixture under target/bench-fixtures/ (the `full` +# scale leaves ~10 GB behind). +bench-fixture-clean: + cargo run -p driven-bench -- fixture clean + lint: cargo fmt --all -- --check cargo clippy --workspace --all-targets -- -D warnings @@ -84,7 +104,7 @@ deny: # (.github/workflows/coverage.yml). Needs `cargo install cargo-llvm-cov`. For # the exact parsed percentages CI compares against main, run ./scripts/coverage.sh. coverage: - cargo llvm-cov --workspace --exclude src-tauri --exclude driven-chaos --summary-only + cargo llvm-cov --workspace --exclude src-tauri --exclude driven-chaos --exclude driven-bench --summary-only pnpm --dir ui run test:coverage # --- sqlx dev helpers (need `cargo install sqlx-cli`) --- diff --git a/scripts/coverage.sh b/scripts/coverage.sh index 20439257..4c85e385 100755 --- a/scripts/coverage.sh +++ b/scripts/coverage.sh @@ -10,6 +10,7 @@ cd "$(dirname "$0")/.." echo "== Rust (library crates) ==" SQLX_OFFLINE=true cargo llvm-cov --workspace --exclude src-tauri --exclude driven-chaos \ + --exclude driven-bench \ --summary-only --json --output-path coverage-rust.json RUST_PCT=$(jq '.data[0].totals.lines.percent' coverage-rust.json)