docs(run-engine): add CK virtual-time A/B benchmark plan and harness
A prod-like A/B benchmark for the concurrency-key virtual-time scheduling change. Includes a method doc (hypotheses, scenarios, metrics, results template), a queue-level micro-benchmark that drives RunQueue with the flag off vs on under identical load (reuses the fairness test harness; inert in CI unless CK_BENCH_REDIS_URL is set), and a deployable end-to-end noisy-neighbor trigger project. All numbers are relative (same box, same load) so the scheduler is isolated from absolute throughput.
This commit is contained in:
Wes Mason
@@ -0,0 +1,253 @@
|
||||
# CK virtual-time scheduling: prod-like A/B benchmark plan
|
||||
|
||||
A method for producing defensible A/B numbers for the concurrency-key
|
||||
virtual-time (SFQ) scheduling change, run on a single prod-shaped box. The two
|
||||
arms are flag OFF (today's age-ordered CK dequeue) and flag ON (virtual-time
|
||||
ordering), under identical load.
|
||||
|
||||
## What the change is (grounded in the branch)
|
||||
|
||||
The concurrency-key dequeue used to serve variants of a base queue in head-message
|
||||
age order. Behind `RUN_ENGINE_CK_VTIME_SCHEDULING_ENABLED` (off by default) it now
|
||||
orders them by start-time fair queueing (SFQ) virtual time:
|
||||
|
||||
- Each CK variant carries a virtual clock in a parallel `:ckVtime` ZSET; a
|
||||
monotonic floor lives in `:ckVtimeFloor`. Both sit under the base queue's
|
||||
`{org}` hash tag, so one atomic Lua script touches all of a queue's state.
|
||||
- The dequeue runs two passes: pass 1 serves the lowest virtual clocks and
|
||||
advances each served variant by `quantum / weight` (weight fixed at 1 today);
|
||||
pass 2 fills any leftover batch slots in today's age order. Pass 2 makes the
|
||||
new command a strict superset of the old one, so it is work-conserving and can
|
||||
never serve fewer runs than today.
|
||||
- Enqueue and nack register a variant into `:ckVtime` at the floor with `NX`, so
|
||||
a brand-new key is reachable from its first enqueue and cannot be parked behind
|
||||
a backlog. This is the case a per-key concurrency cap cannot fix: one tenant
|
||||
sharding work across many keys.
|
||||
- Flag off is byte-identical: the pre-existing Lua scripts run unchanged and no
|
||||
vtime keys are created. The behaviour lives only in new command names.
|
||||
|
||||
Tuning knobs (real env var names, all positive integers, re-clamped in the
|
||||
`RunQueue` constructor):
|
||||
|
||||
| Env var | Default | Meaning |
|
||||
| --- | --- | --- |
|
||||
| `RUN_ENGINE_CK_VTIME_SCHEDULING_ENABLED` | off | master flag |
|
||||
| `RUN_ENGINE_CK_VTIME_QUANTUM` | 1 | virtual-time advance per serve |
|
||||
| `RUN_ENGINE_CK_VTIME_WINDOW_MULTIPLIER` | 3 | pass-1 window = `maxCount * this` |
|
||||
| `RUN_ENGINE_CK_VTIME_STATE_TTL_SECONDS` | 86400 | EXPIRE on the vtime keys |
|
||||
|
||||
Stated limitations this benchmark deliberately probes (from
|
||||
`../../src/run-queue/CK_VTIME_KNOWN_LIMITATIONS.md`): bounded tombstone drift on
|
||||
ack/TTL/DLQ paths; member-name tie-break at equal tags; future-scheduled or
|
||||
retry-backoff variants occupying pass-1 window slots; and the pass-1 window being
|
||||
narrower than the live variant cardinality.
|
||||
|
||||
## Validity: the numbers are RELATIVE, not absolute
|
||||
|
||||
The target is one node on nested/slow storage. Absolute throughput here is NOT
|
||||
prod-scale and must never be reported as such. Every result is a ratio between
|
||||
flag OFF and flag ON, measured under identical load on the same box in the same
|
||||
session. That relative signal is what isolates the scheduler. Two guards keep the
|
||||
arms comparable:
|
||||
|
||||
- Same enqueue set, same timestamps, same step loop / same load generator across
|
||||
OFF and ON.
|
||||
- Fresh queue state between arms (the micro arm FLUSHes a dedicated Redis; the
|
||||
end-to-end arm drains and uses a fresh batch tag), a warmup, and N trials.
|
||||
|
||||
## Two arms
|
||||
|
||||
### Arm 1 (PRIMARY): queue-level micro-benchmark
|
||||
|
||||
Drives the real `RunQueue` directly against a dedicated Redis, comparing OFF vs
|
||||
ON on identical synthetic load. This isolates the scheduler and is where the
|
||||
defensible numbers come from. Because the flag is a `RunQueue` constructor
|
||||
option, one bench process runs both arms in-process: no webapp, no redeploy, no
|
||||
worker clusters. It reuses the existing fairness harness
|
||||
(`../../src/run-queue/tests/ckVtimeFairness.test.ts`): same step loop, same
|
||||
scenario shapes, same conservation checks, plus wall-clock latency, a Redis
|
||||
op-count, N trials, and file output.
|
||||
|
||||
Harness: `../../src/run-queue/bench/ckMicroBench.bench.test.ts`. It is inert in CI
|
||||
(only runs when `CK_BENCH_REDIS_URL` is set) and FLUSHes its target Redis between
|
||||
arms, so it must point only at a dedicated throwaway instance.
|
||||
|
||||
Each step makes one `maxCount = 10` dequeue call, records `(step, key, messageId,
|
||||
wallMs)` per served message, then acks in-flight messages whose logical hold has
|
||||
elapsed. Wait per message = the step it was served at (all load is pre-enqueued
|
||||
at step 0). The logical schedule is deterministic, so step-based metrics are
|
||||
identical across trials (the harness asserts this); trials exist to stabilise the
|
||||
wall-clock latency and op-count.
|
||||
|
||||
### Arm 2 (END-TO-END): deployed tasks on the worker clusters
|
||||
|
||||
The realism check on top of arm 1. A deployed task on a shared base queue with
|
||||
per-run concurrency keys, driven by a noisy-neighbor load generator: tenant A
|
||||
floods across many keys, tenant B sends a few. Each run carries a per-run region
|
||||
so the load also spreads across the three managed worker groups
|
||||
(`trigger-regiona/b/c`), exercising multi-cluster placement. Latency is read back
|
||||
per run as `startedAt - createdAt`.
|
||||
|
||||
Project: `e2e-tasks/` (deployable, secret-free). Contention is forced by pinning
|
||||
the environment concurrency ceiling low (so many keys contend for a few slots and
|
||||
the CK dequeue order decides who starts first); the per-key lane width is 1 in the
|
||||
task config for reproducibility.
|
||||
|
||||
Because the flag is server-side here, arm 2 is a manual OFF-then-ON: set the flag,
|
||||
redeploy the control plane, run the load, collect; flip the flag, redeploy, run
|
||||
the load again, collect. The exact toggle + redeploy belongs to the operator
|
||||
runbook.
|
||||
|
||||
## Hypotheses (tied to the change)
|
||||
|
||||
1. **Bounded wait behind a backlog.** A light key arriving behind a big backlog
|
||||
waits O(number of active keys) under vtime, versus O(backlog size) under the
|
||||
baseline (which drains the backlog first). Micro: `ckSkew`, `ckTrickle`
|
||||
victim wait p95/p99 drops sharply ON vs OFF. E2E: tenant B start-latency p95
|
||||
stays bounded as tenant A's backlog grows.
|
||||
2. **Sharding across many keys cannot starve others.** A tenant fanning out over
|
||||
many concurrency keys (the case a per-key cap cannot fix) does not starve a
|
||||
light key, because the light key registers at the floor and is reachable in
|
||||
pass 1. Micro: `ckSybil` victim first-serve step is small ON (near-immediate)
|
||||
and its wait ratio drops; `ckManyKeys` shows no permanent starvation even
|
||||
when cardinality exceeds the pass-1 window. E2E: tenant B (few keys) is not
|
||||
starved by tenant A's many-key flood.
|
||||
3. **Work conservation.** A lone backlogged tenant still drains at full rate;
|
||||
the fair order adds no idle time when nothing else contends. Micro:
|
||||
`ckHeavyIdle` drain step ON equals OFF exactly. No-harm corollary: the
|
||||
symmetric `ckBalanced` case is not made worse.
|
||||
|
||||
## Scenarios
|
||||
|
||||
### Arm 1 (micro), all ported from the fairness-spike shapes
|
||||
|
||||
| scenario | shape | env limit | hold | probes |
|
||||
| --- | --- | --- | --- | --- |
|
||||
| `ckSkew` | heavy 120 backlog + 4 light x 10 | 1 | 3 | starvation (H1) |
|
||||
| `ckTrickle` | bulk 120 + 2 trickle x 15 | 1 | 3 | starvation (H1) |
|
||||
| `ckSybil` | 20 attacker x 8 + 1 light x 10 | 25 | 3 | sharding/sybil (H2) |
|
||||
| `ckManyKeys` | 60 attacker x 8 (tied head) + 1 light x 10 | 25 | 3 | window limitation, no permanent starvation (H2) |
|
||||
| `ckBalanced` | 4 symmetric x 25 | 4 | 3 | no-harm (H3) |
|
||||
| `ckHeavyIdle` | 1 key x 60 | 25 | 3 | work conservation (H3) |
|
||||
|
||||
### Arm 2 (end-to-end), noisy-neighbor
|
||||
|
||||
- Tenant A: `A_KEYS` (default 40) keys x `A_PER_KEY` (default 5) runs = the flood.
|
||||
- Tenant B: `B_KEYS` (default 2) keys x `B_PER_KEY` (default 5) runs = the victim.
|
||||
- Per-run hold `HOLD_MS` (default 1500). Runs round-robined across the three
|
||||
regions. Environment concurrency ceiling pinned low (e.g. 5) so the keys
|
||||
actually contend.
|
||||
- Optional placement variant: pin tenant A to one region and tenant B to another
|
||||
to separate scheduler effects from cross-cluster effects.
|
||||
|
||||
## Metrics and collection
|
||||
|
||||
| metric | what it shows | source |
|
||||
| --- | --- | --- |
|
||||
| victim wait p50/p95/p99 | starvation relief | micro: serve step; e2e: `startedAt - createdAt` per tenant |
|
||||
| victim first-serve (starvation bound) | reachability at the floor | micro: first serve step for the victim key |
|
||||
| drain step / total served | work conservation, no loss/dup | micro: last serve step + unique messageId count |
|
||||
| Jain's fairness index | share fairness during contention | micro: over per-key contention-window serves |
|
||||
| dequeue call p95 (ms) | scheduler op cost (relative) | micro: wall-clock around each dequeue call |
|
||||
| redis ops (dequeue+ack) | per-dequeue overhead | micro: `CONFIG RESETSTAT` then `INFO commandstats` |
|
||||
|
||||
Jain's index over per-key served counts `x_i`: `J = (sum x_i)^2 / (n * sum
|
||||
x_i^2)`. 1.0 is perfectly fair; `1/n` means one key took everything.
|
||||
|
||||
The micro harness writes `ck-micro-results.json` and `ck-micro-results.md`
|
||||
(the results table below) to `CK_BENCH_OUT`.
|
||||
|
||||
For the end-to-end arm, the primary source is the Runs API by tag
|
||||
(`startedAt - createdAt`), which `collect.ts` reads. Two alternatives give a
|
||||
tighter dequeue-only timestamp if the API delta looks noisy:
|
||||
|
||||
- TRQL `runs` (verify column names with the query schema first): per-run
|
||||
`createdAt` and `startedAt`, filtered by the batch/arm tag.
|
||||
- The run-engine Postgres `TaskRun` timestamps directly (createdAt and the first
|
||||
attempt/started timestamp), if API round-trips add too much jitter.
|
||||
|
||||
## Reproducible A/B procedure
|
||||
|
||||
### Arm 1 (micro)
|
||||
|
||||
1. Stand up a dedicated throwaway Redis reachable from the harness host (a local
|
||||
forward is fine). Nothing else may use it.
|
||||
2. From the run-engine package:
|
||||
|
||||
```bash
|
||||
CK_BENCH_REDIS_URL=redis://127.0.0.1:6399 \
|
||||
CK_BENCH_TRIALS=5 CK_BENCH_OUT=./bench-results \
|
||||
pnpm exec vitest run src/run-queue/bench/ckMicroBench.bench.test.ts
|
||||
```
|
||||
|
||||
Both arms (OFF, then ON) run in one process per scenario, FLUSHing between
|
||||
arms. Knob sweep: add `CK_BENCH_QUANTUM` / `CK_BENCH_WINDOW_MULT`. Scenario
|
||||
subset: `CK_BENCH_SCENARIOS=ckSybil,ckSkew`.
|
||||
3. Read `bench-results/ck-micro-results.md`. The harness fails the run if either
|
||||
arm loses or double-serves a message, so a green run means the comparison is
|
||||
sound.
|
||||
|
||||
### Arm 2 (end-to-end)
|
||||
|
||||
1. Deploy `e2e-tasks/` to the bench project's PROD environment (dev
|
||||
short-circuits worker-group routing, so it must be prod). Pin the prod env
|
||||
concurrency ceiling low.
|
||||
2. Warmup: trigger a handful of runs, confirm they start on each region, discard.
|
||||
3. **Arm OFF:** ensure the flag is off and the control plane is redeployed; then
|
||||
`ARM=off BATCH=<id1> pnpm loadgen`, wait for drain, `ARM=off BATCH=<id1> pnpm
|
||||
collect`.
|
||||
4. **Arm ON:** flip the flag on, redeploy the control plane, drain/clear queue
|
||||
state; then `ARM=on BATCH=<id2> pnpm loadgen`, wait for drain, `ARM=on
|
||||
BATCH=<id2> pnpm collect`.
|
||||
5. Repeat both arms N times with fresh batch ids; compare per-tenant latency.
|
||||
|
||||
The exact deploy, flag-toggle, and secret-key retrieval steps for the specific
|
||||
box are in the operator runbook (kept out of this repo).
|
||||
|
||||
## Results template (paste into the PR)
|
||||
|
||||
Fill from `ck-micro-results.md` (arm 1) and `e2e-summary.md` (arm 2). Keep the
|
||||
"relative only" caveat in the PR text.
|
||||
|
||||
### Arm 1 (micro), quantum 1 / window x3, N trials, dedicated Redis on the box
|
||||
|
||||
| scenario | metric | baseline (OFF) | vtime (ON) | delta |
|
||||
| --- | --- | --- | --- | --- |
|
||||
| **ckSkew** | victim wait p95 (steps) | | | |
|
||||
| | victim wait p99 | | | |
|
||||
| | victim first-serve | | | |
|
||||
| | drain step | | | |
|
||||
| | dequeue call p95 (ms) | | | |
|
||||
| **ckTrickle** | victim wait p95 | | | |
|
||||
| | victim first-serve | | | |
|
||||
| **ckSybil** | victim wait p95 | | | |
|
||||
| | victim first-serve | | | |
|
||||
| | Jain index (contention) | | | |
|
||||
| **ckManyKeys** | victim first-serve | | | |
|
||||
| | drain step | | | |
|
||||
| **ckBalanced** | worst-key wait p95 | | | |
|
||||
| **ckHeavyIdle** | drain step | | | |
|
||||
| (all) | redis ops (dequeue+ack) | | | |
|
||||
|
||||
### Arm 2 (end-to-end), noisy-neighbor across regiona/b/c, env cap N
|
||||
|
||||
| tenant | metric | baseline (OFF) | vtime (ON) | delta |
|
||||
| --- | --- | --- | --- | --- |
|
||||
| B (victim, few keys) | start latency p50 (ms) | | | |
|
||||
| B | start latency p95 | | | |
|
||||
| B | start latency p99 | | | |
|
||||
| A (flood, many keys) | start latency p95 | | | |
|
||||
|
||||
Expected direction: B's p95/p99 drop substantially ON; A's is similar or slightly
|
||||
higher ON (it stops jumping the queue); `ckHeavyIdle` drain step is exactly equal;
|
||||
`ckBalanced` worst-key wait is within noise.
|
||||
|
||||
## How to reproduce on the box (short)
|
||||
|
||||
1. Dedicated Redis up, forwarded locally. Run the arm-1 vitest command above;
|
||||
collect `ck-micro-results.md`.
|
||||
2. Deploy `e2e-tasks/` to prod, pin the env concurrency ceiling, warm up.
|
||||
3. Flag OFF: redeploy control plane, loadgen + collect. Flag ON: redeploy,
|
||||
loadgen + collect. N trials.
|
||||
4. Paste both tables into the PR under a "relative numbers on a single
|
||||
prod-shaped box" heading.
|
||||
@@ -0,0 +1,28 @@
|
||||
# CK virtual-time end-to-end bench tasks
|
||||
|
||||
A tiny self-contained trigger.dev project used as the END-TO-END arm of the CK
|
||||
virtual-time A/B (see `../2026-07-26-ck-vtime-benchmark.md`). It is deployed to a
|
||||
self-hosted instance and is not part of the monorepo build.
|
||||
|
||||
- `src/trigger/ckBench.ts` — one `ck-bench` task on a shared base queue; per-run
|
||||
`concurrencyKey` makes the CK variants; a slot hold forces contention.
|
||||
- `src/loadgen.ts` — noisy-neighbor load: tenant A floods across many keys,
|
||||
tenant B sends a few; each run is tagged and carries a per-run `region`.
|
||||
- `src/collect.ts` — reads per-run `createdAt`/`startedAt` by tag and reports
|
||||
per-tenant enqueue->start latency p50/p95/p99.
|
||||
|
||||
Everything reads credentials from the environment (`TRIGGER_API_URL`,
|
||||
`TRIGGER_SECRET_KEY`); nothing is hard-coded. Deploy with the CLI (`-p <ref>`).
|
||||
The feature flag is server-side and is flipped by the operator between arms, not
|
||||
by this project. Exact instance coordinates and the toggle live in the operator
|
||||
runbook, kept outside this public repo.
|
||||
|
||||
```bash
|
||||
pnpm install
|
||||
# deploy (project ref on the CLI)
|
||||
TRIGGER_PROJECT_REF=<ref> npx trigger.dev@latest deploy --self-hosted --profile <profile> -p <ref>
|
||||
# generate load for one arm (flag already set + redeployed server-side)
|
||||
ARM=off BATCH=run1 pnpm loadgen
|
||||
# collect that arm
|
||||
ARM=off BATCH=run1 pnpm collect
|
||||
```
|
||||
@@ -0,0 +1,19 @@
|
||||
{
|
||||
"name": "ck-vtime-e2e-bench",
|
||||
"version": "0.0.0",
|
||||
"private": true,
|
||||
"type": "module",
|
||||
"description": "End-to-end noisy-neighbor load harness for the CK virtual-time scheduling A/B. Deployed to a self-hosted trigger.dev instance; not part of the monorepo build.",
|
||||
"scripts": {
|
||||
"loadgen": "tsx src/loadgen.ts",
|
||||
"collect": "tsx src/collect.ts"
|
||||
},
|
||||
"dependencies": {
|
||||
"@trigger.dev/sdk": "latest"
|
||||
},
|
||||
"devDependencies": {
|
||||
"trigger.dev": "latest",
|
||||
"tsx": "^4.19.0",
|
||||
"typescript": "^5.5.0"
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,98 @@
|
||||
/**
|
||||
* Collect per-tenant enqueue->start latency for one END-TO-END arm.
|
||||
*
|
||||
* Lists the runs for a batch by tag, reads each run's createdAt (enqueue) and
|
||||
* startedAt (execution start), and reports per-tenant p50/p95/p99 of
|
||||
* (startedAt - createdAt). Run once per arm; point --arm/BATCH at the tags the
|
||||
* loadgen used.
|
||||
*
|
||||
* The headline metric is tenant B's start latency: under the baseline it should
|
||||
* grow with tenant A's backlog (B waits behind the flood); under vtime it should
|
||||
* stay bounded (B takes its fair turn). Tenant A's latency is reported for
|
||||
* context and is expected to be similar or slightly higher under vtime.
|
||||
*
|
||||
* NOTE ON THE TIMESTAMP: startedAt - createdAt is the honest run-start latency a
|
||||
* reviewer cares about (queue wait + dequeue + worker pickup). For a tighter
|
||||
* dequeue-only number, use the Postgres/TRQL alternative in the benchmark doc.
|
||||
*
|
||||
* Auth + config (env):
|
||||
* TRIGGER_API_URL, TRIGGER_SECRET_KEY (prod env secret key)
|
||||
* ARM=off|on BATCH=<id> OUT=./e2e-results
|
||||
*/
|
||||
import { configure, runs } from "@trigger.dev/sdk";
|
||||
import { appendFileSync, mkdirSync, writeFileSync } from "node:fs";
|
||||
|
||||
function pct(sorted: number[], p: number): number {
|
||||
if (sorted.length === 0) return NaN;
|
||||
const idx = Math.min(sorted.length - 1, Math.max(0, Math.ceil((p / 100) * sorted.length) - 1));
|
||||
return sorted[idx]!;
|
||||
}
|
||||
function summarize(xs: number[]) {
|
||||
const s = [...xs].sort((a, b) => a - b);
|
||||
const mean = xs.length ? xs.reduce((a, b) => a + b, 0) / xs.length : NaN;
|
||||
return { count: xs.length, mean, p50: pct(s, 50), p95: pct(s, 95), p99: pct(s, 99) };
|
||||
}
|
||||
|
||||
async function main() {
|
||||
const apiURL = process.env.TRIGGER_API_URL;
|
||||
const accessToken = process.env.TRIGGER_SECRET_KEY;
|
||||
if (!apiURL || !accessToken) throw new Error("Set TRIGGER_API_URL and TRIGGER_SECRET_KEY.");
|
||||
configure({ baseURL: apiURL, accessToken });
|
||||
|
||||
const arm = (process.env.ARM ?? "off") as "off" | "on";
|
||||
const batch = process.env.BATCH;
|
||||
const outDir = process.env.OUT ?? "./e2e-results";
|
||||
if (!batch) throw new Error("Set BATCH=<id> to the loadgen batch id.");
|
||||
|
||||
const waitsByTenant = new Map<string, number[]>();
|
||||
let total = 0;
|
||||
let missingStart = 0;
|
||||
|
||||
// Page through every run carrying this batch tag. BATCH is unique per arm
|
||||
// (e.g. off-1 / on-1), so the single batch tag identifies the arm; filtering
|
||||
// on one tag avoids any multi-tag AND/OR ambiguity in the runs filter.
|
||||
for await (const run of runs.list({ tag: `batch:${batch}`, limit: 100 })) {
|
||||
total++;
|
||||
const detail = await runs.retrieve(run.id);
|
||||
const createdAt = detail.createdAt?.getTime();
|
||||
const startedAt = detail.startedAt?.getTime();
|
||||
if (createdAt === undefined || startedAt === undefined) {
|
||||
missingStart++;
|
||||
continue;
|
||||
}
|
||||
const tenantTag = (detail.tags ?? []).find((t) => t.startsWith("tenant:")) ?? "tenant:?";
|
||||
const tenant = tenantTag.slice("tenant:".length);
|
||||
const wait = startedAt - createdAt;
|
||||
(waitsByTenant.get(tenant) ?? waitsByTenant.set(tenant, []).get(tenant)!).push(wait);
|
||||
}
|
||||
|
||||
const perTenant: Record<string, ReturnType<typeof summarize>> = {};
|
||||
for (const [tenant, xs] of waitsByTenant) perTenant[tenant] = summarize(xs);
|
||||
|
||||
const report = { arm, batch, total, missingStart, unit: "ms (startedAt - createdAt)", perTenant };
|
||||
|
||||
mkdirSync(outDir, { recursive: true });
|
||||
writeFileSync(`${outDir}/e2e-${batch}-${arm}.json`, JSON.stringify(report, null, 2));
|
||||
|
||||
// Append a human row per tenant to a shared markdown file so OFF and ON land
|
||||
// in one table you can eyeball before running the joiner.
|
||||
const mdPath = `${outDir}/e2e-summary.md`;
|
||||
const rows = Object.entries(perTenant)
|
||||
.map(
|
||||
([tenant, s]) =>
|
||||
`| ${batch} | ${arm} | ${tenant} | ${s.count} | ${s.mean.toFixed(0)} | ${s.p50.toFixed(0)} | ${s.p95.toFixed(0)} | ${s.p99.toFixed(0)} |`
|
||||
)
|
||||
.join("\n");
|
||||
appendFileSync(
|
||||
mdPath,
|
||||
`\n<!-- batch ${batch} arm ${arm} -->\n| batch | arm | tenant | runs | mean ms | p50 | p95 | p99 |\n| --- | --- | --- | --- | --- | --- | --- | --- |\n${rows}\n`
|
||||
);
|
||||
|
||||
console.log(`[collect] arm=${arm} batch=${batch} runs=${total} missingStart=${missingStart}`);
|
||||
console.log(JSON.stringify(perTenant, null, 2));
|
||||
}
|
||||
|
||||
main().catch((err) => {
|
||||
console.error(err);
|
||||
process.exit(1);
|
||||
});
|
||||
@@ -0,0 +1,100 @@
|
||||
/**
|
||||
* Noisy-neighbor load generator for the CK virtual-time END-TO-END A/B arm.
|
||||
*
|
||||
* Triggers the deployed `ck-bench` task on a self-hosted instance. Tenant A
|
||||
* floods the base queue across MANY concurrency keys (the sharding/sybil case a
|
||||
* per-key cap cannot fix); tenant B sends a few runs on a couple of keys. Each
|
||||
* run is tagged so `collect.ts` can measure per-tenant enqueue->start latency,
|
||||
* and each run carries a per-run `region` so the load spreads across the managed
|
||||
* worker groups (multi-cluster placement).
|
||||
*
|
||||
* This does NOT flip the feature flag: the flag is server-side (see the operator
|
||||
* runbook). Run this once per arm AFTER the operator has set the flag and
|
||||
* redeployed the control plane, passing the matching --arm so the tags line up.
|
||||
*
|
||||
* Auth (from env):
|
||||
* TRIGGER_API_URL e.g. https://<instance>
|
||||
* TRIGGER_SECRET_KEY the PROD environment secret key of the bench project
|
||||
*
|
||||
* Config (from env, with defaults):
|
||||
* ARM=off|on tag only; must match the server flag state
|
||||
* BATCH=<id> unique per A/B run pair (defaults to a timestamp)
|
||||
* HOLD_MS=1500 per-run slot hold
|
||||
* A_KEYS=40 A_PER_KEY=5 tenant A flood shape
|
||||
* B_KEYS=2 B_PER_KEY=5 tenant B light shape
|
||||
* REGIONS=trigger-regiona,trigger-regionb,trigger-regionc
|
||||
* runs are round-robined across these worker groups
|
||||
*/
|
||||
import { configure, tasks } from "@trigger.dev/sdk";
|
||||
import type { ckBenchTask } from "./trigger/ckBench.js";
|
||||
|
||||
function envInt(name: string, dflt: number): number {
|
||||
const v = process.env[name];
|
||||
return v === undefined ? dflt : Number(v);
|
||||
}
|
||||
|
||||
async function main() {
|
||||
const apiURL = process.env.TRIGGER_API_URL;
|
||||
const accessToken = process.env.TRIGGER_SECRET_KEY;
|
||||
if (!apiURL || !accessToken) {
|
||||
throw new Error("Set TRIGGER_API_URL and TRIGGER_SECRET_KEY (prod env secret key).");
|
||||
}
|
||||
configure({ baseURL: apiURL, accessToken });
|
||||
|
||||
const arm = (process.env.ARM ?? "off") as "off" | "on";
|
||||
const batch = process.env.BATCH ?? `b${Date.now()}`;
|
||||
const holdMs = envInt("HOLD_MS", 1500);
|
||||
const aKeys = envInt("A_KEYS", 40);
|
||||
const aPerKey = envInt("A_PER_KEY", 5);
|
||||
const bKeys = envInt("B_KEYS", 2);
|
||||
const bPerKey = envInt("B_PER_KEY", 5);
|
||||
const regions = (process.env.REGIONS ?? "trigger-regiona,trigger-regionb,trigger-regionc")
|
||||
.split(",")
|
||||
.map((s) => s.trim())
|
||||
.filter(Boolean);
|
||||
|
||||
type Item = {
|
||||
payload: { holdMs: number; tenant: string; key: string; arm: "off" | "on"; batch: string };
|
||||
options: { concurrencyKey: string; region: string; tags: string[] };
|
||||
};
|
||||
|
||||
const items: Item[] = [];
|
||||
let n = 0;
|
||||
const push = (tenant: "A" | "B", key: string) => {
|
||||
const region = regions[n % regions.length]!;
|
||||
n++;
|
||||
items.push({
|
||||
payload: { holdMs, tenant, key, arm, batch },
|
||||
options: {
|
||||
concurrencyKey: key,
|
||||
region,
|
||||
tags: [`ckbench`, `arm:${arm}`, `tenant:${tenant}`, `batch:${batch}`],
|
||||
},
|
||||
});
|
||||
};
|
||||
|
||||
for (let k = 0; k < aKeys; k++) for (let i = 0; i < aPerKey; i++) push("A", `A-${k}`);
|
||||
for (let k = 0; k < bKeys; k++) for (let i = 0; i < bPerKey; i++) push("B", `B-${k}`);
|
||||
|
||||
// Interleave A and B enqueues so B does not simply arrive first; the point is
|
||||
// whether B's few runs start promptly WHILE A's flood is queued.
|
||||
items.sort((x, y) => x.payload.key.localeCompare(y.payload.key));
|
||||
|
||||
console.log(
|
||||
`[loadgen] arm=${arm} batch=${batch} total=${items.length} (A=${aKeys}x${aPerKey}, B=${bKeys}x${bPerKey}) regions=${regions.join(",")} holdMs=${holdMs}`
|
||||
);
|
||||
|
||||
const handle = await tasks.batchTrigger<typeof ckBenchTask>(
|
||||
"ck-bench",
|
||||
items.map((it) => ({ payload: it.payload, options: it.options }))
|
||||
);
|
||||
|
||||
console.log(
|
||||
`[loadgen] batch triggered: ${handle.batchId} (${items.length} runs). BATCH=${batch}`
|
||||
);
|
||||
}
|
||||
|
||||
main().catch((err) => {
|
||||
console.error(err);
|
||||
process.exit(1);
|
||||
});
|
||||
@@ -0,0 +1,42 @@
|
||||
import { logger, queue, task } from "@trigger.dev/sdk";
|
||||
|
||||
// One shared base queue. Per-run concurrencyKey (set at trigger time) creates the
|
||||
// concurrency-key variants whose dequeue ORDER the change under test governs.
|
||||
//
|
||||
// concurrencyLimit is the PER-KEY lane width. Set it to 1 so each key holds one
|
||||
// slot at a time; cross-key contention is then forced by the ENVIRONMENT
|
||||
// concurrency ceiling (pin RuntimeEnvironment.maximumConcurrencyLimit low, e.g.
|
||||
// 5, on the prod env of the bench project). With N keys all wanting to run and
|
||||
// only a few env slots, the CK dequeue decides who starts first: that ordering
|
||||
// is exactly OFF (age) vs ON (virtual time).
|
||||
export const ckBenchQueue = queue({
|
||||
name: "ck-bench",
|
||||
concurrencyLimit: 1,
|
||||
});
|
||||
|
||||
export type CkBenchPayload = {
|
||||
// logical hold: how long the run occupies its slot, in ms
|
||||
holdMs: number;
|
||||
// carried through for grouping in analysis (also set as a tag by the loadgen)
|
||||
tenant: string;
|
||||
key: string;
|
||||
arm: "off" | "on";
|
||||
batch: string;
|
||||
};
|
||||
|
||||
export const ckBenchTask = task({
|
||||
id: "ck-bench",
|
||||
queue: ckBenchQueue,
|
||||
run: async (payload: CkBenchPayload) => {
|
||||
logger.info("ck-bench start", {
|
||||
tenant: payload.tenant,
|
||||
key: payload.key,
|
||||
arm: payload.arm,
|
||||
batch: payload.batch,
|
||||
});
|
||||
// Occupy the slot for the hold so concurrency actually contends. A plain
|
||||
// timer is enough: this task exists only to hold a slot, not to do work.
|
||||
await new Promise((resolve) => setTimeout(resolve, payload.holdMs));
|
||||
return { tenant: payload.tenant, key: payload.key, arm: payload.arm };
|
||||
},
|
||||
});
|
||||
@@ -0,0 +1,11 @@
|
||||
import { defineConfig } from "@trigger.dev/sdk";
|
||||
|
||||
// The project ref is passed on the CLI (`-p <ref>`) at deploy time, so it is not
|
||||
// hard-coded here. This keeps the harness portable and secret-free in the repo.
|
||||
export default defineConfig({
|
||||
project: process.env.TRIGGER_PROJECT_REF ?? "proj_REPLACE_ME",
|
||||
runtime: "node",
|
||||
logLevel: "info",
|
||||
maxDuration: 120,
|
||||
dirs: ["./src/trigger"],
|
||||
});
|
||||
@@ -0,0 +1,13 @@
|
||||
{
|
||||
"compilerOptions": {
|
||||
"target": "ES2022",
|
||||
"module": "ESNext",
|
||||
"moduleResolution": "Bundler",
|
||||
"strict": true,
|
||||
"esModuleInterop": true,
|
||||
"skipLibCheck": true,
|
||||
"types": ["node"],
|
||||
"noEmit": true
|
||||
},
|
||||
"include": ["src/**/*.ts", "trigger.config.ts"]
|
||||
}
|
||||
@@ -0,0 +1,555 @@
|
||||
import { createRedisClient } from "@internal/redis";
|
||||
import { trace } from "@internal/tracing";
|
||||
import { Logger } from "@trigger.dev/core/logger";
|
||||
import { Decimal } from "@trigger.dev/database";
|
||||
import { mkdirSync, writeFileSync } from "node:fs";
|
||||
import { performance } from "node:perf_hooks";
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { FairQueueSelectionStrategy } from "../fairQueueSelectionStrategy.js";
|
||||
import { RunQueue } from "../index.js";
|
||||
import { RunQueueFullKeyProducer } from "../keyProducer.js";
|
||||
import type { InputPayload } from "../types.js";
|
||||
|
||||
// CK virtual-time PRIMARY arm: a queue-level A/B micro-benchmark.
|
||||
//
|
||||
// This drives the REAL RunQueue against an EXTERNAL Redis (not a testcontainer)
|
||||
// and compares flag OFF (age-ordered CK dequeue) vs flag ON (SFQ virtual time)
|
||||
// under identical load. The flag is a RunQueue constructor option, so one bench
|
||||
// process runs both arms in-process: no webapp, no redeploy.
|
||||
//
|
||||
// It is a deliberate, defensible A/B: the two arms enqueue the exact same
|
||||
// messages with the exact same timestamps and drive the exact same step loop.
|
||||
// The isolation and reuse are inherited from tests/ckVtimeFairness.test.ts (the
|
||||
// step loop, scenario shapes, conservation checks are the same); this file adds
|
||||
// wall-clock dequeue latency, a Redis op-count, N trials, and file output.
|
||||
//
|
||||
// It is INERT in CI: the suite only runs when CK_BENCH_REDIS_URL is set, so
|
||||
// `pnpm run test` collects it as a skipped describe and never touches a network.
|
||||
//
|
||||
// Run it (from the run-engine package, pointed at a dedicated throwaway Redis):
|
||||
// CK_BENCH_REDIS_URL=redis://127.0.0.1:6399 \
|
||||
// CK_BENCH_TRIALS=5 CK_BENCH_OUT=./bench-results \
|
||||
// pnpm exec vitest run src/run-queue/bench/ckMicroBench.bench.test.ts
|
||||
//
|
||||
// Knob sweep (optional, defaults match production defaults 1 / 3):
|
||||
// CK_BENCH_QUANTUM=1 CK_BENCH_WINDOW_MULT=3
|
||||
//
|
||||
// WARNING: the bench FLUSHDBs the target Redis between arms. Point it ONLY at a
|
||||
// dedicated throwaway instance, never at a shared or production Redis.
|
||||
|
||||
const REDIS_URL = process.env.CK_BENCH_REDIS_URL;
|
||||
const TRIALS = Math.max(1, Number(process.env.CK_BENCH_TRIALS ?? "5"));
|
||||
const OUT_DIR = process.env.CK_BENCH_OUT ?? "./bench-results";
|
||||
const QUANTUM = Math.max(1, Number(process.env.CK_BENCH_QUANTUM ?? "1"));
|
||||
const WINDOW_MULT = Math.max(1, Number(process.env.CK_BENCH_WINDOW_MULT ?? "3"));
|
||||
const SCENARIO_FILTER = process.env.CK_BENCH_SCENARIOS?.split(",").map((s) => s.trim());
|
||||
|
||||
const keys = new RunQueueFullKeyProducer();
|
||||
|
||||
const testOptions = {
|
||||
name: "rq",
|
||||
tracer: trace.getTracer("rq"),
|
||||
workers: 1,
|
||||
defaultEnvConcurrency: 25,
|
||||
logger: new Logger("RunQueue", "error"),
|
||||
retryOptions: {
|
||||
maxAttempts: 5,
|
||||
factor: 1.1,
|
||||
minTimeoutInMs: 100,
|
||||
maxTimeoutInMs: 1_000,
|
||||
randomize: true,
|
||||
},
|
||||
keys,
|
||||
};
|
||||
|
||||
const authenticatedEnvDev = {
|
||||
id: "e1234",
|
||||
type: "DEVELOPMENT" as const,
|
||||
maximumConcurrencyLimit: 10,
|
||||
concurrencyLimitBurstFactor: new Decimal(2.0),
|
||||
project: { id: "p1234" },
|
||||
organization: { id: "o1234" },
|
||||
};
|
||||
|
||||
function redisConn() {
|
||||
const u = new URL(REDIS_URL!);
|
||||
return { host: u.hostname, port: Number(u.port || "6379") };
|
||||
}
|
||||
|
||||
function createQueue(keyPrefix: string, vtimeEnabled: boolean) {
|
||||
const conn = redisConn();
|
||||
return new RunQueue({
|
||||
...testOptions,
|
||||
masterQueueConsumersDisabled: true,
|
||||
workerOptions: { disabled: true },
|
||||
ckVirtualTimeScheduling: {
|
||||
enabled: vtimeEnabled,
|
||||
quantum: QUANTUM,
|
||||
scanWindowMultiplier: WINDOW_MULT,
|
||||
},
|
||||
queueSelectionStrategy: new FairQueueSelectionStrategy({
|
||||
redis: { keyPrefix, host: conn.host, port: conn.port },
|
||||
keys,
|
||||
}),
|
||||
redis: { keyPrefix, host: conn.host, port: conn.port },
|
||||
});
|
||||
}
|
||||
|
||||
function makeMessage(overrides: Partial<InputPayload> = {}): InputPayload {
|
||||
return {
|
||||
runId: "r1",
|
||||
taskIdentifier: "task/my-task",
|
||||
orgId: "o1234",
|
||||
projectId: "p1234",
|
||||
environmentId: "e1234",
|
||||
environmentType: "DEVELOPMENT",
|
||||
queue: "task/my-task",
|
||||
timestamp: Date.now(),
|
||||
attempt: 0,
|
||||
...overrides,
|
||||
};
|
||||
}
|
||||
|
||||
type ScenarioMessage = { runId: string; ck: string; timestamp: number };
|
||||
|
||||
type Scenario = {
|
||||
name: string;
|
||||
// Human label for the "victim" the fairness hypothesis is about.
|
||||
victimLabel: string;
|
||||
// Classifies a concurrency key as the victim (the light/starved tenant).
|
||||
isVictim: (ck: string) => boolean;
|
||||
messages: ScenarioMessage[];
|
||||
envConcurrencyLimit: number;
|
||||
holdSteps: number;
|
||||
maxSteps: number;
|
||||
};
|
||||
|
||||
type ServeRecord = { step: number; ck: string; messageId: string; wallMs: number };
|
||||
|
||||
type ArmResult = {
|
||||
serves: ServeRecord[];
|
||||
drainStep: number;
|
||||
contentionByCk: Map<string, number>;
|
||||
contentionTotal: number;
|
||||
callLatenciesMs: number[];
|
||||
redisCalls: number;
|
||||
};
|
||||
|
||||
// ---- scenario shapes (ported values from the fairness spike, same as the
|
||||
// tests/ckVtimeFairness.test.ts scenarios; nothing imported from the spike) ----
|
||||
|
||||
function buildScenarios(): Scenario[] {
|
||||
const t0 = Date.now() - 500_000;
|
||||
const all: Scenario[] = [];
|
||||
|
||||
// ckSkew (starvation): heavy 120-msg backlog on an old shared head, 4 light
|
||||
// keys x 10 on later heads. Serialized contention (env limit 1) is where the
|
||||
// baseline's age order starves the light keys.
|
||||
{
|
||||
const messages: ScenarioMessage[] = [];
|
||||
for (let i = 0; i < 120; i++)
|
||||
messages.push({ runId: `heavy-${i}`, ck: "heavy", timestamp: t0 });
|
||||
for (let i = 0; i < 10; i++)
|
||||
for (let k = 0; k < 4; k++)
|
||||
messages.push({
|
||||
runId: `light${k}-${i}`,
|
||||
ck: `light${k}`,
|
||||
timestamp: t0 + 10_000 + i * 4 + k,
|
||||
});
|
||||
all.push({
|
||||
name: "ckSkew",
|
||||
victimLabel: "light keys",
|
||||
isVictim: (ck) => ck.startsWith("light"),
|
||||
messages,
|
||||
envConcurrencyLimit: 1,
|
||||
holdSteps: 3,
|
||||
maxSteps: 1_000,
|
||||
});
|
||||
}
|
||||
|
||||
// ckTrickle (starvation): bulk 120 + 2 trickle keys x 15.
|
||||
{
|
||||
const messages: ScenarioMessage[] = [];
|
||||
for (let i = 0; i < 120; i++) messages.push({ runId: `bulk-${i}`, ck: "bulk", timestamp: t0 });
|
||||
for (let i = 0; i < 15; i++)
|
||||
for (let k = 0; k < 2; k++)
|
||||
messages.push({
|
||||
runId: `trickle${k}-${i}`,
|
||||
ck: `trickle${k}`,
|
||||
timestamp: t0 + 10_000 + i * 2 + k,
|
||||
});
|
||||
all.push({
|
||||
name: "ckTrickle",
|
||||
victimLabel: "trickle keys",
|
||||
isVictim: (ck) => ck.startsWith("trickle"),
|
||||
messages,
|
||||
envConcurrencyLimit: 1,
|
||||
holdSteps: 3,
|
||||
maxSteps: 1_000,
|
||||
});
|
||||
}
|
||||
|
||||
// ckSybil (noisy-neighbor caps cannot fix): 20 attacker keys x 8 (older
|
||||
// heads) + 1 light key x 10 (newer). 21 variants against a batch of 10.
|
||||
{
|
||||
const messages: ScenarioMessage[] = [];
|
||||
for (let i = 0; i < 8; i++)
|
||||
for (let k = 0; k < 20; k++) {
|
||||
const ck = `att${String(k).padStart(2, "0")}`;
|
||||
messages.push({ runId: `${ck}-${i}`, ck, timestamp: t0 + i * 20 + k });
|
||||
}
|
||||
for (let i = 0; i < 10; i++)
|
||||
messages.push({ runId: `light-${i}`, ck: "light", timestamp: t0 + 50_000 + i });
|
||||
all.push({
|
||||
name: "ckSybil",
|
||||
victimLabel: "light key",
|
||||
isVictim: (ck) => ck === "light",
|
||||
messages,
|
||||
envConcurrencyLimit: 25,
|
||||
holdSteps: 3,
|
||||
maxSteps: 300,
|
||||
});
|
||||
}
|
||||
|
||||
// ckManyKeys (cardinality ABOVE the pass-1 window): 60 attacker keys x 8 on a
|
||||
// tied old head + 1 light key x 10. Probes the stated window limitation: the
|
||||
// light key must still drain (no permanent starvation), even though 61
|
||||
// variants exceed the 30-wide pass-1 window.
|
||||
{
|
||||
const messages: ScenarioMessage[] = [];
|
||||
for (let i = 0; i < 8; i++)
|
||||
for (let k = 0; k < 60; k++) {
|
||||
const ck = `att${String(k).padStart(2, "0")}`;
|
||||
messages.push({ runId: `${ck}-${i}`, ck, timestamp: t0 });
|
||||
}
|
||||
for (let i = 0; i < 10; i++)
|
||||
messages.push({ runId: `light-${i}`, ck: "light", timestamp: t0 + 50_000 + i });
|
||||
all.push({
|
||||
name: "ckManyKeys",
|
||||
victimLabel: "light key",
|
||||
isVictim: (ck) => ck === "light",
|
||||
messages,
|
||||
envConcurrencyLimit: 25,
|
||||
holdSteps: 3,
|
||||
maxSteps: 1_000,
|
||||
});
|
||||
}
|
||||
|
||||
// ckBalanced (no-harm mixed multi-tenant): 4 symmetric keys x 25.
|
||||
{
|
||||
const cks = ["bal0", "bal1", "bal2", "bal3"];
|
||||
const messages: ScenarioMessage[] = [];
|
||||
for (let i = 0; i < 25; i++)
|
||||
for (let k = 0; k < cks.length; k++)
|
||||
messages.push({ runId: `${cks[k]}-${i}`, ck: cks[k]!, timestamp: t0 + i * 4 + k });
|
||||
all.push({
|
||||
name: "ckBalanced",
|
||||
victimLabel: "worst symmetric key",
|
||||
isVictim: (ck) => ck.startsWith("bal"),
|
||||
messages,
|
||||
envConcurrencyLimit: 4,
|
||||
holdSteps: 3,
|
||||
maxSteps: 500,
|
||||
});
|
||||
}
|
||||
|
||||
// ckHeavyIdle (work conservation): a lone key with 60 msgs, nothing else
|
||||
// contending. Drain-step ON must equal OFF exactly.
|
||||
{
|
||||
const messages: ScenarioMessage[] = [];
|
||||
for (let i = 0; i < 60; i++)
|
||||
messages.push({ runId: `solo-${i}`, ck: "solo", timestamp: t0 + i });
|
||||
all.push({
|
||||
name: "ckHeavyIdle",
|
||||
victimLabel: "lone key",
|
||||
isVictim: (ck) => ck === "solo",
|
||||
messages,
|
||||
envConcurrencyLimit: 25,
|
||||
holdSteps: 3,
|
||||
maxSteps: 300,
|
||||
});
|
||||
}
|
||||
|
||||
return SCENARIO_FILTER ? all.filter((s) => SCENARIO_FILTER.includes(s.name)) : all;
|
||||
}
|
||||
|
||||
// ---- one arm of one scenario ----
|
||||
|
||||
async function runArm(
|
||||
scenario: Scenario,
|
||||
vtimeEnabled: boolean,
|
||||
trial: number
|
||||
): Promise<ArmResult> {
|
||||
const keyPrefix = `ckbench:${scenario.name}:${vtimeEnabled ? "on" : "off"}:t${trial}:`;
|
||||
const queue = createQueue(keyPrefix, vtimeEnabled);
|
||||
const conn = redisConn();
|
||||
const admin = createRedisClient({ host: conn.host, port: conn.port }, { onError: () => {} });
|
||||
|
||||
try {
|
||||
const env = {
|
||||
...authenticatedEnvDev,
|
||||
maximumConcurrencyLimit: scenario.envConcurrencyLimit,
|
||||
concurrencyLimitBurstFactor: new Decimal(1),
|
||||
};
|
||||
await queue.updateEnvConcurrencyLimits(env);
|
||||
|
||||
for (const msg of scenario.messages) {
|
||||
await queue.enqueueMessage({
|
||||
env,
|
||||
message: makeMessage({
|
||||
runId: msg.runId,
|
||||
concurrencyKey: msg.ck,
|
||||
timestamp: msg.timestamp,
|
||||
}),
|
||||
workerQueue: env.id,
|
||||
skipDequeueProcessing: true,
|
||||
});
|
||||
}
|
||||
|
||||
// Count only steady-state (dequeue + ack) Redis ops, not enqueue.
|
||||
await admin.call("CONFIG", "RESETSTAT");
|
||||
|
||||
const shard = keys.masterQueueShardForEnvironment(env.id, 2);
|
||||
const total = scenario.messages.length;
|
||||
const remaining = new Map<string, number>();
|
||||
for (const m of scenario.messages) remaining.set(m.ck, (remaining.get(m.ck) ?? 0) + 1);
|
||||
|
||||
const serves: ServeRecord[] = [];
|
||||
const inFlight: { messageId: string; servedAtStep: number }[] = [];
|
||||
const contentionByCk = new Map<string, number>();
|
||||
let contentionTotal = 0;
|
||||
let drainStep = -1;
|
||||
const callLatenciesMs: number[] = [];
|
||||
const armStart = performance.now();
|
||||
|
||||
for (let step = 0; step < scenario.maxSteps && serves.length < total; step++) {
|
||||
let keysWithBacklog = 0;
|
||||
for (const count of remaining.values()) if (count > 0) keysWithBacklog++;
|
||||
|
||||
const before = performance.now();
|
||||
const messages = await queue.testDequeueFromMasterQueue(shard, env.id, 10);
|
||||
callLatenciesMs.push(performance.now() - before);
|
||||
|
||||
for (const m of messages) {
|
||||
const ck = m.message.concurrencyKey ?? "";
|
||||
serves.push({ step, ck, messageId: m.messageId, wallMs: performance.now() - armStart });
|
||||
remaining.set(ck, (remaining.get(ck) ?? 0) - 1);
|
||||
inFlight.push({ messageId: m.messageId, servedAtStep: step });
|
||||
if (keysWithBacklog >= 2) {
|
||||
contentionTotal++;
|
||||
contentionByCk.set(ck, (contentionByCk.get(ck) ?? 0) + 1);
|
||||
}
|
||||
if (serves.length === total) drainStep = step;
|
||||
}
|
||||
|
||||
for (let i = inFlight.length - 1; i >= 0; i--) {
|
||||
const entry = inFlight[i]!;
|
||||
if (entry.servedAtStep + scenario.holdSteps <= step) {
|
||||
await queue.acknowledgeMessage(env.organization.id, entry.messageId, {
|
||||
skipDequeueProcessing: true,
|
||||
});
|
||||
inFlight.splice(i, 1);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const stats = await admin.call("INFO", "commandstats");
|
||||
const redisCalls = sumRedisCalls(String(stats));
|
||||
|
||||
return { serves, drainStep, contentionByCk, contentionTotal, callLatenciesMs, redisCalls };
|
||||
} finally {
|
||||
await admin.quit().catch(() => {});
|
||||
await queue.quit();
|
||||
// Clean slate for the next arm: this is a dedicated throwaway Redis.
|
||||
const admin2 = createRedisClient(redisConn(), { onError: () => {} });
|
||||
await admin2.flushdb().catch(() => {});
|
||||
await admin2.quit().catch(() => {});
|
||||
}
|
||||
}
|
||||
|
||||
// ---- metrics ----
|
||||
|
||||
function sumRedisCalls(info: string): number {
|
||||
// lines look like: cmdstat_zadd:calls=123,usec=...,...
|
||||
let total = 0;
|
||||
for (const line of info.split("\n")) {
|
||||
const m = line.match(/cmdstat_[^:]+:calls=(\d+)/);
|
||||
if (m) total += Number(m[1]);
|
||||
}
|
||||
return total;
|
||||
}
|
||||
|
||||
function pct(sorted: number[], p: number): number {
|
||||
if (sorted.length === 0) return NaN;
|
||||
const idx = Math.min(sorted.length - 1, Math.max(0, Math.ceil((p / 100) * sorted.length) - 1));
|
||||
return sorted[idx]!;
|
||||
}
|
||||
|
||||
function stats(xs: number[]) {
|
||||
const s = [...xs].sort((a, b) => a - b);
|
||||
const mean = xs.length ? xs.reduce((a, b) => a + b, 0) / xs.length : NaN;
|
||||
return { mean, p50: pct(s, 50), p95: pct(s, 95), p99: pct(s, 99) };
|
||||
}
|
||||
|
||||
// Jain's fairness index over per-key served counts during contention windows.
|
||||
// 1.0 = perfectly fair; 1/n = one key took everything.
|
||||
function jain(counts: number[]): number {
|
||||
const nonzero = counts.filter((c) => c > 0);
|
||||
if (nonzero.length === 0) return NaN;
|
||||
const sum = nonzero.reduce((a, b) => a + b, 0);
|
||||
const sumSq = nonzero.reduce((a, b) => a + b * b, 0);
|
||||
return (sum * sum) / (nonzero.length * sumSq);
|
||||
}
|
||||
|
||||
function victimWaits(arm: ArmResult, s: Scenario): number[] {
|
||||
return arm.serves.filter((r) => s.isVictim(r.ck)).map((r) => r.step);
|
||||
}
|
||||
|
||||
function firstServe(arm: ArmResult, s: Scenario): number {
|
||||
const first = arm.serves.find((r) => s.isVictim(r.ck));
|
||||
return first ? first.step : -1;
|
||||
}
|
||||
|
||||
// ---- the bench ----
|
||||
|
||||
describe.runIf(!!REDIS_URL)("CK virtual-time micro-benchmark (A/B, external Redis)", () => {
|
||||
it("runs OFF vs ON across scenarios and writes results", { timeout: 30 * 60_000 }, async () => {
|
||||
const scenarios = buildScenarios();
|
||||
const report: any = {
|
||||
generatedAtMs: Date.now(),
|
||||
redisUrl: REDIS_URL,
|
||||
trials: TRIALS,
|
||||
knobs: { quantum: QUANTUM, scanWindowMultiplier: WINDOW_MULT },
|
||||
scenarios: [] as any[],
|
||||
};
|
||||
|
||||
for (const s of scenarios) {
|
||||
// Wall-clock latency and op-count are pooled/aggregated across trials.
|
||||
// Step-based metrics are deterministic, so trial 0 is authoritative and
|
||||
// later trials only assert determinism.
|
||||
const offCalls: number[] = [];
|
||||
const onCalls: number[] = [];
|
||||
const offOps: number[] = [];
|
||||
const onOps: number[] = [];
|
||||
let off0: ArmResult | null = null;
|
||||
let on0: ArmResult | null = null;
|
||||
|
||||
for (let t = 0; t < TRIALS; t++) {
|
||||
const off = await runArm(s, false, t);
|
||||
const on = await runArm(s, true, t);
|
||||
|
||||
// Correctness gate: identical load must serve every message exactly
|
||||
// once in BOTH arms, else the comparison is meaningless.
|
||||
expect(off.serves.length, `${s.name} OFF served != enqueued`).toBe(s.messages.length);
|
||||
expect(on.serves.length, `${s.name} ON served != enqueued`).toBe(s.messages.length);
|
||||
expect(new Set(off.serves.map((r) => r.messageId)).size).toBe(s.messages.length);
|
||||
expect(new Set(on.serves.map((r) => r.messageId)).size).toBe(s.messages.length);
|
||||
|
||||
offCalls.push(...off.callLatenciesMs);
|
||||
onCalls.push(...on.callLatenciesMs);
|
||||
offOps.push(off.redisCalls);
|
||||
onOps.push(on.redisCalls);
|
||||
|
||||
if (t === 0) {
|
||||
off0 = off;
|
||||
on0 = on;
|
||||
} else {
|
||||
// determinism of the logical schedule across trials
|
||||
expect(firstServe(off, s), `${s.name} OFF first-serve not deterministic`).toBe(
|
||||
firstServe(off0!, s)
|
||||
);
|
||||
expect(firstServe(on, s), `${s.name} ON first-serve not deterministic`).toBe(
|
||||
firstServe(on0!, s)
|
||||
);
|
||||
expect(on.drainStep).toBe(on0!.drainStep);
|
||||
}
|
||||
}
|
||||
|
||||
const offWait = stats(victimWaits(off0!, s));
|
||||
const onWait = stats(victimWaits(on0!, s));
|
||||
const median = (xs: number[]) => [...xs].sort((a, b) => a - b)[Math.floor(xs.length / 2)]!;
|
||||
|
||||
const scenarioReport = {
|
||||
name: s.name,
|
||||
victim: s.victimLabel,
|
||||
config: {
|
||||
envConcurrencyLimit: s.envConcurrencyLimit,
|
||||
holdSteps: s.holdSteps,
|
||||
variants: new Set(s.messages.map((m) => m.ck)).size,
|
||||
messages: s.messages.length,
|
||||
},
|
||||
off: {
|
||||
victimWait: offWait,
|
||||
victimFirstServe: firstServe(off0!, s),
|
||||
drainStep: off0!.drainStep,
|
||||
jain: jain([...off0!.contentionByCk.values()]),
|
||||
callLatencyMs: stats(offCalls),
|
||||
redisOpsMedian: median(offOps),
|
||||
},
|
||||
on: {
|
||||
victimWait: onWait,
|
||||
victimFirstServe: firstServe(on0!, s),
|
||||
drainStep: on0!.drainStep,
|
||||
jain: jain([...on0!.contentionByCk.values()]),
|
||||
callLatencyMs: stats(onCalls),
|
||||
redisOpsMedian: median(onOps),
|
||||
},
|
||||
};
|
||||
report.scenarios.push(scenarioReport);
|
||||
// eslint-disable-next-line no-console
|
||||
console.log(
|
||||
`[ckbench] ${s.name}: victim p95 wait OFF=${offWait.p95} ON=${onWait.p95} | drain OFF=${off0!.drainStep} ON=${on0!.drainStep}`
|
||||
);
|
||||
}
|
||||
|
||||
mkdirSync(OUT_DIR, { recursive: true });
|
||||
writeFileSync(`${OUT_DIR}/ck-micro-results.json`, JSON.stringify(report, null, 2));
|
||||
writeFileSync(`${OUT_DIR}/ck-micro-results.md`, renderMarkdown(report));
|
||||
});
|
||||
});
|
||||
|
||||
function fmt(n: number): string {
|
||||
if (Number.isNaN(n)) return "n/a";
|
||||
return Number.isInteger(n) ? String(n) : n.toFixed(2);
|
||||
}
|
||||
function delta(off: number, on: number): string {
|
||||
if (Number.isNaN(off) || Number.isNaN(on)) return "n/a";
|
||||
if (off === 0) return on === 0 ? "0" : "+inf";
|
||||
const pctChange = ((on - off) / off) * 100;
|
||||
return `${pctChange >= 0 ? "+" : ""}${pctChange.toFixed(0)}%`;
|
||||
}
|
||||
|
||||
function renderMarkdown(report: any): string {
|
||||
const lines: string[] = [];
|
||||
lines.push(`# CK virtual-time micro-benchmark results`);
|
||||
lines.push("");
|
||||
lines.push(
|
||||
`Redis \`${report.redisUrl}\`, ${report.trials} trial(s), quantum ${report.knobs.quantum}, window multiplier ${report.knobs.scanWindowMultiplier}.`
|
||||
);
|
||||
lines.push("");
|
||||
lines.push(
|
||||
`Numbers are RELATIVE (same box, same load, flag OFF vs ON). Wait is in logical dequeue steps. Latency is wall-clock per dequeue call on this box and is NOT prod-scale absolute throughput.`
|
||||
);
|
||||
lines.push("");
|
||||
lines.push(`| scenario | metric | baseline (OFF) | vtime (ON) | delta |`);
|
||||
lines.push(`| --- | --- | --- | --- | --- |`);
|
||||
for (const s of report.scenarios) {
|
||||
const rows: [string, number, number][] = [
|
||||
[`victim wait p50 (${s.victim})`, s.off.victimWait.p50, s.on.victimWait.p50],
|
||||
[`victim wait p95`, s.off.victimWait.p95, s.on.victimWait.p95],
|
||||
[`victim wait p99`, s.off.victimWait.p99, s.on.victimWait.p99],
|
||||
[`victim first-serve step (starvation bound)`, s.off.victimFirstServe, s.on.victimFirstServe],
|
||||
[`drain step (work conservation)`, s.off.drainStep, s.on.drainStep],
|
||||
[`Jain fairness index (contention)`, s.off.jain, s.on.jain],
|
||||
[`dequeue call p95 (ms)`, s.off.callLatencyMs.p95, s.on.callLatencyMs.p95],
|
||||
[`redis ops (dequeue+ack)`, s.off.redisOpsMedian, s.on.redisOpsMedian],
|
||||
];
|
||||
rows.forEach(([metric, off, on], i) => {
|
||||
lines.push(
|
||||
`| ${i === 0 ? `**${s.name}**` : ""} | ${metric} | ${fmt(off)} | ${fmt(on)} | ${delta(off, on)} |`
|
||||
);
|
||||
});
|
||||
}
|
||||
lines.push("");
|
||||
return lines.join("\n");
|
||||
}
|
||||
Reference in New Issue
Block a user