diff --git a/internal-packages/run-engine/design/benchmarks/2026-07-28-ck-vtime-resource-cardinality-plan.md b/internal-packages/run-engine/design/benchmarks/2026-07-28-ck-vtime-resource-cardinality-plan.md new file mode 100644 index 000000000..48825d66c --- /dev/null +++ b/internal-packages/run-engine/design/benchmarks/2026-07-28-ck-vtime-resource-cardinality-plan.md @@ -0,0 +1,156 @@ +# CK virtual-time scheduling: Redis CPU + memory vs cardinality test plan + +Answers the review question: how do these changes affect the run-queue Redis +CPU and memory, and how do both react as concurrency-key cardinality grows (e.g. +a base queue that suddenly has 10k distinct concurrency keys)? + +This is a cost/scaling plan, separate from the fairness A/B +(`results-2026-07-27.md`). Same environment constraints: run on a single box, all +numbers **relative** (flag OFF vs ON, identical load, same box), against a +dedicated throwaway Redis. + +## What the change adds to Redis (grounded in the branch) + +Per base queue, flag ON adds: + +- `:ckVtime`, a ZSET whose members are the exact same full CK-variant queue-name + strings the existing `:ckIndex` ZSET already holds, each with an 8-byte double + score. So it is effectively a second copy of `ckIndex`'s membership. +- `:ckVtimeFloor`, a STRING holding one number. Negligible. + +Both are GC'd from the dequeue path when a variant drains, carry a 24h TTL, and +live under the base queue's `{org}` hash tag. The per-call dequeue scan window is +`maxCount * windowMultiplier` (default 30) and does **not** grow with cardinality; +the cardinality-sensitive operations are the ZSET writes/reads (`ZADD` `NX`, +`ZSCORE`, `ZRANGE` by rank, `ZRANGE 0 0`), which are O(log N) on the `ckVtime` +skiplist. + +## Hypotheses + +1. **Memory grows linearly with cardinality, adding roughly one `ckIndex`-sized + ZSET per base queue.** Incremental `used_memory` under ON minus OFF should + track `cardinality x per-member cost`, and `MEMORY USAGE :ckVtime` should be + close to `MEMORY USAGE :ckIndex` for the same queue (same members, one extra + double score). At 10k keys on one queue this is a low single-digit MB for that + queue, bounded and TTL-reclaimed. There is a one-step jump at the + listpack->skiplist encoding boundary (128 entries by default). +2. **Redis CPU per operation grows sub-linearly (about O(log cardinality)), not + linearly.** The fixed 30-entry window scan dominates the vtime-specific work + and does not change with N; the ZSET ops add a `log N` term. So dequeue/enqueue + `usec_per_call` should rise only mildly from 100 to 10k keys, and the ON/OFF + `usec_per_call` ratio should stay roughly flat across the sweep. +3. **Tombstone drift stays bounded under high-cardinality churn.** `ack`, TTL + expiry, and DLQ drain a variant without removing it from `ckVtime` (documented + in `CK_VTIME_KNOWN_LIMITATIONS.md`). Under churn where variants drain via those + paths rather than a vtime dequeue, `ckVtime` may transiently exceed `ckIndex`, + but it should self-heal (next vtime pass GCs empties) or expire (24h TTL), so + `size(ckVtime) / size(ckIndex)` stays bounded and does not grow without limit. + +## Scenarios + +All on a single base queue (worst case for one queue's ZSETs), dedicated Redis, +flag OFF then ON with identical load. + +- **Cardinality sweep (memory).** Enqueue N distinct concurrency keys, one + message each, N in {100, 1_000, 10_000, 50_000}. Measure the Redis memory + footprint at rest for each N, OFF vs ON. This isolates the storage cost with no + dequeue activity. +- **Steady-state load (CPU).** At each N, after building cardinality, run a fixed + 60s workload of enqueue + batched dequeue (`maxCount 10`) + ack at a capped + concurrency, so keys are continuously served and re-registered. Measure Redis + CPU and per-command time over the window, OFF vs ON. +- **Churn / tombstone (memory under adversarial drain).** Build 10k keys, then + drain them via `ack` (not via vtime dequeue) while enqueuing new keys, for a + fixed duration. Sample `size(ckVtime)` and `size(ckIndex)` over time and confirm + the ratio stays bounded (self-heal + TTL), not monotonically growing. + +## Metrics and collection (exact commands) + +Use a second plain Redis client for measurement so it does not perturb the +harness. `redis-cli` shown; the harness issues the same via ioredis. + +Memory: + +- Totals: `INFO memory` -> `used_memory`, `used_memory_dataset`. Delta ON vs OFF + at each N is the incremental footprint. +- Per structure (exact bytes): `MEMORY USAGE {org...}:queue::ckIndex` and + `MEMORY USAGE {org...}:queue::ckVtime`. Report both and the ratio. +- Encoding: `OBJECT ENCODING ` (listpack vs skiplist) at each N, to + mark the transition. + +CPU: + +- Process CPU over the load window: `INFO cpu` -> `used_cpu_user` + + `used_cpu_sys`, sampled before and after the fixed 60s load; the delta is Redis + CPU-seconds consumed. Divide by op count for CPU-per-op. +- Per-command time: `CONFIG RESETSTAT` before the window, then `INFO commandstats` + after -> `cmdstat_zadd`, `cmdstat_zrange`, `cmdstat_zrangebyscore`, + `cmdstat_zscore`, `cmdstat_get`, `cmdstat_set`, `cmdstat_expire` + (`calls`, `usec`, `usec_per_call`). Compare OFF vs ON. +- Cross-check (optional, OS-level): `pidstat -p 1` over the window, or + `redis-cli --latency` / `--latency-history` for command latency. + +Derived: + +- Memory vs N curve (expect linear, slope ~ per-member bytes), OFF and ON. +- CPU-per-op vs N curve (expect flat-ish / log), OFF/ON ratio. +- `size(ckVtime)/size(ckIndex)` over time in the churn scenario (expect bounded). + +## A/B procedure + +1. Dedicated throwaway Redis; `FLUSHDB` between arms and between cardinality + points. Warm up once. +2. For each N in the sweep, for each arm (OFF, ON): + - Build N keys (enqueue N distinct concurrency keys). + - Snapshot memory (`used_memory`, `MEMORY USAGE` of both ZSETs, `OBJECT + ENCODING`). + - `CONFIG RESETSTAT`; run the fixed 60s steady load; snapshot `INFO cpu` and + `INFO commandstats`. + - Record, then `FLUSHDB`. +3. N trials of the load window per point; report median for CPU (memory at rest is + near-deterministic). +4. Run the churn scenario once per arm at N = 10k. + +Only the OFF-vs-ON delta and the shape of the growth curves are reported; absolute +throughput on a single box is not prod scale. + +## Results template + +### Memory at rest, per cardinality (single base queue) + +| keys (N) | used_memory OFF | used_memory ON | delta | MEMORY USAGE ckIndex | MEMORY USAGE ckVtime | ckVtime/ckIndex | ckVtime encoding | +| --- | --- | --- | --- | --- | --- | --- | --- | +| 100 | | | | | | | | +| 1,000 | | | | | | | | +| 10,000 | | | | | | | | +| 50,000 | | | | | | | | + +### Redis CPU under 60s steady load, per cardinality + +| keys (N) | CPU-sec OFF | CPU-sec ON | delta | dequeue usec/call OFF | dequeue usec/call ON | delta | total redis calls OFF/ON | +| --- | --- | --- | --- | --- | --- | --- | --- | +| 100 | | | | | | | | +| 1,000 | | | | | | | | +| 10,000 | | | | | | | | + +### Tombstone drift (N=10k, drain via ack) + +| elapsed | size(ckIndex) | size(ckVtime) | ratio | +| --- | --- | --- | --- | +| 0s | | | | +| 30s | | | | +| 60s | | | | + +## Harness + +Extend the existing micro-benchmark +(`../../src/run-queue/bench/ckMicroBench.bench.test.ts`), which already drives a +real `RunQueue` against an external Redis and reads `INFO commandstats`. Add a +resource/cardinality mode that: builds N variants, snapshots `used_memory` + +`MEMORY USAGE` of the base queue's `ckIndex`/`ckVtime` + `OBJECT ENCODING`, runs a +fixed-duration steady load while sampling `INFO cpu`/`commandstats`, and emits the +tables above. The churn scenario reuses the same enqueue/ack primitives with a +drain-by-ack loop and periodic `ZCARD` sampling of both ZSETs. + +Reuse the same dedicated Redis and the OFF-vs-ON constructor-flag pattern, so this +arm, like the micro-benchmark, needs no webapp or redeploy. diff --git a/internal-packages/run-engine/design/benchmarks/results-cardinality-2026-07-28.md b/internal-packages/run-engine/design/benchmarks/results-cardinality-2026-07-28.md new file mode 100644 index 000000000..e62826e8e --- /dev/null +++ b/internal-packages/run-engine/design/benchmarks/results-cardinality-2026-07-28.md @@ -0,0 +1,82 @@ +# CK virtual-time scheduling: Redis CPU + memory vs cardinality (2026-07-28) + +Answers the review question: how do these changes affect the run-queue Redis CPU +and memory, and how do both react as concurrency-key cardinality grows? + +Run on a **local homelab box, not production**, against a dedicated Redis +configured for measurement (no RDB/AOF in the sampling windows, `maxmemory 0`, +zset listpack thresholds at their defaults so the encoding boundary sits at 128). +All numbers are **relative** (flag OFF vs ON, identical load, same box); absolute +throughput is not prod scale. Method: `2026-07-28-ck-vtime-resource-cardinality-plan.md`. +Server-side metrics (`INFO memory`/`cpu`/`commandstats`, `MEMORY USAGE`, +`OBJECT ENCODING`) are RTT-independent. + +## Memory at rest (single base queue, one queued message per key) + +| keys (N) | used_memory OFF | used_memory ON | ON-OFF delta | ckIndex | ckVtime | ckVtime/ckIndex | ckVtime encoding | +| --- | --- | --- | --- | --- | --- | --- | --- | +| 100 | 1.84 MB | 1.93 MB | +0.09 MB | 6.9 KB | 6.2 KB | 0.89 | listpack | +| 1,000 | 2.42 MB | 2.58 MB | +0.15 MB | 131 KB | 131 KB | 1.00 | skiplist | +| 10,000 | 9.12 MB | 10.4 MB | +1.26 MB | 1.42 MB | 1.42 MB | 1.00 | skiplist | +| 50,000 | 39.1 MB | 45.6 MB | +6.51 MB | 7.55 MB | 7.54 MB | 1.00 | skiplist | + +The `:ckVtime` ZSET is the whole added footprint, and it is essentially a second +copy of `:ckIndex`: same members (the full CK-variant queue names), one extra +8-byte score, so `MEMORY USAGE(ckVtime) ~= MEMORY USAGE(ckIndex)` once past the +listpack boundary. Cost is linear in cardinality at roughly **150 bytes per +concurrency key** on top of the index the queue already keeps: about +1.3 MB for +a queue with 10k keys, +6.5 MB at 50k. It is bounded by live cardinality (entries +are GC'd when a variant drains) and expires on the 24h state TTL. The +`used_memory` delta tracks the direct `MEMORY USAGE(ckVtime)` figure to within +allocator noise. The listpack->skiplist transition lands between 100 and 1,000 +keys as expected. + +## Redis CPU under an identical workload (2,000 rounds, ~42k script calls) + +Same logical workload both arms (enqueue + batched dequeue + ack), so the +`evalsha` call count is identical and the difference is pure vtime overhead. Note +the RunQueue Lua runs as `EVALSHA`, so per-`redis.call` costs inside a script are +not separable in `commandstats`; the reportable signals are total Redis CPU and +aggregate `evalsha` time per call. + +| keys (N) | CPU-sec OFF | CPU-sec ON | delta | overhead | evalsha usec/call OFF | evalsha usec/call ON | +| --- | --- | --- | --- | --- | --- | --- | +| 100 | 1.34 | 1.49 | +0.15 | +12% | 22.3 | 26.2 | +| 1,000 | 1.34 | 1.44 | +0.10 | +7% | 22.2 | 24.8 | +| 10,000 | 1.38 | 1.45 | +0.07 | +5% | 23.8 | 25.8 | + +CPU overhead does not grow with cardinality, it shrinks: +12% at 100 keys down to ++5% at 10k. Per-script cost stays roughly flat (about +2 to +4 usec/call, ~25 +usec/call at every cardinality), which is what the design predicts: the pass-1 +dequeue window is fixed (`maxCount * multiplier`, default 30) and independent of +N, and the added ZSET ops are O(log N), so `log2(10000) ~= 13` adds a negligible +constant. The overhead falls as a percentage because that fixed per-call cost is +amortised over more work as the queue grows. + +## Membership under sustained churn (N = 10,000, flag ON) + +60 rounds of continuous registration + drain (fresh keys enqueued while others +are served and acked), holding cardinality at 10k: + +| round | ckIndex card | ckVtime card | ratio | +| --- | --- | --- | --- | +| 0 | 10,000 | 10,000 | 1.0 | +| 20 | 10,000 | 10,000 | 1.0 | +| 40 | 10,000 | 10,000 | 1.0 | +| 59 | 10,000 | 10,000 | 1.0 | + +`ckVtime` membership tracks `ckIndex` exactly throughout: no tombstone +accumulation, no unbounded growth. The bounded/self-healing drift the limitations +doc calls out (ack/TTL/DLQ draining a variant without a vtime GC) stays reclaimed +by the next vtime pass and the state TTL. + +## Bottom line for the cardinality question + +- **Memory** grows linearly with concurrency-key cardinality, adding one + `ckIndex`-sized ZSET per base queue (~150 B/key): a low-single-digit MB even at + 10k keys on one queue, bounded by live cardinality and TTL-reclaimed. A sudden + 10k-key queue costs about +1.3 MB for that queue. +- **CPU** overhead is small and does not scale with cardinality: ~+5 to +12% on + an identical workload, per-script cost flat at ~+2 to +4 usec/call regardless of + N, because the dequeue scan window is fixed and the ZSET ops are O(log N). +- **No runaway state**: `ckVtime` never outgrows `ckIndex` under sustained churn. diff --git a/internal-packages/run-engine/src/run-queue/bench/ckResourceBench.bench.test.ts b/internal-packages/run-engine/src/run-queue/bench/ckResourceBench.bench.test.ts new file mode 100644 index 000000000..c0c0c3e65 --- /dev/null +++ b/internal-packages/run-engine/src/run-queue/bench/ckResourceBench.bench.test.ts @@ -0,0 +1,390 @@ +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 { 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 RESOURCE arm: Redis CPU + memory vs concurrency-key cardinality. +// +// Answers "how do these changes affect the run-queue Redis CPU/memory, and how do +// both react as cardinality grows (e.g. 10k concurrency keys on one base queue)". +// Drives a real RunQueue against an EXTERNAL dedicated Redis, flag OFF vs ON on +// identical load, and reads server-side metrics (INFO memory/cpu/commandstats, +// MEMORY USAGE, OBJECT ENCODING) that are unaffected by client<->server RTT. +// +// Inert unless CK_BENCH_REDIS_URL is set. FLUSHALLs the target between points, so +// point it ONLY at a dedicated throwaway store (a redis-bench lab store). +// +// export CK_BENCH_REDIS_URL="$(lab store url ckbench1)" +// pnpm exec vitest run src/run-queue/bench/ckResourceBench.bench.test.ts +// +// Env knobs: +// CK_RES_MEM_CARDS=100,1000,10000,50000 memory-at-rest sweep +// CK_RES_CPU_CARDS=100,1000,10000 cpu-under-load sweep +// CK_RES_LOAD_OPS=8000 load rounds per cpu point +// CK_RES_CONCURRENCY=64 client concurrency (beats RTT) +// CK_RES_CHURN_CARD=10000 churn/tombstone cardinality +// CK_RES_CHURN_ROUNDS=60 churn sample rounds +// CK_BENCH_OUT=./bench-results +// +// NOTE: the RunQueue Lua commands run via EVALSHA, so redis.call() ops inside a +// script do NOT show up as separate cmdstat_* lines; they roll up under evalsha. +// The reportable CPU signals are therefore total used_cpu over an identical +// workload and aggregate evalsha usec_per_call, not a per-Redis-command split. + +const REDIS_URL = process.env.CK_BENCH_REDIS_URL; +const MEM_CARDS = (process.env.CK_RES_MEM_CARDS ?? "100,1000,10000,50000") + .split(",") + .map((s) => +s.trim()); +const CPU_CARDS = (process.env.CK_RES_CPU_CARDS ?? "100,1000,10000") + .split(",") + .map((s) => +s.trim()); +const LOAD_OPS = +(process.env.CK_RES_LOAD_OPS ?? "8000"); +const CONCURRENCY = +(process.env.CK_RES_CONCURRENCY ?? "64"); +const CHURN_CARD = +(process.env.CK_RES_CHURN_CARD ?? "10000"); +const CHURN_ROUNDS = +(process.env.CK_RES_CHURN_ROUNDS ?? "60"); +const OUT_DIR = process.env.CK_BENCH_OUT ?? "./bench-results"; + +const keys = new RunQueueFullKeyProducer(); + +const testOptions = { + name: "rq", + tracer: trace.getTracer("rq"), + workers: 1, + defaultEnvConcurrency: 1_000_000, + logger: new Logger("RunQueue", "error"), + retryOptions: { + maxAttempts: 5, + factor: 1.1, + minTimeoutInMs: 100, + maxTimeoutInMs: 1_000, + randomize: true, + }, + keys, +}; + +const env = { + id: "e1234", + type: "PRODUCTION" as const, + maximumConcurrencyLimit: 1_000_000, + concurrencyLimitBurstFactor: new Decimal(1.0), + project: { id: "p1234" }, + organization: { id: "o1234" }, +}; + +function conn() { + const u = new URL(REDIS_URL!); + return { + host: u.hostname, + port: Number(u.port || "6379"), + password: decodeURIComponent(u.password || "") || undefined, + username: decodeURIComponent(u.username || "") || undefined, + }; +} + +function createQueue(keyPrefix: string, vtimeEnabled: boolean) { + const c = conn(); + return new RunQueue({ + ...testOptions, + masterQueueConsumersDisabled: true, + workerOptions: { disabled: true }, + ckVirtualTimeScheduling: { enabled: vtimeEnabled }, + queueSelectionStrategy: new FairQueueSelectionStrategy({ redis: { keyPrefix, ...c }, keys }), + redis: { keyPrefix, ...c }, + }); +} + +function makeMessage(o: Partial = {}): InputPayload { + return { + runId: "r1", + taskIdentifier: "task/my-task", + orgId: "o1234", + projectId: "p1234", + environmentId: "e1234", + environmentType: "PRODUCTION", + queue: "task/my-task", + timestamp: Date.now(), + attempt: 0, + ...o, + }; +} + +// bounded-concurrency runner (beats the ~3.6ms workstation->box RTT) +async function pool(n: number, count: number, fn: (i: number) => Promise) { + let i = 0; + await Promise.all( + Array.from({ length: Math.min(n, count) }, async () => { + while (i < count) { + const idx = i++; + await fn(idx); + } + }) + ); +} + +// ---- server-side metric helpers (separate no-prefix admin client) ---- +function admin() { + return createRedisClient(conn(), { onError: () => {} }); +} +function infoField(info: string, key: string): number { + const line = info.split("\n").find((l) => l.startsWith(key + ":")); + return line ? Number(line.split(":")[1]) : NaN; +} +async function usedMemory(a: any) { + return infoField(await a.info("memory"), "used_memory"); +} +async function usedCpu(a: any) { + const i = await a.info("cpu"); + return infoField(i, "used_cpu_user") + infoField(i, "used_cpu_sys"); +} +function evalsha(info: string) { + const line = info.split("\n").find((l) => l.startsWith("cmdstat_evalsha:")); + if (!line) return { calls: 0, usec: 0, usecPerCall: 0 }; + const g = (k: string) => Number(line.match(new RegExp(`${k}=([0-9.]+)`))?.[1] ?? 0); + return { calls: g("calls"), usec: g("usec"), usecPerCall: g("usec_per_call") }; +} + +// ---- build N distinct concurrency keys (one queued message each) ---- +async function buildCardinality(queue: RunQueue, n: number) { + const t0 = Date.now() - 500_000; + await pool(CONCURRENCY, n, async (i) => { + await queue.enqueueMessage({ + env, + message: makeMessage({ runId: `r-${i}`, concurrencyKey: `k${i}`, timestamp: t0 + i }), + workerQueue: env.id, + skipDequeueProcessing: true, + }); + }); +} + +async function findKey(a: any, prefix: string, suffix: string): Promise { + const found = await a.keys(`${prefix}*:${suffix}`); + return found[0] ?? null; +} + +// ---- the bench ---- +describe.runIf(!!REDIS_URL)("CK virtual-time resource + cardinality benchmark", () => { + it( + "measures Redis memory and CPU vs cardinality, flag OFF vs ON", + { timeout: 60 * 60_000 }, + async () => { + const a = admin(); + const report: any = { generatedAtMs: Date.now(), memory: [], cpu: [], churn: null }; + + // ---------- memory at rest ---------- + for (const n of MEM_CARDS) { + const row: any = { cardinality: n }; + for (const on of [false, true]) { + await a.flushall(); + await a.call("CONFIG", "RESETSTAT"); + const prefix = `ckres:mem:${n}:${on ? "on" : "off"}:`; + const q = createQueue(prefix, on); + try { + await q.updateEnvConcurrencyLimits(env); + await buildCardinality(q, n); + const arm = on ? "on" : "off"; + row[`used_memory_${arm}`] = await usedMemory(a); + const ckIndexKey = await findKey(a, prefix, "ckIndex"); + const ckVtimeKey = await findKey(a, prefix, "ckVtime"); + row[`ckIndex_bytes_${arm}`] = ckIndexKey + ? await a.call("MEMORY", "USAGE", ckIndexKey) + : null; + row[`ckIndex_card_${arm}`] = ckIndexKey ? await a.zcard(ckIndexKey) : 0; + if (on) { + row.ckVtime_bytes = ckVtimeKey ? await a.call("MEMORY", "USAGE", ckVtimeKey) : null; + row.ckVtime_card = ckVtimeKey ? await a.zcard(ckVtimeKey) : 0; + row.ckVtime_encoding = ckVtimeKey + ? await a.call("OBJECT", "ENCODING", ckVtimeKey) + : null; + // ckVtime must mirror ckIndex membership when built via the slow path + expect(row.ckVtime_card).toBe(row.ckIndex_card_on); + } + } finally { + await q.quit(); + } + } + row.used_memory_delta = row.used_memory_on - row.used_memory_off; + row.ckVtime_over_ckIndex = + row.ckIndex_bytes_on && row.ckVtime_bytes + ? +(row.ckVtime_bytes / row.ckIndex_bytes_on).toFixed(2) + : null; + report.memory.push(row); + // eslint-disable-next-line no-console + console.log( + `[ckres] mem N=${n}: used_memory delta=${row.used_memory_delta}B ckVtime=${row.ckVtime_bytes}B (${row.ckVtime_encoding})` + ); + } + + // ---------- CPU under identical load ---------- + for (const n of CPU_CARDS) { + const row: any = { cardinality: n }; + for (const on of [false, true]) { + await a.flushall(); + const prefix = `ckres:cpu:${n}:${on ? "on" : "off"}:`; + const q = createQueue(prefix, on); + try { + await q.updateEnvConcurrencyLimits(env); + await buildCardinality(q, n); + const shard = keys.masterQueueShardForEnvironment(env.id, 2); + + await a.call("CONFIG", "RESETSTAT"); + const cpu0 = await usedCpu(a); + const t0wall = Date.now(); + + // identical workload both arms: LOAD_OPS rounds, each round enqueues + // 10 fresh messages (rotating keys, keeps N populated) and does one + // batched dequeue (maxCount 10) + acks the served set. + let served = 0; + await pool(CONCURRENCY, LOAD_OPS, async (i) => { + for (let j = 0; j < 10; j++) { + await q.enqueueMessage({ + env, + message: makeMessage({ + runId: `L-${i}-${j}`, + concurrencyKey: `k${(i * 10 + j) % n}`, + timestamp: Date.now(), + }), + workerQueue: env.id, + skipDequeueProcessing: true, + }); + } + const msgs = await q.testDequeueFromMasterQueue(shard, env.id, 10); + served += msgs.length; + for (const m of msgs) { + await q.acknowledgeMessage(env.organization.id, m.messageId, { + skipDequeueProcessing: true, + }); + } + }); + + const cpu1 = await usedCpu(a); + const es = evalsha(await a.info("commandstats")); + const arm = on ? "on" : "off"; + row[`cpu_sec_${arm}`] = +(cpu1 - cpu0).toFixed(3); + row[`evalsha_calls_${arm}`] = es.calls; + row[`evalsha_usec_per_call_${arm}`] = +es.usecPerCall.toFixed(2); + row[`wall_ms_${arm}`] = Date.now() - t0wall; + row[`served_${arm}`] = served; + } finally { + await q.quit(); + } + } + row.cpu_sec_delta = +(row.cpu_sec_on - row.cpu_sec_off).toFixed(3); + row.cpu_overhead_pct = row.cpu_sec_off + ? Math.round(((row.cpu_sec_on - row.cpu_sec_off) / row.cpu_sec_off) * 100) + : null; + report.cpu.push(row); + // eslint-disable-next-line no-console + console.log( + `[ckres] cpu N=${n}: cpu OFF=${row.cpu_sec_off}s ON=${row.cpu_sec_on}s (${row.cpu_overhead_pct}%) evalsha usec/call OFF=${row.evalsha_usec_per_call_off} ON=${row.evalsha_usec_per_call_on}` + ); + } + + // ---------- churn: ckVtime membership stays bounded vs ckIndex ---------- + { + await a.flushall(); + const prefix = `ckres:churn:on:`; + const q = createQueue(prefix, true); + const samples: any[] = []; + try { + await q.updateEnvConcurrencyLimits(env); + await buildCardinality(q, CHURN_CARD); + const shard = keys.masterQueueShardForEnvironment(env.id, 2); + const ckIndexKey = (await findKey(a, prefix, "ckIndex"))!; + const ckVtimeKey = (await findKey(a, prefix, "ckVtime"))!; + let nextKey = CHURN_CARD; + for (let r = 0; r < CHURN_ROUNDS; r++) { + // hold cardinality: each iteration enqueues one FRESH key and drains + // one message (maxCount 1), so registration and GC churn continuously + // while total membership stays ~CHURN_CARD. + await pool(CONCURRENCY, 200, async () => { + await q.enqueueMessage({ + env, + message: makeMessage({ + runId: `C-${nextKey}`, + concurrencyKey: `k${nextKey++}`, + timestamp: Date.now(), + }), + workerQueue: env.id, + skipDequeueProcessing: true, + }); + const msgs = await q.testDequeueFromMasterQueue(shard, env.id, 1); + for (const m of msgs) { + await q.acknowledgeMessage(env.organization.id, m.messageId, { + skipDequeueProcessing: true, + }); + } + }); + if (r % 10 === 0 || r === CHURN_ROUNDS - 1) { + const ckIndexCard = await a.zcard(ckIndexKey); + const ckVtimeCard = await a.zcard(ckVtimeKey); + samples.push({ + round: r, + ckIndex: ckIndexCard, + ckVtime: ckVtimeCard, + ratio: ckIndexCard ? +(ckVtimeCard / ckIndexCard).toFixed(2) : null, + }); + } + } + report.churn = { cardinality: CHURN_CARD, rounds: CHURN_ROUNDS, samples }; + // bounded: ckVtime never wildly exceeds ckIndex (allow generous 3x for transient tombstones) + for (const s of samples) if (s.ratio !== null) expect(s.ratio).toBeLessThanOrEqual(3); + } finally { + await q.quit(); + } + } + + await a.flushall(); + await a.quit().catch(() => {}); + + mkdirSync(OUT_DIR, { recursive: true }); + writeFileSync(`${OUT_DIR}/ck-resource-results.json`, JSON.stringify(report, null, 2)); + writeFileSync(`${OUT_DIR}/ck-resource-results.md`, renderMarkdown(report)); + } + ); +}); + +function renderMarkdown(r: any): string { + const L: string[] = []; + const kb = (b: number) => (b == null ? "n/a" : (b / 1024).toFixed(1) + "KB"); + L.push(`# CK virtual-time resource + cardinality results`, ""); + L.push( + `Redis \`${(process.env.CK_BENCH_REDIS_URL || "").replace(/:[^:@/]*@/, ":***@")}\`. Relative OFF-vs-ON on one box; not prod scale.`, + "" + ); + L.push(`## Memory at rest (single base queue, one message per key)`, ""); + L.push( + `| keys (N) | used_memory OFF | used_memory ON | delta | ckIndex bytes | ckVtime bytes | ckVtime/ckIndex | ckVtime encoding |` + ); + L.push(`| --- | --- | --- | --- | --- | --- | --- | --- |`); + for (const m of r.memory) + L.push( + `| ${m.cardinality} | ${kb(m.used_memory_off)} | ${kb(m.used_memory_on)} | ${kb(m.used_memory_delta)} | ${kb(m.ckIndex_bytes_on)} | ${kb(m.ckVtime_bytes)} | ${m.ckVtime_over_ckIndex ?? "n/a"} | ${m.ckVtime_encoding ?? "n/a"} |` + ); + L.push( + "", + `## Redis CPU under identical workload (${process.env.CK_RES_LOAD_OPS ?? "8000"} rounds)`, + "" + ); + L.push( + `| keys (N) | CPU-sec OFF | CPU-sec ON | delta | overhead | evalsha usec/call OFF | evalsha usec/call ON | evalsha calls OFF/ON |` + ); + L.push(`| --- | --- | --- | --- | --- | --- | --- | --- |`); + for (const c of r.cpu) + L.push( + `| ${c.cardinality} | ${c.cpu_sec_off} | ${c.cpu_sec_on} | ${c.cpu_sec_delta} | ${c.cpu_overhead_pct}% | ${c.evalsha_usec_per_call_off} | ${c.evalsha_usec_per_call_on} | ${c.evalsha_calls_off}/${c.evalsha_calls_on} |` + ); + if (r.churn) { + L.push("", `## Tombstone / membership under churn (N=${r.churn.cardinality}, flag ON)`, ""); + L.push(`| round | ckIndex card | ckVtime card | ratio |`, `| --- | --- | --- | --- |`); + for (const s of r.churn.samples) + L.push(`| ${s.round} | ${s.ckIndex} | ${s.ckVtime} | ${s.ratio} |`); + } + L.push(""); + return L.join("\n"); +}