diff --git a/.dockerignore b/.dockerignore index ce8dfe4..643ac8f 100644 --- a/.dockerignore +++ b/.dockerignore @@ -13,6 +13,7 @@ !tidal-net !tidal-server !tidalctl +!tidal-stress !applications # Re-exclude heavy / generated / secret paths nested inside the allowed dirs. diff --git a/CLAUDE.md b/CLAUDE.md index fc55f70..12d3305 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -96,13 +96,13 @@ Dev servers use port range **59520–59529** (e.g. `site/` on 59520). ## Repository Structure This repository is a standalone Cargo workspace (members: `tidal`, `tidal-net`, `tidalctl`, -`tidal-server`, and the `applications/` consumers). **Documentation has exactly two homes:** +`tidal-server`, `tidal-stress`, and the `applications/` consumers). **Documentation has exactly two homes:** the top-level `*.md` files and `docs/`. Do not create per-crate doc mirrors (e.g. `tidal/docs/`, `tidal/ai-lookup/`, `tidal/site/`) — those were a stale duplicate and were consolidated away. ``` . # Workspace root — canonical docs + config -├── Cargo.toml # Workspace manifest (8 members) +├── Cargo.toml # Workspace manifest (9 members) ├── CLAUDE.md AGENTS.md README.md CONTRIBUTING.md CHANGELOG.md ├── VISION.md USE_CASES.md SEQUENCE.md ARCHITECTURE.md ├── API.md QUICKSTART.md CODING_GUIDELINES.md thoughts.md @@ -124,6 +124,7 @@ the top-level `*.md` files and `docs/`. Do not create per-crate doc mirrors (e.g ├── tidal-net/ # Network transport primitives (gRPC, WAL shipping) ├── tidal-server/ # Standalone Axum HTTP server (standalone + cluster modes) ├── tidalctl/ # CLI for inspecting persisted databases +├── tidal-stress/ # Open-loop HTTP load generator + capacity ramp (thepeach feed workload) ├── applications/ # Example consumers (forage, iknowyou) └── site/ # Public marketing site (Next.js, dev on 59520) ``` diff --git a/Cargo.lock b/Cargo.lock index e37957d..51b6488 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3605,6 +3605,21 @@ dependencies = [ "utoipa", ] +[[package]] +name = "tidal-stress" +version = "0.1.0" +dependencies = [ + "clap", + "rand 0.9.2", + "reqwest", + "serde", + "serde_json", + "thiserror 2.0.18", + "tokio", + "tracing", + "tracing-subscriber", +] + [[package]] name = "tidalctl" version = "0.1.0" diff --git a/Cargo.toml b/Cargo.toml index a3f2b4f..ff8c71d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -4,6 +4,7 @@ members = [ "tidal-net", "tidalctl", "tidal-server", + "tidal-stress", "applications/forage/engine", "applications/forage/server", "applications/forage/embedder", diff --git a/docker/stress/Dockerfile b/docker/stress/Dockerfile new file mode 100644 index 0000000..2bfc229 --- /dev/null +++ b/docker/stress/Dockerfile @@ -0,0 +1,37 @@ +# tidal-stress load generator image. +# +# Build context is this repository's root (the workspace Cargo.toml / Cargo.lock +# and every member manifest live there). Build from the repo root: +# +# docker build -f docker/stress/Dockerfile -t tidaldb:stress . +# +# tidal-stress is a pure-Rust HTTP client (tokio + reqwest/rustls): it depends on +# NONE of the engine crates, so this build never compiles tidal/tidal-net/usearch +# and needs no protoc/g++. Pin the builder to bookworm so its glibc matches the +# runtime; trixie's vectorized-math libmvec is handled by the trixie runtime below. +FROM rust:1.91-bookworm AS builder +ARG DEBIAN_FRONTEND=noninteractive +WORKDIR /app + +# Copy the full (already-pruned by .dockerignore) workspace and build only the +# load generator. The unified workspace requires every member manifest present to +# resolve metadata, so copy the whole context rather than a fragile subset. +COPY . . +RUN cargo build -p tidal-stress --release --locked + +FROM debian:trixie-slim +ARG DEBIAN_FRONTEND=noninteractive +WORKDIR /srv + +# ca-certificates only matters if a target is https:// (in-cluster targets are +# plaintext http://); kept for completeness. No libstdc++ (no C++ in this crate). +RUN useradd --system --home /srv stress && \ + apt-get update && apt-get install -y --no-install-recommends ca-certificates && \ + rm -rf /var/lib/apt/lists/* + +COPY --from=builder /app/target/release/tidal-stress /usr/local/bin/tidal-stress + +USER stress + +# ENTRYPOINT is the bare binary so the Job supplies the full ramp/mix/target argv. +ENTRYPOINT ["tidal-stress"] diff --git a/docs/ops/stress-test-thepeach.md b/docs/ops/stress-test-thepeach.md new file mode 100644 index 0000000..6426056 --- /dev/null +++ b/docs/ops/stress-test-thepeach.md @@ -0,0 +1,140 @@ +# Capacity stress test — thepeach feed workload (100k DAU) + +Run with the `tidal-stress` crate (an open-loop, coordinated-omission-corrected +load generator) against the live 3-region multi-process cluster on k3s +(`tidaldb` namespace; 3 pods, one region each, cpu limit `"2"` per pod). Date: +2026-06-10. + +> **TL;DR.** Reads scale trivially (feed p99 <15ms at thousands of rps). The +> *replicated* write path (`/signals` → single leader) saturates at **~90 +> signals/s** — far short of a 100k-DAU peak — because every write funnels to one +> leader, gated by a ~2-worker write pool, each write blocking on WAL fsync + a +> synchronous cross-region gRPC ship. The *sharded* write path +> (`/sharded/signals`, hash-distributed across all 3 regions) cleanly sustains +> **3,669 signals/s at 0 errors and only ~27% cluster CPU** — that is essentially +> the 100k-DAU evening peak, with headroom. Overload degrades **gracefully** +> (HTTP 429 backpressure; zero pod restarts, replication stays caught up). +> **For thepeach's actual plan — embedding the engine in-process — 100k DAU is +> not even a question** (in-process signal write ≈ 82ns ⇒ ~12M signals/s/core). + +## The workload + +thepeach is a character.ai-style AI-companion feed app; its tidaldb integration +is *planned* (experiment E2 / roadmap R8: feed ranking via named tidaldb +profiles). The test models that future hot path — a feed session — mapped onto +the deployed schema (signals `view`/`like`/`skip`, 128-dim `content_vector`): + +- A session = home feed (24 tiles/page, infinite scroll) → `view`/`like`/`skip` + on tiles → occasional search → repeat. Per thepeach's user-graph spec, signals + outnumber feed reads by 1–2 orders of magnitude ("10–1000+ events/session"). +- Default mix (`peach`): ~78% `view`, ~10% `skip`, ~4% `like`, ~6.5% feed reads, + ~1.6% search, a trickle of item/embedding registration. Hot-content power-law + entity selection (a few viral items get most signals). +- 100k-DAU model: ~5 sessions/user/day × ~5 feed pages × ~15 signal-equivalents + ⇒ **~780 signals/s average, ~3,900 signals/s at a 5× evening peak**; feed reads + ~1/20th of that. + +## Methodology + +`tidal-stress` is **open-loop** (constant arrival rate, not N-workers-in-a-loop), +so a server stall inflates the latency percentiles a closed-loop test would hide +(coordinated-omission correction: latency measured from each request's *intended* +send time). It ramps a throttleable rate low→high, reports per-op p50/p90/p99/p999 ++ max latency and a **backpressure-aware** status breakdown (429 = write +backpressure, 408 = concurrency-queue timeout, 503 = leader/region down), and +prints a capacity verdict translated to supported DAU. Run in-cluster as a k8s +Job (a `kubectl port-forward` serializes everything through the API server and +adds tens of ms — useless for capacity numbers). See `tidal-stress/k8s/stress-job.yaml`. + +## Results + +### Reads — never the bottleneck +`/feed` p50 ~3ms, p99 <15ms; `/search` p50 ~2ms — sustained even at 4,000 rps +total. tidalDB's own RETRIEVE SLA (p99 <50ms) holds comfortably over the cluster +network hop. + +### Replicated write path (`/signals` → leader) — ~90 signals/s +`view` ok/s was pinned at **76, 76, 74, 71** across offered rates of 100→4,000 +rps — that flat line *is* the ceiling (~90 signals/s incl. like/skip). Everything +above it shed as **HTTP 429** (honor-retry backpressure). No 5xx, no timeouts, +reads unaffected, replication lag stayed 0–3. + +| offered (total rps) | signal ok/s | errors | feed p99 | +|---|---|---|---| +| 100 | ~90 | 1.6% (429) | 8ms | +| 500 | ~90 | 73% (429) | 5ms | +| 1,500 | ~90 | 86% (429) | 8ms | +| 4,000 | ~90 | 89% (429) | 9ms | + +Bottleneck: a single leader (us-east) + a write pool of `available_parallelism` +clamped to [2,8] (≈2 on a cpu-`"2"` pod), each write blocking on WAL fsync **and +a synchronous gRPC ship to both siblings** (~178ms/write). ⇒ ~2k DAU at peak, +~11k DAU at average. **Not enough for 100k DAU on this path.** + +### Sharded write path (`/sharded/signals`) — 3,669 signals/s, 0 errors +Hash-distributes writes across all 3 regions (applied to the owning region's +local store; **not replicated** — reads use `/sharded/*` scatter-gather to merge +across shards). One generator pod: + +| offered (total rps) | signal ok/s | errors | view p99 | cluster cpu/region | +|---|---|---|---|---| +| 100 | 92 | 0% | 21ms | idle | +| 500 | 458 | 0% | 20ms | ~140m | +| 1,500 | 1,378 | 0% | 22ms | ~300m | +| 4,000 | **3,669** | **0%** | 31ms | ~540m (27%) | + +The knee was **not reached** — the cluster was only ~27% utilized at 3,669 +signals/s ⇒ **~94k DAU at peak, ~470k DAU at average**. ≈ handles the 100k-DAU +peak with headroom. + +### Pushing past one generator — a connection wall, not the engine +A single generator pod tops out ~4–5k rps (its own CPU/connection limit). A +3-pod parallel fleet (aggregate >5k rps) **collapsed with 40s connection +timeouts + 503s while the cluster CPU stayed idle** — successful writes were +still fast (p50 ~12ms) but new connections hung. This is a **connection- +establishment wall** (k8s ClusterIP/conntrack/accept-queue under connection +churn from many pods), hit *before* the engine's compute ceiling. The database +engine was the bottleneck for **neither** the clean run nor the collapse. + +## Verdict — can we handle 100k DAU? more? + +| Deployment mode | 100k DAU? | Ceiling found | Headroom | +|---|---|---|---| +| **Embedded engine** (thepeach's plan: `Arc`, `db.signal()`) | **Trivially yes** | in-process write ≈82ns ⇒ ~12M signals/s/core | RAM-bound, not CPU (capacity-planning.md) | +| **HTTP cluster, sharded writes** | **Yes** | ≥3,669 signals/s @ 27% cpu (knee not reached) | ~2–3× compute; connection wall ~5k aggregate rps | +| **HTTP cluster, replicated `/signals`** | **No** | ~90 signals/s (single leader) | needs sharding or more leader cpu | + +**Can we handle more?** On the sharded/embedded paths, yes — substantially. The +replicated single-leader path does not scale writes by design. + +## Recommendations + +1. **thepeach's planned in-process embedding sidesteps all of this** — no HTTP, + no leader funnel, no connection layer; the engine does millions of signals/s + per core. At 100k–10M DAU the constraint is **RAM for the item/embedding + index** (capacity-planning.md), not signal write throughput. thepeach's real + 1536-dim embeddings cost ~2.3× the RAM/item vs the deployed 128-dim + (`~470MB` vs `~200MB` per 100k items). +2. **If using the HTTP cluster for the signal firehose, use `/sharded/*`** (or + raise the leader pod's CPU to lift the write pool toward its 8-worker clamp — + ~4× the replicated ceiling to ~360/s, still far short of peak; the single- + leader replicated model fundamentally doesn't scale writes). +3. **Graceful under overload**: the cluster sheds excess writes as 429 and never + fell over (0 pod restarts across every run, replication stayed caught up) — a + client that honors the ~50ms retry hint degrades cleanly. +4. **To push past ~5k aggregate HTTP rps**: connection discipline (keep-alive + reuse, bounded pools), more region replicas, or front the regions with an + ingress/mesh that tolerates connection churn — the next wall is networking, + not the database. + +## Reproduce + +```bash +# Build + push the image (cross-compile path), then: +kubectl apply -f tidal-stress/k8s/stress-job.yaml # leader path, peach-100k ramp +kubectl logs -f job/tidal-stress -n tidaldb +# Sharded comparison: set --write-path sharded in the args. +``` + +See `tidal-stress/` for the generator (open-loop scheduler in `scheduler.rs`, +the thepeach workload model in `workload.rs`). diff --git a/tidal-stress/Cargo.toml b/tidal-stress/Cargo.toml new file mode 100644 index 0000000..5ebebcc --- /dev/null +++ b/tidal-stress/Cargo.toml @@ -0,0 +1,56 @@ +[package] +name = "tidal-stress" +version = "0.1.0" +edition.workspace = true +rust-version.workspace = true +description = "Open-loop load generator + capacity ramp for tidalDB's HTTP surface (standalone + multi-process cluster), modeling a thepeach-style feed workload" +license.workspace = true + +# ── tidal-crate lint posture (single source of truth) ────────────────────── +# IDENTICAL block across tidaldb / tidal-net / tidal-server / tidalctl. These +# crates deliberately DO NOT inherit `[workspace.lints]`; they hold the embedded +# recommendation DB + its transport/server/CLI to a stricter correctness bar +# (`unsafe_code = forbid`, `clippy::all = deny`, `unwrap_used = deny`). +# `unwrap_used = "deny"` is kept per-crate rather than in `[workspace.lints]` +# because the workspace also hosts the example/consumer crates under +# `applications/` (not held to the engine's bar). Keep these four blocks BYTE-IDENTICAL. +[lints.rust] +unsafe_code = "forbid" + +[lints.clippy] +all = { level = "deny", priority = -1 } +pedantic = { level = "warn", priority = -1 } +nursery = { level = "warn", priority = -1 } +# Justified allows: latency/rate math is pervasive usize<->u64<->f64 casting (a +# load generator is nothing but counters and durations), so the lossy-cast +# pedantic family is allowed exactly as the engine's ranking math allows it. +cast_possible_truncation = "allow" +cast_possible_wrap = "allow" +cast_precision_loss = "allow" +cast_sign_loss = "allow" +module_name_repetitions = "allow" +# Advisory pedantic/nursery style lints that are noise for a CLI load tool (and +# which the engine crates already run at warn, not deny): help-text URLs/flags +# read fine bare; `out.push_str(&format!(..))` is the clearest way to assemble a +# report table; constructors take their config by value; const-ness of trivial +# accessors churns. Correctness lints (clippy::all) stay DENY. +doc_markdown = "allow" +format_push_string = "allow" +needless_pass_by_value = "allow" +missing_const_for_fn = "allow" +map_unwrap_or = "allow" +unwrap_used = "deny" + +[dependencies] +clap = { version = "4.5", features = ["derive", "env"] } +serde = { version = "1", features = ["derive"] } +serde_json = "1" +thiserror = "2" +tokio = { version = "1", features = ["macros", "rt-multi-thread", "sync", "time", "signal"] } +tracing = "0.1" +tracing-subscriber = { version = "0.3", features = ["env-filter"] } +rand = "0.9" +# Async HTTP only (no blocking client): rustls posture matches tidal-server / +# tidal-net — drops the OpenSSL/native-tls system dependency for a clean static +# build. `json` for the typed request/response bodies. +reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"] } diff --git a/tidal-stress/k8s/stress-job.yaml b/tidal-stress/k8s/stress-job.yaml new file mode 100644 index 0000000..a2a0656 --- /dev/null +++ b/tidal-stress/k8s/stress-job.yaml @@ -0,0 +1,98 @@ +# In-cluster capacity ramp for the tidalDB multi-process cluster. +# +# Runs the load generator AS A POD so requests take the real cluster network path +# (a `kubectl port-forward` serializes everything through the API server and adds +# tens of ms — useless for capacity numbers). Targets the three region ClusterIPs +# directly; reads round-robin across them (each serves locally), leader-path +# writes are pinned to us-east (10.43.99.11) to measure the single-leader funnel. +# +# Apply: kubectl apply -f tidal-stress/k8s/stress-job.yaml +# Watch: kubectl logs -f job/tidal-stress -n tidaldb +# Re-run: kubectl delete job tidal-stress -n tidaldb; kubectl apply -f ... +# +# To compare the horizontally-scaled path, change `--write-path leader` to +# `--write-path sharded` (hash-partitions writes across all 3 regions, no funnel). +apiVersion: batch/v1 +kind: Job +metadata: + name: tidal-stress + namespace: tidaldb + labels: + app.kubernetes.io/name: tidal-stress + app.kubernetes.io/part-of: tidaldb +spec: + backoffLimit: 0 # a load run is not retried — read the logs + ttlSecondsAfterFinished: 7200 + template: + metadata: + labels: + app.kubernetes.io/name: tidal-stress + app.kubernetes.io/part-of: tidaldb + spec: + restartPolicy: Never + automountServiceAccountToken: false + affinity: + podAntiAffinity: + # Keep the generator OFF the leader's node so it never steals CPU from + # the write bottleneck we are measuring. + requiredDuringSchedulingIgnoredDuringExecution: + - labelSelector: + matchLabels: + tidaldb.region: us-east + topologyKey: kubernetes.io/hostname + securityContext: + runAsNonRoot: true + runAsUser: 1000 + runAsGroup: 1000 + seccompProfile: + type: RuntimeDefault + containers: + - name: stress + image: registry.threesix.ai/tidal/stress@sha256:77395cc2857e2adabaa37607b198316d6f1cd0ec04f749caf2de959e6bade55b # :v1 + imagePullPolicy: IfNotPresent + args: + - --target + - http://10.43.99.11:9500 # us-east (leader) + - --target + - http://10.43.99.12:9500 # eu-west + - --target + - http://10.43.99.13:9500 # ap-south + - --leader-url + - http://10.43.99.11:9500 + - --ramp + - peach-100k + - --stage-secs + - "45" + - --mix + - peach + - --write-path + - leader + - --corpus + - "20000" + - --users + - "100000" + - --poll-status + env: + - name: TIDAL_API_KEY + valueFrom: + secretKeyRef: + name: tidaldb-credentials + key: TIDAL_API_KEY + - name: TIDAL_STRESS_LOG + value: warn + resources: + # Small REQUEST (the cluster is request-saturated though ~15% utilised) + # with a high LIMIT: the generator bursts to the cycles it needs on the + # idle node. If it ever CPU-saturates, the report's schedule-lag / + # client-shed will say so — then split the load across multiple Jobs. + requests: + cpu: 250m + memory: 256Mi + limits: + cpu: "3" + memory: 1Gi + securityContext: + allowPrivilegeEscalation: false + readOnlyRootFilesystem: true + capabilities: + drop: ["ALL"] diff --git a/tidal-stress/src/client.rs b/tidal-stress/src/client.rs new file mode 100644 index 0000000..721b97b --- /dev/null +++ b/tidal-stress/src/client.rs @@ -0,0 +1,199 @@ +//! Thin async HTTP client over the tidalDB surface, plus the corpus seeder. +//! +//! The client returns a [`StatusClass`] per request and nothing else — latency is +//! measured by the caller from the request's *intended* send time so the figure +//! is coordinated-omission-corrected (a slow server inflates the number it would +//! otherwise hide). Response bodies are drained so the keep-alive connection is +//! reused (and so the measured time includes full response transfer). + +use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::Duration; + +use rand::Rng; +use tokio::sync::Semaphore; + +use crate::error::{Result, StressError}; +use crate::metrics::StatusClass; +use crate::workload::{HttpMethod, Plan}; + +/// Async client with optional bearer auth. Cheap to clone (Arc inside reqwest). +#[derive(Clone)] +pub struct HttpClient { + inner: reqwest::Client, + api_key: Option, +} + +impl HttpClient { + /// `request_timeout` should sit ABOVE the server's 30s request timeout so the + /// server's own 408 surfaces as a Timeout class rather than a client abort. + pub fn new(request_timeout: Duration, api_key: Option) -> Result { + let inner = reqwest::Client::builder() + .timeout(request_timeout) + .connect_timeout(Duration::from_secs(5)) + // Warm-connection pool sized to comfortably saturate the server's + // 100-concurrency limit per host without exhausting container fds at + // 4 host-pools (≈4×256 idle ceiling). + .pool_max_idle_per_host(256) + .pool_idle_timeout(Duration::from_secs(90)) + .tcp_nodelay(true) + .build() + .map_err(StressError::Client)?; + Ok(Self { inner, api_key }) + } + + fn apply_auth(&self, rb: reqwest::RequestBuilder) -> reqwest::RequestBuilder { + match &self.api_key { + Some(k) => rb.bearer_auth(k), + None => rb, + } + } + + /// Issue one request and classify the result. Never returns an error — a + /// transport fault is a [`StatusClass::Transport`] outcome, not a fatal. + pub async fn execute(&self, plan: &Plan) -> StatusClass { + let rb = match plan.method { + HttpMethod::Get => self.inner.get(&plan.url), + HttpMethod::Post => { + let rb = self.inner.post(&plan.url); + match &plan.body { + Some(b) => rb.json(b), + None => rb, + } + } + }; + match self.apply_auth(rb).send().await { + Ok(resp) => { + let class = StatusClass::from_status(resp.status().as_u16()); + // Drain the body so the connection returns to the pool. + let _ = resp.bytes().await; + class + } + Err(_) => StatusClass::Transport, + } + } + + /// Fetch `GET {base}/cluster/status` (unauthenticated) and return the leader + /// name and the worst replication lag across regions — used to watch whether + /// follower lag grows under write load between ramp stages. `None` on any fault. + pub async fn cluster_status(&self, base: &str) -> Option<(String, i64)> { + let url = format!("{base}/cluster/status"); + let resp = self.inner.get(&url).send().await.ok()?; + let v: serde_json::Value = resp.json().await.ok()?; + let leader = v.get("leader")?.as_str()?.to_string(); + let max_lag = v + .get("regions")? + .as_array()? + .iter() + .filter_map(|r| r.get("lag_events").and_then(serde_json::Value::as_i64)) + .max() + .unwrap_or(0); + Some((leader, max_lag)) + } + + /// One-off POST returning the raw status, for the seeder (which needs to know + /// success vs 429-retry rather than a capacity class). + async fn post_json(&self, url: &str, body: &serde_json::Value) -> Option { + let rb = self.inner.post(url).json(body); + match self.apply_auth(rb).send().await { + Ok(resp) => { + let code = resp.status().as_u16(); + let _ = resp.bytes().await; + Some(code) + } + Err(_) => None, + } + } +} + +/// Register `count` items (id 1..=count) plus a `dim`-wide content embedding for +/// each, against `base` (use the LEADER url so items broadcast to every region +/// and `/feed` on any region can rank them). Bounded-concurrency, retries 429. +/// Returns the number of items confirmed registered. +pub async fn seed_corpus( + client: &HttpClient, + base: &str, + count: u64, + dim: usize, + concurrency: usize, +) -> Result { + let sem = Arc::new(Semaphore::new(concurrency.max(1))); + let done = Arc::new(AtomicU64::new(0)); + let items_url = format!("{base}/items"); + let emb_url = format!("{base}/embeddings"); + let mut handles = Vec::new(); + + for id in 1..=count { + let permit = sem + .clone() + .acquire_owned() + .await + .map_err(|e| StressError::Seed(format!("semaphore closed: {e}")))?; + let client = client.clone(); + let items_url = items_url.clone(); + let emb_url = emb_url.clone(); + let done = done.clone(); + handles.push(tokio::spawn(async move { + let _permit = permit; + let category = SEED_CATEGORIES[(id as usize) % SEED_CATEGORIES.len()]; + // Build bodies and DROP the RNG before the awaits (ThreadRng is !Send). + let (item, emb) = { + let mut rng = rand::rng(); + let values: Vec = (0..dim).map(|_| rng.random::() - 0.5).collect(); + let item = serde_json::json!({ + "entity_id": id, + "metadata": { "title": format!("post-{id}"), "category": category }, + }); + let emb = serde_json::json!({ "entity_id": id, "values": values }); + (item, emb) + }; + + if retry_write(&client, &items_url, &item).await + && retry_write(&client, &emb_url, &emb).await + { + done.fetch_add(1, Ordering::Relaxed); + } + })); + } + for h in handles { + let _ = h.await; + } + Ok(done.load(Ordering::Relaxed)) +} + +const SEED_CATEGORIES: [&str; 12] = [ + "anime", + "gaming", + "fitness", + "music", + "cosplay", + "art", + "fantasy", + "scifi", + "romance", + "comedy", + "horror", + "slice-of-life", +]; + +/// POST with backpressure-aware retry: a 2xx is success; a 429 backs off and +/// retries (the seed must not give up just because the write pool is busy); any +/// other code / transport fault fails fast (a usage bug, not transient load). +async fn retry_write(client: &HttpClient, url: &str, body: &serde_json::Value) -> bool { + const MAX_ATTEMPTS: u32 = 8; + for attempt in 0..MAX_ATTEMPTS { + match client.post_json(url, body).await { + Some(code) if (200..=299).contains(&code) => return true, + Some(429) => { + // Honor the engine's ~50ms hint, with a little growth. + tokio::time::sleep(Duration::from_millis(40 + u64::from(attempt) * 20)).await; + } + Some(503) => { + // Leader briefly unreachable (e.g. a roll) — short wait, retry. + tokio::time::sleep(Duration::from_millis(100)).await; + } + _ => return false, + } + } + false +} diff --git a/tidal-stress/src/error.rs b/tidal-stress/src/error.rs new file mode 100644 index 0000000..d0e1e33 --- /dev/null +++ b/tidal-stress/src/error.rs @@ -0,0 +1,32 @@ +//! Crate error type. Mirrors the `tidal-server` convention: a flat `thiserror` +//! enum + `Result` alias, surfaced by `main` as `error: {err}` + exit 1. No +//! `anyhow` (workspace convention is explicit error enums). + +/// Anything that can go wrong while *setting up or driving* a load run. Per-request +/// failures during a run are NOT errors — they are recorded as outcomes (see +/// [`crate::metrics::Outcome`]); this type is only for fatal setup/teardown faults. +#[derive(Debug, thiserror::Error)] +pub enum StressError { + /// A target URL was missing or unparseable. + #[error("invalid target: {0}")] + Target(String), + + /// The ramp profile (preset name or `rps:secs,...` spec) could not be parsed. + #[error("invalid ramp profile: {0}")] + Ramp(String), + + /// The workload mix (preset name or `op=weight,...` spec) could not be parsed. + #[error("invalid workload mix: {0}")] + Mix(String), + + /// Corpus seeding could not register enough items to run a meaningful read mix. + #[error("seed phase failed: {0}")] + Seed(String), + + /// The HTTP client could not be constructed (TLS/builder fault). + #[error("http client build failed: {0}")] + Client(#[source] reqwest::Error), +} + +/// Crate-local result alias. +pub type Result = std::result::Result; diff --git a/tidal-stress/src/main.rs b/tidal-stress/src/main.rs new file mode 100644 index 0000000..101ef08 --- /dev/null +++ b/tidal-stress/src/main.rs @@ -0,0 +1,403 @@ +//! `tidal-stress` — an open-loop load generator and capacity ramp for tidalDB's +//! HTTP surface, modeling a thepeach-style feed workload. +//! +//! It seeds a corpus, then ramps a throttleable, signal-dominated request mix +//! (the thepeach session: feed reads + view/like/skip signals + search) from a +//! low rate up past a 100k-DAU TikTok-style peak, reporting per-stage latency +//! percentiles and a backpressure-aware status breakdown, and ends with a +//! capacity verdict translated back to supported DAU. +//! +//! Methodology lives in [`scheduler`] (coordinated-omission-corrected open loop); +//! the workload shape and its thepeach derivation live in [`workload`]. + +mod client; +mod error; +mod metrics; +mod scheduler; +mod workload; + +use std::io::Write as _; +use std::sync::Arc; +use std::time::Duration; + +use clap::Parser; + +use crate::client::{HttpClient, seed_corpus}; +use crate::error::{Result, StressError}; +use crate::metrics::StageStats; +use crate::scheduler::{Stage, parse_ramp, run_stage}; +use crate::workload::{OpKind, Workload, WritePath, parse_mix}; + +#[derive(Parser)] +#[command( + version, + about = "Open-loop capacity ramp for tidalDB (thepeach feed workload)" +)] +struct Cli { + /// Region gateway base URL, e.g. http://10.43.99.11:9500 . Repeat for each + /// region; reads round-robin across them (each serves locally), writes too + /// unless --leader-url pins them. + #[arg(long = "target", required = true)] + targets: Vec, + + /// Pin leader-path writes (/signals,/items,/embeddings) to this URL to avoid + /// the follower→leader forward hop. Ignored for --write-path sharded. + #[arg(long)] + leader_url: Option, + + /// Bearer key. The deployed cluster requires it; read from $TIDAL_API_KEY. + #[arg(long, env = "TIDAL_API_KEY")] + api_key: Option, + + /// Ramp: a preset (smoke|quick|peach-100k|max) or `rps:secs,rps:secs,...`. + #[arg(long, default_value = "peach-100k")] + ramp: String, + + /// Seconds to hold each preset ramp stage. + #[arg(long, default_value_t = 45)] + stage_secs: u64, + + /// Workload mix: a preset (peach|reads|writes) or `op=weight,...` + /// (ops: feed,search,view,like,skip,item,embed). + #[arg(long, default_value = "peach")] + mix: String, + + /// Write surface: `leader` (replicated, single-leader funnel) or `sharded` + /// (hash-partitioned across regions, not replicated). + #[arg(long, default_value = "leader")] + write_path: String, + + /// Items (id 1..=N) + embeddings to seed before the ramp. + #[arg(long, default_value_t = 10_000)] + corpus: u64, + + /// Virtual user id space (a 100k-DAU app reuses ids across the run). + #[arg(long, default_value_t = 50_000)] + users: u64, + + /// Concurrent seed writers. + #[arg(long, default_value_t = 64)] + seed_concurrency: usize, + + /// Skip seeding (corpus already registered from a prior run). + #[arg(long, default_value_t = false)] + skip_seed: bool, + + /// Max concurrent in-flight requests before the generator sheds (and reports + /// it). Keep well above the server's 100-concurrency limit so the server's + /// ceiling, not the client's, is what shows. + #[arg(long, default_value_t = 5_000)] + max_inflight: usize, + + /// Power-law concentration of signals onto hot (low-id) items (>1 = more + /// concentrated; 1 = uniform). + #[arg(long, default_value_t = 1.3)] + hot_skew: f64, + + /// Feed page size (thepeach home grid = 24). + #[arg(long, default_value_t = 24)] + feed_limit: u32, + + /// Embedding width. The deployed schema's content_vector is 128 (thepeach's + /// real text-embedding-3-small is 1536 — a schema change, noted in the verdict). + #[arg(long, default_value_t = 128)] + embedding_dim: usize, + + /// Per-request client timeout (s). Keep above the server's 30s so its 408 + /// surfaces rather than a client abort. + #[arg(long, default_value_t = 40)] + request_timeout_secs: u64, + + /// DAU the verdict translates the measured ceiling against. + #[arg(long, default_value_t = 100_000)] + dau: u64, + + /// Poll /cluster/status between stages to watch replication lag under load. + #[arg(long, default_value_t = false)] + poll_status: bool, + + /// Stop the ramp at the first stage that breaches the SLO (error rate >1% or + /// feed p99 >150ms) instead of pushing every stage. + #[arg(long, default_value_t = false)] + stop_on_knee: bool, +} + +// SLO thresholds for the per-stage verdict. tidalDB's own RETRIEVE SLA is p99 +// <50ms in-process; over a cluster network hop we allow 150ms before calling a +// stage degraded, and treat >1% errors (429/408/503/5xx/transport) as the knee. +const SLO_FEED_P99: Duration = Duration::from_millis(150); +const SLO_ERROR_RATE: f64 = 0.01; + +// The thepeach 100k-DAU model: ~5 sessions/user/day × ~5 feed pages × ~15.4 +// signal-equivalents/page ⇒ ~67M signals/day ≈ 780 signals/s average; a TikTok +// evening peak concentrates ~5× ⇒ ~3,900 signals/s peak. Reads are ~1/20th of that. +const SIGNALS_PER_DAU_PER_SEC_AVG: f64 = 780.0 / 100_000.0; +const PEAK_FACTOR: f64 = 5.0; + +#[tokio::main] +async fn main() { + if let Err(err) = run().await { + eprintln!("error: {err}"); + std::process::exit(1); + } +} + +#[allow(clippy::too_many_lines)] +async fn run() -> Result<()> { + let cli = Cli::parse(); + init_tracing(); + + if cli.targets.is_empty() { + return Err(StressError::Target( + "at least one --target is required".into(), + )); + } + let write_path = match cli.write_path.as_str() { + "leader" => WritePath::Leader, + "sharded" => WritePath::Sharded, + other => { + return Err(StressError::Target(format!( + "unknown --write-path '{other}'" + ))); + } + }; + let mix = parse_mix(&cli.mix)?; + let stages = parse_ramp(&cli.ramp, cli.stage_secs)?; + let client = Arc::new(HttpClient::new( + Duration::from_secs(cli.request_timeout_secs), + cli.api_key.clone(), + )?); + + // The leader URL anchors seeding (items must broadcast from the leader so + // every region's /feed can rank them) and, for the leader write path, the + // pinned write target when given. + let leader_url = cli + .leader_url + .clone() + .unwrap_or_else(|| cli.targets[0].clone()); + let read_bases = cli.targets.clone(); + let write_bases = match (write_path, &cli.leader_url) { + (WritePath::Leader, Some(url)) => vec![url.clone()], + _ => cli.targets.clone(), + }; + + println!("tidal-stress — thepeach feed workload"); + println!(" targets : {}", cli.targets.join(", ")); + println!( + " write path : {} ({})", + cli.write_path, + if cli.leader_url.is_some() { + "leader pinned" + } else { + "round-robin gateways" + } + ); + println!(" mix : {}", cli.mix); + println!( + " corpus/users : {} items / {} users", + cli.corpus, cli.users + ); + println!( + " embedding dim : {}{}", + cli.embedding_dim, + if cli.embedding_dim == 128 { + "" + } else { + " (NB: deployed schema is 128)" + } + ); + println!( + " 100k-DAU model: ~{:.0} signals/s avg, ~{:.0} signals/s peak (×{:.0})\n", + SIGNALS_PER_DAU_PER_SEC_AVG * cli.dau as f64, + SIGNALS_PER_DAU_PER_SEC_AVG * cli.dau as f64 * PEAK_FACTOR, + PEAK_FACTOR, + ); + + // ── Seed ────────────────────────────────────────────────────────────────── + if cli.skip_seed { + println!("seed: skipped (--skip-seed)\n"); + } else { + println!( + "seed: registering {} items + {}-dim embeddings via {} ...", + cli.corpus, cli.embedding_dim, leader_url + ); + let t0 = std::time::Instant::now(); + let seeded = seed_corpus( + &client, + &leader_url, + cli.corpus, + cli.embedding_dim, + cli.seed_concurrency, + ) + .await?; + println!( + "seed: {seeded}/{} items in {:.1}s\n", + cli.corpus, + t0.elapsed().as_secs_f64() + ); + if seeded < cli.corpus / 2 { + return Err(StressError::Seed(format!( + "only {seeded}/{} items registered — feed reads would be thin; check auth/schema/leader", + cli.corpus + ))); + } + } + + let workload = Arc::new(Workload::new( + read_bases, + write_bases, + write_path, + mix, + cli.corpus, + cli.users, + cli.hot_skew, + cli.feed_limit, + cli.embedding_dim, + )); + + let _ = std::io::stdout().flush(); + + // ── Ramp ──────────────────────────────────────────────────────────────── + let mut best_pass: Option<(usize, f64, f64)> = None; // (stage idx, total ok/s, signal ok/s) + let mut knee: Option = None; + + for (i, stage) in stages.iter().enumerate() { + if cli.poll_status + && let Some((leader, lag)) = client.cluster_status(&leader_url).await + { + println!(" [pre-stage] leader={leader} max_replication_lag={lag}"); + } + let label = format!("{}/{}", i + 1, stages.len()); + let stats = run_stage(workload.clone(), client.clone(), stage, cli.max_inflight).await; + print!( + "{}", + metrics::render_stage(&label, stage.target_rps, &stats) + ); + // Rust block-buffers stdout to a pipe (k8s log capture); flush so each + // stage is visible live under `kubectl logs -f`, not only at exit. + let _ = std::io::stdout().flush(); + + let (passed, signal_ok_rps) = stage_verdict(&stats); + if passed { + best_pass = Some((i, stats.achieved_rps(), signal_ok_rps)); + } else if knee.is_none() { + knee = Some(i); + println!(" ⚠ SLO breached at this stage (the capacity knee)."); + if cli.stop_on_knee { + println!(" stopping ramp (--stop-on-knee)."); + break; + } + } + } + + if cli.poll_status + && let Some((leader, lag)) = client.cluster_status(&leader_url).await + { + println!("\n [post-ramp] leader={leader} max_replication_lag={lag}"); + } + + print_verdict(&cli, &stages, best_pass, knee); + Ok(()) +} + +/// A stage passes if its error rate is under the SLO and feed-read p99 (when feed +/// reads are in the mix) is under the latency SLO. Returns (passed, signal ok/s). +fn stage_verdict(stats: &StageStats) -> (bool, f64) { + let secs = stats.elapsed.as_secs_f64().max(1e-9); + let signal_ok: u64 = [OpKind::SignalView, OpKind::SignalLike, OpKind::SignalSkip] + .iter() + .map(|k| stats.ops[k.idx()].ok()) + .sum(); + let signal_ok_rps = signal_ok as f64 / secs; + + let feed = &stats.ops[OpKind::FeedRead.idx()]; + let feed_p99_ok = feed.ok() == 0 || feed.hist.percentile(0.99) <= SLO_FEED_P99; + let passed = stats.error_rate() <= SLO_ERROR_RATE && feed_p99_ok && stats.client_shed == 0; + (passed, signal_ok_rps) +} + +fn print_verdict( + cli: &Cli, + stages: &[Stage], + best_pass: Option<(usize, f64, f64)>, + knee: Option, +) { + println!("\n══════════════════════ CAPACITY VERDICT ══════════════════════"); + match best_pass { + None => { + println!( + "No stage held the SLO — even the lowest rate breached it. Investigate before scaling." + ); + } + Some((idx, total_ok_rps, signal_ok_rps)) => { + let peak_per_sec = SIGNALS_PER_DAU_PER_SEC_AVG * PEAK_FACTOR; // signals/s per DAU at peak + let avg_per_sec = SIGNALS_PER_DAU_PER_SEC_AVG; // signals/s per DAU averaged + let dau_peak = signal_ok_rps / peak_per_sec; + let dau_avg = signal_ok_rps / avg_per_sec; + println!( + "Highest stage within SLO: stage {}/{} — target {:.0} rps.", + idx + 1, + stages.len(), + stages[idx].target_rps, + ); + println!( + " sustained OK throughput : {total_ok_rps:.0} req/s total, of which {signal_ok_rps:.0} signal-writes/s", + ); + println!( + " → supports ≈ {:.0}k DAU if provisioned for the 5× evening PEAK,", + dau_peak / 1000.0, + ); + println!( + " or ≈ {:.0}k DAU against AVERAGE load (peaks would shed/queue).", + dau_avg / 1000.0, + ); + let target = cli.dau as f64; + let verdict = if dau_peak >= target { + format!( + "YES — handles {:.0}k DAU at peak with headroom.", + target / 1000.0 + ) + } else if dau_avg >= target { + format!( + "PARTIALLY — handles {:.0}k DAU on average but NOT the 5× peak on this write path; scale or shard.", + target / 1000.0 + ) + } else { + format!( + "NO — below {:.0}k DAU even on average on this write path.", + target / 1000.0 + ) + }; + println!(" 100k-DAU question : {verdict}"); + } + } + if let Some(k) = knee { + println!( + " capacity knee : stage {}/{} (target {:.0} rps) — first SLO breach.", + k + 1, + stages.len(), + stages[k].target_rps + ); + } else { + println!( + " capacity knee : not reached — the cluster never breached the SLO in this ramp. Try --ramp max." + ); + } + println!( + " scaling levers : write path = {} (try --write-path sharded to remove the single-leader funnel);\n leader pod cpu \"2\" caps the write pool at ~2 workers — more CPU on the leader raises it (clamp max 8).", + cli.write_path, + ); + if cli.embedding_dim == 128 { + println!( + " note : thepeach's real embeddings are 1536-dim; a 1536-dim schema ~2.3× the RAM/item (capacity-planning.md)." + ); + } + println!("═══════════════════════════════════════════════════════════════"); +} + +fn init_tracing() { + let env_filter = std::env::var("TIDAL_STRESS_LOG").unwrap_or_else(|_| "warn".into()); + let _ = tracing_subscriber::fmt() + .with_env_filter(env_filter) + .try_init(); +} diff --git a/tidal-stress/src/metrics.rs b/tidal-stress/src/metrics.rs new file mode 100644 index 0000000..974179c --- /dev/null +++ b/tidal-stress/src/metrics.rs @@ -0,0 +1,409 @@ +//! Measurement: a dependency-free latency histogram, backpressure-aware status +//! classification, and per-stage aggregation/reporting. +//! +//! WHY THESE BUCKETS: the server distinguishes 429 (write-pool / WAL +//! backpressure — slow the write rate), 408 (request-timeout from the 100-deep +//! concurrency queue — lower concurrency), and 503 (leader/region down or +//! draining — re-target). A load report that lumps them as "errors" is useless +//! for capacity work, so [`StatusClass`] keeps them apart and the verdict reads +//! the mix to name the ceiling we hit. + +use std::time::Duration; + +use crate::workload::OpKind; + +// ── Latency histogram ──────────────────────────────────────────────────────── +// +// Log-spaced buckets with linear interpolation inside the crossing bucket. Min / +// max / mean are tracked exactly; only the percentiles are bucket-estimated, to +// ~3-4% (≈30 buckets per decade). HdrHistogram would be marginally tighter but a +// new lock entry; this is accurate enough to find a capacity knee and matches the +// repo's dependency-light posture (the engine's own latency metric uses fixed +// 1µs–10ms buckets, docs/ops/monitoring.md). + +const MIN_NS: f64 = 1_000.0; // 1µs — finer than that is noise over a network hop +const GROWTH: f64 = 1.0772; // ~30 buckets/decade +const BUCKETS: usize = 300; // 1µs * 1.0772^300 ≈ 9.4e12ns ≈ 2.6h — far past any real tail + +/// A single operation's latency distribution. Exact count/min/max/sum, bucketed +/// percentiles. +#[derive(Clone)] +pub struct LatencyHistogram { + counts: Vec, + total: u64, + min_ns: u64, + max_ns: u64, +} + +impl Default for LatencyHistogram { + fn default() -> Self { + Self { + counts: vec![0; BUCKETS], + total: 0, + min_ns: u64::MAX, + max_ns: 0, + } + } +} + +impl LatencyHistogram { + fn index(value_ns: u64) -> usize { + let v = value_ns as f64; + if v <= MIN_NS { + return 0; + } + let idx = (v / MIN_NS).log(GROWTH).floor() as usize; + idx.min(BUCKETS - 1) + } + + /// Lower bound (ns) of bucket `i`. + fn bound(i: usize) -> f64 { + MIN_NS * GROWTH.powi(i as i32) + } + + pub fn record(&mut self, latency: Duration) { + let ns = u64::try_from(latency.as_nanos()).unwrap_or(u64::MAX); + self.counts[Self::index(ns)] += 1; + self.total += 1; + self.min_ns = self.min_ns.min(ns); + self.max_ns = self.max_ns.max(ns); + } + + /// Estimated p-th percentile (`p` in 0.0..=1.0), linearly interpolated within + /// the crossing bucket. Returns 0 for an empty histogram. + pub fn percentile(&self, p: f64) -> Duration { + if self.total == 0 { + return Duration::ZERO; + } + let target = (p * self.total as f64).ceil().max(1.0) as u64; + let mut cumulative = 0u64; + for (i, &c) in self.counts.iter().enumerate() { + if c == 0 { + continue; + } + if cumulative + c >= target { + // Interpolate within [bound(i), bound(i+1)) by how far into this + // bucket's mass `target` falls. Clamp to the exact min/max so a + // p100 never reads below the bucket floor or above the real max. + let into = (target - cumulative) as f64 / c as f64; + let lo = Self::bound(i); + let hi = Self::bound(i + 1); + let est = (hi - lo).mul_add(into, lo); + let clamped = est.clamp(self.min_ns as f64, self.max_ns as f64); + return Duration::from_nanos(clamped as u64); + } + cumulative += c; + } + Duration::from_nanos(self.max_ns) + } + + /// Exact worst-case latency (not bucket-estimated) — the tail a load report + /// must surface alongside p999. + pub const fn max(&self) -> Duration { + Duration::from_nanos(if self.total == 0 { 0 } else { self.max_ns }) + } +} + +// ── Status classification ──────────────────────────────────────────────────── + +/// What a single request's result *means* for capacity, not just its code. +#[derive(Clone, Copy, PartialEq, Eq)] +pub enum StatusClass { + /// 2xx — the operation succeeded. + Ok, + /// 429 — write-pool / WAL backpressure (honor retry, slow the write rate). + Backpressure, + /// 408 — request-timeout: queued past 30s behind the 100-concurrency limit. + Timeout, + /// 503 — leader/region unreachable or draining (re-target / wait). + Unavailable, + /// Other 4xx (400/401/413/404…) — a client/usage bug, not load. + ClientError, + /// 5xx other than 503 — a server fault under load. + ServerError, + /// No HTTP response at all: connect refused, TLS, client-side timeout, reset. + Transport, +} + +impl StatusClass { + pub const fn from_status(code: u16) -> Self { + match code { + 200..=299 => Self::Ok, + 429 => Self::Backpressure, + 408 => Self::Timeout, + 503 => Self::Unavailable, + 400..=499 => Self::ClientError, + _ => Self::ServerError, + } + } + + const fn idx(self) -> usize { + match self { + Self::Ok => 0, + Self::Backpressure => 1, + Self::Timeout => 2, + Self::Unavailable => 3, + Self::ClientError => 4, + Self::ServerError => 5, + Self::Transport => 6, + } + } +} + +const STATUS_KINDS: usize = 7; + +/// The result of one issued request, sent from a worker to the collector. +pub struct Outcome { + pub op: OpKind, + pub class: StatusClass, + /// Wall-clock from the request's *intended* send time to its completion — + /// the coordinated-omission-corrected latency (includes any client backlog). + pub latency: Duration, +} + +// ── Per-op + per-stage aggregation ─────────────────────────────────────────── + +#[derive(Clone, Default)] +pub struct OpStats { + pub hist: LatencyHistogram, + pub status: [u64; STATUS_KINDS], +} + +impl OpStats { + fn record(&mut self, class: StatusClass, latency: Duration) { + self.status[class.idx()] += 1; + // Only successful requests' latencies describe service time; a 429/503 + // rejected fast would otherwise flatter the percentiles. Errors are + // counted in `status`, excluded from the latency picture. + if class == StatusClass::Ok { + self.hist.record(latency); + } + } + + pub const fn total(&self) -> u64 { + let mut sum = 0u64; + let mut i = 0; + while i < STATUS_KINDS { + sum += self.status[i]; + i += 1; + } + sum + } + + pub const fn ok(&self) -> u64 { + self.status[0] + } + pub const fn backpressure(&self) -> u64 { + self.status[1] + } + pub const fn timeout(&self) -> u64 { + self.status[2] + } + pub const fn unavailable(&self) -> u64 { + self.status[3] + } + pub const fn client_error(&self) -> u64 { + self.status[4] + } + pub const fn server_error(&self) -> u64 { + self.status[5] + } + pub const fn transport(&self) -> u64 { + self.status[6] + } + pub const fn errors(&self) -> u64 { + self.total() - self.ok() + } +} + +/// Everything one ramp stage produced. Per-op stats plus a roll-up. +pub struct StageStats { + pub ops: Vec, // indexed by OpKind::idx() + pub elapsed: Duration, + /// Requests the client could not even dispatch (in-flight cap hit) — a signal + /// the LOAD GENERATOR is saturated, not the server. Non-zero ⇒ the reported + /// achieved RPS is a client-limited floor, not the server's ceiling. + pub client_shed: u64, + /// Mean delay between a request's intended send time and its actual dispatch. + /// Growing scheduling delay ⇒ the generator is falling behind the target rate. + pub mean_schedule_lag: Duration, +} + +impl StageStats { + pub fn new() -> Self { + Self { + ops: vec![OpStats::default(); OpKind::COUNT], + elapsed: Duration::ZERO, + client_shed: 0, + mean_schedule_lag: Duration::ZERO, + } + } + + pub fn record(&mut self, outcome: &Outcome) { + self.ops[outcome.op.idx()].record(outcome.class, outcome.latency); + } + + pub fn total(&self) -> u64 { + self.ops.iter().map(OpStats::total).sum() + } + pub fn total_ok(&self) -> u64 { + self.ops.iter().map(OpStats::ok).sum() + } + pub fn total_errors(&self) -> u64 { + self.ops.iter().map(OpStats::errors).sum() + } + pub fn error_rate(&self) -> f64 { + let t = self.total(); + if t == 0 { + 0.0 + } else { + self.total_errors() as f64 / t as f64 + } + } + /// Achieved throughput: completed requests per second over the stage. + pub fn achieved_rps(&self) -> f64 { + let s = self.elapsed.as_secs_f64(); + if s <= 0.0 { + 0.0 + } else { + self.total() as f64 / s + } + } + + /// Sum of a status class across every op (for the verdict). + pub fn class_total(&self, pick: fn(&OpStats) -> u64) -> u64 { + self.ops.iter().map(pick).sum() + } +} + +fn fmt_dur(d: Duration) -> String { + let us = d.as_nanos() as f64 / 1000.0; + if us < 1000.0 { + format!("{us:.0}µs") + } else if us < 1_000_000.0 { + format!("{:.2}ms", us / 1000.0) + } else { + format!("{:.2}s", us / 1_000_000.0) + } +} + +/// Render a per-stage report block as a human-readable table. +pub fn render_stage(label: &str, target_rps: f64, stats: &StageStats) -> String { + let mut out = String::new(); + out.push_str(&format!( + "\n── stage {label} (target {target_rps:.0} rps, achieved {:.0} rps, {:.1}s) ──\n", + stats.achieved_rps(), + stats.elapsed.as_secs_f64(), + )); + out.push_str(&format!( + "{:<10} {:>8} {:>8} {:>9} {:>9} {:>9} {:>9} {:>9} {:>5} {:>4} {:>4} {:>4} {:>5} {:>4}\n", + "op", + "count", + "ok/s", + "p50", + "p90", + "p99", + "p999", + "max", + "429", + "408", + "503", + "4xx", + "5xx", + "tx", + )); + for kind in OpKind::ALL { + let s = &stats.ops[kind.idx()]; + if s.total() == 0 { + continue; + } + let ok_rps = s.ok() as f64 / stats.elapsed.as_secs_f64().max(1e-9); + out.push_str(&format!( + "{:<10} {:>8} {:>8.0} {:>9} {:>9} {:>9} {:>9} {:>9} {:>5} {:>4} {:>4} {:>4} {:>5} {:>4}\n", + kind.label(), + s.total(), + ok_rps, + fmt_dur(s.hist.percentile(0.50)), + fmt_dur(s.hist.percentile(0.90)), + fmt_dur(s.hist.percentile(0.99)), + fmt_dur(s.hist.percentile(0.999)), + fmt_dur(s.hist.max()), + s.backpressure(), + s.timeout(), + s.unavailable(), + s.client_error(), + s.server_error(), + s.transport(), + )); + } + out.push_str(&format!( + "{:<10} {:>8} {:>8.0} {:>52} {:>5} {:>4} {:>4} {:>4} {:>5} {:>4}\n", + "ALL", + stats.total(), + stats.total_ok() as f64 / stats.elapsed.as_secs_f64().max(1e-9), + "", + stats.class_total(OpStats::backpressure), + stats.class_total(OpStats::timeout), + stats.class_total(OpStats::unavailable), + stats.class_total(OpStats::client_error), + stats.class_total(OpStats::server_error), + stats.class_total(OpStats::transport), + )); + out.push_str(&format!( + "error rate {:.2}% | client-shed {} | schedule-lag {}\n", + stats.error_rate() * 100.0, + stats.client_shed, + fmt_dur(stats.mean_schedule_lag), + )); + out +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn histogram_percentiles_are_ordered_and_bounded() { + let mut h = LatencyHistogram::default(); + for ns in 1..=10_000u64 { + h.record(Duration::from_nanos(ns * 1000)); // 1µs..10ms + } + let p50 = h.percentile(0.50); + let p99 = h.percentile(0.99); + assert!(p50 < p99, "p50 {p50:?} should be < p99 {p99:?}"); + assert!(p99 <= h.max(), "p99 {p99:?} above exact max {:?}", h.max()); + assert!( + h.max() >= Duration::from_millis(9), + "max should be ~10ms, got {:?}", + h.max() + ); + // 50th percentile of a uniform 1µs..10ms ≈ 5ms; allow the bucket error. + let p50_ms = p50.as_secs_f64() * 1000.0; + assert!((p50_ms - 5.0).abs() < 0.5, "p50 {p50_ms}ms not ~5ms"); + } + + #[test] + fn status_class_maps_backpressure_codes() { + assert!(matches!( + StatusClass::from_status(429), + StatusClass::Backpressure + )); + assert!(matches!( + StatusClass::from_status(408), + StatusClass::Timeout + )); + assert!(matches!( + StatusClass::from_status(503), + StatusClass::Unavailable + )); + assert!(matches!(StatusClass::from_status(201), StatusClass::Ok)); + assert!(matches!( + StatusClass::from_status(400), + StatusClass::ClientError + )); + assert!(matches!( + StatusClass::from_status(500), + StatusClass::ServerError + )); + } +} diff --git a/tidal-stress/src/scheduler.rs b/tidal-stress/src/scheduler.rs new file mode 100644 index 0000000..a166302 --- /dev/null +++ b/tidal-stress/src/scheduler.rs @@ -0,0 +1,207 @@ +//! Open-loop, constant-arrival-rate load scheduler — the methodology that makes +//! the numbers trustworthy. +//! +//! A CLOSED-loop generator (N workers each doing `send(); await; repeat`) sends +//! FEWER requests when the server slows down, so it silently under-measures both +//! load and latency — the "coordinated omission" error. This scheduler instead +//! fires requests at a fixed target rate regardless of outstanding responses, +//! and measures each request's latency from its *intended* send time, so a server +//! stall inflates the very percentiles a closed-loop test would hide. +//! +//! When the in-flight cap is reached we COUNT a client-shed rather than block — +//! blocking would re-introduce the closed-loop coupling. A non-zero shed count is +//! reported as "the generator, not the server, is the limit here". + +use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::Duration; + +use tokio::sync::{Semaphore, mpsc}; +use tokio::time::Instant; + +use crate::client::HttpClient; +use crate::error::{Result, StressError}; +use crate::metrics::{Outcome, StageStats}; +use crate::workload::Workload; + +/// One step of the ramp: hold `target_rps` for `duration`. +pub struct Stage { + pub target_rps: f64, + pub duration: Duration, +} + +/// Run one stage open-loop and return its aggregated stats. +pub async fn run_stage( + workload: Arc, + client: Arc, + stage: &Stage, + max_inflight: usize, +) -> StageStats { + let (tx, mut rx) = mpsc::unbounded_channel::(); + let collector = tokio::spawn(async move { + let mut stats = StageStats::new(); + while let Some(o) = rx.recv().await { + stats.record(&o); + } + stats + }); + + let sem = Arc::new(Semaphore::new(max_inflight)); + let shed = Arc::new(AtomicU64::new(0)); + let lag_sum_ns = Arc::new(AtomicU64::new(0)); + let lag_count = Arc::new(AtomicU64::new(0)); + + let start = Instant::now(); + let deadline = start + stage.duration; + let period = Duration::from_secs_f64(1.0 / stage.target_rps.max(1e-9)); + let mut next = start; + let mut dispatched: u64 = 0; + + while Instant::now() < deadline { + let now = Instant::now(); + if next <= now { + match sem.clone().try_acquire_owned() { + Ok(permit) => { + let intended = next; + let lag = now.saturating_duration_since(intended); + lag_sum_ns.fetch_add(lag.as_nanos() as u64, Ordering::Relaxed); + lag_count.fetch_add(1, Ordering::Relaxed); + + let workload = workload.clone(); + let client = client.clone(); + let tx = tx.clone(); + tokio::spawn(async move { + let _permit = permit; + // Build the plan and DROP the RNG before any await: + // `ThreadRng` is !Send (holds an `Rc`), so it must not be + // live across `.await` or the task future isn't `Send`. + let plan = { + let mut rng = rand::rng(); + workload.next(&mut rng) + }; + let op = plan.op; + let class = client.execute(&plan).await; + // CO-corrected latency: from when the request was DUE, not + // when it was sent — so client backlog counts against us. + let latency = Instant::now().saturating_duration_since(intended); + let _ = tx.send(Outcome { op, class, latency }); + }); + } + Err(_) => { + shed.fetch_add(1, Ordering::Relaxed); + } + } + next += period; + dispatched += 1; + // Let the local runtime worker service spawned tasks during a + // catch-up burst instead of monopolising it. + if dispatched.is_multiple_of(256) { + tokio::task::yield_now().await; + } + } else { + tokio::time::sleep_until(next).await; + } + } + let elapsed = start.elapsed(); + + // Dropping the producer's sender lets the collector finish once every in-flight + // task (each holding its own sender clone) completes — bounded by the client + // request timeout, so this can never hang indefinitely. + drop(tx); + let mut stats = collector.await.unwrap_or_else(|_| StageStats::new()); + stats.elapsed = elapsed; + stats.client_shed = shed.load(Ordering::Relaxed); + let lc = lag_count.load(Ordering::Relaxed); + stats.mean_schedule_lag = if lc == 0 { + Duration::ZERO + } else { + Duration::from_nanos(lag_sum_ns.load(Ordering::Relaxed) / lc) + }; + stats +} + +// ── Ramp parsing ───────────────────────────────────────────────────────────── + +/// Parse a ramp: a preset name or an explicit `rps:secs,rps:secs,...` spec. +/// +/// Presets are pinned to the thepeach 100k-DAU model (see crate docs): at the +/// `peach` mix (~90% signals) a 100k-DAU TikTok-style evening peak is ≈3,900 +/// total req/s, so `peach-100k` brackets and then doubles past that to answer +/// "can we handle more?". `secs` defaults from the preset. +pub fn parse_ramp(spec: &str, stage_secs: u64) -> Result> { + let mk = |rates: &[f64]| -> Vec { + rates + .iter() + .map(|&r| Stage { + target_rps: r, + duration: Duration::from_secs(stage_secs), + }) + .collect() + }; + match spec { + // Local validation through a port-forward (low, short). + "smoke" => Ok(vec![ + Stage { + target_rps: 10.0, + duration: Duration::from_secs(10), + }, + Stage { + target_rps: 40.0, + duration: Duration::from_secs(10), + }, + ]), + "quick" => Ok(mk(&[100.0, 500.0, 1500.0, 4000.0])), + // The headline ramp: low → past a 100k-DAU peak → 2× beyond. + "peach-100k" | "default" => Ok(mk(&[ + 50.0, 150.0, 400.0, 800.0, 1500.0, 3000.0, 5000.0, 8000.0, + ])), + // Push to find the hard ceiling. + "max" => Ok(mk(&[ + 500.0, 1500.0, 3000.0, 6000.0, 10000.0, 15000.0, 20000.0, + ])), + _ => { + let mut stages = Vec::new(); + for part in spec.split(',') { + let part = part.trim(); + if part.is_empty() { + continue; + } + let (r, s) = part + .split_once(':') + .ok_or_else(|| StressError::Ramp(format!("expected rps:secs, got '{part}'")))?; + let target_rps: f64 = r + .trim() + .parse() + .map_err(|_| StressError::Ramp(format!("bad rps '{r}'")))?; + let secs: u64 = s + .trim() + .parse() + .map_err(|_| StressError::Ramp(format!("bad secs '{s}'")))?; + stages.push(Stage { + target_rps, + duration: Duration::from_secs(secs), + }); + } + if stages.is_empty() { + return Err(StressError::Ramp("empty ramp".into())); + } + Ok(stages) + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn ramp_presets_and_specs_parse() { + assert_eq!(parse_ramp("peach-100k", 45).expect("preset").len(), 8); + let custom = parse_ramp("100:30,500:30,2000:60", 45).expect("spec"); + assert_eq!(custom.len(), 3); + assert!((custom[2].target_rps - 2000.0).abs() < f64::EPSILON); + assert_eq!(custom[2].duration.as_secs(), 60); + assert!(parse_ramp("oops", 45).is_err()); + assert!(parse_ramp("100-30", 45).is_err()); + } +} diff --git a/tidal-stress/src/workload.rs b/tidal-stress/src/workload.rs new file mode 100644 index 0000000..984cd84 --- /dev/null +++ b/tidal-stress/src/workload.rs @@ -0,0 +1,472 @@ +//! The workload model: a thepeach feed session expressed as a weighted mix of +//! tidalDB operations, with realistic "hot content" entity selection. +//! +//! WHY THIS SHAPE (from thepeach's specs): a session is anon-mint → home feed +//! (24 tiles/page, infinite scroll) → `view`/`like`/`skip` on tiles → occasional +//! search → repeat. thepeach's user-graph spec says behaviour is observed at +//! "10–1000+ events per session" — signals outnumber feed reads by 1–2 orders of +//! magnitude — so the default mix is signal-dominated. The deployed schema only +//! declares `view`/`like`/`skip` and a 128-dim `content_vector`, so those are the +//! only signal names and the only embedding width we emit. + +use std::sync::atomic::{AtomicUsize, Ordering}; + +use rand::Rng; + +use crate::error::{Result, StressError}; + +/// The operations a feed session performs against tidalDB. +#[derive(Clone, Copy, PartialEq, Eq)] +pub enum OpKind { + /// GET /feed — a home-feed page render (the read the experiment will swap a + /// tidaldb ranking profile into). + FeedRead, + /// GET /search — handle/tag search. + Search, + /// POST /signals view — a tile entering the viewport (the firehose). + SignalView, + /// POST /signals like — a reaction (positive engagement). + SignalLike, + /// POST /signals skip — a dismissal / sub-3s skip (strong negative, UC-01). + SignalSkip, + /// POST /items — a creator publishing a post/companion (rare in a read-heavy app). + RegisterItem, + /// POST /embeddings — the post's content vector (paired with RegisterItem). + RegisterEmbedding, +} + +impl OpKind { + pub const ALL: [Self; 7] = [ + Self::FeedRead, + Self::Search, + Self::SignalView, + Self::SignalLike, + Self::SignalSkip, + Self::RegisterItem, + Self::RegisterEmbedding, + ]; + pub const COUNT: usize = Self::ALL.len(); + + pub const fn idx(self) -> usize { + match self { + Self::FeedRead => 0, + Self::Search => 1, + Self::SignalView => 2, + Self::SignalLike => 3, + Self::SignalSkip => 4, + Self::RegisterItem => 5, + Self::RegisterEmbedding => 6, + } + } + + pub const fn label(self) -> &'static str { + match self { + Self::FeedRead => "feed", + Self::Search => "search", + Self::SignalView => "view", + Self::SignalLike => "like", + Self::SignalSkip => "skip", + Self::RegisterItem => "item", + Self::RegisterEmbedding => "embed", + } + } + + fn from_label(s: &str) -> Option { + Self::ALL.into_iter().find(|k| k.label() == s) + } +} + +/// Which write surface the signal/item/embedding writes target. +#[derive(Clone, Copy, PartialEq, Eq)] +pub enum WritePath { + /// `/signals` etc. — every write funnels to the single leader (us-east); a + /// write to a follower adds a forward hop. Replicated (the real feedback loop). + Leader, + /// `/sharded/*` — hash-partitioned across all 3 regions, no leader funnel. + /// NOT replicated (single owner per entity) — a horizontal-scale comparison. + Sharded, +} + +impl WritePath { + fn signal_path(self) -> &'static str { + match self { + Self::Leader => "/signals", + Self::Sharded => "/sharded/signals", + } + } + fn item_path(self) -> &'static str { + match self { + Self::Leader => "/items", + Self::Sharded => "/sharded/items", + } + } + fn embedding_path(self) -> &'static str { + match self { + Self::Leader => "/embeddings", + Self::Sharded => "/sharded/embeddings", + } + } +} + +#[derive(Clone, Copy)] +pub enum HttpMethod { + Get, + Post, +} + +/// One concrete request to issue. +pub struct Plan { + pub op: OpKind, + pub method: HttpMethod, + pub url: String, + pub body: Option, +} + +/// Round-robin over a set of base URLs (e.g. the three region gateways). +struct RoundRobin { + bases: Vec, + next: AtomicUsize, +} + +impl RoundRobin { + fn new(bases: Vec) -> Self { + Self { + bases, + next: AtomicUsize::new(0), + } + } + fn pick(&self) -> &str { + let i = self.next.fetch_add(1, Ordering::Relaxed) % self.bases.len(); + &self.bases[i] + } +} + +/// Cumulative-weight picker over the seven op kinds (no `rand_distr` dependency). +struct Mix { + cum: Vec<(f64, OpKind)>, + total: f64, +} + +impl Mix { + fn from_weights(weights: &[(OpKind, f64)]) -> Self { + let mut cum = Vec::with_capacity(weights.len()); + let mut acc = 0.0; + for &(op, w) in weights { + if w <= 0.0 { + continue; + } + acc += w; + cum.push((acc, op)); + } + Self { cum, total: acc } + } + + fn pick(&self, rng: &mut impl Rng) -> OpKind { + let r = rng.random::() * self.total; + for &(threshold, op) in &self.cum { + if r < threshold { + return op; + } + } + self.cum.last().map_or(OpKind::FeedRead, |&(_, op)| op) + } +} + +/// The session model + corpus parameters, shared (immutable) across all workers. +pub struct Workload { + reads: RoundRobin, + writes: RoundRobin, + write_path: WritePath, + mix: Mix, + feed_profiles: Vec<(f64, &'static str)>, // cumulative + profile_total: f64, + corpus: u64, // item ids 1..=corpus already seeded + users: u64, // virtual user ids 1..=users + hot_skew: f64, // >1 concentrates signals on low (hot) ids + feed_limit: u32, + embedding_dim: usize, + categories: Vec<&'static str>, +} + +/// Default category vocabulary — stands in for thepeach companion/post tags so +/// `category` keyword filters and `/search` queries hit real keyword values. +const CATEGORIES: [&str; 12] = [ + "anime", + "gaming", + "fitness", + "music", + "cosplay", + "art", + "fantasy", + "scifi", + "romance", + "comedy", + "horror", + "slice-of-life", +]; + +impl Workload { + #[allow(clippy::too_many_arguments)] + pub fn new( + read_bases: Vec, + write_bases: Vec, + write_path: WritePath, + mix_weights: Vec<(OpKind, f64)>, + corpus: u64, + users: u64, + hot_skew: f64, + feed_limit: u32, + embedding_dim: usize, + ) -> Self { + // for_you is the deployed default and the one E2 will A/B; trending/hot/new + // exercise the other built-in sort paths. (related/following need + // relationship data we don't seed; date_saved needs a user — both omitted.) + let feed_profiles = vec![ + ("for_you", 70.0), + ("trending", 15.0), + ("hot", 10.0), + ("new", 5.0), + ]; + let mut cum = Vec::new(); + let mut acc = 0.0; + for (name, w) in feed_profiles { + acc += w; + cum.push((acc, name)); + } + Self { + reads: RoundRobin::new(read_bases), + writes: RoundRobin::new(write_bases), + write_path, + mix: Mix::from_weights(&mix_weights), + feed_profiles: cum, + profile_total: acc, + corpus, + users, + hot_skew, + feed_limit, + embedding_dim, + categories: CATEGORIES.to_vec(), + } + } + + fn pick_user(&self, rng: &mut impl Rng) -> u64 { + rng.random_range(1..=self.users.max(1)) + } + + /// Power-law item pick: `id = corpus * u^skew` concentrates engagement on the + /// low (hot) ids — a few viral posts get most of the views, a long tail gets + /// the rest, like a real feed. + fn pick_item(&self, rng: &mut impl Rng) -> u64 { + let u: f64 = rng.random(); + let scaled = (self.corpus as f64 * u.powf(self.hot_skew)) as u64; + scaled.clamp(1, self.corpus.max(1)) + } + + fn pick_profile(&self, rng: &mut impl Rng) -> &'static str { + let r = rng.random::() * self.profile_total; + for &(threshold, name) in &self.feed_profiles { + if r < threshold { + return name; + } + } + "for_you" + } + + fn pick_category(&self, rng: &mut impl Rng) -> &'static str { + let i = rng.random_range(0..self.categories.len()); + self.categories[i] + } + + fn random_embedding(&self, rng: &mut impl Rng) -> Vec { + // A non-zero vector (the engine rejects zero-norm); exact distribution is + // irrelevant to the write-path cost we're measuring. + (0..self.embedding_dim) + .map(|_| rng.random::() - 0.5) + .collect() + } + + /// Build the next request to issue. + pub fn next(&self, rng: &mut impl Rng) -> Plan { + let op = self.mix.pick(rng); + match op { + OpKind::FeedRead => { + let user = self.pick_user(rng); + let profile = self.pick_profile(rng); + let url = format!( + "{}/feed?profile={profile}&user_id={user}&limit={}", + self.reads.pick(), + self.feed_limit, + ); + Plan { + op, + method: HttpMethod::Get, + url, + body: None, + } + } + OpKind::Search => { + let user = self.pick_user(rng); + let q = self.pick_category(rng); + let url = format!( + "{}/search?query={q}&user_id={user}&limit=20", + self.reads.pick(), + ); + Plan { + op, + method: HttpMethod::Get, + url, + body: None, + } + } + OpKind::SignalView | OpKind::SignalLike | OpKind::SignalSkip => { + let (name, weight) = match op { + OpKind::SignalLike => ("like", 1.0), + OpKind::SignalSkip => ("skip", 1.0), + _ => ("view", 1.0), + }; + let url = format!("{}{}", self.writes.pick(), self.write_path.signal_path()); + let body = serde_json::json!({ + "entity_id": self.pick_item(rng), + "signal": name, + "weight": weight, + }); + Plan { + op, + method: HttpMethod::Post, + url, + body: Some(body), + } + } + OpKind::RegisterItem => { + let id = self.pick_item(rng); + let url = format!("{}{}", self.writes.pick(), self.write_path.item_path()); + let body = serde_json::json!({ + "entity_id": id, + "metadata": { "title": format!("post-{id}"), "category": self.pick_category(rng) }, + }); + Plan { + op, + method: HttpMethod::Post, + url, + body: Some(body), + } + } + OpKind::RegisterEmbedding => { + let url = format!("{}{}", self.writes.pick(), self.write_path.embedding_path()); + let body = serde_json::json!({ + "entity_id": self.pick_item(rng), + "values": self.random_embedding(rng), + }); + Plan { + op, + method: HttpMethod::Post, + url, + body: Some(body), + } + } + } + } +} + +// ── Preset / spec parsing ──────────────────────────────────────────────────── + +/// Parse the workload mix: a preset name or an explicit `op=weight,...` spec. +/// +/// Default preset `peach` (≈ a thepeach feed session): per home-feed page the +/// user views ~12 of 24 tiles, likes ~0.6, skips ~1.5, searches ~0.25, and a +/// trickle of creator posts. The result is ~78% views, ~10% skips, ~6.5% feed +/// reads — the signal-dominated shape thepeach's user-graph spec describes. +pub fn parse_mix(spec: &str) -> Result> { + match spec { + "peach" | "tiktok" | "default" => Ok(vec![ + (OpKind::FeedRead, 1.0), + (OpKind::Search, 0.25), + (OpKind::SignalView, 12.0), + (OpKind::SignalLike, 0.6), + (OpKind::SignalSkip, 1.5), + (OpKind::RegisterItem, 0.02), + (OpKind::RegisterEmbedding, 0.02), + ]), + // Read-only sanity profile (capacity of the local-read path alone). + "reads" => Ok(vec![(OpKind::FeedRead, 4.0), (OpKind::Search, 1.0)]), + // Write-only firehose (isolates the leader/sharded write ceiling). + "writes" => Ok(vec![ + (OpKind::SignalView, 12.0), + (OpKind::SignalLike, 0.6), + (OpKind::SignalSkip, 1.5), + ]), + _ => { + let mut weights = Vec::new(); + for part in spec.split(',') { + let part = part.trim(); + if part.is_empty() { + continue; + } + let (name, w) = part + .split_once('=') + .ok_or_else(|| StressError::Mix(format!("expected op=weight, got '{part}'")))?; + let op = OpKind::from_label(name.trim()) + .ok_or_else(|| StressError::Mix(format!("unknown op '{name}'")))?; + let w: f64 = w + .trim() + .parse() + .map_err(|_| StressError::Mix(format!("bad weight '{w}'")))?; + weights.push((op, w)); + } + if weights.is_empty() { + return Err(StressError::Mix("empty mix".into())); + } + Ok(weights) + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn peach_mix_is_signal_dominated() { + let w = parse_mix("peach").expect("preset parses"); + let total: f64 = w.iter().map(|&(_, x)| x).sum(); + let views = w + .iter() + .find(|&&(o, _)| o == OpKind::SignalView) + .map(|&(_, x)| x) + .unwrap_or(0.0); + assert!(views / total > 0.6, "views should dominate the mix"); + } + + #[test] + fn explicit_mix_spec_parses() { + let w = parse_mix("feed=1,view=10").expect("spec parses"); + assert_eq!(w.len(), 2); + assert!(parse_mix("feed=oops").is_err()); + assert!(parse_mix("bogus=1").is_err()); + } + + #[test] + fn hot_skew_biases_low_ids() { + let wl = Workload::new( + vec!["http://x".into()], + vec!["http://x".into()], + WritePath::Leader, + parse_mix("peach").expect("mix"), + 1000, + 100, + 1.5, + 24, + 128, + ); + let mut rng = rand::rng(); + let mut low = 0; + for _ in 0..10_000 { + if wl.pick_item(&mut rng) <= 200 { + low += 1; + } + } + // Uniform would put ~2000 picks in the bottom 20% of ids; skew 1.5 + // concentrates ~34% (≈3400) there — clearly biased toward hot content. + assert!(low > 3_000, "hot-skew not biasing low ids: {low}"); + } +}