jordan/tidaldb
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(tidal-stress): open-loop capacity load generator (thepeach feed workload)

Browse Source 
New workspace crate: an open-loop, coordinated-omission-corrected HTTP load
generator + capacity ramp for the standalone and multi-process cluster surfaces,
modeling a thepeach feed session (feed reads + view/like/skip signals + search,
signal-dominated per their user-graph spec). Throttleable target rate, ramp
presets (smoke/quick/peach-100k/max) or rps:secs specs, peach/reads/writes/custom
mixes, leader vs sharded write paths, per-op p50/p90/p99/p999/max latency, a
backpressure-aware status breakdown (429/408/503/4xx/5xx/transport), and a verdict
translated to supported DAU. Runs in-cluster as a k8s Job (tidal-stress/k8s/).

Open-loop scheduler (scheduler.rs) fires at a fixed arrival rate and measures
latency from each request's intended send time, so a server stall inflates the
percentiles a closed-loop test hides; it shed-and-counts rather than blocking when
the in-flight cap is reached. Pure-Rust (tokio + reqwest/rustls), no engine deps.

Findings on the live 3-region k3s cluster (docs/ops/stress-test-thepeach.md):
reads scale to thousands/s at <15ms p99; the replicated /signals path saturates at
~90 signals/s (single-leader funnel + 2-worker write pool + synchronous gRPC ship);
the sharded path sustains 3,669 signals/s at 0 errors and ~27% cluster CPU (≈ the
100k-DAU peak, knee not reached). Overload degrades gracefully (429; 0 pod
restarts). thepeach's planned in-process embedding sidesteps all of it (write ≈82ns).
This commit is contained in:
jx12n 2026-06-10 21:53:17 -06:00
parent 6f17409f40
commit f640764d89
14 changed files with 2073 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.dockerignore
View File

@ -13,6 +13,7 @@
!tidal-net
!tidal-server
!tidalctl
!tidal-stress
!applications
# Re-exclude heavy / generated / secret paths nested inside the allowed dirs.

5
CLAUDE.md
View File

@ -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)
```

15
Cargo.lock generated
View File

@ -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"

1
Cargo.toml
View File

@ -4,6 +4,7 @@ members = [
"tidal-net",
"tidalctl",
"tidal-server",
"tidal-stress",
"applications/forage/engine",
"applications/forage/server",
"applications/forage/embedder",

37
docker/stress/Dockerfile Normal file
View File

@ -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"]

140
docs/ops/stress-test-thepeach.md Normal file
View File

@ -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<TidalDb>`, `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`).

56
tidal-stress/Cargo.toml Normal file
View File

@ -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"] }

98
tidal-stress/k8s/stress-job.yaml Normal file
View File

@ -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"]

199
tidal-stress/src/client.rs Normal file
View File

@ -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<String>,
}
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<String>) -> Result<Self> {
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<u16> {
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<u64> {
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<f32> = (0..dim).map(|_| rng.random::<f32>() - 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
}

32
tidal-stress/src/error.rs Normal file
View File

@ -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<T> = std::result::Result<T, StressError>;

403
tidal-stress/src/main.rs Normal file
View File

@ -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<String>,
/// 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<String>,
/// Bearer key. The deployed cluster requires it; read from $TIDAL_API_KEY.
#[arg(long, env = "TIDAL_API_KEY")]
api_key: Option<String>,
/// 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<usize> = 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<usize>,
) {
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();
}

409
tidal-stress/src/metrics.rs Normal file
View File

@ -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<u64>,
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<OpStats>, // 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
));
}
}

207
tidal-stress/src/scheduler.rs Normal file
View File

@ -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<Workload>,
client: Arc<HttpClient>,
stage: &Stage,
max_inflight: usize,
) -> StageStats {
let (tx, mut rx) = mpsc::unbounded_channel::<Outcome>();
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<Vec<Stage>> {
let mk = |rates: &[f64]| -> Vec<Stage> {
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());
}
}

472
tidal-stress/src/workload.rs Normal file
View File

@ -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> {
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<serde_json::Value>,
}
/// Round-robin over a set of base URLs (e.g. the three region gateways).
struct RoundRobin {
bases: Vec<String>,
next: AtomicUsize,
}
impl RoundRobin {
fn new(bases: Vec<String>) -> 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::<f64>() * 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<String>,
write_bases: Vec<String>,
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::<f64>() * 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<f32> {
// 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::<f32>() - 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<Vec<(OpKind, f64)>> {
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}");
}
}