Build a Distributed LLM Scheduler
Build a Redis-backed fair scheduler for multi-tenant LLM workloads.
Introduction
30 Second Summary
When one customer floods a shared AI service with requests, quieter customers can end up waiting indefinitely. A fast system still feels broken when its capacity is distributed unfairly.
In this project, you will build a local multi-tenant LLM workload simulator in TypeScript. Its benchmark exposes priority starvation before proving Redis-backed weighted deficit round-robin fairness across two workers.
What You'll Build
You'll watch the same workload expose starvation before a final Markdown report proves fair service across two workers.
By the end of this project, you'll have:
- A deterministic starvation demo that shows the noisy tenant dominating lower-priority work under BullMQ priority scheduling.
- A distributed fairness run where atomic reservations keep two workers inside global concurrency, tenant concurrency, and token-budget ceilings.
- An automated benchmark report that records Jain's fairness index, latency percentiles, reservation states, controller events, and orphan checks.
- Secret Mission: Reproduce a two-controller recovery race before repairing it with one atomic Redis Lua decision.
Are there any prerequisites?
You need Node.js 22 installed before you begin. Step 1 covers Docker Desktop setup plus the local queue stack.
Before We Start
Before any code, let's lock in the failure this scheduler must fix: naive priority dispatch should starve lower-priority interactive and standard work. You will accept the weighted deficit round-robin result only when every backlogged tenant receives service during a continuously backlogged steady-state window with enough samples. A materially higher Jain fairness index provides the final proof.
Set Up the Local Queue Stack
Trustworthy scheduler results depend on a repeatable local environment. Your benchmark needs a running Redis server before it can coordinate queues or measure fairness.
You will use Docker Desktop to run Redis locally. You will also create a strict TypeScript repository with every dependency pinned to an exact version.
In this step, get ready to:
- Confirm Docker Desktop can run local containers.
- Start Redis in a container with its standard port mapped to your computer.
- Create the scheduler repository with pinned dependencies and a passing type check.
Confirm Docker Desktop can run containers
Docker Desktop provides the local container runtime for Redis. The Docker command-line check reveals whether that runtime is available on your computer.
- Press Cmd+Space on macOS or the Windows key on Windows to open system search.
- Type Terminal on macOS or PowerShell on Windows.
- Press Enter to open the terminal.
- Check whether Docker is available by running this command:
docker --versionWhat does this command do?
The command asks the Docker command-line tool to report its installed version. A version line confirms that your terminal can reach Docker.
✔️ I see a Docker version
Docker is installed. Start Docker Desktop so its container engine is ready for Redis.
- Press Cmd+Space on macOS or the Windows key on Windows to open system search.
- Type Docker into system search.
- Press Enter to start Docker Desktop.
- Wait until Docker Desktop reports that its engine is running.
The first start can take a few minutes while Docker prepares its local engine. The wait means the container runtime is getting ready.
ⓧ Docker is missing
Docker Desktop is free for personal use or education. Larger commercial organizations should check Docker's subscription terms before accepting them.
- Open the official Docker Desktop installation guide in your browser.
- Download the Docker.dmg installer for your Mac.
- Open Docker.dmg from your Downloads folder.
- Drag Docker into the Applications folder.
- Start Docker from /Applications/Docker.app.
- Accept the applicable terms when Docker Desktop asks.
- Wait until Docker Desktop reports that its engine is running.
- Return to your terminal.
- Confirm the installation by running this command:
docker --versionWhat does this command confirm?
A Docker version line confirms that the installation is available from your terminal. That connection lets the next command create the Redis container.
Docker still unavailable?
Close your terminal after Docker Desktop finishes starting. Open a new terminal so it can load the updated command path.
If the command still fails, use this prompt to help me troubleshoot Docker Desktop installation.
You now have a local container engine ready to host the scheduler's shared state. That gives every benchmark run the same Redis starting point.
Start the local Redis container
Redis stores queue coordination data outside the worker processes. Mapping port 6379 exposes the container to the TypeScript application on your computer.
- Create the Redis container by running this command:
docker run -d --name my-redis -p 6379:6379 redis:latestWhat does this command do?
- The -d option starts Redis in the background.
- The --name my-redis option gives the container a stable name.
- The -p 6379:6379 option maps your computer's port to the Redis port inside the container.
- The redis:latest image supplies Redis Open Source.
Before you check the container, do you expect Docker to show my-redis as running with port 6379 mapped?
- Inspect the running containers by running this command:
docker psWhat should you see?
The output should include a running container named my-redis. Its port details should show host port 6379 mapped to Redis port 6379.
Redis container missing?
Confirm that Docker Desktop still reports its engine as running. A stopped engine cannot create or list containers.
A previous container with the same name can also block creation. Use this prompt to help me inspect the existing my-redis container.
Great progress. Redis is now running locally with the exact name and port the scheduler configuration expects.
Build the pinned TypeScript repository
The benchmark needs one shared configuration for tenant weights or resource ceilings. Exact dependency pins keep queue behavior consistent across runs.
- Move to your Desktop and create the repository folders by running these commands:
cd ~/Desktop
mkdir distributed-wdrr-scheduler
cd distributed-wdrr-scheduler
mkdir srcWhat do these commands create?
The commands create a distributed-wdrr-scheduler folder on your Desktop. The nested src folder holds the TypeScript source files.
- Launch the code editor you already use through system search.
- Use your editor's folder picker to open distributed-wdrr-scheduler from your Desktop.
- Create package.json inside distributed-wdrr-scheduler.
- Paste this package definition into package.json.
{
"name": "distributed-wdrr-scheduler",
"version": "1.0.0",
"private": true,
"type": "module",
"scripts": {
"typecheck": "tsc --noEmit",
"baseline": "tsx src/cli.ts baseline",
"fair": "tsx src/cli.ts fair",
"benchmark": "tsx src/cli.ts benchmark"
},
"dependencies": {
"bullmq": "6.3.19",
"ioredis": "6.0.0"
},
"devDependencies": {
"@types/node": "26.6.5",
"tsx": "4.23.15",
"typescript": "7.0.2"
}
}What does this package define?
The package uses the ESM module format. Its scripts expose type checking plus the three benchmark modes used later.
Every dependency has an exact version. This keeps the scheduler API surface stable while you compare benchmark results.
- Save package.json.
- Confirm that package.json appears in your editor's file sidebar.
- Install the pinned dependencies by running this command in the repository terminal:
npm installWhat does this installation add?
npm reads the exact versions from package.json. It installs BullMQ 6.3.19 plus ioredis 6.0.0 for queue and Redis access.
It also installs TypeScript 7.0.2, tsx 4.23.15, and the Node declarations at 26.6.5.
You should see npm finish without an installation failure. Your editor should now list node_modules plus a lock file.
Dependency installation failed?
Check that the terminal path ends with distributed-wdrr-scheduler. npm must run beside package.json.
Use this prompt to help me diagnose the npm installation.
- Create tsconfig.json inside distributed-wdrr-scheduler.
- Paste this compiler configuration into tsconfig.json.
{
"compilerOptions": {
"target": "ES2022",
"module": "NodeNext",
"moduleResolution": "NodeNext",
"allowImportingTsExtensions": true,
"noEmit": true,
"strict": true,
"skipLibCheck": true,
"types": ["node"]
},
"include": ["src/**/*.ts"]
}How does TypeScript check this project?
The compiler checks strict NodeNext modules against ES2022. The noEmit setting turns the compiler into a validation tool without generating JavaScript files.
The include pattern limits checks to TypeScript files under src. That keeps the benchmark source under one predictable boundary.
- Save tsconfig.json.
- Confirm that tsconfig.json appears beside package.json in your editor's file sidebar.
- Create README.md inside distributed-wdrr-scheduler.
- Paste the first README section into README.md.
# Distributed Weighted Deficit Scheduler
A TypeScript, Redis, and BullMQ simulation of fair multi-tenant LLM request scheduling.
## What it demonstrates
- A deterministic noisy-neighbor failure under naive priority scheduling
- Weighted deficit round-robin based on task token cost and tenant weight
- Atomic Redis concurrency and in-flight token reservations
- Two independent BullMQ worker processes under one global ceiling
- Observable reservation IDs and lifecycle states
- A steady-state fairness gate that rejects demand-limited or undersampled runs
- Bounded adaptive concurrency under deterministic provider throttling
- Automated throughput, latency, starvation, isolation, and Jain fairness assertions
## Run it
```bash
npm install
npm run typecheck
WORKER_COUNT=1 npm run baseline
WORKER_COUNT=1 npm run fair
npm run fair
npm run benchmark
```
The full benchmark writes `benchmark-report.md`.What does this section document?
The opening section states the scheduler's observable behaviors. The command list gives the repository a repeatable path from installation to the complete benchmark.
- Append the remaining README section directly below the first section.
## Fairness definition
For each saturated tenant, service is completed token cost divided by configured weight. Jain's index is evaluated only over the 1000-4000 ms observation window, after at least 30 successful starts and only when every tenant is backlogged in at least 80 percent of samples.
Targets:
- Fair Jain index at least 0.85
- Improvement over baseline at least 0.15
- Fair throughput at least 85 percent of baseline
- Zero global, tenant concurrency, or token-budget violations
- Both worker processes represented
- Zero normal-run handoff orphans
## Reservation lifecycle
`reserved` -> `enqueued` -> `active` -> `completed` or `throttled`
The Redis reservation and BullMQ add are not one transaction. A reservation that remains in `reserved` beyond the audit threshold is printed by ID as a potential handoff orphan. Production recovery is intentionally deferred.
## Production extensions
- Reservation leases and crash reconciliation
- Transactional outbox for Redis-to-BullMQ handoff
- Redis Cluster deployment and leadership control
- Deadline scheduling and per-tenant SLA policies
- Durable metrics and dashboards
- Tenant-specific adaptive controlsWhy document the fairness gate?
The fairness definition records when a result is eligible for comparison. It also makes the Redis-to-BullMQ handoff gap visible as a measured lifecycle state.
The production extensions keep the local benchmark's scope honest. They show which recovery and deployment concerns remain outside this simulation.
- Save README.md.
- Confirm that the README preview contains the run commands and reservation lifecycle.
- Create config.ts inside the src folder.
- Paste the connection settings and first tenant section into src/config.ts.
export const CONNECTION = {
host: "127.0.0.1",
port: 6379,
maxRetriesPerRequest: null,
};
export type TenantId = "interactive" | "standard" | "noisy";
export interface TenantConfig {
id: TenantId;
weight: number;
arrivalEveryMs: number;
durationMs: number;
tokenCost: number;
priority: number;
concurrency: number;
tokenBudget: number;
}
export const TENANTS: readonly TenantConfig[] = [
{
id: "interactive",
weight: 2,
arrivalEveryMs: 25,
durationMs: 90,
tokenCost: 400,
priority: 5,
concurrency: 2,
tokenBudget: 800,
},What does this configuration establish?
The connection points the application at the Redis container through local port 6379. The tenant type limits scheduler identifiers to the three configured workloads.
The interactive tenant receives weight 2 with its own arrival rate and resource ceilings. These values later become inputs to fairness and isolation checks.
- Append the standard and noisy tenant definitions below the interactive tenant.
{
id: "standard",
weight: 1,
arrivalEveryMs: 40,
durationMs: 180,
tokenCost: 800,
priority: 10,
concurrency: 1,
tokenBudget: 800,
},
{
id: "noisy",
weight: 1,
arrivalEveryMs: 8,
durationMs: 100,
tokenCost: 200,
priority: 1,
concurrency: 1,
tokenBudget: 400,
},
];How do the tenant profiles differ?
Standard work is slower and more token-heavy. Noisy work arrives every 8 milliseconds with the strongest BullMQ priority because lower nonzero values run first.
The unequal costs make job-count rotation unsuitable for this workload. Token cost gives the fair scheduler a service unit it can normalize by weight.
- Append the scheduler limits and task shape below the tenant array.
export const HARD_GLOBAL_LIMIT = 4;
export const MIN_ADAPTIVE_LIMIT = 1;
export const QUANTUM = 200;
export const RUN_MS = 4_500;
export const OBSERVATION_START_MS = 1_000;
export const OBSERVATION_END_MS = 4_000;
export const SAMPLE_EVERY_MS = 100;
export const MIN_OBSERVATION_STARTS = 30;
export const MIN_BACKLOG_RATIO = 0.8;
export const ORPHAN_AGE_MS = 250;
export interface Task {
id: string;
tenantId: TenantId;
arrivalMs: number;
durationMs: number;
tokenCost: number;
createdAt: number;
reservationId?: string;
}What do these limits control?
The constants define capacity and the fixed benchmark timeline. A run only qualifies for fairness analysis after enough successful starts and sufficient tenant backlog.
The task shape carries timing plus token cost through the queue. Its optional reservation ID links fair dispatch to the later lifecycle audit.
- Append the worker result and reservation record shapes below Task.
export interface WorkerResult {
taskId: string;
tenantId: TenantId;
tokenCost: number;
workerId: string;
createdAt: number;
startedAt: number;
completedAt: number;
status: "ok" | "throttled";
reservationId?: string;
}
export interface ReservationRecord {
reservationId: string;
taskId: string;
tenantId: TenantId;
tokenCost: number;
state: "reserved" | "enqueued" | "active" | "completed" | "throttled";
reservedAt: number;
stateChangedAt: number;
jobId?: string;
workerId?: string;
releasedAt?: number;
}Why keep result and reservation records?
Worker results capture who processed a task and how long it spent in the system. Reservation records expose every fair-dispatch handoff state.
Together they support latency measurements plus orphan detection. They also connect worker activity back to a specific reservation ID.
- Append the tenant lookup and first group of Redis keys below the interfaces.
export function tenantById(id: TenantId): TenantConfig {
const tenant = TENANTS.find((candidate) => candidate.id === id);
if (!tenant) {
throw new Error(`Unknown tenant: ${id}`);
}
return tenant;
}
export function runKeys(runId: string) {
const base = `nw:{${runId}}`;
return {
base,
pending: {
interactive: `${base}:pending:interactive`,
standard: `${base}:pending:standard`,
noisy: `${base}:pending:noisy`,
} satisfies Record<TenantId, string>,
deficit: `${base}:deficit`,
cursor: `${base}:cursor`,
inflight: `${base}:inflight`,
tokens: `${base}:tokens`,
peakInflight: `${base}:peak-inflight`,
peakTokens: `${base}:peak-tokens`,
reservations: `${base}:reservations`,
lifecycle: `${base}:lifecycle`,How are Redis keys organized?
The lookup function returns one tenant's configuration from the shared array. It also stops immediately when an unknown tenant ID appears.
Each benchmark run receives a namespaced key prefix. The shared run identifier keeps related Redis keys grouped together.
- Append the remaining run keys below the lifecycle key.
capacity: `${base}:capacity`,
submitted: `${base}:submitted`,
started: `${base}:started`,
actual: `${base}:actual`,
actualPeak: `${base}:actual-peak`,
results: `${base}:results`,
runStartedAt: `${base}:run-started-at`,
throttleEnabled: `${base}:throttle-enabled`,
throttles: `${base}:throttles`,
successes: `${base}:successes`,
controllerEvents: `${base}:controller-events`,
errors: `${base}:errors`,
};
}What do the remaining keys measure?
These keys store submitted work and actual activity. They also retain benchmark results plus adaptive-controller events.
Keeping the measurements under the same run namespace lets teardown remove one scenario without disturbing another.
- Append the cleanup key list at the bottom of src/config.ts.
export function allRunKeys(runId: string): string[] {
const keys = runKeys(runId);
return [
...Object.values(keys.pending),
keys.deficit,
keys.cursor,
keys.inflight,
keys.tokens,
keys.peakInflight,
keys.peakTokens,
keys.reservations,
keys.lifecycle,
keys.capacity,
keys.submitted,
keys.started,
keys.actual,
keys.actualPeak,
keys.results,
keys.runStartedAt,
keys.throttleEnabled,
keys.throttles,
keys.successes,
keys.controllerEvents,
keys.errors,
];
}Why collect every run key?
The helper returns every Redis key created for one benchmark run. Later teardown can remove temporary queue measurements without touching another run's data.
- Save src/config.ts.
- Confirm that config.ts appears inside the src folder in your editor's file sidebar.
Before the final checks, do you expect Docker to report Redis as running and TypeScript to accept the shared configuration?
- Return to the repository terminal.
- Confirm the Redis container is still running by executing this command:
docker psWhat does this check prove?
The running container list should still include my-redis with port 6379 mapped. This proves the repository can reach the local Redis service expected by CONNECTION.
- Validate the TypeScript surface by running this command:
npm run typecheckWhat does the type check prove?
npm runs the tsc --noEmit script defined in package.json. A clean exit confirms that the strict NodeNext configuration accepts src/config.ts.
You should return to the command prompt without TypeScript errors. That clean exit is the repository's setup checkpoint.
Type check showing errors?
Compare the reported filename and line number with the matching code section above. Missing braces or commas often cause nearby lines to be highlighted.
Confirm that dependency installation completed beside package.json. Use this prompt to help me debug the type-check output.
✔️ Awesome, I've got everything!
Your Redis container is running. Your pinned TypeScript repository also passes its first strict check.
ⓧ I'd like to double check the full code
Compare each file with the complete setup below. Save any corrections before running the type check again.
{
"name": "distributed-wdrr-scheduler",
"version": "1.0.0",
"private": true,
"type": "module",
"scripts": {
"typecheck": "tsc --noEmit",
"baseline": "tsx src/cli.ts baseline",
"fair": "tsx src/cli.ts fair",
"benchmark": "tsx src/cli.ts benchmark"
},
"dependencies": {
"bullmq": "6.3.19",
"ioredis": "6.0.0"
},
"devDependencies": {
"@types/node": "26.6.5",
"tsx": "4.23.15",
"typescript": "7.0.2"
}
}{
"compilerOptions": {
"target": "ES2022",
"module": "NodeNext",
"moduleResolution": "NodeNext",
"allowImportingTsExtensions": true,
"noEmit": true,
"strict": true,
"skipLibCheck": true,
"types": ["node"]
},
"include": ["src/**/*.ts"]
}export const CONNECTION = {
host: "127.0.0.1",
port: 6379,
maxRetriesPerRequest: null,
};
export type TenantId = "interactive" | "standard" | "noisy";
export interface TenantConfig {
id: TenantId;
weight: number;
arrivalEveryMs: number;
durationMs: number;
tokenCost: number;
priority: number;
concurrency: number;
tokenBudget: number;
}
export const TENANTS: readonly TenantConfig[] = [
{
id: "interactive",
weight: 2,
arrivalEveryMs: 25,
durationMs: 90,
tokenCost: 400,
priority: 5,
concurrency: 2,
tokenBudget: 800,
},
{
id: "standard",
weight: 1,
arrivalEveryMs: 40,
durationMs: 180,
tokenCost: 800,
priority: 10,
concurrency: 1,
tokenBudget: 800,
},
{
id: "noisy",
weight: 1,
arrivalEveryMs: 8,
durationMs: 100,
tokenCost: 200,
priority: 1,
concurrency: 1,
tokenBudget: 400,
},
];
export const HARD_GLOBAL_LIMIT = 4;
export const MIN_ADAPTIVE_LIMIT = 1;
export const QUANTUM = 200;
export const RUN_MS = 4_500;
export const OBSERVATION_START_MS = 1_000;
export const OBSERVATION_END_MS = 4_000;
export const SAMPLE_EVERY_MS = 100;
export const MIN_OBSERVATION_STARTS = 30;
export const MIN_BACKLOG_RATIO = 0.8;
export const ORPHAN_AGE_MS = 250;
export interface Task {
id: string;
tenantId: TenantId;
arrivalMs: number;
durationMs: number;
tokenCost: number;
createdAt: number;
reservationId?: string;
}
export interface WorkerResult {
taskId: string;
tenantId: TenantId;
tokenCost: number;
workerId: string;
createdAt: number;
startedAt: number;
completedAt: number;
status: "ok" | "throttled";
reservationId?: string;
}
export interface ReservationRecord {
reservationId: string;
taskId: string;
tenantId: TenantId;
tokenCost: number;
state: "reserved" | "enqueued" | "active" | "completed" | "throttled";
reservedAt: number;
stateChangedAt: number;
jobId?: string;
workerId?: string;
releasedAt?: number;
}
export function tenantById(id: TenantId): TenantConfig {
const tenant = TENANTS.find((candidate) => candidate.id === id);
if (!tenant) {
throw new Error(`Unknown tenant: ${id}`);
}
return tenant;
}
export function runKeys(runId: string) {
const base = `nw:{${runId}}`;
return {
base,
pending: {
interactive: `${base}:pending:interactive`,
standard: `${base}:pending:standard`,
noisy: `${base}:pending:noisy`,
} satisfies Record<TenantId, string>,
deficit: `${base}:deficit`,
cursor: `${base}:cursor`,
inflight: `${base}:inflight`,
tokens: `${base}:tokens`,
peakInflight: `${base}:peak-inflight`,
peakTokens: `${base}:peak-tokens`,
reservations: `${base}:reservations`,
lifecycle: `${base}:lifecycle`,
capacity: `${base}:capacity`,
submitted: `${base}:submitted`,
started: `${base}:started`,
actual: `${base}:actual`,
actualPeak: `${base}:actual-peak`,
results: `${base}:results`,
runStartedAt: `${base}:run-started-at`,
throttleEnabled: `${base}:throttle-enabled`,
throttles: `${base}:throttles`,
successes: `${base}:successes`,
controllerEvents: `${base}:controller-events`,
errors: `${base}:errors`,
};
}
export function allRunKeys(runId: string): string[] {
const keys = runKeys(runId);
return [
...Object.values(keys.pending),
keys.deficit,
keys.cursor,
keys.inflight,
keys.tokens,
keys.peakInflight,
keys.peakTokens,
keys.reservations,
keys.lifecycle,
keys.capacity,
keys.submitted,
keys.started,
keys.actual,
keys.actualPeak,
keys.results,
keys.runStartedAt,
keys.throttleEnabled,
keys.throttles,
keys.successes,
keys.controllerEvents,
keys.errors,
];
}# Distributed Weighted Deficit Scheduler
A TypeScript, Redis, and BullMQ simulation of fair multi-tenant LLM request scheduling.
## What it demonstrates
- A deterministic noisy-neighbor failure under naive priority scheduling
- Weighted deficit round-robin based on task token cost and tenant weight
- Atomic Redis concurrency and in-flight token reservations
- Two independent BullMQ worker processes under one global ceiling
- Observable reservation IDs and lifecycle states
- A steady-state fairness gate that rejects demand-limited or undersampled runs
- Bounded adaptive concurrency under deterministic provider throttling
- Automated throughput, latency, starvation, isolation, and Jain fairness assertions
## Run it
```bash
npm install
npm run typecheck
WORKER_COUNT=1 npm run baseline
WORKER_COUNT=1 npm run fair
npm run fair
npm run benchmark
```
The full benchmark writes `benchmark-report.md`.
## Fairness definition
For each saturated tenant, service is completed token cost divided by configured weight. Jain's index is evaluated only over the 1000-4000 ms observation window, after at least 30 successful starts and only when every tenant is backlogged in at least 80 percent of samples.
Targets:
- Fair Jain index at least 0.85
- Improvement over baseline at least 0.15
- Fair throughput at least 85 percent of baseline
- Zero global, tenant concurrency, or token-budget violations
- Both worker processes represented
- Zero normal-run handoff orphans
## Reservation lifecycle
`reserved` -> `enqueued` -> `active` -> `completed` or `throttled`
The Redis reservation and BullMQ add are not one transaction. A reservation that remains in `reserved` beyond the audit threshold is printed by ID as a potential handoff orphan. Production recovery is intentionally deferred.
## Production extensions
- Reservation leases and crash reconciliation
- Transactional outbox for Redis-to-BullMQ handoff
- Redis Cluster deployment and leadership control
- Deadline scheduling and per-tenant SLA policies
- Durable metrics and dashboards
- Tenant-specific adaptive controlsYour reproducible queue stack is ready. Next, you will run the deterministic workload and watch naive priority dispatch starve backlogged tenants.
Reproduce Priority Starvation
Your local Redis stack can now accept queued work. The next question is whether every tenant receives a fair share when demand stays high.
BullMQ processes lower priority numbers first. The noisy tenant submits priority 1 work every 8 milliseconds. This step makes that noisy-neighbor problem visible before you add a fairness layer.
In this step, get ready to:
- Inspect the deterministic tenant workload that keeps every priority explicitly nonzero.
- Create the worker and benchmark paths that measure backlog during steady state.
- Run the baseline with one worker and capture evidence of priority starvation.
Inspect the deterministic workload
A deterministic workload sends the same tenant pattern into every scenario. This gives the later fair run a meaningful baseline for comparison.
- Return to src/config.ts in your repository.
- Find the TENANTS array.
- Confirm that interactive arrives every 25 milliseconds with priority 5.
- Confirm that standard arrives every 40 milliseconds with priority 10.
- Confirm that noisy arrives every 8 milliseconds with priority 1.
Why Keep Every Priority Nonzero?
BullMQ accepts prioritized values from 1 through 2 097 151. Lower numbers run first.
Priority 0 represents unprioritized work. Keeping all three values nonzero isolates the behavior of priority ordering.
What Makes the Workload Saturated?
The arrival intervals submit work faster than one worker can complete every task. Submitted work therefore accumulates behind started work.
The benchmark samples that difference as backlog. A tenant with sustained backlog has demand waiting for service.
Create the baseline execution path
The worker records when each task starts. It also records which worker handled the task. Redis holds those measurements outside the worker process.
- Create src/lua.ts inside the existing src folder.
- Create src/worker.ts inside the same folder.
- Use the second tab below to copy the exact contents into both files.
✔️ I created the worker files
Good. Save both files before checking their types.
ⓧ I'd like to double check the full code
Compare your src/lua.ts file with this reference.
export const DISPATCH_LUA = String.raw`
local configs = cjson.decode(ARGV[1])
local quantum = tonumber(ARGV[2])
local reservationId = ARGV[3]
local now = tonumber(ARGV[4])
local count = #configs
local capacity = tonumber(redis.call('GET', KEYS[12]) or '1')
local globalInflight = tonumber(redis.call('HGET', KEYS[6], '__global') or '0')
if globalInflight >= capacity then
return nil
end
local cursor = tonumber(redis.call('GET', KEYS[5]) or '0')
local function updatePeak(hashKey, field, value)
local peak = tonumber(redis.call('HGET', hashKey, field) or '0')
if value > peak then
redis.call('HSET', hashKey, field, value)
end
end
for offset = 0, count - 1 do
local index = (cursor + offset) % count
local config = configs[index + 1]
local pendingKey = KEYS[index + 1]
if tonumber(redis.call('LLEN', pendingKey)) > 0 then
local tenantInflight = tonumber(redis.call('HGET', KEYS[6], config.id) or '0')
local tenantTokens = tonumber(redis.call('HGET', KEYS[7], config.id) or '0')
if tenantInflight < config.concurrency then
local rawTask = redis.call('LINDEX', pendingKey, 0)
local task = cjson.decode(rawTask)
if tenantTokens + task.tokenCost <= config.tokenBudget then
local credit = redis.call(
'HINCRBY',
KEYS[4],
config.id,
config.weight * quantum
)
if tonumber(credit) >= task.tokenCost then
redis.call('LPOP', pendingKey)
redis.call('HINCRBY', KEYS[4], config.id, -task.tokenCost)
local nextGlobal = redis.call('HINCRBY', KEYS[6], '__global', 1)
local nextTenant = redis.call('HINCRBY', KEYS[6], config.id, 1)
local nextTokens = redis.call('HINCRBY', KEYS[7], config.id, task.tokenCost)
updatePeak(KEYS[8], '__global', tonumber(nextGlobal))
updatePeak(KEYS[8], config.id, tonumber(nextTenant))
updatePeak(KEYS[9], config.id, tonumber(nextTokens))
local record = {
reservationId = reservationId,
taskId = task.id,
tenantId = config.id,
tokenCost = task.tokenCost,
state = 'reserved',
reservedAt = now,
stateChangedAt = now
}
redis.call('HSET', KEYS[10], reservationId, cjson.encode(record))
redis.call('RPUSH', KEYS[11], cjson.encode({
reservationId = reservationId,
taskId = task.id,
tenantId = config.id,
state = 'reserved',
at = now
}))
redis.call('SET', KEYS[5], tostring((index + 1) % count))
return {rawTask, reservationId}
end
end
end
end
end
return nil
`;
export const MARK_RESERVATION_LUA = String.raw`
local raw = redis.call('HGET', KEYS[1], ARGV[1])
if not raw then
return 'missing'
end
local record = cjson.decode(raw)
local nextState = ARGV[2]
if nextState == 'enqueued' and record.state ~= 'reserved' then
return record.state
end
if nextState == 'active' and record.state ~= 'reserved' and record.state ~= 'enqueued' then
return record.state
end
record.state = nextState
record.stateChangedAt = tonumber(ARGV[3])
if ARGV[4] ~= '' then
record.jobId = ARGV[4]
end
if ARGV[5] ~= '' then
record.workerId = ARGV[5]
end
redis.call('HSET', KEYS[1], ARGV[1], cjson.encode(record))
redis.call('RPUSH', KEYS[2], cjson.encode({
reservationId = ARGV[1],
taskId = record.taskId,
tenantId = record.tenantId,
state = nextState,
at = tonumber(ARGV[3]),
jobId = record.jobId,
workerId = record.workerId
}))
return nextState
`;
export const RELEASE_RESERVATION_LUA = String.raw`
local raw = redis.call('HGET', KEYS[3], ARGV[1])
if not raw then
return 'missing'
end
local record = cjson.decode(raw)
if record.releasedAt then
return record.state
end
redis.call('HINCRBY', KEYS[1], '__global', -1)
redis.call('HINCRBY', KEYS[1], record.tenantId, -1)
redis.call('HINCRBY', KEYS[2], record.tenantId, -record.tokenCost)
record.state = ARGV[2]
record.stateChangedAt = tonumber(ARGV[3])
record.releasedAt = tonumber(ARGV[3])
record.workerId = ARGV[4]
redis.call('HSET', KEYS[3], ARGV[1], cjson.encode(record))
redis.call('RPUSH', KEYS[4], cjson.encode({
reservationId = ARGV[1],
taskId = record.taskId,
tenantId = record.tenantId,
state = ARGV[2],
at = tonumber(ARGV[3]),
workerId = ARGV[4]
}))
return record.state
`;
export const ACTIVITY_START_LUA = String.raw`
local global = redis.call('HINCRBY', KEYS[1], '__global', 1)
local tenant = redis.call('HINCRBY', KEYS[1], ARGV[1], 1)
local peakGlobal = tonumber(redis.call('HGET', KEYS[2], '__global') or '0')
local peakTenant = tonumber(redis.call('HGET', KEYS[2], ARGV[1]) or '0')
if tonumber(global) > peakGlobal then
redis.call('HSET', KEYS[2], '__global', global)
end
if tonumber(tenant) > peakTenant then
redis.call('HSET', KEYS[2], ARGV[1], tenant)
end
return {global, tenant}
`;
export const ACTIVITY_END_LUA = String.raw`
redis.call('HINCRBY', KEYS[1], '__global', -1)
redis.call('HINCRBY', KEYS[1], ARGV[1], -1)
return 1
`;Compare your src/worker.ts file with this reference.
import { Worker } from "bullmq";
import { Redis } from "ioredis";
import {
CONNECTION,
HARD_GLOBAL_LIMIT,
type Task,
type WorkerResult,
runKeys,
} from "./config.ts";
import {
ACTIVITY_END_LUA,
ACTIVITY_START_LUA,
MARK_RESERVATION_LUA,
RELEASE_RESERVATION_LUA,
} from "./lua.ts";
const [, , queueName, runId, workerId] = process.argv;
if (!queueName || !runId || !workerId) {
throw new Error("worker.ts requires queueName, runId, and workerId");
}
const redis = new Redis(CONNECTION);
const keys = runKeys(runId);
const sleep = (milliseconds: number) =>
new Promise<void>((resolve) => setTimeout(resolve, milliseconds));
async function callNumber(command: string, ...args: Array<string | number>) {
const value = await redis.call(command, ...args);
return Number(value ?? 0);
}
const worker = new Worker<Task, WorkerResult>(
queueName,
async (job) => {
const task = job.data;
const startedAt = Date.now();
let terminalState: "completed" | "throttled" = "completed";
await redis.call("HINCRBY", keys.started, task.tenantId, 1);
if (task.reservationId) {
await redis.eval(
MARK_RESERVATION_LUA,
2,
keys.reservations,
keys.lifecycle,
task.reservationId,
"active",
startedAt,
job.id ?? task.id,
workerId,
);
}
const activity = (await redis.eval(
ACTIVITY_START_LUA,
2,
keys.actual,
keys.actualPeak,
task.tenantId,
)) as [number, number];
try {
const runStartedAt = callNumber("GET", keys.runStartedAt);
const throttleEnabled = callNumber("GET", keys.throttleEnabled);
const [runStart, enabled] = await Promise.all([
runStartedAt,
throttleEnabled,
]);
const elapsed = startedAt - runStart;
const providerThreshold = elapsed >= 1_500 && elapsed < 2_700 ? 2 : 4;
const throttled = enabled === 1 && Number(activity[0]) > providerThreshold;
if (throttled) {
terminalState = "throttled";
await redis.call("INCR", keys.throttles);
await sleep(20);
} else {
await sleep(task.durationMs);
await redis.call("INCR", keys.successes);
}
const result: WorkerResult = {
taskId: task.id,
tenantId: task.tenantId,
tokenCost: task.tokenCost,
workerId,
createdAt: task.createdAt,
startedAt,
completedAt: Date.now(),
status: throttled ? "throttled" : "ok",
reservationId: task.reservationId,
};
await redis.call("RPUSH", keys.results, JSON.stringify(result));
return result;
} finally {
const finishedAt = Date.now();
await redis.eval(
ACTIVITY_END_LUA,
1,
keys.actual,
task.tenantId,
);
if (task.reservationId) {
await redis.eval(
RELEASE_RESERVATION_LUA,
4,
keys.inflight,
keys.tokens,
keys.reservations,
keys.lifecycle,
task.reservationId,
terminalState,
finishedAt,
workerId,
);
}
}
},
{
connection: CONNECTION,
concurrency: HARD_GLOBAL_LIMIT,
},
);
worker.on("error", (error) => {
console.error(`[${workerId}]`, error);
});
console.log(`[${workerId}] ready for ${queueName}`);
let closing = false;
async function shutdown() {
if (closing) return;
closing = true;
await worker.close();
await redis.quit();
process.exit(0);
}
process.on("SIGTERM", () => void shutdown());
process.on("SIGINT", () => void shutdown());How Does the Worker Measure Service?
- The worker increments the tenant's started counter before processing each task.
- The activity scripts update shared concurrency values inside Redis.
- Each WorkerResult stores timestamps and the worker ID for later analysis.
- The shutdown handlers close the BullMQ worker before closing its Redis connection.
- Save src/lua.ts.
- Save src/worker.ts.
- Check the worker files by running:
npm run typecheckTypeScript should exit without reporting an error. That confirms the worker can use the shared configuration and Lua helpers.
Type Check Failing on the Worker?
Confirm that both files are inside the existing src folder. Check that the import ends with ./lua.ts.
Compare the exported Lua constant names with the worker imports. A missing constant prevents TypeScript from resolving the module.
Help me diagnose the worker type-check error.
The benchmark supplies tasks to that worker. It also samples backlog throughout the observation window so demand-limited tenants cannot produce a misleading fairness score.
- Create src/benchmark.ts inside the existing src folder.
- Create src/cli.ts inside the same folder.
- Use the second tab below to copy the exact contents into both files.
✔️ I created the benchmark files
Great. Save both files before running the repository-wide type check.
ⓧ I'd like to double check the full code
Compare your src/benchmark.ts file with this reference.
import assert from "node:assert/strict";
import { spawn, type ChildProcess } from "node:child_process";
import { once } from "node:events";
import * as fs from "node:fs/promises";
import { Queue } from "bullmq";
import { Redis } from "ioredis";
import {
allRunKeys,
CONNECTION,
HARD_GLOBAL_LIMIT,
MIN_ADAPTIVE_LIMIT,
MIN_BACKLOG_RATIO,
MIN_OBSERVATION_STARTS,
OBSERVATION_END_MS,
OBSERVATION_START_MS,
ORPHAN_AGE_MS,
QUANTUM,
RUN_MS,
SAMPLE_EVERY_MS,
TENANTS,
type ReservationRecord,
type Task,
type TenantId,
type WorkerResult,
runKeys,
tenantById,
} from "./config.ts";
import { DISPATCH_LUA, MARK_RESERVATION_LUA } from "./lua.ts";
type ScenarioMode = "baseline" | "fair";
interface BacklogSample {
at: number;
backlog: Record<TenantId, number>;
}
interface ControllerEvent {
at: number;
from: number;
to: number;
reason: "throttle" | "recovery";
}
interface TenantMetrics {
successfulStarts: number;
serviceTokens: number;
normalizedService: number;
starvationWindows: number;
queueWaitP95: number | null;
endToEndP95: number | null;
backlogRatio: number;
}
export interface ScenarioReport {
label: string;
mode: ScenarioMode;
adaptive: boolean;
runId: string;
observation: { startMs: number; endMs: number; successfulStarts: number };
evaluable: boolean;
evaluationNotes: string[];
fairness: number | null;
throughputPerSecond: number;
queueWait: { p50: number | null; p95: number | null; p99: number | null };
endToEnd: { p50: number | null; p95: number | null; p99: number | null };
tenants: Record<TenantId, TenantMetrics>;
workerIds: string[];
reservationStates: Record<string, number>;
potentialOrphans: string[];
peakReservedConcurrency: Record<string, number>;
peakReservedTokens: Record<string, number>;
peakActualConcurrency: Record<string, number>;
controllerEvents: ControllerEvent[];
}
const sleep = (milliseconds: number) =>
new Promise<void>((resolve) => setTimeout(resolve, milliseconds));
async function redisNumber(
redis: Redis,
command: string,
...args: Array<string | number>
) {
const value = await redis.call(command, ...args);
return Number(value ?? 0);
}
function generateWorkload(): Task[] {
const tasks: Task[] = [];
for (const tenant of TENANTS) {
let sequence = 0;
for (
let arrivalMs = 0;
arrivalMs < RUN_MS - 400;
arrivalMs += tenant.arrivalEveryMs
) {
sequence += 1;
tasks.push({
id: `${tenant.id}-${sequence}`,
tenantId: tenant.id,
arrivalMs,
durationMs: tenant.durationMs,
tokenCost: tenant.tokenCost,
createdAt: 0,
});
}
}
return tasks.sort(
(left, right) =>
left.arrivalMs - right.arrivalMs || left.id.localeCompare(right.id),
);
}
function spawnWorkers(queueName: string, runId: string, count: number) {
const children: ChildProcess[] = [];
for (let index = 1; index <= count; index += 1) {
const child = spawn(
process.execPath,
["--import=tsx", "src/worker.ts", queueName, runId, `worker-${index}`],
{ stdio: "inherit", env: process.env },
);
children.push(child);
}
return children;
}
async function stopWorkers(children: ChildProcess[]) {
await Promise.all(
children.map(async (child) => {
child.kill("SIGTERM");
await once(child, "close");
}),
);
}
async function produce(
redis: Redis,
queue: Queue<Task, WorkerResult>,
runId: string,
runStartedAt: number,
mode: ScenarioMode,
) {
const keys = runKeys(runId);
await Promise.all(
generateWorkload().map(async (template) => {
await sleep(Math.max(0, runStartedAt + template.arrivalMs - Date.now()));
const task = { ...template, createdAt: Date.now() };
await redis.call("HINCRBY", keys.submitted, task.tenantId, 1);
if (mode === "baseline") {
const tenant = tenantById(task.tenantId);
await queue.add("simulate-llm", task, {
jobId: task.id,
priority: tenant.priority,
removeOnComplete: true,
removeOnFail: true,
});
} else {
await redis.call(
"RPUSH",
keys.pending[task.tenantId],
JSON.stringify(task),
);
}
}),
);
}
async function dispatchFair(
redis: Redis,
queue: Queue<Task, WorkerResult>,
runId: string,
shouldStop: () => boolean,
) {
const keys = runKeys(runId);
const dispatchKeys = [
keys.pending.interactive,
keys.pending.standard,
keys.pending.noisy,
keys.deficit,
keys.cursor,
keys.inflight,
keys.tokens,
keys.peakInflight,
keys.peakTokens,
keys.reservations,
keys.lifecycle,
keys.capacity,
];
const configs = JSON.stringify(
TENANTS.map(({ id, weight, concurrency, tokenBudget }) => ({
id,
weight,
concurrency,
tokenBudget,
})),
);
let reservationSequence = 0;
while (!shouldStop()) {
reservationSequence += 1;
const reservationId = `${runId}-r${reservationSequence}`;
const result = (await redis.eval(
DISPATCH_LUA,
dispatchKeys.length,
...dispatchKeys,
configs,
QUANTUM,
reservationId,
Date.now(),
)) as [string, string] | null;
if (!result) {
await sleep(2);
continue;
}
const [rawTask, selectedReservationId] = result;
const task = JSON.parse(rawTask) as Task;
const reservedTask = { ...task, reservationId: selectedReservationId };
try {
const job = await queue.add("simulate-llm", reservedTask, {
jobId: task.id,
priority: 5,
removeOnComplete: true,
removeOnFail: true,
});
await redis.eval(
MARK_RESERVATION_LUA,
2,
keys.reservations,
keys.lifecycle,
selectedReservationId,
"enqueued",
Date.now(),
job.id ?? task.id,
"",
);
} catch (error) {
await redis.call(
"RPUSH",
keys.errors,
JSON.stringify({
reservationId: selectedReservationId,
message: error instanceof Error ? error.message : String(error),
at: Date.now(),
}),
);
}
}
}
async function sampleBacklog(
redis: Redis,
runId: string,
runStartedAt: number,
shouldStop: () => boolean,
samples: BacklogSample[],
) {
const keys = runKeys(runId);
while (!shouldStop()) {
const at = Date.now() - runStartedAt;
if (at >= OBSERVATION_START_MS && at < OBSERVATION_END_MS) {
const backlog = {} as Record<TenantId, number>;
for (const tenant of TENANTS) {
const submitted = await redisNumber(
redis,
"HGET",
keys.submitted,
tenant.id,
);
const started = await redisNumber(
redis,
"HGET",
keys.started,
tenant.id,
);
backlog[tenant.id] = Math.max(0, submitted - started);
}
samples.push({ at, backlog });
}
await sleep(SAMPLE_EVERY_MS);
}
}
async function runAdaptiveController(
redis: Redis,
queue: Queue<Task, WorkerResult>,
runId: string,
runStartedAt: number,
shouldStop: () => boolean,
events: ControllerEvent[],
) {
const keys = runKeys(runId);
let capacity = HARD_GLOBAL_LIMIT;
let seenThrottles = 0;
let cleanTicks = 0;
let lastChangeAt = 0;
while (!shouldStop()) {
await sleep(250);
const now = Date.now();
const throttles = await redisNumber(redis, "GET", keys.throttles);
const newThrottles = throttles - seenThrottles;
seenThrottles = throttles;
let next = capacity;
let reason: ControllerEvent["reason"] | null = null;
if (newThrottles > 0 && now - lastChangeAt >= 800) {
next = Math.max(MIN_ADAPTIVE_LIMIT, Math.floor(capacity / 2));
cleanTicks = 0;
reason = "throttle";
} else if (newThrottles === 0) {
cleanTicks += 1;
if (
cleanTicks >= 4 &&
capacity < HARD_GLOBAL_LIMIT &&
now - lastChangeAt >= 800
) {
next = capacity + 1;
cleanTicks = 0;
reason = "recovery";
}
} else {
cleanTicks = 0;
}
if (reason && next !== capacity) {
const event: ControllerEvent = {
at: now - runStartedAt,
from: capacity,
to: next,
reason,
};
capacity = next;
lastChangeAt = now;
events.push(event);
await queue.setGlobalConcurrency(capacity);
await redis.call("SET", keys.capacity, capacity);
await redis.call("RPUSH", keys.controllerEvents, JSON.stringify(event));
}
}
}
function objectFromRedisReply(reply: unknown): Record<string, string> {
if (!reply) return {};
if (!Array.isArray(reply)) return reply as Record<string, string>;
const object: Record<string, string> = {};
for (let index = 0; index < reply.length; index += 2) {
object[String(reply[index])] = String(reply[index + 1]);
}
return object;
}
async function readHash(redis: Redis, key: string) {
return objectFromRedisReply(await redis.call("HGETALL", key));
}
async function drainResults(redis: Redis, key: string) {
const results: WorkerResult[] = [];
while (true) {
const raw = (await redis.call("LPOP", key)) as string | null;
if (!raw) break;
results.push(JSON.parse(raw) as WorkerResult);
}
return results;
}
function percentile(values: number[], percentileValue: number): number | null {
if (values.length === 0) return null;
const sorted = [...values].sort((left, right) => left - right);
const index = Math.max(
0,
Math.ceil((percentileValue / 100) * sorted.length) - 1,
);
return sorted[index];
}
function jainIndex(values: number[]): number {
const sum = values.reduce((total, value) => total + value, 0);
const squares = values.reduce((total, value) => total + value * value, 0);
return squares === 0 ? 0 : (sum * sum) / (values.length * squares);
}
function countStarvationWindows(
tenantId: TenantId,
results: WorkerResult[],
samples: BacklogSample[],
) {
let count = 0;
for (
let start = OBSERVATION_START_MS;
start < OBSERVATION_END_MS;
start += 500
) {
const end = start + 500;
const windowSamples = samples.filter(
(sample) => sample.at >= start && sample.at < end,
);
const backlogged =
windowSamples.length > 0 &&
windowSamples.filter((sample) => sample.backlog[tenantId] > 0).length /
windowSamples.length >=
0.5;
const starts = results.filter(
(result) =>
result.tenantId === tenantId &&
result.status === "ok" &&
result.startedAt >= start &&
result.startedAt < end,
).length;
if (backlogged && starts === 0) count += 1;
}
return count;
}
function numericHash(hash: Record<string, string>) {
return Object.fromEntries(
Object.entries(hash).map(([key, value]) => [key, Number(value)]),
);
}
function analyze(
label: string,
mode: ScenarioMode,
adaptive: boolean,
runId: string,
runStartedAt: number,
results: WorkerResult[],
samples: BacklogSample[],
reservations: Record<string, string>,
peakInflight: Record<string, string>,
peakTokens: Record<string, string>,
actualPeak: Record<string, string>,
controllerEvents: ControllerEvent[],
): ScenarioReport {
const successfulWindow = results.filter((result) => {
const relativeStart = result.startedAt - runStartedAt;
return (
result.status === "ok" &&
relativeStart >= OBSERVATION_START_MS &&
relativeStart < OBSERVATION_END_MS
);
});
const evaluationNotes: string[] = [];
const backlogRatios = {} as Record<TenantId, number>;
for (const tenant of TENANTS) {
backlogRatios[tenant.id] =
samples.length === 0
? 0
: samples.filter((sample) => sample.backlog[tenant.id] > 0).length /
samples.length;
if (backlogRatios[tenant.id] < MIN_BACKLOG_RATIO) {
evaluationNotes.push(
`${tenant.id} backlog ratio ${backlogRatios[tenant.id].toFixed(2)} is below ${MIN_BACKLOG_RATIO.toFixed(2)}; demand may be insufficient for a fairness claim.`,
);
}
}
if (successfulWindow.length < MIN_OBSERVATION_STARTS) {
evaluationNotes.push(
`Only ${successfulWindow.length} successful starts were observed; at least ${MIN_OBSERVATION_STARTS} are required.`,
);
}
const evaluable = evaluationNotes.length === 0;
const service = Object.fromEntries(
TENANTS.map((tenant) => [
tenant.id,
successfulWindow
.filter((result) => result.tenantId === tenant.id)
.reduce((total, result) => total + result.tokenCost, 0),
]),
) as Record<TenantId, number>;
const normalized = Object.fromEntries(
TENANTS.map((tenant) => [tenant.id, service[tenant.id] / tenant.weight]),
) as Record<TenantId, number>;
const fairness = evaluable ? jainIndex(Object.values(normalized)) : null;
const queueWaitValues = successfulWindow.map(
(result) => result.startedAt - result.createdAt,
);
const endToEndValues = successfulWindow.map(
(result) => result.completedAt - result.createdAt,
);
const shiftedResults = results.map((result) => ({
...result,
startedAt: result.startedAt - runStartedAt,
}));
const tenants = {} as Record<TenantId, TenantMetrics>;
for (const tenant of TENANTS) {
const tenantResults = successfulWindow.filter(
(result) => result.tenantId === tenant.id,
);
tenants[tenant.id] = {
successfulStarts: tenantResults.length,
serviceTokens: service[tenant.id],
normalizedService: normalized[tenant.id],
starvationWindows: countStarvationWindows(
tenant.id,
shiftedResults,
samples,
),
queueWaitP95: percentile(
tenantResults.map((result) => result.startedAt - result.createdAt),
95,
),
endToEndP95: percentile(
tenantResults.map((result) => result.completedAt - result.createdAt),
95,
),
backlogRatio: backlogRatios[tenant.id],
};
}
const reservationRecords = Object.values(reservations).map(
(raw) => JSON.parse(raw) as ReservationRecord,
);
const reservationStates: Record<string, number> = {};
for (const record of reservationRecords) {
reservationStates[record.state] = (reservationStates[record.state] ?? 0) + 1;
}
const runEndedAt = runStartedAt + RUN_MS;
const potentialOrphans = reservationRecords
.filter(
(record) =>
record.state === "reserved" &&
runEndedAt - record.reservedAt >= ORPHAN_AGE_MS,
)
.map((record) => record.reservationId);
return {
label,
mode,
adaptive,
runId,
observation: {
startMs: OBSERVATION_START_MS,
endMs: OBSERVATION_END_MS,
successfulStarts: successfulWindow.length,
},
evaluable,
evaluationNotes,
fairness,
throughputPerSecond:
successfulWindow.length /
((OBSERVATION_END_MS - OBSERVATION_START_MS) / 1_000),
queueWait: {
p50: percentile(queueWaitValues, 50),
p95: percentile(queueWaitValues, 95),
p99: percentile(queueWaitValues, 99),
},
endToEnd: {
p50: percentile(endToEndValues, 50),
p95: percentile(endToEndValues, 95),
p99: percentile(endToEndValues, 99),
},
tenants,
workerIds: [...new Set(successfulWindow.map((result) => result.workerId))].sort(),
reservationStates,
potentialOrphans,
peakReservedConcurrency: numericHash(peakInflight),
peakReservedTokens: numericHash(peakTokens),
peakActualConcurrency: numericHash(actualPeak),
controllerEvents,
};
}
export async function runScenario(
label: string,
mode: ScenarioMode,
adaptive: boolean,
): Promise<ScenarioReport> {
const runId = `${label}-${Date.now()}-${process.pid}`;
const queueName = `wdrr-${runId}`;
const redis = new Redis(CONNECTION);
const queue = new Queue<Task, WorkerResult>(queueName, {
connection: CONNECTION,
});
const keys = runKeys(runId);
const workerCount = Number(process.env.WORKER_COUNT ?? "2");
const runStartedAt = Date.now() + 250;
const samples: BacklogSample[] = [];
const controllerEvents: ControllerEvent[] = [];
let stopped = false;
await queue.setGlobalConcurrency(HARD_GLOBAL_LIMIT);
await redis.call("SET", keys.capacity, HARD_GLOBAL_LIMIT);
await redis.call("SET", keys.runStartedAt, runStartedAt);
await redis.call("SET", keys.throttleEnabled, adaptive ? 1 : 0);
const children = spawnWorkers(queueName, runId, workerCount);
const producer = produce(redis, queue, runId, runStartedAt, mode);
const sampler = sampleBacklog(
redis,
runId,
runStartedAt,
() => stopped,
samples,
);
const dispatcher =
mode === "fair"
? dispatchFair(redis, queue, runId, () => stopped)
: Promise.resolve();
const controller = adaptive
? runAdaptiveController(
redis,
queue,
runId,
runStartedAt,
() => stopped,
controllerEvents,
)
: Promise.resolve();
await sleep(Math.max(0, runStartedAt + RUN_MS - Date.now()));
await producer;
stopped = true;
await Promise.all([sampler, dispatcher, controller]);
await sleep(350);
await stopWorkers(children);
const [
results,
reservations,
peakInflight,
peakTokens,
actualPeak,
] = await Promise.all([
drainResults(redis, keys.results),
readHash(redis, keys.reservations),
readHash(redis, keys.peakInflight),
readHash(redis, keys.peakTokens),
readHash(redis, keys.actualPeak),
]);
const report = analyze(
label,
mode,
adaptive,
runId,
runStartedAt,
results,
samples,
reservations,
peakInflight,
peakTokens,
actualPeak,
controllerEvents,
);
await queue.obliterate({ force: true });
await queue.close();
await redis.call("DEL", ...allRunKeys(runId));
await redis.quit();
return report;
}
function value(value: number | null, digits = 2) {
return value === null ? "n/a" : value.toFixed(digits);
}
export function printReport(report: ScenarioReport) {
console.log(`\n${report.label}`);
console.log(
`window=${report.observation.startMs}-${report.observation.endMs}ms evaluable=${report.evaluable} starts=${report.observation.successfulStarts}`,
);
console.log(
`jain=${value(report.fairness, 3)} throughput=${value(report.throughputPerSecond)} tasks/s workers=${report.workerIds.join(",")}`,
);
console.log(
`queue wait p50/p95/p99=${value(report.queueWait.p50, 0)}/${value(report.queueWait.p95, 0)}/${value(report.queueWait.p99, 0)}ms`,
);
for (const tenant of TENANTS) {
const metrics = report.tenants[tenant.id];
console.log(
`${tenant.id}: starts=${metrics.successfulStarts} normalized=${value(metrics.normalizedService, 0)} backlog=${value(metrics.backlogRatio)} starvationWindows=${metrics.starvationWindows}`,
);
}
console.log(`reservation states=${JSON.stringify(report.reservationStates)}`);
console.log(`potential handoff orphans=${report.potentialOrphans.length}`);
if (!report.evaluable) {
console.log(`evaluation notes=${report.evaluationNotes.join(" | ")}`);
}
}
function assertProject(
baseline: ScenarioReport,
fair: ScenarioReport,
adaptive: ScenarioReport,
) {
assert.equal(baseline.evaluable, true, baseline.evaluationNotes.join(" | "));
assert.equal(fair.evaluable, true, fair.evaluationNotes.join(" | "));
assert.notEqual(baseline.fairness, null);
assert.notEqual(fair.fairness, null);
assert.ok(fair.fairness! >= 0.85, `fair Jain index was ${fair.fairness}`);
assert.ok(
fair.fairness! - baseline.fairness! >= 0.15,
`fairness improvement was ${fair.fairness! - baseline.fairness!}`,
);
assert.ok(
fair.throughputPerSecond >= baseline.throughputPerSecond * 0.85,
"fair throughput fell more than 15 percent below baseline",
);
assert.ok(
Object.values(fair.tenants).every((tenant) => tenant.successfulStarts > 0),
"a backlogged tenant received no fair-mode service",
);
assert.ok(
Object.values(fair.tenants).reduce(
(total, tenant) => total + tenant.starvationWindows,
0,
) <
Object.values(baseline.tenants).reduce(
(total, tenant) => total + tenant.starvationWindows,
0,
),
"fair mode did not reduce starvation windows",
);
assert.ok(
Number(fair.peakReservedConcurrency.__global ?? 0) <= HARD_GLOBAL_LIMIT,
);
assert.ok(
Number(fair.peakActualConcurrency.__global ?? 0) <= HARD_GLOBAL_LIMIT,
);
for (const tenant of TENANTS) {
assert.ok(
Number(fair.peakReservedConcurrency[tenant.id] ?? 0) <= tenant.concurrency,
`${tenant.id} exceeded its concurrency limit`,
);
assert.ok(
Number(fair.peakReservedTokens[tenant.id] ?? 0) <= tenant.tokenBudget,
`${tenant.id} exceeded its token budget`,
);
}
assert.deepEqual(fair.workerIds, ["worker-1", "worker-2"]);
assert.equal(fair.potentialOrphans.length, 0);
assert.ok(
adaptive.controllerEvents.some((event) => event.reason === "throttle"),
"adaptive run never decreased capacity",
);
assert.ok(
adaptive.controllerEvents.some((event) => event.reason === "recovery"),
"adaptive run never recovered capacity",
);
for (const event of adaptive.controllerEvents) {
assert.ok(event.to >= MIN_ADAPTIVE_LIMIT && event.to <= HARD_GLOBAL_LIMIT);
if (event.reason === "recovery") {
assert.equal(event.to - event.from, 1);
}
}
assert.equal(adaptive.potentialOrphans.length, 0);
}
function reportMarkdown(
baseline: ScenarioReport,
fair: ScenarioReport,
adaptive: ScenarioReport,
) {
const fairnessImprovement = fair.fairness! - baseline.fairness!;
return `# Distributed WDRR Scheduler Benchmark
## Result
The fair scheduler passed its steady-state eligibility gate and improved Jain's fairness index from ${value(baseline.fairness, 3)} to ${value(fair.fairness, 3)}, an improvement of ${value(fairnessImprovement, 3)}.
| Scenario | Evaluable | Jain fairness | Throughput tasks/s | Queue wait p95 ms | End-to-end p95 ms | Workers | Potential orphans |
| --- | --- | ---: | ---: | ---: | ---: | --- | ---: |
| Baseline priority | ${baseline.evaluable} | ${value(baseline.fairness, 3)} | ${value(baseline.throughputPerSecond)} | ${value(baseline.queueWait.p95, 0)} | ${value(baseline.endToEnd.p95, 0)} | ${baseline.workerIds.join(", ")} | ${baseline.potentialOrphans.length} |
| Fair WDRR | ${fair.evaluable} | ${value(fair.fairness, 3)} | ${value(fair.throughputPerSecond)} | ${value(fair.queueWait.p95, 0)} | ${value(fair.endToEnd.p95, 0)} | ${fair.workerIds.join(", ")} | ${fair.potentialOrphans.length} |
| Fair adaptive | ${adaptive.evaluable} | ${value(adaptive.fairness, 3)} | ${value(adaptive.throughputPerSecond)} | ${value(adaptive.queueWait.p95, 0)} | ${value(adaptive.endToEnd.p95, 0)} | ${adaptive.workerIds.join(", ")} | ${adaptive.potentialOrphans.length} |
## Steady-state validity
- Observation window: ${OBSERVATION_START_MS}-${OBSERVATION_END_MS} ms.
- Minimum successful starts: ${MIN_OBSERVATION_STARTS}.
- Every tenant must be backlogged in at least ${(MIN_BACKLOG_RATIO * 100).toFixed(0)} percent of samples.
- A zero-service tenant is counted as algorithmic starvation only when it passes the backlog gate.
## Tenant evidence
${TENANTS.map((tenant) => {
const before = baseline.tenants[tenant.id];
const after = fair.tenants[tenant.id];
return `- ${tenant.id}: normalized service ${value(before.normalizedService, 0)} -> ${value(after.normalizedService, 0)}, starvation windows ${before.starvationWindows} -> ${after.starvationWindows}, fair backlog ratio ${value(after.backlogRatio)}`;
}).join("\n")}
## Isolation evidence
- Fair peak reserved concurrency: ${JSON.stringify(fair.peakReservedConcurrency)}
- Fair peak reserved tokens: ${JSON.stringify(fair.peakReservedTokens)}
- Fair peak actual concurrency: ${JSON.stringify(fair.peakActualConcurrency)}
- Reservation terminal states: ${JSON.stringify(fair.reservationStates)}
- Potential Redis-to-BullMQ handoff orphans: ${fair.potentialOrphans.length}
## Adaptive controller trace
${adaptive.controllerEvents.map((event) => `- ${event.at} ms: ${event.reason} ${event.from} -> ${event.to}`).join("\n")}
## Architecture
Tenant arrivals enter separate Redis lists. A short Lua script performs weighted deficit selection, concurrency checks, token-budget checks, reservation creation, and cursor movement atomically. Admitted jobs enter one BullMQ execution queue. Two worker processes update actual activity, simulate provider work, publish measurements, and release reservations.
## Trade-offs and deferred work
- The Redis reservation and BullMQ add are intentionally observable but not one transaction. A reservation left in reserved state is surfaced by ID.
- Production crash recovery would add reservation leases, an outbox, or reconciliation.
- The short local benchmark demonstrates scheduler behavior, not production capacity.
- Strict deadlines, Redis Cluster deployment, leader election, durable metrics, and tenant-specific adaptive control are deferred.
`;
}
export async function runFullBenchmark() {
const baseline = await runScenario("baseline", "baseline", false);
const fair = await runScenario("fair", "fair", false);
const adaptive = await runScenario("adaptive", "fair", true);
printReport(baseline);
printReport(fair);
printReport(adaptive);
assertProject(baseline, fair, adaptive);
await fs.writeFile(
"benchmark-report.md",
reportMarkdown(baseline, fair, adaptive),
"utf8",
);
console.log("\nAll assertions passed. Wrote benchmark-report.md");
}Compare your src/cli.ts file with this reference.
import {
printReport,
runFullBenchmark,
runScenario,
} from "./benchmark.ts";
const command = process.argv[2];
if (command === "baseline") {
const report = await runScenario("baseline", "baseline", false);
printReport(report);
} else if (command === "fair") {
const report = await runScenario("fair", "fair", false);
printReport(report);
} else if (command === "benchmark") {
await runFullBenchmark();
} else {
throw new Error("Use baseline, fair, or benchmark");
}How Does the Baseline Prove Starvation?
- The workload generator creates arrivals from the existing TENANTS configuration.
- The baseline producer adds each task directly to one BullMQ queue with its tenant priority.
- The sampler subtracts started work from submitted work to estimate each tenant's backlog.
- The analysis counts service only between OBSERVATION_START_MS and OBSERVATION_END_MS.
Why Gate Jain's Fairness Index?
Jain's fairness index ranges from 1/n to 1 for nonnegative allocations. A value of 1 represents equal normalized allocations.
The benchmark requires at least 30 successful starts. It also requires every tenant to be backlogged in at least 80 percent of samples.
These gates separate scheduler starvation from a tenant that simply has no work waiting.
- Save src/benchmark.ts.
- Save src/cli.ts.
- Validate the complete TypeScript surface by running:
npm run typecheckTypeScript should exit without errors. Your queue producer, worker process, measurements, and CLI now agree on the same shared types.
Type Check Failing on the Benchmark?
Check that src/benchmark.ts imports from ./config.ts and ./lua.ts with the exact file extensions.
Check that src/cli.ts imports printReport, runFullBenchmark, and runScenario from ./benchmark.ts.
Help me trace the benchmark type-check error.
Run the priority baseline
The baseline sends every arrival directly into BullMQ priority scheduling. No tenant entitlement changes the dispatch order.
Before you run it, which tenant do you expect to dominate the successful starts?
- Run the baseline with one worker process by entering:
WORKER_COUNT=1 npm run baselineExpect the terminal to stay active for several seconds while tasks arrive and the observation window completes. The worker then closes before the report prints.
You should see evaluable=true with at least 30 successful starts. You should also see backlog ratios that satisfy the saturation gate.
The noisy tenant should dominate service because priority 1 runs before priorities 5 and 10. Backlogged lower-priority tenants can show starvation windows.
What Should the Report Tell You?
- The window line confirms that measurements came from the steady-state interval.
- The jain value should sit near the unfair end of its range.
- Each tenant line reports successful starts, normalized service, backlog ratio, and starvation windows.
- The worker list should contain only worker-1 for this run.
Baseline Not Evaluable?
Confirm that the my-redis container is still running. A stopped Redis container prevents the workload from recording samples.
Check that src/config.ts still contains the original arrival intervals and observation constants. Changing them can remove sustained backlog or reduce the sample.
Help me diagnose a non-evaluable baseline report.
You have reproduced the noisy-neighbor failure under sustained demand. Next, you will allocate cost-normalized service before work reaches BullMQ.
Dispatch Work with Weighted Deficit Round-Robin
Your baseline exposed the noisy-neighbor problem. Continuously backlogged tenants could wait while priority-1 noisy work dominated the queue.
This step adds a weighted deficit round-robin dispatcher before BullMQ. The dispatcher allocates cost-normalized service before admitted work enters the execution queue.
In this step, get ready to:
- Stage each tenant's arrivals in separate Redis pending lists.
- Select work with an atomic weighted-deficit reservation script.
- Track reservation lifecycles and verify fair service for every saturated tenant.
Create the atomic reservation scripts
The fair dispatcher stores pending tasks in Redis lists. A short Lua script can inspect shared limits and reserve one task as a single atomic operation.
Why weighted deficit round-robin?
The tenants submit tasks with different token costs. A job-count rotation could give each tenant the same number of starts while allocating very different amounts of service.
Each tenant earns credit from its configured weight. The dispatcher spends that credit using the selected task's token cost, which makes the configured entitlement measurable.
- Select the src folder in your editor's file sidebar.
- Create src/lua.ts with your editor's new-file control.
- Use the full-code tab below to add the atomic dispatch and lifecycle scripts to src/lua.ts.
✔️ I've added the Lua scripts
Confirm that src/lua.ts contains all five exported script constants.
ⓧ I'd like to double check the full code
Compare src/lua.ts with this complete file.
export const DISPATCH_LUA = String.raw`
local configs = cjson.decode(ARGV[1])
local quantum = tonumber(ARGV[2])
local reservationId = ARGV[3]
local now = tonumber(ARGV[4])
local count = #configs
local capacity = tonumber(redis.call('GET', KEYS[12]) or '1')
local globalInflight = tonumber(redis.call('HGET', KEYS[6], '__global') or '0')
if globalInflight >= capacity then
return nil
end
local cursor = tonumber(redis.call('GET', KEYS[5]) or '0')
local function updatePeak(hashKey, field, value)
local peak = tonumber(redis.call('HGET', hashKey, field) or '0')
if value > peak then
redis.call('HSET', hashKey, field, value)
end
end
for offset = 0, count - 1 do
local index = (cursor + offset) % count
local config = configs[index + 1]
local pendingKey = KEYS[index + 1]
if tonumber(redis.call('LLEN', pendingKey)) > 0 then
local tenantInflight = tonumber(redis.call('HGET', KEYS[6], config.id) or '0')
local tenantTokens = tonumber(redis.call('HGET', KEYS[7], config.id) or '0')
if tenantInflight < config.concurrency then
local rawTask = redis.call('LINDEX', pendingKey, 0)
local task = cjson.decode(rawTask)
if tenantTokens + task.tokenCost <= config.tokenBudget then
local credit = redis.call(
'HINCRBY',
KEYS[4],
config.id,
config.weight * quantum
)
if tonumber(credit) >= task.tokenCost then
redis.call('LPOP', pendingKey)
redis.call('HINCRBY', KEYS[4], config.id, -task.tokenCost)
local nextGlobal = redis.call('HINCRBY', KEYS[6], '__global', 1)
local nextTenant = redis.call('HINCRBY', KEYS[6], config.id, 1)
local nextTokens = redis.call('HINCRBY', KEYS[7], config.id, task.tokenCost)
updatePeak(KEYS[8], '__global', tonumber(nextGlobal))
updatePeak(KEYS[8], config.id, tonumber(nextTenant))
updatePeak(KEYS[9], config.id, tonumber(nextTokens))
local record = {
reservationId = reservationId,
taskId = task.id,
tenantId = config.id,
tokenCost = task.tokenCost,
state = 'reserved',
reservedAt = now,
stateChangedAt = now
}
redis.call('HSET', KEYS[10], reservationId, cjson.encode(record))
redis.call('RPUSH', KEYS[11], cjson.encode({
reservationId = reservationId,
taskId = task.id,
tenantId = config.id,
state = 'reserved',
at = now
}))
redis.call('SET', KEYS[5], tostring((index + 1) % count))
return {rawTask, reservationId}
end
end
end
end
end
return nil
`;
export const MARK_RESERVATION_LUA = String.raw`
local raw = redis.call('HGET', KEYS[1], ARGV[1])
if not raw then
return 'missing'
end
local record = cjson.decode(raw)
local nextState = ARGV[2]
if nextState == 'enqueued' and record.state ~= 'reserved' then
return record.state
end
if nextState == 'active' and record.state ~= 'reserved' and record.state ~= 'enqueued' then
return record.state
end
record.state = nextState
record.stateChangedAt = tonumber(ARGV[3])
if ARGV[4] ~= '' then
record.jobId = ARGV[4]
end
if ARGV[5] ~= '' then
record.workerId = ARGV[5]
end
redis.call('HSET', KEYS[1], ARGV[1], cjson.encode(record))
redis.call('RPUSH', KEYS[2], cjson.encode({
reservationId = ARGV[1],
taskId = record.taskId,
tenantId = record.tenantId,
state = nextState,
at = tonumber(ARGV[3]),
jobId = record.jobId,
workerId = record.workerId
}))
return nextState
`;
export const RELEASE_RESERVATION_LUA = String.raw`
local raw = redis.call('HGET', KEYS[3], ARGV[1])
if not raw then
return 'missing'
end
local record = cjson.decode(raw)
if record.releasedAt then
return record.state
end
redis.call('HINCRBY', KEYS[1], '__global', -1)
redis.call('HINCRBY', KEYS[1], record.tenantId, -1)
redis.call('HINCRBY', KEYS[2], record.tenantId, -record.tokenCost)
record.state = ARGV[2]
record.stateChangedAt = tonumber(ARGV[3])
record.releasedAt = tonumber(ARGV[3])
record.workerId = ARGV[4]
redis.call('HSET', KEYS[3], ARGV[1], cjson.encode(record))
redis.call('RPUSH', KEYS[4], cjson.encode({
reservationId = ARGV[1],
taskId = record.taskId,
tenantId = record.tenantId,
state = ARGV[2],
at = tonumber(ARGV[3]),
workerId = ARGV[4]
}))
return record.state
`;
export const ACTIVITY_START_LUA = String.raw`
local global = redis.call('HINCRBY', KEYS[1], '__global', 1)
local tenant = redis.call('HINCRBY', KEYS[1], ARGV[1], 1)
local peakGlobal = tonumber(redis.call('HGET', KEYS[2], '__global') or '0')
local peakTenant = tonumber(redis.call('HGET', KEYS[2], ARGV[1]) or '0')
if tonumber(global) > peakGlobal then
redis.call('HSET', KEYS[2], '__global', global)
end
if tonumber(tenant) > peakTenant then
redis.call('HSET', KEYS[2], ARGV[1], tenant)
end
return {global, tenant}
`;
export const ACTIVITY_END_LUA = String.raw`
redis.call('HINCRBY', KEYS[1], '__global', -1)
redis.call('HINCRBY', KEYS[1], ARGV[1], -1)
return 1
`;What do these scripts protect?
- The dispatcher checks the queue-wide capacity before selecting more work.
- The tenant checks enforce both concurrent-job limits and in-flight token budgets.
- The reservation scripts record every handoff state under one reservation ID.
- The release script ignores a repeated release after the first terminal transition.
- Save src/lua.ts.
- Check the new script exports by running:
npm run typecheckTypeScript exits without errors. The scheduler now has atomic primitives for selection, reservation state changes, release, and activity accounting.
Type check failing after adding the scripts?
Check that every template literal starts with String.raw` and ends with a matching backtick. A missing delimiter makes the remaining Lua text look like TypeScript.
Confirm that the five export names match the full-code tab exactly.
Help me compare my Lua script file with the required exports.
Wire the reservation lifecycle
A reservation begins before the job enters BullMQ. The worker advances that same record when execution starts and releases its shared capacity when execution reaches a terminal state.
- Select src/worker.ts in your editor's file sidebar.
- Use the full-code tab below to replace src/worker.ts with the lifecycle-aware worker.
✔️ My worker tracks reservations
Confirm that the worker marks reservations active before simulated provider work begins.
Confirm that its finally block always ends activity accounting and releases a fair reservation.
ⓧ I'd like to double check the full code
Compare src/worker.ts with this complete file.
import { Worker } from "bullmq";
import { Redis } from "ioredis";
import {
CONNECTION,
HARD_GLOBAL_LIMIT,
type Task,
type WorkerResult,
runKeys,
} from "./config.ts";
import {
ACTIVITY_END_LUA,
ACTIVITY_START_LUA,
MARK_RESERVATION_LUA,
RELEASE_RESERVATION_LUA,
} from "./lua.ts";
const [, , queueName, runId, workerId] = process.argv;
if (!queueName || !runId || !workerId) {
throw new Error("worker.ts requires queueName, runId, and workerId");
}
const redis = new Redis(CONNECTION);
const keys = runKeys(runId);
const sleep = (milliseconds: number) =>
new Promise<void>((resolve) => setTimeout(resolve, milliseconds));
async function callNumber(command: string, ...args: Array<string | number>) {
const value = await redis.call(command, ...args);
return Number(value ?? 0);
}
const worker = new Worker<Task, WorkerResult>(
queueName,
async (job) => {
const task = job.data;
const startedAt = Date.now();
let terminalState: "completed" | "throttled" = "completed";
await redis.call("HINCRBY", keys.started, task.tenantId, 1);
if (task.reservationId) {
await redis.eval(
MARK_RESERVATION_LUA,
2,
keys.reservations,
keys.lifecycle,
task.reservationId,
"active",
startedAt,
job.id ?? task.id,
workerId,
);
}
const activity = (await redis.eval(
ACTIVITY_START_LUA,
2,
keys.actual,
keys.actualPeak,
task.tenantId,
)) as [number, number];
try {
const runStartedAt = callNumber("GET", keys.runStartedAt);
const throttleEnabled = callNumber("GET", keys.throttleEnabled);
const [runStart, enabled] = await Promise.all([
runStartedAt,
throttleEnabled,
]);
const elapsed = startedAt - runStart;
const providerThreshold = elapsed >= 1_500 && elapsed < 2_700 ? 2 : 4;
const throttled = enabled === 1 && Number(activity[0]) > providerThreshold;
if (throttled) {
terminalState = "throttled";
await redis.call("INCR", keys.throttles);
await sleep(20);
} else {
await sleep(task.durationMs);
await redis.call("INCR", keys.successes);
}
const result: WorkerResult = {
taskId: task.id,
tenantId: task.tenantId,
tokenCost: task.tokenCost,
workerId,
createdAt: task.createdAt,
startedAt,
completedAt: Date.now(),
status: throttled ? "throttled" : "ok",
reservationId: task.reservationId,
};
await redis.call("RPUSH", keys.results, JSON.stringify(result));
return result;
} finally {
const finishedAt = Date.now();
await redis.eval(
ACTIVITY_END_LUA,
1,
keys.actual,
task.tenantId,
);
if (task.reservationId) {
await redis.eval(
RELEASE_RESERVATION_LUA,
4,
keys.inflight,
keys.tokens,
keys.reservations,
keys.lifecycle,
task.reservationId,
terminalState,
finishedAt,
workerId,
);
}
}
},
{
connection: CONNECTION,
concurrency: HARD_GLOBAL_LIMIT,
},
);
worker.on("error", (error) => {
console.error(`[${workerId}]`, error);
});
console.log(`[${workerId}] ready for ${queueName}`);
let closing = false;
async function shutdown() {
if (closing) return;
closing = true;
await worker.close();
await redis.quit();
process.exit(0);
}
process.on("SIGTERM", () => void shutdown());
process.on("SIGINT", () => void shutdown());How does the worker close the lifecycle?
- The worker records the reservation as active before simulating the request.
- The activity scripts measure real execution separately from reserved capacity.
- The terminal state records whether work completed or hit the deterministic throttle.
- The finally block releases counters even when processing exits through an error path.
- Save src/worker.ts.
- Select src/benchmark.ts in your editor's file sidebar.
Fair-mode arrivals need a staging path before they reach BullMQ. The dispatcher repeatedly asks Redis for one admissible reservation and marks the reservation as enqueued only after the queue accepts the job.
- Add the fair arrival branch below the existing baseline branch in produce() by using this code:
} else {
await redis.call(
"RPUSH",
keys.pending[task.tenantId],
JSON.stringify(task),
);
}What does the fair arrival branch do?
Baseline jobs still enter BullMQ immediately. Fair-mode jobs enter the pending list belonging to their tenant, which gives the dispatcher one explicit selection point.
- Add dispatchFair() below produce() by using this complete function:
async function dispatchFair(
redis: Redis,
queue: Queue<Task, WorkerResult>,
runId: string,
shouldStop: () => boolean,
) {
const keys = runKeys(runId);
const dispatchKeys = [
keys.pending.interactive,
keys.pending.standard,
keys.pending.noisy,
keys.deficit,
keys.cursor,
keys.inflight,
keys.tokens,
keys.peakInflight,
keys.peakTokens,
keys.reservations,
keys.lifecycle,
keys.capacity,
];
const configs = JSON.stringify(
TENANTS.map(({ id, weight, concurrency, tokenBudget }) => ({
id,
weight,
concurrency,
tokenBudget,
})),
);
let reservationSequence = 0;
while (!shouldStop()) {
reservationSequence += 1;
const reservationId = `${runId}-r${reservationSequence}`;
const result = (await redis.eval(
DISPATCH_LUA,
dispatchKeys.length,
...dispatchKeys,
configs,
QUANTUM,
reservationId,
Date.now(),
)) as [string, string] | null;
if (!result) {
await sleep(2);
continue;
}
const [rawTask, selectedReservationId] = result;
const task = JSON.parse(rawTask) as Task;
const reservedTask = { ...task, reservationId: selectedReservationId };
try {
const job = await queue.add("simulate-llm", reservedTask, {
jobId: task.id,
priority: 5,
removeOnComplete: true,
removeOnFail: true,
});
await redis.eval(
MARK_RESERVATION_LUA,
2,
keys.reservations,
keys.lifecycle,
selectedReservationId,
"enqueued",
Date.now(),
job.id ?? task.id,
"",
);
} catch (error) {
await redis.call(
"RPUSH",
keys.errors,
JSON.stringify({
reservationId: selectedReservationId,
message: error instanceof Error ? error.message : String(error),
at: Date.now(),
}),
);
}
}
}How does fair dispatch work?
- The ordered key list gives the Lua script every Redis key it can access.
- The tenant configuration passes weights and resource ceilings into the atomic decision.
- Each successful selection returns the serialized task with its reservation ID.
- The BullMQ handoff advances the lifecycle from reserved to enqueued only after the job is accepted.
- Import DISPATCH_LUA and MARK_RESERVATION_LUA from src/lua.ts with this line:
import { DISPATCH_LUA, MARK_RESERVATION_LUA } from "./lua.ts";Why import both scripts?
The dispatch script creates the reservation before the queue handoff. The marking script records that BullMQ accepted the selected task.
- Add the fair dispatcher beside the existing sampler setup in runScenario() with this code:
const dispatcher =
mode === "fair"
? dispatchFair(redis, queue, runId, () => stopped)
: Promise.resolve();What starts the dispatcher?
Fair scenarios run the dispatcher loop beside the producer and backlog sampler. Baseline scenarios resolve immediately because they continue to enqueue work directly.
- Update the scenario shutdown wait so the dispatcher finishes with the other loops:
await Promise.all([sampler, dispatcher, controller]);Why wait for the dispatcher?
The stop flag ends each background loop at the same scenario boundary. Waiting for all loops prevents Redis activity from leaking into analysis or cleanup.
- Save src/benchmark.ts.
- Validate the fair-dispatch integration by running:
npm run typecheckTypeScript exits without errors. The producer, dispatcher, worker, and lifecycle scripts now agree on the same task and reservation shapes.
Seeing missing script names or type errors?
Confirm that the Lua import sits with the other imports at the top of src/benchmark.ts. Check that dispatchFair() is outside produce().
Confirm that the fair producer branch uses keys.pending[task.tenantId] so each task reaches the correct tenant list.
Help me debug the fair dispatcher TypeScript integration.
Run and compare fair mode
The baseline measured entitlement only after priority scheduling had already decided who ran. Fair mode now decides entitlement first, while BullMQ remains responsible for executing admitted jobs.
What makes the fairness result valid?
The same steady-state eligibility gate still applies. The report needs at least 30 successful starts during the observation window.
Every tenant must also remain backlogged in at least 80 percent of samples. This ensures that a higher fairness score represents scheduling behavior under sustained demand.
Before you run fair mode, do you expect every saturated tenant to receive service this time?
- Run fair mode with one worker process by running:
WORKER_COUNT=1 npm run fairThe terminal reports an evaluable 1000-4000 ms window. All three tenants have successful starts, while normalized service values move closer to their weighted entitlements.
You also see reservation state counts and zero potential handoff orphans during the normal run. Jain fairness reaches at least 0.85 for the saturated workload.
How should you read the comparison?
Normalized service divides completed token cost by each tenant's weight. Similar normalized values show that the dispatcher respected the configured 2:1:1 entitlement.
Lower starvation-window counts show that backlogged tenants no longer disappear for repeated observation intervals. The reservation summary proves that admitted tasks passed through observable lifecycle states.
Fair mode not reaching the target?
Confirm that the my-redis container is still running. A stopped local Redis server prevents the worker and dispatcher from coordinating.
Check that QUANTUM remains 200 and that each tenant keeps the weight from src/config.ts.
Help me diagnose an unfair or non-evaluable fair-mode report.
You have replaced priority-based tenant entitlement with atomic cost-aware dispatch. Next, you will run two independent workers and prove that the shared resource ceilings survive distributed execution.
Prove Isolation Across Two Workers
Your weighted deficit round-robin dispatcher now gives every saturated tenant a share of service. One worker has already confirmed that the fairness policy works.
Local counters can hide races between independent processes. Two Node.js processes now run the same BullMQ worker while shared Redis accounting protects every capacity limit.
In this step, get ready to:
- Trace how two worker processes share one queue-wide concurrency ceiling.
- Inspect the shared activity accounting and idempotent reservation release.
- Run fair mode to audit the distributed isolation evidence.
Trace the two-process execution boundary
Each worker needs its own operating-system process for this test to expose process-local assumptions. The benchmark creates those processes with unique worker IDs.
- Return to src/benchmark.ts in your editor.
- Find the spawnWorkers() helper.
- Compare the helper with this process-spawning reference:
function spawnWorkers(queueName: string, runId: string, count: number) {
const children: ChildProcess[] = [];
for (let index = 1; index <= count; index += 1) {
const child = spawn(
process.execPath,
["--import=tsx", "src/worker.ts", queueName, runId, `worker-${index}`],
{ stdio: "inherit", env: process.env },
);
children.push(child);
}
return children;
}How does this create independent workers?
- The loop creates the requested number of child processes.
- Each process runs the same src/worker.ts implementation.
- The final argument assigns a unique ID such as worker-1 or worker-2.
- Inherited terminal output lets you see readiness messages from both processes.
- Scroll to the beginning of runScenario().
- Locate the worker-count and queue-capacity setup shown here:
const workerCount = Number(process.env.WORKER_COUNT ?? "2");
const runStartedAt = Date.now() + 250;
const samples: BacklogSample[] = [];
const controllerEvents: ControllerEvent[] = [];
let stopped = false;
await queue.setGlobalConcurrency(HARD_GLOBAL_LIMIT);
await redis.call("SET", keys.capacity, HARD_GLOBAL_LIMIT);
await redis.call("SET", keys.runStartedAt, runStartedAt);
await redis.call("SET", keys.throttleEnabled, adaptive ? 1 : 0);
const children = spawnWorkers(queueName, runId, workerCount);Why are there two concurrency settings?
Each worker can process up to HARD_GLOBAL_LIMIT jobs locally. BullMQ applies the same constant as a queue-wide ceiling across every worker.
Redis stores that capacity for the reservation script. Both admission and execution therefore use the shared limit of 4.
Inspect shared activity and release accounting
A shared reservation can stay within its limits while the real worker activity drifts higher. The activity scripts maintain a second set of counters for the jobs that are actually running.
- Switch back to src/lua.ts from earlier.
- Find the actual activity scripts shown here:
export const ACTIVITY_START_LUA = String.raw`
local global = redis.call('HINCRBY', KEYS[1], '__global', 1)
local tenant = redis.call('HINCRBY', KEYS[1], ARGV[1], 1)
local peakGlobal = tonumber(redis.call('HGET', KEYS[2], '__global') or '0')
local peakTenant = tonumber(redis.call('HGET', KEYS[2], ARGV[1]) or '0')
if tonumber(global) > peakGlobal then
redis.call('HSET', KEYS[2], '__global', global)
end
if tonumber(tenant) > peakTenant then
redis.call('HSET', KEYS[2], ARGV[1], tenant)
end
return {global, tenant}
`;
export const ACTIVITY_END_LUA = String.raw`
redis.call('HINCRBY', KEYS[1], '__global', -1)
redis.call('HINCRBY', KEYS[1], ARGV[1], -1)
return 1
`;What do the activity scripts measure?
- The start script increments the shared global counter for every active job.
- The same script increments the active counter for the job's tenant.
- The peak hash preserves the highest global value and tenant value observed during the run.
- The end script removes the activity after the worker finishes its attempt.
- Switch back to src/worker.ts from earlier.
- Find the finally block at the end of the worker processor.
- Compare the cleanup path with this reference:
} finally {
const finishedAt = Date.now();
await redis.eval(
ACTIVITY_END_LUA,
1,
keys.actual,
task.tenantId,
);
if (task.reservationId) {
await redis.eval(
RELEASE_RESERVATION_LUA,
4,
keys.inflight,
keys.tokens,
keys.reservations,
keys.lifecycle,
task.reservationId,
terminalState,
finishedAt,
workerId,
);
}
}Why does cleanup run in finally?
The finally block runs after successful work or throttled work. Actual activity is removed before the reservation reaches its terminal state.
RELEASE_RESERVATION_LUA checks releasedAt before changing counters. A duplicate release returns the existing state without subtracting capacity twice.
Run the distributed isolation check
The final audit combines worker identity with reserved peaks and actual peaks. It also keeps old reserved records visible as possible handoff orphans.
- Return to the analyze() function in src/benchmark.ts.
- Locate these distributed evidence fields in the returned report:
workerIds: [...new Set(successfulWindow.map((result) => result.workerId))].sort(),
reservationStates,
potentialOrphans,
peakReservedConcurrency: numericHash(peakInflight),
peakReservedTokens: numericHash(peakTokens),
peakActualConcurrency: numericHash(actualPeak),
controllerEvents,What does the report preserve?
- workerIds proves which independent processes completed work during the observation window.
- peakReservedConcurrency records the largest capacity reservation globally and per tenant.
- peakReservedTokens records each tenant's highest in-flight token total.
- peakActualConcurrency records how many jobs truly ran at the same time.
- potentialOrphans lists reservation IDs that remained reserved beyond ORPHAN_AGE_MS.
Before you run this, predict whether both worker processes can stay inside the shared ceiling.
- Run fair mode with its default two-worker configuration by running:
npm run fairWhat should this prove?
- You should see readiness lines for worker-1 and worker-2.
- The summary should include workers=worker-1,worker-2.
- The global entries in peakReservedConcurrency and peakActualConcurrency stay at or below 4.
- Each tenant peak stays within its configured concurrency and tokenBudget values.
- The orphan audit should report potential handoff orphans=0.
That is the distributed boundary proven: both worker processes handled jobs inside the shared reservation lifecycle.
Only seeing one worker or a failed run?
If only one worker appears, remove any persistent WORKER_COUNT override from your terminal session. Rerun fair mode so the default value creates two processes.
If the workers cannot connect, confirm that the my-redis container is still running in Docker Desktop.
help me diagnose my two-worker isolation benchmark
Two independent workers now share one enforceable capacity boundary. Next, you will test bounded recovery under provider throttling and generate the benchmark report.
Add Adaptive Recovery and Generate the Report
Your distributed BullMQ workers now share capacity without breaking global limits or tenant isolation. The scheduler can allocate fair service across both processes.
A fixed capacity cannot react when a shared provider starts throttling requests. This step adds bounded recovery to the scheduler. It also turns the benchmark evidence into a reusable report.
In this step, get ready to:
- Simulate a deterministic provider throttle window inside each worker.
- Decrease shared capacity after throttles and recover one slot at a time.
- Run every scenario and generate the benchmark report.
Simulate provider throttling
The adaptive controller needs a repeatable pressure signal. Each worker reads the run clock from Redis and applies the same temporary provider ceiling when adaptive mode is enabled.
- Switch back to src/worker.ts from the previous step.
- Add a terminal-state variable below the existing startedAt declaration.
- Set its initial value to completed with this line:
let terminalState: "completed" | "throttled" = "completed";Why track a terminal state?
The worker releases every reservation in its finally block. The terminal state lets that shared release path record whether provider work completed or was throttled.
- Inside the processor's existing try block, replace the direct task delay with this deterministic throttle logic:
try {
const runStartedAt = callNumber("GET", keys.runStartedAt);
const throttleEnabled = callNumber("GET", keys.throttleEnabled);
const [runStart, enabled] = await Promise.all([
runStartedAt,
throttleEnabled,
]);
const elapsed = startedAt - runStart;
const providerThreshold = elapsed >= 1_500 && elapsed < 2_700 ? 2 : 4;
const throttled = enabled === 1 && Number(activity[0]) > providerThreshold;
if (throttled) {
terminalState = "throttled";
await redis.call("INCR", keys.throttles);
await sleep(20);
} else {
await sleep(task.durationMs);
await redis.call("INCR", keys.successes);
}How does the throttle simulation work?
- The run clock makes every worker evaluate the same elapsed-time window.
- The temporary threshold drops from four active requests to two between 1_500 and 2_700 milliseconds.
- The counters give the controller a shared signal without coupling it to either worker process.
- Update the result object's status field to use the new throttled value.
- Pass terminalState to RELEASE_RESERVATION_LUA in the existing finally block.
- Save src/worker.ts.
Before you run fair mode, do you expect the throttle counters to change while adaptive mode remains disabled?
- Confirm the ordinary fair scenario still runs through both workers by running:
npm run fairWhat should this prove?
You should see both worker-1 and worker-2 in the report. You should also see zero potential handoff orphans.
Adaptive mode remains off during this run. The deterministic throttle window only affects the later adaptive scenario.
Does fair mode stop early?
Check that terminalState is declared inside the worker processor. Confirm that the result status uses the local throttled value.
Ask for help with the worker changes if the TypeScript process exits before the report appears: help me debug the deterministic throttle logic in src/worker.ts
Add bounded adaptive recovery
A throttle causes multiplicative decrease, which cuts capacity quickly. Recovery uses additive increase, so each clean period restores only one slot. The cooldown prevents repeated changes from turning temporary pressure into capacity oscillation.
- Switch back to src/benchmark.ts from the previous step.
- Add MIN_ADAPTIVE_LIMIT to the existing import from ./config.ts.
- Add this controller-event shape below BacklogSample:
interface ControllerEvent {
at: number;
from: number;
to: number;
reason: "throttle" | "recovery";
}What does the event record capture?
Each event records when capacity changed. It also records the previous limit, the next limit, and the reason for the transition.
These records make the controller testable. They also become the recovery trace in the final report.
The controller is one asynchronous loop. Add the next three code sections consecutively so they form one complete function.
- Add the controller signature and its shared-counter reads before objectFromRedisReply():
async function runAdaptiveController(
redis: Redis,
queue: Queue<Task, WorkerResult>,
runId: string,
runStartedAt: number,
shouldStop: () => boolean,
events: ControllerEvent[],
) {
const keys = runKeys(runId);
let capacity = HARD_GLOBAL_LIMIT;
let seenThrottles = 0;
let cleanTicks = 0;
let lastChangeAt = 0;
while (!shouldStop()) {
await sleep(250);
const now = Date.now();
const throttles = await redisNumber(redis, "GET", keys.throttles);
const newThrottles = throttles - seenThrottles;
seenThrottles = throttles;How does the controller observe pressure?
The loop samples the shared throttle counter every 250 milliseconds. Subtracting the previous count isolates throttles observed during the latest controller tick.
- Continue the same function with the decrease and recovery decisions:
let next = capacity;
let reason: ControllerEvent["reason"] | null = null;
if (newThrottles > 0 && now - lastChangeAt >= 800) {
next = Math.max(MIN_ADAPTIVE_LIMIT, Math.floor(capacity / 2));
cleanTicks = 0;
reason = "throttle";
} else if (newThrottles === 0) {
cleanTicks += 1;
if (
cleanTicks >= 4 &&
capacity < HARD_GLOBAL_LIMIT &&
now - lastChangeAt >= 800
) {
next = capacity + 1;
cleanTicks = 0;
reason = "recovery";
}
} else {
cleanTicks = 0;
}Why is recovery bounded?
A throttle halves capacity without crossing MIN_ADAPTIVE_LIMIT. Four clean ticks are required before recovery can begin.
Each accepted recovery adds exactly one slot. The 800 millisecond cooldown limits how quickly another transition can occur.
- Finish the function by applying accepted changes to BullMQ and the run-specific Redis state:
if (reason && next !== capacity) {
const event: ControllerEvent = {
at: now - runStartedAt,
from: capacity,
to: next,
reason,
};
capacity = next;
lastChangeAt = now;
events.push(event);
await queue.setGlobalConcurrency(capacity);
await redis.call("SET", keys.capacity, capacity);
await redis.call("RPUSH", keys.controllerEvents, JSON.stringify(event));
}
}
}Where does the capacity change apply?
The queue-wide BullMQ limit changes first. The run-specific Redis capacity then gives the fair dispatcher the same admission ceiling.
Appending the event preserves the evidence used by the benchmark assertions and report.
- Update runScenario() so throttleEnabled is set from its adaptive argument.
- Start runAdaptiveController() only when adaptive is true.
- Include the controller promise in the existing scenario shutdown wait.
- Extend assertProject() with the adaptive decrease, one-slot recovery, capacity-bound, and zero-orphan assertions.
- Add reportMarkdown() and runFullBenchmark() from the complete-file comparison below.
- Save src/benchmark.ts.
- Check the complete worker and benchmark files against the target implementation:
✔️ Awesome, I've got everything!
Great. Save src/worker.ts and src/benchmark.ts before running the type check.
ⓧ I'd like to double check the full code
Your complete src/worker.ts should match this file:
import { Worker } from "bullmq";
import { Redis } from "ioredis";
import {
CONNECTION,
HARD_GLOBAL_LIMIT,
type Task,
type WorkerResult,
runKeys,
} from "./config.ts";
import {
ACTIVITY_END_LUA,
ACTIVITY_START_LUA,
MARK_RESERVATION_LUA,
RELEASE_RESERVATION_LUA,
} from "./lua.ts";
const [, , queueName, runId, workerId] = process.argv;
if (!queueName || !runId || !workerId) {
throw new Error("worker.ts requires queueName, runId, and workerId");
}
const redis = new Redis(CONNECTION);
const keys = runKeys(runId);
const sleep = (milliseconds: number) =>
new Promise<void>((resolve) => setTimeout(resolve, milliseconds));
async function callNumber(command: string, ...args: Array<string | number>) {
const value = await redis.call(command, ...args);
return Number(value ?? 0);
}
const worker = new Worker<Task, WorkerResult>(
queueName,
async (job) => {
const task = job.data;
const startedAt = Date.now();
let terminalState: "completed" | "throttled" = "completed";
await redis.call("HINCRBY", keys.started, task.tenantId, 1);
if (task.reservationId) {
await redis.eval(
MARK_RESERVATION_LUA,
2,
keys.reservations,
keys.lifecycle,
task.reservationId,
"active",
startedAt,
job.id ?? task.id,
workerId,
);
}
const activity = (await redis.eval(
ACTIVITY_START_LUA,
2,
keys.actual,
keys.actualPeak,
task.tenantId,
)) as [number, number];
try {
const runStartedAt = callNumber("GET", keys.runStartedAt);
const throttleEnabled = callNumber("GET", keys.throttleEnabled);
const [runStart, enabled] = await Promise.all([
runStartedAt,
throttleEnabled,
]);
const elapsed = startedAt - runStart;
const providerThreshold = elapsed >= 1_500 && elapsed < 2_700 ? 2 : 4;
const throttled = enabled === 1 && Number(activity[0]) > providerThreshold;
if (throttled) {
terminalState = "throttled";
await redis.call("INCR", keys.throttles);
await sleep(20);
} else {
await sleep(task.durationMs);
await redis.call("INCR", keys.successes);
}
const result: WorkerResult = {
taskId: task.id,
tenantId: task.tenantId,
tokenCost: task.tokenCost,
workerId,
createdAt: task.createdAt,
startedAt,
completedAt: Date.now(),
status: throttled ? "throttled" : "ok",
reservationId: task.reservationId,
};
await redis.call("RPUSH", keys.results, JSON.stringify(result));
return result;
} finally {
const finishedAt = Date.now();
await redis.eval(
ACTIVITY_END_LUA,
1,
keys.actual,
task.tenantId,
);
if (task.reservationId) {
await redis.eval(
RELEASE_RESERVATION_LUA,
4,
keys.inflight,
keys.tokens,
keys.reservations,
keys.lifecycle,
task.reservationId,
terminalState,
finishedAt,
workerId,
);
}
}
},
{
connection: CONNECTION,
concurrency: HARD_GLOBAL_LIMIT,
},
);
worker.on("error", (error) => {
console.error(`[${workerId}]`, error);
});
console.log(`[${workerId}] ready for ${queueName}`);
let closing = false;
async function shutdown() {
if (closing) return;
closing = true;
await worker.close();
await redis.quit();
process.exit(0);
}
process.on("SIGTERM", () => void shutdown());
process.on("SIGINT", () => void shutdown());What should match in the worker?
The worker records actual activity before checking the deterministic provider threshold. Its finally block always ends activity tracking and releases a reservation with the correct terminal state.
Your complete src/benchmark.ts should match this file:
import assert from "node:assert/strict";
import { spawn, type ChildProcess } from "node:child_process";
import { once } from "node:events";
import * as fs from "node:fs/promises";
import { Queue } from "bullmq";
import { Redis } from "ioredis";
import {
allRunKeys,
CONNECTION,
HARD_GLOBAL_LIMIT,
MIN_ADAPTIVE_LIMIT,
MIN_BACKLOG_RATIO,
MIN_OBSERVATION_STARTS,
OBSERVATION_END_MS,
OBSERVATION_START_MS,
ORPHAN_AGE_MS,
QUANTUM,
RUN_MS,
SAMPLE_EVERY_MS,
TENANTS,
type ReservationRecord,
type Task,
type TenantId,
type WorkerResult,
runKeys,
tenantById,
} from "./config.ts";
import { DISPATCH_LUA, MARK_RESERVATION_LUA } from "./lua.ts";
type ScenarioMode = "baseline" | "fair";
interface BacklogSample {
at: number;
backlog: Record<TenantId, number>;
}
interface ControllerEvent {
at: number;
from: number;
to: number;
reason: "throttle" | "recovery";
}
interface TenantMetrics {
successfulStarts: number;
serviceTokens: number;
normalizedService: number;
starvationWindows: number;
queueWaitP95: number | null;
endToEndP95: number | null;
backlogRatio: number;
}
export interface ScenarioReport {
label: string;
mode: ScenarioMode;
adaptive: boolean;
runId: string;
observation: { startMs: number; endMs: number; successfulStarts: number };
evaluable: boolean;
evaluationNotes: string[];
fairness: number | null;
throughputPerSecond: number;
queueWait: { p50: number | null; p95: number | null; p99: number | null };
endToEnd: { p50: number | null; p95: number | null; p99: number | null };
tenants: Record<TenantId, TenantMetrics>;
workerIds: string[];
reservationStates: Record<string, number>;
potentialOrphans: string[];
peakReservedConcurrency: Record<string, number>;
peakReservedTokens: Record<string, number>;
peakActualConcurrency: Record<string, number>;
controllerEvents: ControllerEvent[];
}
const sleep = (milliseconds: number) =>
new Promise<void>((resolve) => setTimeout(resolve, milliseconds));
async function redisNumber(
redis: Redis,
command: string,
...args: Array<string | number>
) {
const value = await redis.call(command, ...args);
return Number(value ?? 0);
}
function generateWorkload(): Task[] {
const tasks: Task[] = [];
for (const tenant of TENANTS) {
let sequence = 0;
for (
let arrivalMs = 0;
arrivalMs < RUN_MS - 400;
arrivalMs += tenant.arrivalEveryMs
) {
sequence += 1;
tasks.push({
id: `${tenant.id}-${sequence}`,
tenantId: tenant.id,
arrivalMs,
durationMs: tenant.durationMs,
tokenCost: tenant.tokenCost,
createdAt: 0,
});
}
}
return tasks.sort(
(left, right) =>
left.arrivalMs - right.arrivalMs || left.id.localeCompare(right.id),
);
}
function spawnWorkers(queueName: string, runId: string, count: number) {
const children: ChildProcess[] = [];
for (let index = 1; index <= count; index += 1) {
const child = spawn(
process.execPath,
["--import=tsx", "src/worker.ts", queueName, runId, `worker-${index}`],
{ stdio: "inherit", env: process.env },
);
children.push(child);
}
return children;
}
async function stopWorkers(children: ChildProcess[]) {
await Promise.all(
children.map(async (child) => {
child.kill("SIGTERM");
await once(child, "close");
}),
);
}
async function produce(
redis: Redis,
queue: Queue<Task, WorkerResult>,
runId: string,
runStartedAt: number,
mode: ScenarioMode,
) {
const keys = runKeys(runId);
await Promise.all(
generateWorkload().map(async (template) => {
await sleep(Math.max(0, runStartedAt + template.arrivalMs - Date.now()));
const task = { ...template, createdAt: Date.now() };
await redis.call("HINCRBY", keys.submitted, task.tenantId, 1);
if (mode === "baseline") {
const tenant = tenantById(task.tenantId);
await queue.add("simulate-llm", task, {
jobId: task.id,
priority: tenant.priority,
removeOnComplete: true,
removeOnFail: true,
});
} else {
await redis.call(
"RPUSH",
keys.pending[task.tenantId],
JSON.stringify(task),
);
}
}),
);
}
async function dispatchFair(
redis: Redis,
queue: Queue<Task, WorkerResult>,
runId: string,
shouldStop: () => boolean,
) {
const keys = runKeys(runId);
const dispatchKeys = [
keys.pending.interactive,
keys.pending.standard,
keys.pending.noisy,
keys.deficit,
keys.cursor,
keys.inflight,
keys.tokens,
keys.peakInflight,
keys.peakTokens,
keys.reservations,
keys.lifecycle,
keys.capacity,
];
const configs = JSON.stringify(
TENANTS.map(({ id, weight, concurrency, tokenBudget }) => ({
id,
weight,
concurrency,
tokenBudget,
})),
);
let reservationSequence = 0;
while (!shouldStop()) {
reservationSequence += 1;
const reservationId = `${runId}-r${reservationSequence}`;
const result = (await redis.eval(
DISPATCH_LUA,
dispatchKeys.length,
...dispatchKeys,
configs,
QUANTUM,
reservationId,
Date.now(),
)) as [string, string] | null;
if (!result) {
await sleep(2);
continue;
}
const [rawTask, selectedReservationId] = result;
const task = JSON.parse(rawTask) as Task;
const reservedTask = { ...task, reservationId: selectedReservationId };
try {
const job = await queue.add("simulate-llm", reservedTask, {
jobId: task.id,
priority: 5,
removeOnComplete: true,
removeOnFail: true,
});
await redis.eval(
MARK_RESERVATION_LUA,
2,
keys.reservations,
keys.lifecycle,
selectedReservationId,
"enqueued",
Date.now(),
job.id ?? task.id,
"",
);
} catch (error) {
await redis.call(
"RPUSH",
keys.errors,
JSON.stringify({
reservationId: selectedReservationId,
message: error instanceof Error ? error.message : String(error),
at: Date.now(),
}),
);
}
}
}
async function sampleBacklog(
redis: Redis,
runId: string,
runStartedAt: number,
shouldStop: () => boolean,
samples: BacklogSample[],
) {
const keys = runKeys(runId);
while (!shouldStop()) {
const at = Date.now() - runStartedAt;
if (at >= OBSERVATION_START_MS && at < OBSERVATION_END_MS) {
const backlog = {} as Record<TenantId, number>;
for (const tenant of TENANTS) {
const submitted = await redisNumber(
redis,
"HGET",
keys.submitted,
tenant.id,
);
const started = await redisNumber(
redis,
"HGET",
keys.started,
tenant.id,
);
backlog[tenant.id] = Math.max(0, submitted - started);
}
samples.push({ at, backlog });
}
await sleep(SAMPLE_EVERY_MS);
}
}
async function runAdaptiveController(
redis: Redis,
queue: Queue<Task, WorkerResult>,
runId: string,
runStartedAt: number,
shouldStop: () => boolean,
events: ControllerEvent[],
) {
const keys = runKeys(runId);
let capacity = HARD_GLOBAL_LIMIT;
let seenThrottles = 0;
let cleanTicks = 0;
let lastChangeAt = 0;
while (!shouldStop()) {
await sleep(250);
const now = Date.now();
const throttles = await redisNumber(redis, "GET", keys.throttles);
const newThrottles = throttles - seenThrottles;
seenThrottles = throttles;
let next = capacity;
let reason: ControllerEvent["reason"] | null = null;
if (newThrottles > 0 && now - lastChangeAt >= 800) {
next = Math.max(MIN_ADAPTIVE_LIMIT, Math.floor(capacity / 2));
cleanTicks = 0;
reason = "throttle";
} else if (newThrottles === 0) {
cleanTicks += 1;
if (
cleanTicks >= 4 &&
capacity < HARD_GLOBAL_LIMIT &&
now - lastChangeAt >= 800
) {
next = capacity + 1;
cleanTicks = 0;
reason = "recovery";
}
} else {
cleanTicks = 0;
}
if (reason && next !== capacity) {
const event: ControllerEvent = {
at: now - runStartedAt,
from: capacity,
to: next,
reason,
};
capacity = next;
lastChangeAt = now;
events.push(event);
await queue.setGlobalConcurrency(capacity);
await redis.call("SET", keys.capacity, capacity);
await redis.call("RPUSH", keys.controllerEvents, JSON.stringify(event));
}
}
}
function objectFromRedisReply(reply: unknown): Record<string, string> {
if (!reply) return {};
if (!Array.isArray(reply)) return reply as Record<string, string>;
const object: Record<string, string> = {};
for (let index = 0; index < reply.length; index += 2) {
object[String(reply[index])] = String(reply[index + 1]);
}
return object;
}
async function readHash(redis: Redis, key: string) {
return objectFromRedisReply(await redis.call("HGETALL", key));
}
async function drainResults(redis: Redis, key: string) {
const results: WorkerResult[] = [];
while (true) {
const raw = (await redis.call("LPOP", key)) as string | null;
if (!raw) break;
results.push(JSON.parse(raw) as WorkerResult);
}
return results;
}
function percentile(values: number[], percentileValue: number): number | null {
if (values.length === 0) return null;
const sorted = [...values].sort((left, right) => left - right);
const index = Math.max(
0,
Math.ceil((percentileValue / 100) * sorted.length) - 1,
);
return sorted[index];
}
function jainIndex(values: number[]): number {
const sum = values.reduce((total, value) => total + value, 0);
const squares = values.reduce((total, value) => total + value * value, 0);
return squares === 0 ? 0 : (sum * sum) / (values.length * squares);
}
function countStarvationWindows(
tenantId: TenantId,
results: WorkerResult[],
samples: BacklogSample[],
) {
let count = 0;
for (
let start = OBSERVATION_START_MS;
start < OBSERVATION_END_MS;
start += 500
) {
const end = start + 500;
const windowSamples = samples.filter(
(sample) => sample.at >= start && sample.at < end,
);
const backlogged =
windowSamples.length > 0 &&
windowSamples.filter((sample) => sample.backlog[tenantId] > 0).length /
windowSamples.length >=
0.5;
const starts = results.filter(
(result) =>
result.tenantId === tenantId &&
result.status === "ok" &&
result.startedAt >= start &&
result.startedAt < end,
).length;
if (backlogged && starts === 0) count += 1;
}
return count;
}
function numericHash(hash: Record<string, string>) {
return Object.fromEntries(
Object.entries(hash).map(([key, value]) => [key, Number(value)]),
);
}
function analyze(
label: string,
mode: ScenarioMode,
adaptive: boolean,
runId: string,
runStartedAt: number,
results: WorkerResult[],
samples: BacklogSample[],
reservations: Record<string, string>,
peakInflight: Record<string, string>,
peakTokens: Record<string, string>,
actualPeak: Record<string, string>,
controllerEvents: ControllerEvent[],
): ScenarioReport {
const successfulWindow = results.filter((result) => {
const relativeStart = result.startedAt - runStartedAt;
return (
result.status === "ok" &&
relativeStart >= OBSERVATION_START_MS &&
relativeStart < OBSERVATION_END_MS
);
});
const evaluationNotes: string[] = [];
const backlogRatios = {} as Record<TenantId, number>;
for (const tenant of TENANTS) {
backlogRatios[tenant.id] =
samples.length === 0
? 0
: samples.filter((sample) => sample.backlog[tenant.id] > 0).length /
samples.length;
if (backlogRatios[tenant.id] < MIN_BACKLOG_RATIO) {
evaluationNotes.push(
`${tenant.id} backlog ratio ${backlogRatios[tenant.id].toFixed(2)} is below ${MIN_BACKLOG_RATIO.toFixed(2)}; demand may be insufficient for a fairness claim.`,
);
}
}
if (successfulWindow.length < MIN_OBSERVATION_STARTS) {
evaluationNotes.push(
`Only ${successfulWindow.length} successful starts were observed; at least ${MIN_OBSERVATION_STARTS} are required.`,
);
}
const evaluable = evaluationNotes.length === 0;
const service = Object.fromEntries(
TENANTS.map((tenant) => [
tenant.id,
successfulWindow
.filter((result) => result.tenantId === tenant.id)
.reduce((total, result) => total + result.tokenCost, 0),
]),
) as Record<TenantId, number>;
const normalized = Object.fromEntries(
TENANTS.map((tenant) => [tenant.id, service[tenant.id] / tenant.weight]),
) as Record<TenantId, number>;
const fairness = evaluable ? jainIndex(Object.values(normalized)) : null;
const queueWaitValues = successfulWindow.map(
(result) => result.startedAt - result.createdAt,
);
const endToEndValues = successfulWindow.map(
(result) => result.completedAt - result.createdAt,
);
const shiftedResults = results.map((result) => ({
...result,
startedAt: result.startedAt - runStartedAt,
}));
const tenants = {} as Record<TenantId, TenantMetrics>;
for (const tenant of TENANTS) {
const tenantResults = successfulWindow.filter(
(result) => result.tenantId === tenant.id,
);
tenants[tenant.id] = {
successfulStarts: tenantResults.length,
serviceTokens: service[tenant.id],
normalizedService: normalized[tenant.id],
starvationWindows: countStarvationWindows(
tenant.id,
shiftedResults,
samples,
),
queueWaitP95: percentile(
tenantResults.map((result) => result.startedAt - result.createdAt),
95,
),
endToEndP95: percentile(
tenantResults.map((result) => result.completedAt - result.createdAt),
95,
),
backlogRatio: backlogRatios[tenant.id],
};
}
const reservationRecords = Object.values(reservations).map(
(raw) => JSON.parse(raw) as ReservationRecord,
);
const reservationStates: Record<string, number> = {};
for (const record of reservationRecords) {
reservationStates[record.state] = (reservationStates[record.state] ?? 0) + 1;
}
const runEndedAt = runStartedAt + RUN_MS;
const potentialOrphans = reservationRecords
.filter(
(record) =>
record.state === "reserved" &&
runEndedAt - record.reservedAt >= ORPHAN_AGE_MS,
)
.map((record) => record.reservationId);
return {
label,
mode,
adaptive,
runId,
observation: {
startMs: OBSERVATION_START_MS,
endMs: OBSERVATION_END_MS,
successfulStarts: successfulWindow.length,
},
evaluable,
evaluationNotes,
fairness,
throughputPerSecond:
successfulWindow.length /
((OBSERVATION_END_MS - OBSERVATION_START_MS) / 1_000),
queueWait: {
p50: percentile(queueWaitValues, 50),
p95: percentile(queueWaitValues, 95),
p99: percentile(queueWaitValues, 99),
},
endToEnd: {
p50: percentile(endToEndValues, 50),
p95: percentile(endToEndValues, 95),
p99: percentile(endToEndValues, 99),
},
tenants,
workerIds: [...new Set(successfulWindow.map((result) => result.workerId))].sort(),
reservationStates,
potentialOrphans,
peakReservedConcurrency: numericHash(peakInflight),
peakReservedTokens: numericHash(peakTokens),
peakActualConcurrency: numericHash(actualPeak),
controllerEvents,
};
}
export async function runScenario(
label: string,
mode: ScenarioMode,
adaptive: boolean,
): Promise<ScenarioReport> {
const runId = `${label}-${Date.now()}-${process.pid}`;
const queueName = `wdrr-${runId}`;
const redis = new Redis(CONNECTION);
const queue = new Queue<Task, WorkerResult>(queueName, {
connection: CONNECTION,
});
const keys = runKeys(runId);
const workerCount = Number(process.env.WORKER_COUNT ?? "2");
const runStartedAt = Date.now() + 250;
const samples: BacklogSample[] = [];
const controllerEvents: ControllerEvent[] = [];
let stopped = false;
await queue.setGlobalConcurrency(HARD_GLOBAL_LIMIT);
await redis.call("SET", keys.capacity, HARD_GLOBAL_LIMIT);
await redis.call("SET", keys.runStartedAt, runStartedAt);
await redis.call("SET", keys.throttleEnabled, adaptive ? 1 : 0);
const children = spawnWorkers(queueName, runId, workerCount);
const producer = produce(redis, queue, runId, runStartedAt, mode);
const sampler = sampleBacklog(
redis,
runId,
runStartedAt,
() => stopped,
samples,
);
const dispatcher =
mode === "fair"
? dispatchFair(redis, queue, runId, () => stopped)
: Promise.resolve();
const controller = adaptive
? runAdaptiveController(
redis,
queue,
runId,
runStartedAt,
() => stopped,
controllerEvents,
)
: Promise.resolve();
await sleep(Math.max(0, runStartedAt + RUN_MS - Date.now()));
await producer;
stopped = true;
await Promise.all([sampler, dispatcher, controller]);
await sleep(350);
await stopWorkers(children);
const [
results,
reservations,
peakInflight,
peakTokens,
actualPeak,
] = await Promise.all([
drainResults(redis, keys.results),
readHash(redis, keys.reservations),
readHash(redis, keys.peakInflight),
readHash(redis, keys.peakTokens),
readHash(redis, keys.actualPeak),
]);
const report = analyze(
label,
mode,
adaptive,
runId,
runStartedAt,
results,
samples,
reservations,
peakInflight,
peakTokens,
actualPeak,
controllerEvents,
);
await queue.obliterate({ force: true });
await queue.close();
await redis.call("DEL", ...allRunKeys(runId));
await redis.quit();
return report;
}
function value(value: number | null, digits = 2) {
return value === null ? "n/a" : value.toFixed(digits);
}
export function printReport(report: ScenarioReport) {
console.log(`\n${report.label}`);
console.log(
`window=${report.observation.startMs}-${report.observation.endMs}ms evaluable=${report.evaluable} starts=${report.observation.successfulStarts}`,
);
console.log(
`jain=${value(report.fairness, 3)} throughput=${value(report.throughputPerSecond)} tasks/s workers=${report.workerIds.join(",")}`,
);
console.log(
`queue wait p50/p95/p99=${value(report.queueWait.p50, 0)}/${value(report.queueWait.p95, 0)}/${value(report.queueWait.p99, 0)}ms`,
);
for (const tenant of TENANTS) {
const metrics = report.tenants[tenant.id];
console.log(
`${tenant.id}: starts=${metrics.successfulStarts} normalized=${value(metrics.normalizedService, 0)} backlog=${value(metrics.backlogRatio)} starvationWindows=${metrics.starvationWindows}`,
);
}
console.log(`reservation states=${JSON.stringify(report.reservationStates)}`);
console.log(`potential handoff orphans=${report.potentialOrphans.length}`);
if (!report.evaluable) {
console.log(`evaluation notes=${report.evaluationNotes.join(" | ")}`);
}
}
function assertProject(
baseline: ScenarioReport,
fair: ScenarioReport,
adaptive: ScenarioReport,
) {
assert.equal(baseline.evaluable, true, baseline.evaluationNotes.join(" | "));
assert.equal(fair.evaluable, true, fair.evaluationNotes.join(" | "));
assert.notEqual(baseline.fairness, null);
assert.notEqual(fair.fairness, null);
assert.ok(fair.fairness! >= 0.85, `fair Jain index was ${fair.fairness}`);
assert.ok(
fair.fairness! - baseline.fairness! >= 0.15,
`fairness improvement was ${fair.fairness! - baseline.fairness!}`,
);
assert.ok(
fair.throughputPerSecond >= baseline.throughputPerSecond * 0.85,
"fair throughput fell more than 15 percent below baseline",
);
assert.ok(
Object.values(fair.tenants).every((tenant) => tenant.successfulStarts > 0),
"a backlogged tenant received no fair-mode service",
);
assert.ok(
Object.values(fair.tenants).reduce(
(total, tenant) => total + tenant.starvationWindows,
0,
) <
Object.values(baseline.tenants).reduce(
(total, tenant) => total + tenant.starvationWindows,
0,
),
"fair mode did not reduce starvation windows",
);
assert.ok(
Number(fair.peakReservedConcurrency.__global ?? 0) <= HARD_GLOBAL_LIMIT,
);
assert.ok(
Number(fair.peakActualConcurrency.__global ?? 0) <= HARD_GLOBAL_LIMIT,
);
for (const tenant of TENANTS) {
assert.ok(
Number(fair.peakReservedConcurrency[tenant.id] ?? 0) <= tenant.concurrency,
`${tenant.id} exceeded its concurrency limit`,
);
assert.ok(
Number(fair.peakReservedTokens[tenant.id] ?? 0) <= tenant.tokenBudget,
`${tenant.id} exceeded its token budget`,
);
}
assert.deepEqual(fair.workerIds, ["worker-1", "worker-2"]);
assert.equal(fair.potentialOrphans.length, 0);
assert.ok(
adaptive.controllerEvents.some((event) => event.reason === "throttle"),
"adaptive run never decreased capacity",
);
assert.ok(
adaptive.controllerEvents.some((event) => event.reason === "recovery"),
"adaptive run never recovered capacity",
);
for (const event of adaptive.controllerEvents) {
assert.ok(event.to >= MIN_ADAPTIVE_LIMIT && event.to <= HARD_GLOBAL_LIMIT);
if (event.reason === "recovery") {
assert.equal(event.to - event.from, 1);
}
}
assert.equal(adaptive.potentialOrphans.length, 0);
}
function reportMarkdown(
baseline: ScenarioReport,
fair: ScenarioReport,
adaptive: ScenarioReport,
) {
const fairnessImprovement = fair.fairness! - baseline.fairness!;
return `# Distributed WDRR Scheduler Benchmark
## Result
The fair scheduler passed its steady-state eligibility gate and improved Jain's fairness index from ${value(baseline.fairness, 3)} to ${value(fair.fairness, 3)}, an improvement of ${value(fairnessImprovement, 3)}.
| Scenario | Evaluable | Jain fairness | Throughput tasks/s | Queue wait p95 ms | End-to-end p95 ms | Workers | Potential orphans |
| --- | --- | ---: | ---: | ---: | ---: | --- | ---: |
| Baseline priority | ${baseline.evaluable} | ${value(baseline.fairness, 3)} | ${value(baseline.throughputPerSecond)} | ${value(baseline.queueWait.p95, 0)} | ${value(baseline.endToEnd.p95, 0)} | ${baseline.workerIds.join(", ")} | ${baseline.potentialOrphans.length} |
| Fair WDRR | ${fair.evaluable} | ${value(fair.fairness, 3)} | ${value(fair.throughputPerSecond)} | ${value(fair.queueWait.p95, 0)} | ${value(fair.endToEnd.p95, 0)} | ${fair.workerIds.join(", ")} | ${fair.potentialOrphans.length} |
| Fair adaptive | ${adaptive.evaluable} | ${value(adaptive.fairness, 3)} | ${value(adaptive.throughputPerSecond)} | ${value(adaptive.queueWait.p95, 0)} | ${value(adaptive.endToEnd.p95, 0)} | ${adaptive.workerIds.join(", ")} | ${adaptive.potentialOrphans.length} |
## Steady-state validity
- Observation window: ${OBSERVATION_START_MS}-${OBSERVATION_END_MS} ms.
- Minimum successful starts: ${MIN_OBSERVATION_STARTS}.
- Every tenant must be backlogged in at least ${(MIN_BACKLOG_RATIO * 100).toFixed(0)} percent of samples.
- A zero-service tenant is counted as algorithmic starvation only when it passes the backlog gate.
## Tenant evidence
${TENANTS.map((tenant) => {
const before = baseline.tenants[tenant.id];
const after = fair.tenants[tenant.id];
return `- ${tenant.id}: normalized service ${value(before.normalizedService, 0)} -> ${value(after.normalizedService, 0)}, starvation windows ${before.starvationWindows} -> ${after.starvationWindows}, fair backlog ratio ${value(after.backlogRatio)}`;
}).join("\n")}
## Isolation evidence
- Fair peak reserved concurrency: ${JSON.stringify(fair.peakReservedConcurrency)}
- Fair peak reserved tokens: ${JSON.stringify(fair.peakReservedTokens)}
- Fair peak actual concurrency: ${JSON.stringify(fair.peakActualConcurrency)}
- Reservation terminal states: ${JSON.stringify(fair.reservationStates)}
- Potential Redis-to-BullMQ handoff orphans: ${fair.potentialOrphans.length}
## Adaptive controller trace
${adaptive.controllerEvents.map((event) => `- ${event.at} ms: ${event.reason} ${event.from} -> ${event.to}`).join("\n")}
## Architecture
Tenant arrivals enter separate Redis lists. A short Lua script performs weighted deficit selection, concurrency checks, token-budget checks, reservation creation, and cursor movement atomically. Admitted jobs enter one BullMQ execution queue. Two worker processes update actual activity, simulate provider work, publish measurements, and release reservations.
## Trade-offs and deferred work
- The Redis reservation and BullMQ add are intentionally observable but not one transaction. A reservation left in reserved state is surfaced by ID.
- Production crash recovery would add reservation leases, an outbox, or reconciliation.
- The short local benchmark demonstrates scheduler behavior, not production capacity.
- Strict deadlines, Redis Cluster deployment, leader election, durable metrics, and tenant-specific adaptive control are deferred.
`;
}
export async function runFullBenchmark() {
const baseline = await runScenario("baseline", "baseline", false);
const fair = await runScenario("fair", "fair", false);
const adaptive = await runScenario("adaptive", "fair", true);
printReport(baseline);
printReport(fair);
printReport(adaptive);
assertProject(baseline, fair, adaptive);
await fs.writeFile(
"benchmark-report.md",
reportMarkdown(baseline, fair, adaptive),
"utf8",
);
console.log("\nAll assertions passed. Wrote benchmark-report.md");
}What should match in the benchmark?
The complete file runs baseline, fair, and adaptive scenarios with fresh run identifiers. It checks fairness, throughput, isolation, worker coverage, orphan counts, and controller bounds before writing the report.
- Validate both edited TypeScript files by running:
npm run typecheckWhat does this check prove?
A clean exit confirms that the controller event type, scenario wiring, assertions, report builder, and worker status changes agree across the repository.
Does the type check fail?
Check that MIN_ADAPTIVE_LIMIT appears in the import from ./config.ts. Confirm that every runScenario() call includes its adaptive Boolean.
Use the full-file comparison to check misplaced braces around runAdaptiveController(). The three controller snippets form one function.
Ask for targeted help if the compiler still reports a mismatch: help me resolve the TypeScript errors in the adaptive benchmark controller
Run the benchmark and review the report
The final command runs three fresh scenarios. It accepts the fairness claim only when the steady-state workload remains backlogged and the observation sample is large enough.
The run takes several timed windows to finish, so the terminal stays busy while the workers process each scenario. That pause is the benchmark collecting evidence.
Before you run it, do you expect every assertion to pass while the adaptive trace shows both a decrease and a one-slot recovery?
- Run the complete benchmark by running:
npm run benchmarkWhat should the benchmark prove?
You should see baseline, fair, and adaptive results. The final line should report All assertions passed. Wrote benchmark-report.md.
The fair Jain's fairness index should be at least 0.85. It should also improve on the baseline by at least 0.15.
- Return to the Explorer sidebar in your editor.
- Open benchmark-report.md from the repository folder.
- Confirm the scenario table includes Baseline priority, Fair WDRR, and Fair adaptive.
- Confirm the report contains steady-state validity and tenant evidence.
- Confirm the report contains isolation evidence and an adaptive controller trace.
- Confirm the report ends with architecture and deferred production work.
Do the benchmark assertions fail?
Confirm that the my-redis container is still running. A stopped local Redis server prevents every scenario from completing.
Check the first assertion message in the terminal. It identifies whether the run failed its demand gate, fairness target, isolation check, worker check, orphan audit, or controller bound.
Ask for help with the reported assertion before changing a benchmark threshold: help me diagnose the failed distributed scheduler benchmark assertion
That completes the scheduler. You now have a reproducible noisy-neighbor baseline, distributed fair dispatch, bounded adaptive recovery, and a Markdown report that records the evidence.
Secret mission
Repair a Distributed Controller Race
Two recovery controllers can observe the same capacity before either records its decision. Reproduce the resulting two-slot jump, then move the recovery decision into one atomic Redis Lua script so only one controller accepts the shared slot.
Clean Up Your Resources
Clean Up Your Resources
The project makes no external API calls. It creates no cloud resources, so there are no ongoing usage costs.
Licensing for Docker Desktop depends on your usage context. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.
Resources you used:
- The local Redis container named my-redis from redis:latest with host port 6379 mapped.
- The project dependencies installed in node_modules inside your scheduler repository.
Your source files remain available for future experiments. The generated benchmark-report.md remains as evidence from your benchmark.
Keep everything running
No action is needed. Choose this option if you want to continue testing the scheduler or rerun its benchmarks.
- Keep my-redis running on port 6379 for future scheduler experiments.
- Keep node_modules so the benchmark scripts remain ready to run.
- Retain the source files plus benchmark-report.md for future reference.
Pause - I'll come back to this later
Stop the Redis container to free its mapped port. The container remains available for your next benchmark session.
- Switch back to the terminal from your benchmark run.
- Stop the local Redis container by running this command:
docker stop my-redisWhat Does This Command Do?
The docker stop command stops the running my-redis container. Its local container state remains available for a later restart.
- Return to Docker Desktop to check the state of my-redis.
You should see my-redis stopped instead of running.
- Start my-redis from Docker Desktop before you rerun the benchmark.
Delete - I don't want to use this again
Removing my-redis permanently clears the Redis data stored in that container. Your source files plus benchmark-report.md stay untouched.
- Return to Docker Desktop.
- Locate my-redis among your local containers.
- Stop my-redis with its stop control.
- Remove my-redis with its delete control.
- Refresh the local container list.
You should no longer see my-redis among your local containers.
- Open the existing scheduler repository in Finder on macOS or File Explorer on Windows.
- Select the node_modules folder.
- Remove node_modules with the file manager's deletion control.
- Refresh the repository folder.
You should no longer see node_modules. The project source plus benchmark-report.md remain available.
Nice Work!
Nice Work!
You did it. Your distributed TypeScript scheduler now turns a visible noisy-neighbor failure into measurable weighted fairness.
You've learned how to:
- Create a deterministic noisy-neighbor workload that drives three saturated tenants through the same arrival pattern. Apply a steady-state fairness gate only after every tenant meets the backlog threshold. Require enough successful starts before accepting the result.
- Implement weighted deficit round-robin to allocate token-normalized service by tenant weight. Use Redis Lua to select work atomically. Reserve shared capacity in the same decision. Track each reservation ID from reserved through its terminal lifecycle state.
- Prove distributed isolation across two independent BullMQ workers without breaching global or tenant ceilings. Calculate Jain's fairness index from cost-normalized service. Add bounded adaptive concurrency that recovers one slot at a time. Generate benchmark-report.md as the measured evidence behind the passing assertions.
- Secret Mission: Reproduce a two-controller recovery race with a Redis barrier that drives capacity from 2 to 4. Repair the race with ATOMIC_RECOVERY_LUA. Exactly one controller accepts recovery from 2 to 3 while the other observes the shared cooldown.
Ready to quiz yourself?