Build a Budget-Aware LLM Dispatch Queue

Introduction

30 Second Summary

When several AI requests arrive at once, a fast response can hide a costly problem: every request may believe the same budget is available. Average response times can also disguise how long the slowest requests wait.

In this project, you will build a local LLM dispatch laboratory that sends simulated jobs through two independent workers. A BullMQ queue backed by Redis will compare naive budget checks with an atomic Lua reservation.

What You'll Build

Picture one terminal command producing a table where both worker IDs, three latency distributions, and explicit safety checks are visible together.

By the end of this project, you'll have:

• A reproducible dispatch benchmark where the naive gate admits 400 tokens from a 200-token budget before Lua safely caps the same race at 200 tokens.
• A shared-control proof showing work split across both workers while combined concurrency stays at four.
• A benchmark-results.json report with throughput, budget utilization, and separate p50/p95/p99 measurements for queue wait, processing, and end-to-end latency.
• Secret Mission: Simulate a provider throttle that delays the queue before returning the affected job to waiting without recording a normal failure.

Are there any prerequisites?

You should be comfortable with TypeScript.

The project expects Node.js 20 or newer, Docker Desktop, and VS Code on macOS.

Before We Start

Before the hands-on work begins, lock in why you are building a measurable distributed LLM dispatch system. Every mechanism will connect to a production risk such as overload, unfair scheduling, latency, or budget overspend.

Set Up the Redis Benchmark Lab

Distributed measurements only mean something when every process reaches the same Redis instance. Pinned dependencies also keep the benchmark behavior reproducible across separate runs.

In this step, you will confirm that Node.js meets the required version. You will create a pinned TypeScript project before starting Redis through Docker Compose.

In this step, get ready to:

• Verify that Node.js reports major version 20 or newer.
• Create the project configuration files before installing the pinned npm packages.
• Start a local Redis container with a persistent volume plus a health check.

Verify Node.js and create the project files

The Redis client in this project requires Node.js major version 20 or newer. Checking now prevents a dependency failure during installation.

• Press Cmd+Space on macOS or the Windows key on Windows to open your search bar.
• Type Visual Studio Code and press Enter to open it.
• Click View in the top menu bar.
• Select Terminal to display the integrated terminal.
• Check your installed Node.js version by running this command:

node --version

What does this check prove?

The command prints the Node.js version available to the VS Code terminal. The first number after v is the major version used for this requirement.

✔️ I see version 20 or newer

Your Node.js runtime meets the project requirement. You can continue with the pinned dependencies.

ⓧ I see an older version

Your installed Node.js runtime is below the required major version. Install a supported release before creating the project.

• Open the official Node.js download page.
• Choose a supported release with major version 20 or newer.
• Complete the macOS installer for that release.
• Return to the VS Code terminal after installation.
• Check the installed version again by running:

node --version

What should I see?

The first number after v should now be 20 or newer. That result confirms the updated runtime is available in your terminal.

ⓧ Command not found

The terminal cannot currently find Node.js. Install a supported release through the official download page.

• Open the official Node.js download page.
• Choose a supported release with major version 20 or newer.
• Complete the macOS installer for that release.
• Return to the VS Code terminal after installation.
• Confirm that Node.js is now available by running:

node --version

What should I see?

You should now see a version beginning with v20 or a newer major number. That output confirms the terminal can find Node.js.

A dedicated folder keeps the dependency manifest beside the compiler settings. It also gives Docker Compose one clear place to find its configuration.

• Click File in the VS Code menu bar.
• Select Open Folder....
• Choose your Desktop as the location.
• Select New Folder.
• Enter budget-aware-llm-dispatch as the folder name.
• Press Enter to create the folder.
• Select Open to open the folder as your VS Code workspace.
• Select Yes, I trust the authors if the Workspace Trust dialog appears.

The package.json manifest pins every dependency used by the benchmark. Its scripts also define the three benchmark modes you will build in later steps.

• Select the Explorer view in the left Activity Bar.
• Select New File... in the Explorer toolbar.
• Enter package.json as the file name.
• Add the pinned project manifest by pasting this code:

{
  "name": "budget-aware-llm-dispatch",
  "version": "1.0.0",
  "private": true,
  "type": "module",
  "engines": { "node": ">=20" },
  "scripts": {
    "build": "tsc -p tsconfig.json",
    "benchmark:naive": "npm run build && node dist/benchmark.js naive",
    "benchmark:atomic": "npm run build && node dist/benchmark.js atomic",
    "benchmark": "npm run build && node dist/benchmark.js compare"
  },
  "dependencies": { "bullmq": "6.3.17", "ioredis": "6.0.0" },
  "devDependencies": { "@types/node": "26.6.5", "typescript": "7.0.2" }
}

What does this manifest configure?

• The engines field records the Node.js requirement as >=20.
• The scripts field builds the TypeScript source before each benchmark mode runs.
• BullMQ is pinned to version 6.3.17.
• ioredis is pinned to version 6.0.0.
• The TypeScript compiler is pinned to version 7.0.2.
• The Node.js type definitions are pinned to version 26.6.5.

• Save package.json.
• Confirm that package.json is listed inside budget-aware-llm-dispatch in the Explorer sidebar.

Seeing a JSON warning?

• Check that every property except the final property in an object ends with a comma.
• Confirm that every key uses double quotes.
• Compare the opening braces with the closing braces.
• Ask for help with the manifest warning: Help me find the JSON syntax problem in my package.json file.

The compiler configuration turns on strict checking. It also sends compiled JavaScript from src into dist.

• Select New File... in the Explorer toolbar.
• Enter tsconfig.json as the file name.
• Add the TypeScript compiler configuration by pasting this code:

{
  "compilerOptions": {
    "target": "ES2022",
    "module": "NodeNext",
    "moduleResolution": "NodeNext",
    "rootDir": "src",
    "outDir": "dist",
    "strict": true,
    "esModuleInterop": true,
    "skipLibCheck": true,
    "types": ["node"]
  },
  "include": ["src/**/*.ts"]
}

What does this compiler configuration do?

• The NodeNext settings align TypeScript modules with the project manifest.
• The rootDir setting tells the compiler to read source files from src.
• The outDir setting sends compiled files to dist.
• The strict setting enables stronger type checks.

• Save tsconfig.json.
• Confirm that tsconfig.json appears beside package.json in the Explorer sidebar.

Seeing a compiler configuration warning?

• Confirm that compilerOptions is an object enclosed by braces.
• Check that include is outside the compilerOptions object.
• Ask for help with the configuration warning: Help me debug my tsconfig.json structure.

Generated dependencies plus compiled output do not belong in the source history. The benchmark result is also excluded because each run creates a fresh artifact.

• Select New File... in the Explorer toolbar.
• Enter .gitignore as the file name.
• Add the generated paths by pasting this code:

node_modules/
dist/
benchmark-results.json

What does this file exclude?

The entries exclude installed packages from source control. They also exclude compiled files plus the generated benchmark report.

• Save .gitignore.
• Confirm that .gitignore appears in the Explorer sidebar.

The Compose configuration gives the benchmark one local Redis service. Its health check blocks later work until Redis answers successfully.

• Select New File... in the Explorer toolbar.
• Enter compose.yaml as the file name.
• Add the local Redis service by pasting this code:

services:
  redis:
    image: redis:8.10.2-alpine
    ports:
      - "127.0.0.1:6379:6379"
    volumes:
      - redis-data:/data
    healthcheck:
      test: ["CMD", "redis-cli", "PING"]
      interval: 2s
      timeout: 1s
      retries: 10

volumes:
  redis-data:

What does this Compose file configure?

• The service uses the Redis image tag 8.10.2-alpine.
• The port mapping binds Redis to 127.0.0.1:6379 on your machine.
• The redis-data named volume persists data when the container is recreated.
• The health check runs redis-cli PING inside the container.
• A successful Redis health check receives PONG.

• Save compose.yaml.
• Confirm that compose.yaml appears beside the other project files in the Explorer sidebar.

Seeing a YAML warning?

• Use spaces for every indentation level in compose.yaml.
• Keep volumes at the far left when declaring the named volume.
• Ask for help with the YAML warning: Help me fix the indentation in my compose.yaml file.

✔️ Awesome, I've got everything!

Good progress. Confirm that all four project files are saved in budget-aware-llm-dispatch.

ⓧ I'd like to double check the full code

Compare each saved file with the complete configuration below.

{
  "name": "budget-aware-llm-dispatch",
  "version": "1.0.0",
  "private": true,
  "type": "module",
  "engines": { "node": ">=20" },
  "scripts": {
    "build": "tsc -p tsconfig.json",
    "benchmark:naive": "npm run build && node dist/benchmark.js naive",
    "benchmark:atomic": "npm run build && node dist/benchmark.js atomic",
    "benchmark": "npm run build && node dist/benchmark.js compare"
  },
  "dependencies": { "bullmq": "6.3.17", "ioredis": "6.0.0" },
  "devDependencies": { "@types/node": "26.6.5", "typescript": "7.0.2" }
}

Package manifest checkpoint

This file should contain the pinned dependencies plus all four project scripts.

{
  "compilerOptions": {
    "target": "ES2022",
    "module": "NodeNext",
    "moduleResolution": "NodeNext",
    "rootDir": "src",
    "outDir": "dist",
    "strict": true,
    "esModuleInterop": true,
    "skipLibCheck": true,
    "types": ["node"]
  },
  "include": ["src/**/*.ts"]
}

Compiler configuration checkpoint

This file should compile strict TypeScript from src into dist.

node_modules/
dist/
benchmark-results.json

Generated files checkpoint

This file should exclude the dependency folder plus both generated outputs.

services:
  redis:
    image: redis:8.10.2-alpine
    ports:
      - "127.0.0.1:6379:6379"
    volumes:
      - redis-data:/data
    healthcheck:
      test: ["CMD", "redis-cli", "PING"]
      interval: 2s
      timeout: 1s
      retries: 10

volumes:
  redis-data:

Redis configuration checkpoint

This file should define one Redis service plus the persistent redis-data volume.

Install the pinned npm packages

The manifest names exact package versions. Running npm now installs those packages before generating package-lock.json as the resolved dependency record.

• Return to the integrated terminal panel in VS Code.
• Install the project dependencies by running:

npm install

What does this command do?

The command reads package.json before downloading the declared dependencies. It creates node_modules plus package-lock.json.

• Wait for npm to return control to the terminal prompt.
• Confirm that the terminal finishes without a dependency error.
• Confirm that package-lock.json appears in the Explorer sidebar.
• Confirm that node_modules appears in the Explorer sidebar.

That installation is complete. Your lockfile now records the dependency tree generated from the pinned manifest.

Did npm installation fail?

• Confirm that your Node.js major version is 20 or newer.
• Check package.json for a missing comma or closing brace.
• Run the installation again after correcting the reported issue.
• Ask for help with the installation output: Help me diagnose my npm install failure.

Start Redis and confirm its health

The benchmark needs one shared Redis process for every producer plus worker. Docker Compose uses the saved health check to wait for a real Redis response before reporting success.

Docker Desktop may need about a minute to make its engine ready. That startup pause is expected.

• Press Cmd+Space on macOS or the Windows key on Windows to open your search bar.
• Type Docker Desktop and press Enter to open it.
• Wait until Docker Desktop shows that its engine is ready.
• Return to the VS Code terminal panel.

Before you run the final check, do you expect Compose to return immediately or wait for Redis to answer its health check?

• Start Redis in the background before waiting for its healthy state by running:

docker compose up -d --wait

What does this command do?

• Docker Compose reads compose.yaml before creating the Redis container.
• The -d option leaves the container running in the background.
• The --wait option keeps the command active until the service is running or healthy.
• The health check sends PING through redis-cli.
• Redis answers that health check with PONG.

You should see the Redis service reach a healthy state before the terminal prompt returns. This proves that the container can answer redis-cli PING with PONG.

Your local benchmark backbone is running. Redis is available only through 127.0.0.1:6379 while the redis-data volume preserves its data.

Redis did not become healthy?

• Confirm that Docker Desktop reports its engine as ready.
• Check that compose.yaml uses spaces for indentation.
• Confirm that no other local process is already using port 6379.
• Ask for help with the Compose output: Help me diagnose why my Redis container is not becoming healthy.

Your pinned project now has a healthy shared Redis instance behind it. Next up, you will send one simulated LLM job across the queue boundary and inspect its timing metadata.

Dispatch a Visible Simulated LLM Job

Your local Redis service is healthy. That proves the shared data store can answer requests.

Now you will send one typed simulated LLM job through BullMQ. You will see the job become active before a worker returns timing metadata.

In this step, get ready to:

• Define the typed job payload plus the result returned by the worker.
• Build a worker that records queue wait time separately from simulated processing time.
• Dispatch one job through Redis to confirm the complete queue path works.

Define the shared job contract

A distributed queue needs one shared definition of the data entering each job. It also needs one shared definition of the result leaving the worker.

• Use the VS Code Explorer sidebar to create a folder named src inside budget-aware-llm-dispatch.
• Create shared.ts inside the new src folder.
• Add the shared connection settings plus the typed dispatch contract by replacing the contents of src/shared.ts with this code:

import { Redis } from "ioredis";

export const CONNECTION = { host: "127.0.0.1", port: 6379 } as const;
export const GLOBAL_CONCURRENCY = 4;
export const LOCAL_CONCURRENCY = 3;
export const RATE_LIMIT_MAX = 4;
export const RATE_LIMIT_DURATION_MS = 1_000;
export type Mode = "naive" | "atomic";
export type Phase = "load" | "race";

export interface JobData {
  phase: Phase;
  index: number;
  tokenCost: number;
  simulatedMs: number;
  enqueuedAt: number;
  budgetKey: string;
  barrierKey?: string;
  barrierParticipants?: number;
}

export interface JobResult {
  jobId: string;
  phase: Phase;
  workerId: string;
  priority: number;
  enqueuedAt: number;
  startedAt: number;
  completedAt: number;
  queueWaitMs: number;
  serviceMs: number;
  endToEndMs: number;
  tokenCost: number;
  reserved: boolean;
  remainingBudget: number;
  status: "completed" | "budget_rejected";
}

const ATOMIC_RESERVATION_SCRIPT = `
local remaining = tonumber(redis.call('GET', KEYS[1]) or '0')
local cost = tonumber(ARGV[1])
if remaining < cost then return {0, remaining} end
local next_remaining = remaining - cost
redis.call('SET', KEYS[1], next_remaining)
return {1, next_remaining}
`;

export function sleep(milliseconds: number): Promise<void> {
  return new Promise((resolve) => setTimeout(resolve, milliseconds));
}

async function waitAtBarrier(redis: Redis, key: string, participants: number): Promise<void> {
  await redis.incr(key);
  const deadline = Date.now() + 5_000;
  while (Number((await redis.get(key)) ?? "0") < participants) {
    if (Date.now() > deadline) throw new Error(`Barrier timed out: ${key}`);
    await sleep(5);
  }
}

export async function reserveBudget(redis: Redis, mode: Mode, data: JobData) {
  if (mode === "naive") {
    const observed = Number((await redis.get(data.budgetKey)) ?? "0");
    if (observed < data.tokenCost) return { reserved: false, remaining: observed };
    if (data.barrierKey && data.barrierParticipants) {
      await waitAtBarrier(redis, data.barrierKey, data.barrierParticipants);
    }
    const remaining = observed - data.tokenCost;
    await redis.set(data.budgetKey, remaining.toString());
    return { reserved: true, remaining };
  }
  if (data.barrierKey && data.barrierParticipants) {
    await waitAtBarrier(redis, data.barrierKey, data.barrierParticipants);
  }
  const reply = (await redis.eval(
    ATOMIC_RESERVATION_SCRIPT,
    1,
    data.budgetKey,
    data.tokenCost.toString(),
  )) as [number, number];
  return { reserved: Number(reply[0]) === 1, remaining: Number(reply[1]) };
}

What does this shared module do?

• The CONNECTION object gives every process the same local Redis address.
• The JobData interface defines the payload stored in each queue job.
• The JobResult interface keeps waiting time separate from processing time.
• The sleep() helper creates deterministic simulated processing time without calling a live provider.
• The reserveBudget() function gives the worker one typed budget reservation interface.

• Save src/shared.ts.
• Compile the shared module by running this command in the VS Code terminal:

npm run build

What does this command prove?

The build script runs TypeScript using the strict configuration from the previous step. A successful build confirms that the shared types agree with the pinned packages.

The command should finish without a TypeScript error. You will also see a compiled shared.js file inside dist.

Seeing a TypeScript error?

• Confirm that shared.ts is inside the src folder.
• Check that the import uses Redis with the same capitalization shown in the code.
• Compare the closing braces around JobData plus JobResult.
• Help me debug the strict TypeScript errors in src/shared.ts.

Build the observable worker

A queue worker owns the processing side of the dispatch. This worker records timestamps around its reservation plus its simulated LLM call.

Why separate the three timings?

Queue wait measures the delay before processing starts. Service time measures the worker's own processing window.

End-to-end latency covers the complete journey from enqueue to completion. Separate measurements show where a slow job actually spent its time.

• Create worker.ts inside the src folder.
• Add the observable worker by replacing the contents of src/worker.ts with this code:

import { Worker } from "bullmq";
import { Redis } from "ioredis";
import {
  CONNECTION,
  LOCAL_CONCURRENCY,
  RATE_LIMIT_DURATION_MS,
  RATE_LIMIT_MAX,
  type JobData,
  type JobResult,
  type Mode,
  reserveBudget,
  sleep,
} from "./shared.js";

const [queueName, modeValue, workerId] = process.argv.slice(2);
if (!queueName || !workerId || (modeValue !== "naive" && modeValue !== "atomic")) {
  throw new Error("Usage: worker.js <queue-name> <naive|atomic> <worker-id>");
}
const mode: Mode = modeValue;
const redis = new Redis(CONNECTION);

const worker = new Worker<JobData, JobResult, "llm-call">(
  queueName,
  async (job) => {
    const startedAt = Date.now();
    console.log(JSON.stringify({
      event: "active", jobId: job.id, phase: job.data.phase, workerId,
      priority: job.priority, attemptsStarted: job.attemptsStarted,
      enqueuedAt: job.data.enqueuedAt, startedAt,
    }));

    const reservation = await reserveBudget(redis, mode, job.data);
    if (!reservation.reserved) {
      const completedAt = Date.now();
      const result: JobResult = {
        jobId: job.id ?? "unknown", phase: job.data.phase, workerId,
        priority: job.priority, enqueuedAt: job.data.enqueuedAt,
        startedAt, completedAt,
        queueWaitMs: startedAt - job.data.enqueuedAt,
        serviceMs: completedAt - startedAt,
        endToEndMs: completedAt - job.data.enqueuedAt,
        tokenCost: job.data.tokenCost, reserved: false,
        remainingBudget: reservation.remaining, status: "budget_rejected",
      };
      console.log(JSON.stringify({ event: "completed", ...result }));
      return result;
    }

    await sleep(job.data.simulatedMs);
    const completedAt = Date.now();
    const result: JobResult = {
      jobId: job.id ?? "unknown", phase: job.data.phase, workerId,
      priority: job.priority, enqueuedAt: job.data.enqueuedAt,
      startedAt, completedAt,
      queueWaitMs: startedAt - job.data.enqueuedAt,
      serviceMs: completedAt - startedAt,
      endToEndMs: completedAt - job.data.enqueuedAt,
      tokenCost: job.data.tokenCost, reserved: true,
      remainingBudget: reservation.remaining, status: "completed",
    };
    console.log(JSON.stringify({ event: "completed", ...result }));
    return result;
  },
  {
    connection: CONNECTION,
    name: workerId,
    concurrency: LOCAL_CONCURRENCY,
    limiter: { max: RATE_LIMIT_MAX, duration: RATE_LIMIT_DURATION_MS },
  },
);

worker.on("failed", (job, error) => console.error(JSON.stringify({
  event: "worker_failed", jobId: job?.id, workerId, error: error.message,
})));
worker.on("error", (error) => console.error(JSON.stringify({
  event: "worker_error", workerId, error: error.message,
})));
await worker.waitUntilReady();
process.send?.({ type: "ready", workerId });

async function shutdown(): Promise<void> {
  await worker.close();
  await redis.quit();
  process.disconnect?.();
}
process.on("message", (message: unknown) => {
  if (typeof message === "object" && message !== null && "type" in message && message.type === "shutdown") {
    void shutdown();
  }
});
process.on("SIGTERM", () => void shutdown());

How does the worker make a job observable?

• The Worker accepts only jobs named llm-call with the shared payload type.
• The first JSON log uses the active event to capture when processing begins.
• The worker reserves the job's token cost before simulating provider latency.
• The final JSON log uses the completed event to expose the three timing measurements.
• The worker sends a readiness message after its Redis connection can process jobs.

• Save src/worker.ts.
• Compile the worker by running this command in the VS Code terminal:

npm run build

What does this build check?

TypeScript checks the worker against the shared job contract. A successful build confirms that every returned result contains the required timing plus budget fields.

The command should finish without an error. You will see worker.js inside dist.

Worker failing to compile?

• Confirm that the shared import ends with ./shared.js.
• Check that both budget_rejected plus completed match the status values in JobResult.
• Verify that serviceMs uses the interval from startedAt to completedAt.
• Help me fix the TypeScript errors in src/worker.ts.

Dispatch and verify one job

The producer completes the queue path. It creates one job before waiting for the global completion event that carries the worker's result.

• Create benchmark.ts inside the src folder.
• Add the initial one-job benchmark by replacing the contents of src/benchmark.ts with this code:

import { fork } from "node:child_process";
import { once } from "node:events";
import { Queue, QueueEvents } from "bullmq";
import { Redis } from "ioredis";
import {
  CONNECTION,
  type JobData,
  type JobResult,
  type Mode,
} from "./shared.js";

const mode: Mode = "atomic";
const queueName = `llm-dispatch-${mode}-${Date.now()}`;
const budgetKey = `budget:${queueName}:load`;
const redis = new Redis(CONNECTION);
const queue = new Queue<JobData, JobResult, "llm-call">(queueName, { connection: CONNECTION });
const events = new QueueEvents<JobResult>(queueName, { connection: CONNECTION });
const worker = fork(new URL("./worker.js", import.meta.url), [queueName, mode, "worker-a"], {
  stdio: ["inherit", "inherit", "inherit", "ipc"],
});

await new Promise<void>((resolve, reject) => {
  const onMessage = (message: unknown): void => {
    if (typeof message === "object" && message !== null && "type" in message && message.type === "ready") {
      cleanup();
      resolve();
    }
  };
  const onExit = (code: number | null): void => {
    cleanup();
    reject(new Error(`worker-a exited before ready: ${code}`));
  };
  const cleanup = (): void => {
    worker.off("message", onMessage);
    worker.off("exit", onExit);
  };
  worker.on("message", onMessage);
  worker.on("exit", onExit);
});

try {
  await Promise.all([queue.waitUntilReady(), events.waitUntilReady()]);
  await redis.set(budgetKey, "100");
  const job = await queue.add("llm-call", {
    phase: "load",
    index: 0,
    tokenCost: 100,
    simulatedMs: 100,
    enqueuedAt: Date.now(),
    budgetKey,
  }, { priority: 1, attempts: 1 });
  const result = await job.waitUntilFinished(events, 45_000);
  console.log(JSON.stringify({ event: "benchmark_result", ...result }));
} finally {
  if (worker.connected) worker.send({ type: "shutdown" });
  await once(worker, "exit");
  await events.close();
  await queue.obliterate({ force: true });
  await queue.close();
  await redis.del(budgetKey);
  await redis.quit();
}

How does the dispatch work?

• The benchmark starts worker-a as an independent Node.js process.
• The producer waits for the worker's readiness message before creating the job.
• The job carries a priority plus a simulated duration of 100 milliseconds.
• The call to waitUntilFinished() resolves with the typed JobResult returned by the worker.
• The cleanup block closes the worker plus every Redis-backed connection created by this run.

✔️ Awesome, I've got everything!

Your three source files are ready. Double check that you saved src/shared.ts, src/worker.ts, plus src/benchmark.ts.

ⓧ I'd like to double check the full code

Compare each source file with the complete versions below. Keep the files in this order when tracing the dispatch path.

import { Redis } from "ioredis";

export const CONNECTION = { host: "127.0.0.1", port: 6379 } as const;
export const GLOBAL_CONCURRENCY = 4;
export const LOCAL_CONCURRENCY = 3;
export const RATE_LIMIT_MAX = 4;
export const RATE_LIMIT_DURATION_MS = 1_000;
export type Mode = "naive" | "atomic";
export type Phase = "load" | "race";

export interface JobData {
  phase: Phase;
  index: number;
  tokenCost: number;
  simulatedMs: number;
  enqueuedAt: number;
  budgetKey: string;
  barrierKey?: string;
  barrierParticipants?: number;
}

export interface JobResult {
  jobId: string;
  phase: Phase;
  workerId: string;
  priority: number;
  enqueuedAt: number;
  startedAt: number;
  completedAt: number;
  queueWaitMs: number;
  serviceMs: number;
  endToEndMs: number;
  tokenCost: number;
  reserved: boolean;
  remainingBudget: number;
  status: "completed" | "budget_rejected";
}

const ATOMIC_RESERVATION_SCRIPT = `
local remaining = tonumber(redis.call('GET', KEYS[1]) or '0')
local cost = tonumber(ARGV[1])
if remaining < cost then return {0, remaining} end
local next_remaining = remaining - cost
redis.call('SET', KEYS[1], next_remaining)
return {1, next_remaining}
`;

export function sleep(milliseconds: number): Promise<void> {
  return new Promise((resolve) => setTimeout(resolve, milliseconds));
}

async function waitAtBarrier(redis: Redis, key: string, participants: number): Promise<void> {
  await redis.incr(key);
  const deadline = Date.now() + 5_000;
  while (Number((await redis.get(key)) ?? "0") < participants) {
    if (Date.now() > deadline) throw new Error(`Barrier timed out: ${key}`);
    await sleep(5);
  }
}

export async function reserveBudget(redis: Redis, mode: Mode, data: JobData) {
  if (mode === "naive") {
    const observed = Number((await redis.get(data.budgetKey)) ?? "0");
    if (observed < data.tokenCost) return { reserved: false, remaining: observed };
    if (data.barrierKey && data.barrierParticipants) {
      await waitAtBarrier(redis, data.barrierKey, data.barrierParticipants);
    }
    const remaining = observed - data.tokenCost;
    await redis.set(data.budgetKey, remaining.toString());
    return { reserved: true, remaining };
  }
  if (data.barrierKey && data.barrierParticipants) {
    await waitAtBarrier(redis, data.barrierKey, data.barrierParticipants);
  }
  const reply = (await redis.eval(
    ATOMIC_RESERVATION_SCRIPT,
    1,
    data.budgetKey,
    data.tokenCost.toString(),
  )) as [number, number];
  return { reserved: Number(reply[0]) === 1, remaining: Number(reply[1]) };
}

Shared file checkpoint

This file should define the connection, queue constants, shared types, deterministic delay, barrier helper, plus the reservation function.

import { Worker } from "bullmq";
import { Redis } from "ioredis";
import {
  CONNECTION,
  LOCAL_CONCURRENCY,
  RATE_LIMIT_DURATION_MS,
  RATE_LIMIT_MAX,
  type JobData,
  type JobResult,
  type Mode,
  reserveBudget,
  sleep,
} from "./shared.js";

const [queueName, modeValue, workerId] = process.argv.slice(2);
if (!queueName || !workerId || (modeValue !== "naive" && modeValue !== "atomic")) {
  throw new Error("Usage: worker.js <queue-name> <naive|atomic> <worker-id>");
}
const mode: Mode = modeValue;
const redis = new Redis(CONNECTION);

const worker = new Worker<JobData, JobResult, "llm-call">(
  queueName,
  async (job) => {
    const startedAt = Date.now();
    console.log(JSON.stringify({
      event: "active", jobId: job.id, phase: job.data.phase, workerId,
      priority: job.priority, attemptsStarted: job.attemptsStarted,
      enqueuedAt: job.data.enqueuedAt, startedAt,
    }));

    const reservation = await reserveBudget(redis, mode, job.data);
    if (!reservation.reserved) {
      const completedAt = Date.now();
      const result: JobResult = {
        jobId: job.id ?? "unknown", phase: job.data.phase, workerId,
        priority: job.priority, enqueuedAt: job.data.enqueuedAt,
        startedAt, completedAt,
        queueWaitMs: startedAt - job.data.enqueuedAt,
        serviceMs: completedAt - startedAt,
        endToEndMs: completedAt - job.data.enqueuedAt,
        tokenCost: job.data.tokenCost, reserved: false,
        remainingBudget: reservation.remaining, status: "budget_rejected",
      };
      console.log(JSON.stringify({ event: "completed", ...result }));
      return result;
    }

    await sleep(job.data.simulatedMs);
    const completedAt = Date.now();
    const result: JobResult = {
      jobId: job.id ?? "unknown", phase: job.data.phase, workerId,
      priority: job.priority, enqueuedAt: job.data.enqueuedAt,
      startedAt, completedAt,
      queueWaitMs: startedAt - job.data.enqueuedAt,
      serviceMs: completedAt - startedAt,
      endToEndMs: completedAt - job.data.enqueuedAt,
      tokenCost: job.data.tokenCost, reserved: true,
      remainingBudget: reservation.remaining, status: "completed",
    };
    console.log(JSON.stringify({ event: "completed", ...result }));
    return result;
  },
  {
    connection: CONNECTION,
    name: workerId,
    concurrency: LOCAL_CONCURRENCY,
    limiter: { max: RATE_LIMIT_MAX, duration: RATE_LIMIT_DURATION_MS },
  },
);

worker.on("failed", (job, error) => console.error(JSON.stringify({
  event: "worker_failed", jobId: job?.id, workerId, error: error.message,
})));
worker.on("error", (error) => console.error(JSON.stringify({
  event: "worker_error", workerId, error: error.message,
})));
await worker.waitUntilReady();
process.send?.({ type: "ready", workerId });

async function shutdown(): Promise<void> {
  await worker.close();
  await redis.quit();
  process.disconnect?.();
}
process.on("message", (message: unknown) => {
  if (typeof message === "object" && message !== null && "type" in message && message.type === "shutdown") {
    void shutdown();
  }
});
process.on("SIGTERM", () => void shutdown());

Worker file checkpoint

This file should validate its process arguments, process typed jobs, log lifecycle events, calculate three timings, plus close cleanly.

import { fork } from "node:child_process";
import { once } from "node:events";
import { Queue, QueueEvents } from "bullmq";
import { Redis } from "ioredis";
import {
  CONNECTION,
  type JobData,
  type JobResult,
  type Mode,
} from "./shared.js";

const mode: Mode = "atomic";
const queueName = `llm-dispatch-${mode}-${Date.now()}`;
const budgetKey = `budget:${queueName}:load`;
const redis = new Redis(CONNECTION);
const queue = new Queue<JobData, JobResult, "llm-call">(queueName, { connection: CONNECTION });
const events = new QueueEvents<JobResult>(queueName, { connection: CONNECTION });
const worker = fork(new URL("./worker.js", import.meta.url), [queueName, mode, "worker-a"], {
  stdio: ["inherit", "inherit", "inherit", "ipc"],
});

await new Promise<void>((resolve, reject) => {
  const onMessage = (message: unknown): void => {
    if (typeof message === "object" && message !== null && "type" in message && message.type === "ready") {
      cleanup();
      resolve();
    }
  };
  const onExit = (code: number | null): void => {
    cleanup();
    reject(new Error(`worker-a exited before ready: ${code}`));
  };
  const cleanup = (): void => {
    worker.off("message", onMessage);
    worker.off("exit", onExit);
  };
  worker.on("message", onMessage);
  worker.on("exit", onExit);
});

try {
  await Promise.all([queue.waitUntilReady(), events.waitUntilReady()]);
  await redis.set(budgetKey, "100");
  const job = await queue.add("llm-call", {
    phase: "load",
    index: 0,
    tokenCost: 100,
    simulatedMs: 100,
    enqueuedAt: Date.now(),
    budgetKey,
  }, { priority: 1, attempts: 1 });
  const result = await job.waitUntilFinished(events, 45_000);
  console.log(JSON.stringify({ event: "benchmark_result", ...result }));
} finally {
  if (worker.connected) worker.send({ type: "shutdown" });
  await once(worker, "exit");
  await events.close();
  await queue.obliterate({ force: true });
  await queue.close();
  await redis.del(budgetKey);
  await redis.quit();
}

Benchmark file checkpoint

This file should launch one worker, enqueue one typed job, wait for its result, print that result, plus close every connection.

• Save src/benchmark.ts.

Before you run the benchmark, do you expect the job to report more time waiting in the queue or more time inside the simulated worker?

• Compile the project plus run the initial atomic dispatch by running this command in the VS Code terminal:

npm run benchmark:atomic

What does this benchmark command do?

The package script compiles the TypeScript source before starting the atomic benchmark. The benchmark launches the worker process before placing one job on the queue.

You should first see an active JSON event. You should then see a completed event plus a benchmark_result event.

The result should include jobId, workerId, priority, queueWaitMs, serviceMs, endToEndMs, plus remainingBudget.

That is the first complete dispatch working. Your typed job now crosses the Redis queue boundary before returning observable worker metadata.

Benchmark not completing?

• Confirm that the Redis service from the previous step is still healthy.
• Check that all three source files are saved inside src.
• Verify that the worker import ends with ./shared.js.
• Help me debug a one-job BullMQ benchmark that does not complete.

Your first simulated LLM job now moves through a typed queue plus an observable worker. Next, you will run two independent workers under shared scheduling controls.

Benchmark Two Workers Under Shared Controls

Your first BullMQ job crossed the queue boundary and returned timing data from one worker. That visible dispatch proves the basic path works.

One worker hides the coordination problem found in a distributed system. This step launches two independent workers against the same Redis queue. The combined results prove whether shared scheduling controls govern both processes.

In this step, get ready to:

• Launch two independent worker processes against one queue.
• Enqueue high-priority work alongside standard-priority work.
• Measure shared concurrency, rate spacing, throughput, priority order, and latency percentiles.

Add shared measurement helpers

The benchmark needs one set of constants for both workers. It also needs typed references to each child process so it can wait for readiness and shut every process down cleanly.

• Switch back to src/benchmark.ts in the VS Code editor.
• Replace the current import section with these imports and workload constants:

import { fork, type ChildProcess } from "node:child_process";
import { once } from "node:events";
import { Queue, QueueEvents, type Job } from "bullmq";
import { Redis } from "ioredis";
import {
  CONNECTION,
  GLOBAL_CONCURRENCY,
  RATE_LIMIT_DURATION_MS,
  RATE_LIMIT_MAX,
  type JobData,
  type JobResult,
  type Mode,
} from "./shared.js";

const LOAD_JOBS = 24;
const HIGH_PRIORITY_JOBS = 8;
const TOKEN_COST = 100;
type LaunchedWorker = { child: ChildProcess; ready: Promise<void> };

What does this code prepare?

• The fork import lets the benchmark start each worker as an independent Node.js process.
• The once import creates a promise that resolves when a child process emits its exit event.
• The workload constants define 24 jobs. Eight jobs receive the higher priority.
• The LaunchedWorker type keeps the child process beside its readiness promise.

• Save src/benchmark.ts.
• Confirm the existing single-job path still compiles by running:

npm run benchmark:naive

You should still see one active event followed by one completed event. This confirms the expanded imports preserve the working dispatch path.

Seeing an import error?

• Check that each Node.js import starts with the exact node: prefix shown above.
• Confirm the shared import ends with ./shared.js because the project uses NodeNext module resolution.
• Use this prompt for targeted help: Help me diagnose the TypeScript import error in src/benchmark.ts after adding child process support.

Latency needs separate distributions for time spent waiting and time spent processing. A nearest-rank summary sorts each measurement before selecting its p50, p95, and p99 values.

• Add these latency and overlap helpers below the LaunchedWorker type:

function round(value: number): number { return Number(value.toFixed(2)); }
function summarize(values: number[]) {
  const sorted = [...values].sort((a, b) => a - b);
  const at = (p: number): number => sorted[Math.max(0, Math.ceil((p / 100) * sorted.length) - 1)] ?? 0;
  return { p50: at(50), p95: at(95), p99: at(99) };
}
function maximumConcurrency(results: JobResult[]): number {
  const points = results.flatMap((r) => [
    { time: r.startedAt, delta: 1 }, { time: r.completedAt, delta: -1 },
  ]).sort((a, b) => a.time - b.time || a.delta - b.delta);
  let active = 0;
  let maximum = 0;
  for (const point of points) { active += point.delta; maximum = Math.max(maximum, active); }
  return maximum;
}

How are the measurements calculated?

• The summarize() helper sorts a copy of one latency series. It selects the nearest-ranked observation for each percentile.
• The maximumConcurrency() helper converts every job interval into start and finish points.
• Start points increase the active count. Finish points decrease it.
• The highest active count proves how many jobs overlapped across both workers.

• Save src/benchmark.ts.
• Confirm the measurement helpers compile by running:

npm run benchmark:naive

You should see the same completed single-job result. The unchanged result confirms the new helpers compile without disturbing the existing processor.

Seeing a duplicate declaration?

• Search src/benchmark.ts for a second copy of round or summarize.
• Keep one copy of each helper directly below the workload constants.
• Use this prompt for targeted help: Help me find the duplicate helper declaration in src/benchmark.ts.

The rate check compares each start with the start four positions earlier. A gap of at least 900 milliseconds shows that the globally coordinated limit of four starts per 1,000 milliseconds is holding within a small scheduling tolerance.

• Add the rate-spacing and priority checks below maximumConcurrency():

function minimumRateSpacing(results: JobResult[]): number {
  const starts = results.map((r) => r.startedAt).sort((a, b) => a - b);
  const spacings: number[] = [];
  for (let index = RATE_LIMIT_MAX; index < starts.length; index += 1) {
    spacings.push(starts[index] - starts[index - RATE_LIMIT_MAX]);
  }
  return Math.min(...spacings);
}
function priorityWasEnforced(results: JobResult[]): boolean {
  const high = results.filter((r) => r.priority === 1).map((r) => r.startedAt);
  const low = results.filter((r) => r.priority === 10).map((r) => r.startedAt);
  return Math.max(...high) <= Math.min(...low);
}

What do these checks prove?

• The minimumRateSpacing() helper orders all starts from both worker processes.
• Each fifth start is compared with the first start in its group of five.
• The priorityWasEnforced() helper confirms every priority-1 job starts before any priority-10 job.

• Save src/benchmark.ts.
• Confirm the scheduling checks compile by running:

npm run benchmark:naive

You should again see the single simulated job complete. Your benchmark now has the calculations required for the distributed workload.

Seeing a type error in a helper?

• Confirm both helper parameters use JobResult[].
• Check that the priority comparison uses numeric values 1 and 10.
• Use this prompt for targeted help: Help me diagnose the TypeScript type error in my rate-spacing or priority helper.

Launch two worker processes

Each worker must start in its own process for the benchmark to test shared coordination. An IPC readiness message prevents the producer from treating process creation as proof that a worker can already accept jobs.

• Add the worker launcher below priorityWasEnforced():

function launchWorker(queueName: string, mode: Mode, workerId: string): LaunchedWorker {
  const child = fork(new URL("./worker.js", import.meta.url), [queueName, mode, workerId], {
    stdio: ["inherit", "inherit", "inherit", "ipc"],
  });
  const ready = new Promise<void>((resolve, reject) => {
    const onMessage = (message: unknown): void => {
      if (typeof message === "object" && message !== null && "type" in message && message.type === "ready") {
        cleanup(); resolve();
      }
    };
    const onExit = (code: number | null): void => { cleanup(); reject(new Error(`${workerId} exited before ready: ${code}`)); };
    const cleanup = (): void => { child.off("message", onMessage); child.off("exit", onExit); };
    child.on("message", onMessage); child.on("exit", onExit);
  });
  return { child, ready };
}

How does worker readiness work?

• The fork() call starts the compiled worker.js module with the queue name, mode, and worker ID.
• The ipc channel carries the worker's existing readiness message back to the benchmark.
• The readiness promise rejects if a worker exits before sending that message.
• The cleanup callback removes temporary event listeners after either outcome.

• Save src/benchmark.ts.
• Confirm the launcher compiles by running:

npm run benchmark:naive

You should see the original one-job dispatch complete. The new launcher is ready for the expanded orchestration code.

Seeing an IPC or fork type error?

• Confirm the stdio array contains one ipc entry.
• Check that ChildProcess is imported as a type from node:child_process.
• Use this prompt for targeted help: Help me fix the child-process IPC type error in launchWorker().

Child processes stay alive while their workers listen for more jobs. The parent sends a shutdown message after measurement so both workers can close their BullMQ and Redis connections.

• Add the shutdown and completion-wait helpers below launchWorker():

async function stopWorkers(workers: LaunchedWorker[]): Promise<void> {
  const live = workers.filter(({ child }) => child.exitCode === null);
  const exits = live.map(({ child }) => once(child, "exit"));
  for (const { child } of live) if (child.connected) child.send({ type: "shutdown" });
  await Promise.all(exits);
}
async function waitForJobs(jobs: Job<JobData, JobResult, string>[], events: QueueEvents<JobResult>) {
  return Promise.all(jobs.map((job) => job.waitUntilFinished(events, 45_000)));
}

How does the parent control the lifecycle?

• The stopWorkers() helper ignores processes that already exited.
• It registers every exit promise before sending shutdown messages.
• The waitForJobs() helper gathers each typed processor result through the shared QueueEvents connection.
• The 45-second timeout prevents the benchmark from waiting forever for a lost job.

• Save src/benchmark.ts.
• Confirm the lifecycle helpers compile by running:

npm run benchmark:naive

You should see the single-job result complete normally. This proves the lifecycle helpers type-check alongside the existing queue code.

Seeing a QueueEvents type error?

• Confirm waitForJobs() receives QueueEvents<JobResult>.
• Check that each job uses Job<JobData, JobResult, string>.
• Use this prompt for targeted help: Help me align the BullMQ generic types in waitForJobs().

All jobs enter the queue before either worker starts. This lets BullMQ choose the eight priority-1 jobs before the sixteen priority-10 jobs from the same waiting set.

• Add the load-job producer below waitForJobs():

async function addLoadJobs(queue: Queue<JobData, JobResult, string>, budgetKey: string) {
  const jobs: Job<JobData, JobResult, string>[] = [];
  for (let index = HIGH_PRIORITY_JOBS; index < LOAD_JOBS; index += 1) {
    jobs.push(await queue.add("llm-call", {
      phase: "load", index, tokenCost: TOKEN_COST,
      simulatedMs: 40 + (index % 3) * 20, enqueuedAt: Date.now(), budgetKey,
    }, { priority: 10, attempts: 1 }));
  }
  for (let index = 0; index < HIGH_PRIORITY_JOBS; index += 1) {
    jobs.push(await queue.add("llm-call", {
      phase: "load", index, tokenCost: TOKEN_COST,
      simulatedMs: 40 + (index % 3) * 20, enqueuedAt: Date.now(), budgetKey,
    }, { priority: 1, attempts: 1 }));
  }
  return jobs;
}

How is the workload shaped?

• The first loop enqueues sixteen priority-10 jobs.
• The second loop enqueues eight priority-1 jobs.
• Each simulated duration follows a repeatable 40, 60, or 80 millisecond pattern.
• Each job allows one processing attempt so retries cannot distort the timing comparison.

• Save src/benchmark.ts.
• Confirm the workload producer compiles by running:

npm run benchmark:naive

You should see the original one-job result complete one final time. The next replacement connects the new workload to both workers.

Seeing a job-data type error?

• Confirm both job payloads include phase, index, tokenCost, simulatedMs, enqueuedAt, and budgetKey.
• Check that both queue additions use the job name llm-call.
• Use this prompt for targeted help: Help me fix the typed BullMQ payload in addLoadJobs().

Run the distributed benchmark

The final orchestration creates one disposable queue for each run. It applies the global concurrency ceiling before launching workers. It then calculates all metrics from the combined results.

• Select the remaining top-level code that runs the old single-job benchmark.
• Replace that selected code with the following five parts in the order shown:

async function runScenario(mode: Mode) {
  const runId = `${mode}-${Date.now()}`;
  const queueName = `llm-dispatch-${runId}`;
  const loadBudgetKey = `budget:${runId}:load`;
  const loadInitialBudget = LOAD_JOBS * TOKEN_COST;
  const redis = new Redis(CONNECTION);
  const queue = new Queue<JobData, JobResult, string>(queueName, { connection: CONNECTION });
  const events = new QueueEvents<JobResult>(queueName, { connection: CONNECTION });
  const workers: LaunchedWorker[] = [];

  events.on("completed", ({ jobId }) => console.log(JSON.stringify({ event: "queue_completed", queueName, jobId })));
  events.on("failed", ({ jobId, failedReason }) => console.error(JSON.stringify({ event: "queue_failed", queueName, jobId, failedReason })));
  events.on("error", (error) => console.error(JSON.stringify({ event: "queue_events_error", error: error.message })));

  try {
    await Promise.all([queue.waitUntilReady(), events.waitUntilReady()]);
    await queue.setGlobalConcurrency(GLOBAL_CONCURRENCY);
    await redis.set(loadBudgetKey, loadInitialBudget.toString());
    const loadJobs = await addLoadJobs(queue, loadBudgetKey);
    workers.push(launchWorker(queueName, mode, "worker-a"), launchWorker(queueName, mode, "worker-b"));
    await Promise.all(workers.map((worker) => worker.ready));
    const loadResults = await waitForJobs(loadJobs, events);

What does the scenario setup do?

• The timestamped queue name isolates each benchmark run.
• QueueEvents records queue-wide completion and failure lifecycle events.
• The queue receives its global concurrency ceiling before either worker begins processing.
• Both readiness promises resolve before the benchmark waits for completed results.

• Continue the same replacement with the metric calculations and checks:

    const loadRemaining = Number((await redis.get(loadBudgetKey)) ?? "0");
    const loadAdmitted = loadResults.filter((r) => r.reserved).reduce((sum, r) => sum + r.tokenCost, 0);
    const firstStart = Math.min(...loadResults.map((r) => r.startedAt));
    const lastEnd = Math.max(...loadResults.map((r) => r.completedAt));
    const maxConcurrent = maximumConcurrency(loadResults);
    const rateSpacing = minimumRateSpacing(loadResults);
    const workerIds = [...new Set(loadResults.map((r) => r.workerId))].sort();
    const sharedConcurrencyCeilingHeld = workerIds.length === 2 && maxConcurrent <= GLOBAL_CONCURRENCY;
    const checks = {
      usedTwoWorkers: workerIds.length === 2,
      priorityOrderEnforced: priorityWasEnforced(loadResults),
      sharedConcurrencyCeilingHeld,
      globalRateLimitObserved: rateSpacing >= 900,
      allPassed: false,
    };
    checks.allPassed = Object.entries(checks)
      .filter(([name]) => name !== "allPassed")
      .every(([, passed]) => passed);

How are the checks combined?

• The benchmark calculates throughput from the first start through the final completion.
• A set of worker IDs proves whether results came from both child processes.
• The shared concurrency check requires two worker IDs and no more than four overlapping intervals.
• The final flag becomes true only when every individual check passes.

• Continue the replacement with the workload and metrics report:

    return {
      mode,
      workload: {
        loadJobs: LOAD_JOBS, workerProcesses: 2,
        globalConcurrency: GLOBAL_CONCURRENCY,
        rateLimitMax: RATE_LIMIT_MAX, rateLimitDurationMs: RATE_LIMIT_DURATION_MS,
      },
      metrics: {
        throughputJobsPerSecond: round(LOAD_JOBS / Math.max((lastEnd - firstStart) / 1_000, 0.001)),
        maxConcurrent,
        minimumMsBetweenEveryFifthStart: rateSpacing,
        queueWaitMs: summarize(loadResults.map((r) => r.queueWaitMs)),
        processingMs: summarize(loadResults.map((r) => r.serviceMs)),
        endToEndMs: summarize(loadResults.map((r) => r.endToEndMs)),
        workerIds,
      },
      loadBudget: {
        initial: loadInitialBudget, remaining: loadRemaining, admittedTokens: loadAdmitted,
        admittedUtilizationPct: round((loadAdmitted / loadInitialBudget) * 100),
        ledgerUtilizationPct: round(((loadInitialBudget - loadRemaining) / loadInitialBudget) * 100),
      },
      checks,
    };

Why are there three latency summaries?

• The queueWaitMs distribution measures time spent waiting before a worker starts the job.
• The processingMs distribution measures simulated LLM service time after processing begins.
• The endToEndMs distribution measures the full interval from enqueue to completion.
• Keeping these measurements separate reveals whether a slow result came from queue pressure or simulated processing.

• Continue the replacement with the cleanup path:

  } finally {
    await stopWorkers(workers);
    await events.close();
    await queue.obliterate({ force: true });
    await queue.close();
    await redis.quit();
  }
}

Why does cleanup use finally?

The finally block runs after success or failure. It closes both workers before destroying the disposable queue and releasing the remaining connections.

• Finish the replacement with the command selection and comparison table:

const requested = process.argv[2] ?? "compare";
if (!new Set(["naive", "atomic", "compare"]).has(requested)) throw new Error("Use naive, atomic, or compare");
const modes: Mode[] = requested === "compare" ? ["naive", "atomic"] : [requested as Mode];
const reports = [];
for (const mode of modes) reports.push(await runScenario(mode));

console.table(reports.map((report) => ({
  mode: report.mode,
  jobsPerSecond: report.metrics.throughputJobsPerSecond,
  queueWaitP50: report.metrics.queueWaitMs.p50,
  queueWaitP95: report.metrics.queueWaitMs.p95,
  queueWaitP99: report.metrics.queueWaitMs.p99,
  processingP50: report.metrics.processingMs.p50,
  processingP95: report.metrics.processingMs.p95,
  processingP99: report.metrics.processingMs.p99,
  endToEndP50: report.metrics.endToEndMs.p50,
  endToEndP95: report.metrics.endToEndMs.p95,
  endToEndP99: report.metrics.endToEndMs.p99,
  maxConcurrent: report.metrics.maxConcurrent,
  sharedConcurrencyCeilingHeld: report.checks.sharedConcurrencyCeilingHeld,
  allChecksPassed: report.checks.allPassed,
})));
if (reports.some((report) => !report.checks.allPassed)) process.exitCode = 1;

What does the table reveal?

• The table keeps queue wait, processing, and end-to-end percentiles in separate columns.
• The maxConcurrent column exposes the highest combined overlap across both workers.
• The shared concurrency flag confirms that both worker IDs participated without exceeding the global ceiling.
• A failed correctness check gives the command a nonzero exit status.

• Save src/benchmark.ts.

Before you run the distributed workload, do you expect each worker's local concurrency of three to produce six active jobs at once, or will the queue-wide ceiling hold the combined total lower?

• Run the naive distributed benchmark with this command:

npm run benchmark:naive

You should see active and completed JSON events from both worker-a and worker-b. The final table should report maxConcurrent no greater than 4.

The minimumMsBetweenEveryFifthStart value should be at least 900. You should also see independent p50, p95, and p99 columns for queue wait, processing, and end-to-end latency.

What did the benchmark prove?

Both workers could each process three jobs locally. The queue-wide ceiling kept their combined overlap at four or fewer.

The shared limiter also governed starts across both processes. Priority-1 work started before priority-10 work because all jobs were waiting before the workers began.

Checks not passing?

• Confirm the output contains both worker-a and worker-b before investigating the concurrency calculation.
• Check that setGlobalConcurrency receives GLOBAL_CONCURRENCY before either worker launches.
• Confirm the priority-10 loop runs before the priority-1 loop so all jobs are queued before processing begins.
• Use this prompt for targeted help: Help me diagnose a failed distributed benchmark check.

✔️ Awesome, I've got everything!

Great work. Your benchmark now measures one workload across two independent workers while enforcing shared scheduling controls.

• Double check that src/benchmark.ts is saved.

ⓧ I'd like to double check the full code

Compare your saved src/benchmark.ts with this cumulative version.

import { fork, type ChildProcess } from "node:child_process";
import { once } from "node:events";
import { Queue, QueueEvents, type Job } from "bullmq";
import { Redis } from "ioredis";
import {
  CONNECTION,
  GLOBAL_CONCURRENCY,
  RATE_LIMIT_DURATION_MS,
  RATE_LIMIT_MAX,
  type JobData,
  type JobResult,
  type Mode,
} from "./shared.js";

const LOAD_JOBS = 24;
const HIGH_PRIORITY_JOBS = 8;
const TOKEN_COST = 100;
type LaunchedWorker = { child: ChildProcess; ready: Promise<void> };

function round(value: number): number { return Number(value.toFixed(2)); }
function summarize(values: number[]) {
  const sorted = [...values].sort((a, b) => a - b);
  const at = (p: number): number => sorted[Math.max(0, Math.ceil((p / 100) * sorted.length) - 1)] ?? 0;
  return { p50: at(50), p95: at(95), p99: at(99) };
}
function maximumConcurrency(results: JobResult[]): number {
  const points = results.flatMap((r) => [
    { time: r.startedAt, delta: 1 }, { time: r.completedAt, delta: -1 },
  ]).sort((a, b) => a.time - b.time || a.delta - b.delta);
  let active = 0;
  let maximum = 0;
  for (const point of points) { active += point.delta; maximum = Math.max(maximum, active); }
  return maximum;
}
function minimumRateSpacing(results: JobResult[]): number {
  const starts = results.map((r) => r.startedAt).sort((a, b) => a - b);
  const spacings: number[] = [];
  for (let index = RATE_LIMIT_MAX; index < starts.length; index += 1) {
    spacings.push(starts[index] - starts[index - RATE_LIMIT_MAX]);
  }
  return Math.min(...spacings);
}
function priorityWasEnforced(results: JobResult[]): boolean {
  const high = results.filter((r) => r.priority === 1).map((r) => r.startedAt);
  const low = results.filter((r) => r.priority === 10).map((r) => r.startedAt);
  return Math.max(...high) <= Math.min(...low);
}
function launchWorker(queueName: string, mode: Mode, workerId: string): LaunchedWorker {
  const child = fork(new URL("./worker.js", import.meta.url), [queueName, mode, workerId], {
    stdio: ["inherit", "inherit", "inherit", "ipc"],
  });
  const ready = new Promise<void>((resolve, reject) => {
    const onMessage = (message: unknown): void => {
      if (typeof message === "object" && message !== null && "type" in message && message.type === "ready") {
        cleanup(); resolve();
      }
    };
    const onExit = (code: number | null): void => { cleanup(); reject(new Error(`${workerId} exited before ready: ${code}`)); };
    const cleanup = (): void => { child.off("message", onMessage); child.off("exit", onExit); };
    child.on("message", onMessage); child.on("exit", onExit);
  });
  return { child, ready };
}
async function stopWorkers(workers: LaunchedWorker[]): Promise<void> {
  const live = workers.filter(({ child }) => child.exitCode === null);
  const exits = live.map(({ child }) => once(child, "exit"));
  for (const { child } of live) if (child.connected) child.send({ type: "shutdown" });
  await Promise.all(exits);
}
async function waitForJobs(jobs: Job<JobData, JobResult, string>[], events: QueueEvents<JobResult>) {
  return Promise.all(jobs.map((job) => job.waitUntilFinished(events, 45_000)));
}
async function addLoadJobs(queue: Queue<JobData, JobResult, string>, budgetKey: string) {
  const jobs: Job<JobData, JobResult, string>[] = [];
  for (let index = HIGH_PRIORITY_JOBS; index < LOAD_JOBS; index += 1) {
    jobs.push(await queue.add("llm-call", {
      phase: "load", index, tokenCost: TOKEN_COST,
      simulatedMs: 40 + (index % 3) * 20, enqueuedAt: Date.now(), budgetKey,
    }, { priority: 10, attempts: 1 }));
  }
  for (let index = 0; index < HIGH_PRIORITY_JOBS; index += 1) {
    jobs.push(await queue.add("llm-call", {
      phase: "load", index, tokenCost: TOKEN_COST,
      simulatedMs: 40 + (index % 3) * 20, enqueuedAt: Date.now(), budgetKey,
    }, { priority: 1, attempts: 1 }));
  }
  return jobs;
}

async function runScenario(mode: Mode) {
  const runId = `${mode}-${Date.now()}`;
  const queueName = `llm-dispatch-${runId}`;
  const loadBudgetKey = `budget:${runId}:load`;
  const loadInitialBudget = LOAD_JOBS * TOKEN_COST;
  const redis = new Redis(CONNECTION);
  const queue = new Queue<JobData, JobResult, string>(queueName, { connection: CONNECTION });
  const events = new QueueEvents<JobResult>(queueName, { connection: CONNECTION });
  const workers: LaunchedWorker[] = [];

  events.on("completed", ({ jobId }) => console.log(JSON.stringify({ event: "queue_completed", queueName, jobId })));
  events.on("failed", ({ jobId, failedReason }) => console.error(JSON.stringify({ event: "queue_failed", queueName, jobId, failedReason })));
  events.on("error", (error) => console.error(JSON.stringify({ event: "queue_events_error", error: error.message })));

  try {
    await Promise.all([queue.waitUntilReady(), events.waitUntilReady()]);
    await queue.setGlobalConcurrency(GLOBAL_CONCURRENCY);
    await redis.set(loadBudgetKey, loadInitialBudget.toString());
    const loadJobs = await addLoadJobs(queue, loadBudgetKey);
    workers.push(launchWorker(queueName, mode, "worker-a"), launchWorker(queueName, mode, "worker-b"));
    await Promise.all(workers.map((worker) => worker.ready));
    const loadResults = await waitForJobs(loadJobs, events);

    const loadRemaining = Number((await redis.get(loadBudgetKey)) ?? "0");
    const loadAdmitted = loadResults.filter((r) => r.reserved).reduce((sum, r) => sum + r.tokenCost, 0);
    const firstStart = Math.min(...loadResults.map((r) => r.startedAt));
    const lastEnd = Math.max(...loadResults.map((r) => r.completedAt));
    const maxConcurrent = maximumConcurrency(loadResults);
    const rateSpacing = minimumRateSpacing(loadResults);
    const workerIds = [...new Set(loadResults.map((r) => r.workerId))].sort();
    const sharedConcurrencyCeilingHeld = workerIds.length === 2 && maxConcurrent <= GLOBAL_CONCURRENCY;
    const checks = {
      usedTwoWorkers: workerIds.length === 2,
      priorityOrderEnforced: priorityWasEnforced(loadResults),
      sharedConcurrencyCeilingHeld,
      globalRateLimitObserved: rateSpacing >= 900,
      allPassed: false,
    };
    checks.allPassed = Object.entries(checks)
      .filter(([name]) => name !== "allPassed")
      .every(([, passed]) => passed);

    return {
      mode,
      workload: {
        loadJobs: LOAD_JOBS, workerProcesses: 2,
        globalConcurrency: GLOBAL_CONCURRENCY,
        rateLimitMax: RATE_LIMIT_MAX, rateLimitDurationMs: RATE_LIMIT_DURATION_MS,
      },
      metrics: {
        throughputJobsPerSecond: round(LOAD_JOBS / Math.max((lastEnd - firstStart) / 1_000, 0.001)),
        maxConcurrent,
        minimumMsBetweenEveryFifthStart: rateSpacing,
        queueWaitMs: summarize(loadResults.map((r) => r.queueWaitMs)),
        processingMs: summarize(loadResults.map((r) => r.serviceMs)),
        endToEndMs: summarize(loadResults.map((r) => r.endToEndMs)),
        workerIds,
      },
      loadBudget: {
        initial: loadInitialBudget, remaining: loadRemaining, admittedTokens: loadAdmitted,
        admittedUtilizationPct: round((loadAdmitted / loadInitialBudget) * 100),
        ledgerUtilizationPct: round(((loadInitialBudget - loadRemaining) / loadInitialBudget) * 100),
      },
      checks,
    };
  } finally {
    await stopWorkers(workers);
    await events.close();
    await queue.obliterate({ force: true });
    await queue.close();
    await redis.quit();
  }
}

const requested = process.argv[2] ?? "compare";
if (!new Set(["naive", "atomic", "compare"]).has(requested)) throw new Error("Use naive, atomic, or compare");
const modes: Mode[] = requested === "compare" ? ["naive", "atomic"] : [requested as Mode];
const reports = [];
for (const mode of modes) reports.push(await runScenario(mode));

console.table(reports.map((report) => ({
  mode: report.mode,
  jobsPerSecond: report.metrics.throughputJobsPerSecond,
  queueWaitP50: report.metrics.queueWaitMs.p50,
  queueWaitP95: report.metrics.queueWaitMs.p95,
  queueWaitP99: report.metrics.queueWaitMs.p99,
  processingP50: report.metrics.processingMs.p50,
  processingP95: report.metrics.processingMs.p95,
  processingP99: report.metrics.processingMs.p99,
  endToEndP50: report.metrics.endToEndMs.p50,
  endToEndP95: report.metrics.endToEndMs.p95,
  endToEndP99: report.metrics.endToEndMs.p99,
  maxConcurrent: report.metrics.maxConcurrent,
  sharedConcurrencyCeilingHeld: report.checks.sharedConcurrencyCeilingHeld,
  allChecksPassed: report.checks.allPassed,
})));
if (reports.some((report) => !report.checks.allPassed)) process.exitCode = 1;

That is the coordination proof complete. Your two workers share one concurrency ceiling and one rate limiter while producing separate latency measurements.

Next, you will keep the distributed workload and force its read-then-write budget gate into a reproducible race.

Force the Naive Budget Race

Your BullMQ benchmark now coordinates two workers while measuring scheduling performance. Those healthy metrics can still hide a shared-budget correctness bug.

This step stages four simultaneous reservations against one Redis balance. A barrier makes every worker read before any worker writes, so you can reproduce the same race condition on every run.

In this step, get ready to:

• Implement the naive budget path with separate GET and SET operations.
• Synchronize four 100-token reservations against a 200-token balance.
• Run the deterministic race and inspect the admitted total against the final ledger.

Synchronize the stale budget reads

A naive reservation makes its decision from a value that can become stale before the write. The shared barrier pauses all four jobs after that decision point.

• Switch back to src/shared.ts in the VS Code editor.
• Find the reserveBudget() function.
• Replace its naive branch with this code:

  if (mode === "naive") {
    const observed = Number((await redis.get(data.budgetKey)) ?? "0");
    if (observed < data.tokenCost) return { reserved: false, remaining: observed };
    if (data.barrierKey && data.barrierParticipants) {
      await waitAtBarrier(redis, data.barrierKey, data.barrierParticipants);
    }
    const remaining = observed - data.tokenCost;
    await redis.set(data.budgetKey, remaining.toString());
    return { reserved: true, remaining };
  }

What does this branch do?

• The first Redis read stores a local snapshot in observed.
• The budget check accepts the reservation when that snapshot covers data.tokenCost.
• The waitAtBarrier() call holds each race job until all four jobs have reached the same point.
• The final Redis write uses the local snapshot to calculate the new balance.

• Save src/shared.ts.
• Confirm the shared module still passes strict TypeScript checking by running:

npm run build

The terminal returns to the prompt without a TypeScript error. That successful build confirms the naive path matches the existing job types.

Seeing a TypeScript error?

• Confirm the naive branch remains inside reserveBudget().
• Check that waitAtBarrier() still appears above reserveBudget().
• Compare the property names against budgetKey, barrierKey, and barrierParticipants in JobData.
• Help me debug the naive reservation branch in src/shared.ts.

Add the four-job race workload

The benchmark needs a small workload with an impossible budget target. Four jobs each request 100 tokens from a ledger containing only 200 tokens.

• Switch back to src/benchmark.ts in the VS Code editor.
• Find the workload constants near the top of the file.
• Replace that constants group with this code:

const LOAD_JOBS = 24;
const HIGH_PRIORITY_JOBS = 8;
const TOKEN_COST = 100;
const RACE_JOBS = 4;
const RACE_BUDGET = 200;

What do these constants control?

• The existing load workload remains at 24 jobs.
• The race workload contains four jobs through RACE_JOBS.
• Each reservation costs 100 tokens through TOKEN_COST.
• The shared race ledger starts with 200 tokens through RACE_BUDGET.

• Find the existing addLoadJobs() function.
• Add this new function immediately below it:

async function addRaceJobs(queue: Queue<JobData, JobResult, string>, budgetKey: string, barrierKey: string) {
  const jobs: Job<JobData, JobResult, string>[] = [];
  for (let index = 0; index < RACE_JOBS; index += 1) {
    jobs.push(await queue.add("llm-call", {
      phase: "race", index, tokenCost: TOKEN_COST, simulatedMs: 50,
      enqueuedAt: Date.now(), budgetKey, barrierKey, barrierParticipants: RACE_JOBS,
    }, { priority: 5, attempts: 1 }));
  }
  return jobs;
}

How does the race workload stay deterministic?

• Every job receives the same budgetKey.
• Every job receives the same barrierKey.
• The barrierParticipants value requires all four jobs to arrive before any reservation continues.
• The single attempt keeps retries from changing the reservation experiment.

• Save src/benchmark.ts.
• Check the new workload helper by running:

npm run build

The build completes without a TypeScript error. Your benchmark can now construct the four synchronized race jobs.

Does the race helper fail to compile?

• Confirm addRaceJobs() appears outside addLoadJobs().
• Check that the race job uses phase: "race".
• Verify that barrierParticipants receives RACE_JOBS.
• Help me fix the addRaceJobs function in src/benchmark.ts.

Each scenario now needs separate keys for the load budget, race budget, and synchronization barrier. Unique per-run keys keep one benchmark run from contaminating the next.

• Find the opening declarations inside runScenario().
• Replace the run-specific declaration group with this code:

  const runId = `${mode}-${Date.now()}`;
  const queueName = `llm-dispatch-${runId}`;
  const loadBudgetKey = `budget:${runId}:load`;
  const raceBudgetKey = `budget:${runId}:race`;
  const barrierKey = `barrier:${runId}:race`;
  const loadInitialBudget = LOAD_JOBS * TOKEN_COST;

Why use three run-specific keys?

• The load key preserves the budget used by the performance workload.
• The race key isolates the 200-token correctness experiment.
• The barrier key counts how many race jobs have reached the synchronized decision point.

• Find the setup statements near the start of the try block.
• Replace the setup group through loadJobs with this code:

    await queue.setGlobalConcurrency(GLOBAL_CONCURRENCY);
    await redis.set(loadBudgetKey, loadInitialBudget.toString());
    await redis.set(raceBudgetKey, RACE_BUDGET.toString());
    await redis.del(barrierKey);
    const loadJobs = await addLoadJobs(queue, loadBudgetKey);

What does this setup establish?

• The existing shared concurrency ceiling remains active.
• The load workload receives enough budget for every performance job.
• The race workload receives exactly enough budget for two reservations.
• Deleting the barrier key resets its participant count before the workers start.

• Find the line that waits for loadJobs to finish.
• Replace that line with this two-line sequence:

    const loadResults = await waitForJobs(loadJobs, events);
    const raceResults = await waitForJobs(await addRaceJobs(queue, raceBudgetKey, barrierKey), events);

Why run the race after the load workload?

The benchmark records its priority and latency workload first. It then uses the same workers for a separate correctness experiment.

• Save src/benchmark.ts.
• Confirm the scenario wiring compiles by running:

npm run build

The build completes without a TypeScript error. The benchmark now schedules its race immediately after the measured load workload.

Does the scenario wiring fail to compile?

• Confirm raceBudgetKey and barrierKey are declared inside runScenario().
• Check that addRaceJobs() receives the queue followed by both race keys.
• Verify that raceResults is created after loadResults.
• Help me debug the race setup inside runScenario.

Measure the unsafe outcome

A shared ledger can look plausible even when admitted work exceeds its capacity. The report therefore compares accepted token cost with the balance left in Redis.

• Find the metrics calculations after raceResults inside runScenario().
• Replace that calculation group through raceOutcomeMatchedExpectation with this code:

    const loadRemaining = Number((await redis.get(loadBudgetKey)) ?? "0");
    const raceRemaining = Number((await redis.get(raceBudgetKey)) ?? "0");
    const loadAdmitted = loadResults.filter((r) => r.reserved).reduce((sum, r) => sum + r.tokenCost, 0);
    const raceAdmitted = raceResults.filter((r) => r.reserved).reduce((sum, r) => sum + r.tokenCost, 0);
    const firstStart = Math.min(...loadResults.map((r) => r.startedAt));
    const lastEnd = Math.max(...loadResults.map((r) => r.completedAt));
    const maxConcurrent = maximumConcurrency(loadResults);
    const rateSpacing = minimumRateSpacing(loadResults);
    const workerIds = [...new Set(loadResults.map((r) => r.workerId))].sort();
    const raceAccepted = raceResults.filter((r) => r.reserved).length;
    const sharedConcurrencyCeilingHeld = workerIds.length === 2 && maxConcurrent <= GLOBAL_CONCURRENCY;
    const raceOutcomeMatchedExpectation = mode === "naive"
      ? raceAccepted === 4 && raceRemaining === 100
      : raceAccepted === 2 && raceRemaining === 0;

What does the race measurement compare?

• The raceAdmitted total counts the token cost of every accepted reservation.
• The raceRemaining value reads the final shared ledger directly from Redis.
• The raceAccepted count shows how many jobs passed the budget decision.
• The naive expectation requires four accepted reservations with 100 tokens still recorded.

• Find the checks object immediately below those calculations.
• Replace that object with this code:

    const checks = {
      usedTwoWorkers: workerIds.length === 2,
      priorityOrderEnforced: priorityWasEnforced(loadResults),
      sharedConcurrencyCeilingHeld,
      globalRateLimitObserved: rateSpacing >= 900,
      raceOutcomeMatchedExpectation,
      allPassed: false,
    };

Why count the race as an expected result?

This check proves that the benchmark reproduced the intended failure deterministically. A passing reproduction check does not make the budget state safe.

• Save src/benchmark.ts.
• Confirm the new calculations compile by running:

npm run build

The build completes without a TypeScript error. The report can now distinguish admitted work from the value stored in the ledger.

Do the race calculations fail to compile?

• Confirm raceResults is declared before the calculation group.
• Check that raceAccepted uses the reserved result field.
• Verify that every property in checks matches a calculation above it.
• Help me fix the race metrics in src/benchmark.ts.

• Find the returned report object inside runScenario().
• Add this race budget section immediately below loadBudget:

      raceBudget: {
        initial: RACE_BUDGET, remaining: raceRemaining,
        acceptedReservations: raceAccepted, rejectedReservations: RACE_JOBS - raceAccepted,
        admittedTokens: raceAdmitted,
        admittedUtilizationPct: round((raceAdmitted / RACE_BUDGET) * 100),
        ledgerUtilizationPct: round(((RACE_BUDGET - raceRemaining) / RACE_BUDGET) * 100),
      },

How are the two utilization values different?

• Admitted utilization divides accepted token cost by the initial race budget.
• Ledger utilization calculates consumption from the balance Redis retained.
• A difference between those percentages exposes a ledger that no longer represents admitted work.

• Find the Redis cleanup call in the finally block.
• Update that call to remove all three run-specific keys:

    await redis.del(loadBudgetKey, raceBudgetKey, barrierKey);

Why clean up the race keys?

The benchmark generates new keys for every run. Removing the budget and barrier keys keeps the local Redis instance tidy after the report is complete.

• Find the object passed to console.table() near the bottom of src/benchmark.ts.
• Add these race fields after maxConcurrent:

  raceAccepted: report.raceBudget.acceptedReservations,
  raceRemaining: report.raceBudget.remaining,
  admittedBudgetPct: report.raceBudget.admittedUtilizationPct,
  ledgerBudgetPct: report.raceBudget.ledgerUtilizationPct,
  sharedConcurrencyCeilingHeld: report.checks.sharedConcurrencyCeilingHeld,
  allChecksPassed: report.checks.allPassed,

What does the table reveal?

The table places the accepted count beside both views of utilization. You can compare the shared-control result with the budget race without losing the existing latency measurements.

• Save src/benchmark.ts.

Before you run the benchmark, do you think four workers can all accept a reservation from a balance that only covers two?

• Run the naive benchmark to reveal the synchronized outcome:

npm run benchmark:naive

You will see lifecycle logs from both worker IDs. The comparison table reports four accepted race reservations and 400 admitted tokens.

Admitted utilization reaches 200 percent while the shared ledger still reports 100 tokens remaining. You have reproduced the time-of-check-to-time-of-use bug on demand.

Why is the ledger unsafe?

All four jobs read the same 200-token balance. Each job independently calculates a remaining balance of 100 tokens.

The four writes store the same stale calculation. The ledger records one reservation even though the workers admitted four.

Does the race hang or show a different result?

• Confirm both worker-a and worker-b reported active race jobs.
• Check that global concurrency remains 4.
• Verify that every race job uses barrierParticipants: RACE_JOBS.
• Help me debug the deterministic naive budget race.

✔️ Awesome, I've got everything!

Great work. Your distributed benchmark now reproduces a shared-budget race while preserving its scheduling and latency measurements.

ⓧ I'd like to double check the full code

Compare your cumulative src/shared.ts with this reference:

import { Redis } from "ioredis";

export const CONNECTION = { host: "127.0.0.1", port: 6379 } as const;
export const GLOBAL_CONCURRENCY = 4;
export const LOCAL_CONCURRENCY = 3;
export const RATE_LIMIT_MAX = 4;
export const RATE_LIMIT_DURATION_MS = 1_000;
export type Mode = "naive" | "atomic";
export type Phase = "load" | "race";

export interface JobData {
  phase: Phase;
  index: number;
  tokenCost: number;
  simulatedMs: number;
  enqueuedAt: number;
  budgetKey: string;
  barrierKey?: string;
  barrierParticipants?: number;
}

export interface JobResult {
  jobId: string;
  phase: Phase;
  workerId: string;
  priority: number;
  enqueuedAt: number;
  startedAt: number;
  completedAt: number;
  queueWaitMs: number;
  serviceMs: number;
  endToEndMs: number;
  tokenCost: number;
  reserved: boolean;
  remainingBudget: number;
  status: "completed" | "budget_rejected";
}

const ATOMIC_RESERVATION_SCRIPT = `
local remaining = tonumber(redis.call('GET', KEYS[1]) or '0')
local cost = tonumber(ARGV[1])
if remaining < cost then return {0, remaining} end
local next_remaining = remaining - cost
redis.call('SET', KEYS[1], next_remaining)
return {1, next_remaining}
`;

export function sleep(milliseconds: number): Promise<void> {
  return new Promise((resolve) => setTimeout(resolve, milliseconds));
}

async function waitAtBarrier(redis: Redis, key: string, participants: number): Promise<void> {
  await redis.incr(key);
  const deadline = Date.now() + 5_000;
  while (Number((await redis.get(key)) ?? "0") < participants) {
    if (Date.now() > deadline) throw new Error(`Barrier timed out: ${key}`);
    await sleep(5);
  }
}

export async function reserveBudget(redis: Redis, mode: Mode, data: JobData) {
  if (mode === "naive") {
    const observed = Number((await redis.get(data.budgetKey)) ?? "0");
    if (observed < data.tokenCost) return { reserved: false, remaining: observed };
    if (data.barrierKey && data.barrierParticipants) {
      await waitAtBarrier(redis, data.barrierKey, data.barrierParticipants);
    }
    const remaining = observed - data.tokenCost;
    await redis.set(data.budgetKey, remaining.toString());
    return { reserved: true, remaining };
  }
  if (data.barrierKey && data.barrierParticipants) {
    await waitAtBarrier(redis, data.barrierKey, data.barrierParticipants);
  }
  const reply = (await redis.eval(
    ATOMIC_RESERVATION_SCRIPT,
    1,
    data.budgetKey,
    data.tokenCost.toString(),
  )) as [number, number];
  return { reserved: Number(reply[0]) === 1, remaining: Number(reply[1]) };
}

This file contains the shared job types, barrier, and reservation paths used by both worker processes.

Compare your cumulative src/benchmark.ts with this reference:

import { fork, type ChildProcess } from "node:child_process";
import { once } from "node:events";
import { Queue, QueueEvents, type Job } from "bullmq";
import { Redis } from "ioredis";
import {
  CONNECTION,
  GLOBAL_CONCURRENCY,
  RATE_LIMIT_DURATION_MS,
  RATE_LIMIT_MAX,
  type JobData,
  type JobResult,
  type Mode,
} from "./shared.js";

const LOAD_JOBS = 24;
const HIGH_PRIORITY_JOBS = 8;
const TOKEN_COST = 100;
const RACE_JOBS = 4;
const RACE_BUDGET = 200;
type LaunchedWorker = { child: ChildProcess; ready: Promise<void> };

function round(value: number): number { return Number(value.toFixed(2)); }
function summarize(values: number[]) {
  const sorted = [...values].sort((a, b) => a - b);
  const at = (p: number): number => sorted[Math.max(0, Math.ceil((p / 100) * sorted.length) - 1)] ?? 0;
  return { p50: at(50), p95: at(95), p99: at(99) };
}
function maximumConcurrency(results: JobResult[]): number {
  const points = results.flatMap((r) => [
    { time: r.startedAt, delta: 1 }, { time: r.completedAt, delta: -1 },
  ]).sort((a, b) => a.time - b.time || a.delta - b.delta);
  let active = 0;
  let maximum = 0;
  for (const point of points) { active += point.delta; maximum = Math.max(maximum, active); }
  return maximum;
}
function minimumRateSpacing(results: JobResult[]): number {
  const starts = results.map((r) => r.startedAt).sort((a, b) => a - b);
  const spacings: number[] = [];
  for (let index = RATE_LIMIT_MAX; index < starts.length; index += 1) {
    spacings.push(starts[index] - starts[index - RATE_LIMIT_MAX]);
  }
  return Math.min(...spacings);
}
function priorityWasEnforced(results: JobResult[]): boolean {
  const high = results.filter((r) => r.priority === 1).map((r) => r.startedAt);
  const low = results.filter((r) => r.priority === 10).map((r) => r.startedAt);
  return Math.max(...high) <= Math.min(...low);
}
function launchWorker(queueName: string, mode: Mode, workerId: string): LaunchedWorker {
  const child = fork(new URL("./worker.js", import.meta.url), [queueName, mode, workerId], {
    stdio: ["inherit", "inherit", "inherit", "ipc"],
  });
  const ready = new Promise<void>((resolve, reject) => {
    const onMessage = (message: unknown): void => {
      if (typeof message === "object" && message !== null && "type" in message && message.type === "ready") {
        cleanup(); resolve();
      }
    };
    const onExit = (code: number | null): void => { cleanup(); reject(new Error(`${workerId} exited before ready: ${code}`)); };
    const cleanup = (): void => { child.off("message", onMessage); child.off("exit", onExit); };
    child.on("message", onMessage); child.on("exit", onExit);
  });
  return { child, ready };
}
async function stopWorkers(workers: LaunchedWorker[]): Promise<void> {
  const live = workers.filter(({ child }) => child.exitCode === null);
  const exits = live.map(({ child }) => once(child, "exit"));
  for (const { child } of live) if (child.connected) child.send({ type: "shutdown" });
  await Promise.all(exits);
}
async function waitForJobs(jobs: Job<JobData, JobResult, string>[], events: QueueEvents<JobResult>) {
  return Promise.all(jobs.map((job) => job.waitUntilFinished(events, 45_000)));
}
async function addLoadJobs(queue: Queue<JobData, JobResult, string>, budgetKey: string) {
  const jobs: Job<JobData, JobResult, string>[] = [];
  for (let index = HIGH_PRIORITY_JOBS; index < LOAD_JOBS; index += 1) {
    jobs.push(await queue.add("llm-call", {
      phase: "load", index, tokenCost: TOKEN_COST,
      simulatedMs: 40 + (index % 3) * 20, enqueuedAt: Date.now(), budgetKey,
    }, { priority: 10, attempts: 1 }));
  }
  for (let index = 0; index < HIGH_PRIORITY_JOBS; index += 1) {
    jobs.push(await queue.add("llm-call", {
      phase: "load", index, tokenCost: TOKEN_COST,
      simulatedMs: 40 + (index % 3) * 20, enqueuedAt: Date.now(), budgetKey,
    }, { priority: 1, attempts: 1 }));
  }
  return jobs;
}
async function addRaceJobs(queue: Queue<JobData, JobResult, string>, budgetKey: string, barrierKey: string) {
  const jobs: Job<JobData, JobResult, string>[] = [];
  for (let index = 0; index < RACE_JOBS; index += 1) {
    jobs.push(await queue.add("llm-call", {
      phase: "race", index, tokenCost: TOKEN_COST, simulatedMs: 50,
      enqueuedAt: Date.now(), budgetKey, barrierKey, barrierParticipants: RACE_JOBS,
    }, { priority: 5, attempts: 1 }));
  }
  return jobs;
}

async function runScenario(mode: Mode) {
  const runId = `${mode}-${Date.now()}`;
  const queueName = `llm-dispatch-${runId}`;
  const loadBudgetKey = `budget:${runId}:load`;
  const raceBudgetKey = `budget:${runId}:race`;
  const barrierKey = `barrier:${runId}:race`;
  const loadInitialBudget = LOAD_JOBS * TOKEN_COST;
  const redis = new Redis(CONNECTION);
  const queue = new Queue<JobData, JobResult, string>(queueName, { connection: CONNECTION });
  const events = new QueueEvents<JobResult>(queueName, { connection: CONNECTION });
  const workers: LaunchedWorker[] = [];

  events.on("completed", ({ jobId }) => console.log(JSON.stringify({ event: "queue_completed", queueName, jobId })));
  events.on("failed", ({ jobId, failedReason }) => console.error(JSON.stringify({ event: "queue_failed", queueName, jobId, failedReason })));
  events.on("error", (error) => console.error(JSON.stringify({ event: "queue_events_error", error: error.message })));

  try {
    await Promise.all([queue.waitUntilReady(), events.waitUntilReady()]);
    await queue.setGlobalConcurrency(GLOBAL_CONCURRENCY);
    await redis.set(loadBudgetKey, loadInitialBudget.toString());
    await redis.set(raceBudgetKey, RACE_BUDGET.toString());
    await redis.del(barrierKey);
    const loadJobs = await addLoadJobs(queue, loadBudgetKey);
    workers.push(launchWorker(queueName, mode, "worker-a"), launchWorker(queueName, mode, "worker-b"));
    await Promise.all(workers.map((worker) => worker.ready));
    const loadResults = await waitForJobs(loadJobs, events);
    const raceResults = await waitForJobs(await addRaceJobs(queue, raceBudgetKey, barrierKey), events);

    const loadRemaining = Number((await redis.get(loadBudgetKey)) ?? "0");
    const raceRemaining = Number((await redis.get(raceBudgetKey)) ?? "0");
    const loadAdmitted = loadResults.filter((r) => r.reserved).reduce((sum, r) => sum + r.tokenCost, 0);
    const raceAdmitted = raceResults.filter((r) => r.reserved).reduce((sum, r) => sum + r.tokenCost, 0);
    const firstStart = Math.min(...loadResults.map((r) => r.startedAt));
    const lastEnd = Math.max(...loadResults.map((r) => r.completedAt));
    const maxConcurrent = maximumConcurrency(loadResults);
    const rateSpacing = minimumRateSpacing(loadResults);
    const workerIds = [...new Set(loadResults.map((r) => r.workerId))].sort();
    const raceAccepted = raceResults.filter((r) => r.reserved).length;
    const sharedConcurrencyCeilingHeld = workerIds.length === 2 && maxConcurrent <= GLOBAL_CONCURRENCY;
    const raceOutcomeMatchedExpectation = mode === "naive"
      ? raceAccepted === 4 && raceRemaining === 100
      : raceAccepted === 2 && raceRemaining === 0;
    const checks = {
      usedTwoWorkers: workerIds.length === 2,
      priorityOrderEnforced: priorityWasEnforced(loadResults),
      sharedConcurrencyCeilingHeld,
      globalRateLimitObserved: rateSpacing >= 900,
      raceOutcomeMatchedExpectation,
      allPassed: false,
    };
    checks.allPassed = Object.entries(checks)
      .filter(([name]) => name !== "allPassed")
      .every(([, passed]) => passed);

    return {
      mode,
      workload: {
        loadJobs: LOAD_JOBS, workerProcesses: 2,
        globalConcurrency: GLOBAL_CONCURRENCY,
        rateLimitMax: RATE_LIMIT_MAX, rateLimitDurationMs: RATE_LIMIT_DURATION_MS,
      },
      metrics: {
        throughputJobsPerSecond: round(LOAD_JOBS / Math.max((lastEnd - firstStart) / 1_000, 0.001)),
        maxConcurrent,
        minimumMsBetweenEveryFifthStart: rateSpacing,
        queueWaitMs: summarize(loadResults.map((r) => r.queueWaitMs)),
        processingMs: summarize(loadResults.map((r) => r.serviceMs)),
        endToEndMs: summarize(loadResults.map((r) => r.endToEndMs)),
        workerIds,
      },
      loadBudget: {
        initial: loadInitialBudget, remaining: loadRemaining, admittedTokens: loadAdmitted,
        admittedUtilizationPct: round((loadAdmitted / loadInitialBudget) * 100),
        ledgerUtilizationPct: round(((loadInitialBudget - loadRemaining) / loadInitialBudget) * 100),
      },
      raceBudget: {
        initial: RACE_BUDGET, remaining: raceRemaining,
        acceptedReservations: raceAccepted, rejectedReservations: RACE_JOBS - raceAccepted,
        admittedTokens: raceAdmitted,
        admittedUtilizationPct: round((raceAdmitted / RACE_BUDGET) * 100),
        ledgerUtilizationPct: round(((RACE_BUDGET - raceRemaining) / RACE_BUDGET) * 100),
      },
      checks,
    };
  } finally {
    await stopWorkers(workers);
    await events.close();
    await queue.obliterate({ force: true });
    await queue.close();
    await redis.del(loadBudgetKey, raceBudgetKey, barrierKey);
    await redis.quit();
  }
}

const requested = process.argv[2] ?? "compare";
if (!new Set(["naive", "atomic", "compare"]).has(requested)) throw new Error("Use naive, atomic, or compare");
const modes: Mode[] = requested === "compare" ? ["naive", "atomic"] : [requested as Mode];
const reports = [];
for (const mode of modes) reports.push(await runScenario(mode));

console.table(reports.map((report) => ({
  mode: report.mode,
  jobsPerSecond: report.metrics.throughputJobsPerSecond,
  queueWaitP50: report.metrics.queueWaitMs.p50,
  queueWaitP95: report.metrics.queueWaitMs.p95,
  queueWaitP99: report.metrics.queueWaitMs.p99,
  processingP50: report.metrics.processingMs.p50,
  processingP95: report.metrics.processingMs.p95,
  processingP99: report.metrics.processingMs.p99,
  endToEndP50: report.metrics.endToEndMs.p50,
  endToEndP95: report.metrics.endToEndMs.p95,
  endToEndP99: report.metrics.endToEndMs.p99,
  maxConcurrent: report.metrics.maxConcurrent,
  raceAccepted: report.raceBudget.acceptedReservations,
  raceRemaining: report.raceBudget.remaining,
  admittedBudgetPct: report.raceBudget.admittedUtilizationPct,
  ledgerBudgetPct: report.raceBudget.ledgerUtilizationPct,
  sharedConcurrencyCeilingHeld: report.checks.sharedConcurrencyCeilingHeld,
  allChecksPassed: report.checks.allPassed,
})));
if (reports.some((report) => !report.checks.allPassed)) process.exitCode = 1;

This file runs the load workload first. It then measures the synchronized budget race through the same two worker processes.

You have proved that healthy throughput does not guarantee a safe budget gate. Next, you will move the reservation decision into one atomic Redis operation and compare both outcomes.

Replace the Race with an Atomic Reservation

Your two-worker benchmark now reproduces the budget race on demand. Four jobs reserve 400 tokens from a 200-token budget because every process acts on the same stale Redis value.

The baseline over-admits because its policy check and ledger update happen in separate round trips. This step uses a short Lua operation to make that decision atomic.

In this step, get ready to:

• Run the staged atomic reservation path against the deterministic race.
• Compare naive and atomic modes under the same queue controls.
• Export both reports with explicit correctness checks.

Prove the atomic reservation

The atomic branch in src/shared.ts waits at the same four-job barrier used by the naive branch. It then passes the budget key and token cost to one redis.eval() call.

Why Use Lua Here?

Redis runs a script atomically. The script reads the balance before deciding whether to write the reduced value.

A WATCH transaction can abort during contention. The Lua operation makes the complete decision in one server-side critical section.

• Switch back to src/shared.ts in the VS Code editor.
• Locate ATOMIC_RESERVATION_SCRIPT below the shared job types.
• Confirm the script reads KEYS[1] before checking ARGV[1].
• Confirm the atomic branch calls redis.eval() after waitAtBarrier().
• Switch back to src/benchmark.ts in the VS Code editor.
• Find the section that defines raceAccepted.
• Replace that correctness section through the checks.allPassed assignment with this code:

    const raceAccepted = raceResults.filter((r) => r.reserved).length;
    const sharedConcurrencyCeilingHeld = workerIds.length === 2 && maxConcurrent <= GLOBAL_CONCURRENCY;
    const atomicBudgetNeverOverspent = mode !== "atomic" || (
      raceAdmitted <= RACE_BUDGET && raceRemaining >= 0 &&
      raceAccepted === 2 && RACE_JOBS - raceAccepted === 2
    );
    const raceOutcomeMatchedExpectation = mode === "naive"
      ? raceAccepted === 4 && raceRemaining === 100
      : raceAccepted === 2 && raceRemaining === 0;
    const checks = {
      usedTwoWorkers: workerIds.length === 2,
      priorityOrderEnforced: priorityWasEnforced(loadResults),
      sharedConcurrencyCeilingHeld,
      globalRateLimitObserved: rateSpacing >= 900,
      atomicBudgetNeverOverspent,
      raceOutcomeMatchedExpectation,
      allPassed: false,
    };
    checks.allPassed = Object.entries(checks)
      .filter(([name]) => name !== "allPassed")
      .every(([, passed]) => passed);

What Do These Checks Prove?

• The shared-concurrency check requires results from both worker processes.
• The same check requires maximum combined concurrency to stay at or below 4.
• The atomic safety check requires two accepted reservations.
• It also requires two rejections with a nonnegative ledger balance.

• Find the mode-selection block near the bottom of src/benchmark.ts.
• Replace the block that reads process.argv[2] with this code:

const requested = process.argv[2] ?? "compare";
if (!new Set(["naive", "atomic", "compare"]).has(requested)) throw new Error("Use naive, atomic, or compare");
const modes: Mode[] = requested === "compare" ? ["naive", "atomic"] : [requested as Mode];
const reports = [];
for (const mode of modes) reports.push(await runScenario(mode));

How Are Scenarios Selected?

A specific mode produces one report. The compare mode runs naive first before running atomic through the same runScenario() function.

Before you run the atomic mode, do you expect all four race jobs to reserve 100 tokens again?

• Save src/benchmark.ts.
• Run the atomic race in the VS Code terminal with this command:

npm run benchmark:atomic

What Does This Command Run?

The command compiles the strict TypeScript project. It then sends the existing load workload and deterministic race through the atomic reservation mode.

You should see exactly 2 accepted race reservations. You should see 2 rejected reservations with 0 tokens remaining.

The report should show sharedConcurrencyCeilingHeld as true. It should also show atomicBudgetNeverOverspent as true.

Atomic Check Still Failing?

• Confirm the atomic branch calls redis.eval() after the barrier.
• Check that data.budgetKey is the script's only key.
• Verify the atomic expectation requires two accepted reservations with a zero balance.
• Help me debug the atomic reservation result.

Export and verify the comparison

One atomic run proves the safe result. A useful comparison artifact also preserves the unsafe baseline under the same priorities and shared controls.

• Scroll to the import group at the top of src/benchmark.ts.
• Add this file-writing import below the node:events import:

import { writeFile } from "node:fs/promises";

Why Import writeFile?

The benchmark keeps its scenario reports in memory. The promise-based writer saves those reports to benchmark-results.json.

• Save src/benchmark.ts.
• Confirm the new import compiles by running the atomic scenario again:

npm run benchmark:atomic

What Does This Check Confirm?

The command recompiles src/benchmark.ts with the new import. The atomic result should still accept two race reservations before finishing with zero tokens.

Seeing an Import Error?

• Confirm the import uses node:fs/promises.
• Check that writeFile uses the same capitalization in the import.
• Help me fix the file-writing import.

• Find the output section that begins with console.table at the bottom of src/benchmark.ts.
• Replace that output section with this code:

console.table(reports.map((report) => ({
  mode: report.mode,
  jobsPerSecond: report.metrics.throughputJobsPerSecond,
  queueWaitP50: report.metrics.queueWaitMs.p50,
  queueWaitP95: report.metrics.queueWaitMs.p95,
  queueWaitP99: report.metrics.queueWaitMs.p99,
  processingP50: report.metrics.processingMs.p50,
  processingP95: report.metrics.processingMs.p95,
  processingP99: report.metrics.processingMs.p99,
  endToEndP50: report.metrics.endToEndMs.p50,
  endToEndP95: report.metrics.endToEndMs.p95,
  endToEndP99: report.metrics.endToEndMs.p99,
  maxConcurrent: report.metrics.maxConcurrent,
  raceAccepted: report.raceBudget.acceptedReservations,
  raceRemaining: report.raceBudget.remaining,
  admittedBudgetPct: report.raceBudget.admittedUtilizationPct,
  ledgerBudgetPct: report.raceBudget.ledgerUtilizationPct,
  sharedConcurrencyCeilingHeld: report.checks.sharedConcurrencyCeilingHeld,
  atomicBudgetNeverOverspent: report.checks.atomicBudgetNeverOverspent,
  allChecksPassed: report.checks.allPassed,
})));
await writeFile("benchmark-results.json", `${JSON.stringify(reports, null, 2)}\n`);
console.log("Wrote benchmark-results.json");
if (reports.some((report) => !report.checks.allPassed)) process.exitCode = 1;

What Does the Output Preserve?

• The table places the naive report beside the atomic report.
• Each row shows throughput with all nine latency percentiles.
• The JSON artifact preserves workload settings and budget utilization.
• A failed correctness check gives the benchmark a nonzero exit code.

✔️ Awesome, I've got everything!

Your atomic checks and comparison output are ready. Double check that src/benchmark.ts is saved.

ⓧ Double check src/shared.ts

import { Redis } from "ioredis";

export const CONNECTION = { host: "127.0.0.1", port: 6379 } as const;
export const GLOBAL_CONCURRENCY = 4;
export const LOCAL_CONCURRENCY = 3;
export const RATE_LIMIT_MAX = 4;
export const RATE_LIMIT_DURATION_MS = 1_000;
export type Mode = "naive" | "atomic";
export type Phase = "load" | "race";

export interface JobData {
  phase: Phase;
  index: number;
  tokenCost: number;
  simulatedMs: number;
  enqueuedAt: number;
  budgetKey: string;
  barrierKey?: string;
  barrierParticipants?: number;
}

export interface JobResult {
  jobId: string;
  phase: Phase;
  workerId: string;
  priority: number;
  enqueuedAt: number;
  startedAt: number;
  completedAt: number;
  queueWaitMs: number;
  serviceMs: number;
  endToEndMs: number;
  tokenCost: number;
  reserved: boolean;
  remainingBudget: number;
  status: "completed" | "budget_rejected";
}

const ATOMIC_RESERVATION_SCRIPT = `
local remaining = tonumber(redis.call('GET', KEYS[1]) or '0')
local cost = tonumber(ARGV[1])
if remaining < cost then return {0, remaining} end
local next_remaining = remaining - cost
redis.call('SET', KEYS[1], next_remaining)
return {1, next_remaining}
`;

export function sleep(milliseconds: number): Promise<void> {
  return new Promise((resolve) => setTimeout(resolve, milliseconds));
}

async function waitAtBarrier(redis: Redis, key: string, participants: number): Promise<void> {
  await redis.incr(key);
  const deadline = Date.now() + 5_000;
  while (Number((await redis.get(key)) ?? "0") < participants) {
    if (Date.now() > deadline) throw new Error(`Barrier timed out: ${key}`);
    await sleep(5);
  }
}

export async function reserveBudget(redis: Redis, mode: Mode, data: JobData) {
  if (mode === "naive") {
    const observed = Number((await redis.get(data.budgetKey)) ?? "0");
    if (observed < data.tokenCost) return { reserved: false, remaining: observed };
    if (data.barrierKey && data.barrierParticipants) {
      await waitAtBarrier(redis, data.barrierKey, data.barrierParticipants);
    }
    const remaining = observed - data.tokenCost;
    await redis.set(data.budgetKey, remaining.toString());
    return { reserved: true, remaining };
  }
  if (data.barrierKey && data.barrierParticipants) {
    await waitAtBarrier(redis, data.barrierKey, data.barrierParticipants);
  }
  const reply = (await redis.eval(
    ATOMIC_RESERVATION_SCRIPT,
    1,
    data.budgetKey,
    data.tokenCost.toString(),
  )) as [number, number];
  return { reserved: Number(reply[0]) === 1, remaining: Number(reply[1]) };
}

How to Use This Reference

Compare this file with your saved src/shared.ts from top to bottom. Pay close attention to the barrier placement before redis.eval().

ⓧ Double check src/benchmark.ts

import { fork, type ChildProcess } from "node:child_process";
import { once } from "node:events";
import { writeFile } from "node:fs/promises";
import { Queue, QueueEvents, type Job } from "bullmq";
import { Redis } from "ioredis";
import {
  CONNECTION,
  GLOBAL_CONCURRENCY,
  RATE_LIMIT_DURATION_MS,
  RATE_LIMIT_MAX,
  type JobData,
  type JobResult,
  type Mode,
} from "./shared.js";

const LOAD_JOBS = 24;
const HIGH_PRIORITY_JOBS = 8;
const TOKEN_COST = 100;
const RACE_JOBS = 4;
const RACE_BUDGET = 200;
type LaunchedWorker = { child: ChildProcess; ready: Promise<void> };

function round(value: number): number { return Number(value.toFixed(2)); }
function summarize(values: number[]) {
  const sorted = [...values].sort((a, b) => a - b);
  const at = (p: number): number => sorted[Math.max(0, Math.ceil((p / 100) * sorted.length) - 1)] ?? 0;
  return { p50: at(50), p95: at(95), p99: at(99) };
}
function maximumConcurrency(results: JobResult[]): number {
  const points = results.flatMap((r) => [
    { time: r.startedAt, delta: 1 }, { time: r.completedAt, delta: -1 },
  ]).sort((a, b) => a.time - b.time || a.delta - b.delta);
  let active = 0;
  let maximum = 0;
  for (const point of points) { active += point.delta; maximum = Math.max(maximum, active); }
  return maximum;
}
function minimumRateSpacing(results: JobResult[]): number {
  const starts = results.map((r) => r.startedAt).sort((a, b) => a - b);
  const spacings: number[] = [];
  for (let index = RATE_LIMIT_MAX; index < starts.length; index += 1) {
    spacings.push(starts[index] - starts[index - RATE_LIMIT_MAX]);
  }
  return Math.min(...spacings);
}
function priorityWasEnforced(results: JobResult[]): boolean {
  const high = results.filter((r) => r.priority === 1).map((r) => r.startedAt);
  const low = results.filter((r) => r.priority === 10).map((r) => r.startedAt);
  return Math.max(...high) <= Math.min(...low);
}
function launchWorker(queueName: string, mode: Mode, workerId: string): LaunchedWorker {
  const child = fork(new URL("./worker.js", import.meta.url), [queueName, mode, workerId], {
    stdio: ["inherit", "inherit", "inherit", "ipc"],
  });
  const ready = new Promise<void>((resolve, reject) => {
    const onMessage = (message: unknown): void => {
      if (typeof message === "object" && message !== null && "type" in message && message.type === "ready") {
        cleanup(); resolve();
      }
    };
    const onExit = (code: number | null): void => { cleanup(); reject(new Error(`${workerId} exited before ready: ${code}`)); };
    const cleanup = (): void => { child.off("message", onMessage); child.off("exit", onExit); };
    child.on("message", onMessage); child.on("exit", onExit);
  });
  return { child, ready };
}
async function stopWorkers(workers: LaunchedWorker[]): Promise<void> {
  const live = workers.filter(({ child }) => child.exitCode === null);
  const exits = live.map(({ child }) => once(child, "exit"));
  for (const { child } of live) if (child.connected) child.send({ type: "shutdown" });
  await Promise.all(exits);
}
async function waitForJobs(jobs: Job<JobData, JobResult, string>[], events: QueueEvents<JobResult>) {
  return Promise.all(jobs.map((job) => job.waitUntilFinished(events, 45_000)));
}
async function addLoadJobs(queue: Queue<JobData, JobResult, string>, budgetKey: string) {
  const jobs: Job<JobData, JobResult, string>[] = [];
  for (let index = HIGH_PRIORITY_JOBS; index < LOAD_JOBS; index += 1) {
    jobs.push(await queue.add("llm-call", {
      phase: "load", index, tokenCost: TOKEN_COST,
      simulatedMs: 40 + (index % 3) * 20, enqueuedAt: Date.now(), budgetKey,
    }, { priority: 10, attempts: 1 }));
  }
  for (let index = 0; index < HIGH_PRIORITY_JOBS; index += 1) {
    jobs.push(await queue.add("llm-call", {
      phase: "load", index, tokenCost: TOKEN_COST,
      simulatedMs: 40 + (index % 3) * 20, enqueuedAt: Date.now(), budgetKey,
    }, { priority: 1, attempts: 1 }));
  }
  return jobs;
}
async function addRaceJobs(queue: Queue<JobData, JobResult, string>, budgetKey: string, barrierKey: string) {
  const jobs: Job<JobData, JobResult, string>[] = [];
  for (let index = 0; index < RACE_JOBS; index += 1) {
    jobs.push(await queue.add("llm-call", {
      phase: "race", index, tokenCost: TOKEN_COST, simulatedMs: 50,
      enqueuedAt: Date.now(), budgetKey, barrierKey, barrierParticipants: RACE_JOBS,
    }, { priority: 5, attempts: 1 }));
  }
  return jobs;
}

async function runScenario(mode: Mode) {
  const runId = `${mode}-${Date.now()}`;
  const queueName = `llm-dispatch-${runId}`;
  const loadBudgetKey = `budget:${runId}:load`;
  const raceBudgetKey = `budget:${runId}:race`;
  const barrierKey = `barrier:${runId}:race`;
  const loadInitialBudget = LOAD_JOBS * TOKEN_COST;
  const redis = new Redis(CONNECTION);
  const queue = new Queue<JobData, JobResult, string>(queueName, { connection: CONNECTION });
  const events = new QueueEvents<JobResult>(queueName, { connection: CONNECTION });
  const workers: LaunchedWorker[] = [];

  events.on("completed", ({ jobId }) => console.log(JSON.stringify({ event: "queue_completed", queueName, jobId })));
  events.on("failed", ({ jobId, failedReason }) => console.error(JSON.stringify({ event: "queue_failed", queueName, jobId, failedReason })));
  events.on("error", (error) => console.error(JSON.stringify({ event: "queue_events_error", error: error.message })));

  try {
    await Promise.all([queue.waitUntilReady(), events.waitUntilReady()]);
    await queue.setGlobalConcurrency(GLOBAL_CONCURRENCY);
    await redis.set(loadBudgetKey, loadInitialBudget.toString());
    await redis.set(raceBudgetKey, RACE_BUDGET.toString());
    await redis.del(barrierKey);
    const loadJobs = await addLoadJobs(queue, loadBudgetKey);
    workers.push(launchWorker(queueName, mode, "worker-a"), launchWorker(queueName, mode, "worker-b"));
    await Promise.all(workers.map((worker) => worker.ready));
    const loadResults = await waitForJobs(loadJobs, events);
    const raceResults = await waitForJobs(await addRaceJobs(queue, raceBudgetKey, barrierKey), events);

    const loadRemaining = Number((await redis.get(loadBudgetKey)) ?? "0");
    const raceRemaining = Number((await redis.get(raceBudgetKey)) ?? "0");
    const loadAdmitted = loadResults.filter((r) => r.reserved).reduce((sum, r) => sum + r.tokenCost, 0);
    const raceAdmitted = raceResults.filter((r) => r.reserved).reduce((sum, r) => sum + r.tokenCost, 0);
    const firstStart = Math.min(...loadResults.map((r) => r.startedAt));
    const lastEnd = Math.max(...loadResults.map((r) => r.completedAt));
    const maxConcurrent = maximumConcurrency(loadResults);
    const rateSpacing = minimumRateSpacing(loadResults);
    const workerIds = [...new Set(loadResults.map((r) => r.workerId))].sort();
    const raceAccepted = raceResults.filter((r) => r.reserved).length;
    const sharedConcurrencyCeilingHeld = workerIds.length === 2 && maxConcurrent <= GLOBAL_CONCURRENCY;
    const atomicBudgetNeverOverspent = mode !== "atomic" || (
      raceAdmitted <= RACE_BUDGET && raceRemaining >= 0 &&
      raceAccepted === 2 && RACE_JOBS - raceAccepted === 2
    );
    const raceOutcomeMatchedExpectation = mode === "naive"
      ? raceAccepted === 4 && raceRemaining === 100
      : raceAccepted === 2 && raceRemaining === 0;
    const checks = {
      usedTwoWorkers: workerIds.length === 2,
      priorityOrderEnforced: priorityWasEnforced(loadResults),
      sharedConcurrencyCeilingHeld,
      globalRateLimitObserved: rateSpacing >= 900,
      atomicBudgetNeverOverspent,
      raceOutcomeMatchedExpectation,
      allPassed: false,
    };
    checks.allPassed = Object.entries(checks)
      .filter(([name]) => name !== "allPassed")
      .every(([, passed]) => passed);

    return {
      mode,
      workload: {
        loadJobs: LOAD_JOBS, workerProcesses: 2,
        globalConcurrency: GLOBAL_CONCURRENCY,
        rateLimitMax: RATE_LIMIT_MAX, rateLimitDurationMs: RATE_LIMIT_DURATION_MS,
      },
      metrics: {
        throughputJobsPerSecond: round(LOAD_JOBS / Math.max((lastEnd - firstStart) / 1_000, 0.001)),
        maxConcurrent,
        minimumMsBetweenEveryFifthStart: rateSpacing,
        queueWaitMs: summarize(loadResults.map((r) => r.queueWaitMs)),
        processingMs: summarize(loadResults.map((r) => r.serviceMs)),
        endToEndMs: summarize(loadResults.map((r) => r.endToEndMs)),
        workerIds,
      },
      loadBudget: {
        initial: loadInitialBudget, remaining: loadRemaining, admittedTokens: loadAdmitted,
        admittedUtilizationPct: round((loadAdmitted / loadInitialBudget) * 100),
        ledgerUtilizationPct: round(((loadInitialBudget - loadRemaining) / loadInitialBudget) * 100),
      },
      raceBudget: {
        initial: RACE_BUDGET, remaining: raceRemaining,
        acceptedReservations: raceAccepted, rejectedReservations: RACE_JOBS - raceAccepted,
        admittedTokens: raceAdmitted,
        admittedUtilizationPct: round((raceAdmitted / RACE_BUDGET) * 100),
        ledgerUtilizationPct: round(((RACE_BUDGET - raceRemaining) / RACE_BUDGET) * 100),
      },
      checks,
    };
  } finally {
    await stopWorkers(workers);
    await events.close();
    await queue.obliterate({ force: true });
    await queue.close();
    await redis.del(loadBudgetKey, raceBudgetKey, barrierKey);
    await redis.quit();
  }
}

const requested = process.argv[2] ?? "compare";
if (!new Set(["naive", "atomic", "compare"]).has(requested)) throw new Error("Use naive, atomic, or compare");
const modes: Mode[] = requested === "compare" ? ["naive", "atomic"] : [requested as Mode];
const reports = [];
for (const mode of modes) reports.push(await runScenario(mode));

console.table(reports.map((report) => ({
  mode: report.mode,
  jobsPerSecond: report.metrics.throughputJobsPerSecond,
  queueWaitP50: report.metrics.queueWaitMs.p50,
  queueWaitP95: report.metrics.queueWaitMs.p95,
  queueWaitP99: report.metrics.queueWaitMs.p99,
  processingP50: report.metrics.processingMs.p50,
  processingP95: report.metrics.processingMs.p95,
  processingP99: report.metrics.processingMs.p99,
  endToEndP50: report.metrics.endToEndMs.p50,
  endToEndP95: report.metrics.endToEndMs.p95,
  endToEndP99: report.metrics.endToEndMs.p99,
  maxConcurrent: report.metrics.maxConcurrent,
  raceAccepted: report.raceBudget.acceptedReservations,
  raceRemaining: report.raceBudget.remaining,
  admittedBudgetPct: report.raceBudget.admittedUtilizationPct,
  ledgerBudgetPct: report.raceBudget.ledgerUtilizationPct,
  sharedConcurrencyCeilingHeld: report.checks.sharedConcurrencyCeilingHeld,
  atomicBudgetNeverOverspent: report.checks.atomicBudgetNeverOverspent,
  allChecksPassed: report.checks.allPassed,
})));
await writeFile("benchmark-results.json", `${JSON.stringify(reports, null, 2)}\n`);
console.log("Wrote benchmark-results.json");
if (reports.some((report) => !report.checks.allPassed)) process.exitCode = 1;

How to Use This Reference

Compare this complete file with your saved src/benchmark.ts from top to bottom. Check the imports and final output section first.

Before you run the final comparison, which race row do you expect to show 200 percent admitted utilization?

• Save src/benchmark.ts.
• Run both scenarios through the same benchmark with this command:

npm run benchmark

What Does the Comparison Run?

The command builds the project once. It then runs the naive scenario before running the atomic scenario.

Both modes retain the shared concurrency ceiling and queue-wide rate limiter. The finished report writes both scenario objects to benchmark-results.json.

You should see a two-row comparison table. Both rows should show sharedConcurrencyCeilingHeld as true.

The naive row should accept 4 race reservations. It should report 200 percent admitted utilization with 100 tokens remaining.

The atomic row should accept 2 race reservations. It should reject 2 reservations with 0 tokens remaining.

• Select benchmark-results.json from the VS Code Explorer sidebar.
• Confirm the array contains one naive report.
• Confirm the array contains one atomic report.
• Check that each report contains queueWaitMs percentiles.
• Check that each report contains processingMs percentiles.
• Check that each report contains endToEndMs percentiles.

That closes the correctness gap. Your benchmark now proves shared scheduling behavior while the atomic ledger stays inside its 200-token budget.

How Should You Read the Comparison?

The configured rate limiter dominates throughput and tail queue wait. This run validates shared coordination with budget correctness.

The Lua script introduces a short serialization point around each reservation. Redis keeps that critical section atomic.

What Changes When Jobs Retry?

A worker can crash after a successful reservation. A later retry could reserve the same job twice without an idempotency ledger.

A production ledger tracks reserve and commit states. It also tracks refund and reconciliation states for recovery.

Comparison Command Exits With a Failure?

• Check the table for the first correctness value that reports false.
• Confirm both worker IDs appear in each report.
• Verify the atomic race contains two accepted reservations.
• Verify the atomic race contains two rejected reservations.
• Help me diagnose a failed benchmark check.

Your dispatch laboratory now exposes the unsafe race and proves the atomic fix under the same distributed workload. The exported artifact preserves the evidence behind both conclusions.

Secret mission

React to Simulated Provider Throttling

A provider throttle should slow the whole queue without becoming an ordinary job failure. Add a deterministic first-attempt throttle that returns one atomic load job to waiting before it reserves budget.

Clean Up Your Resources

Clean Up Your Resources

All resources run locally. Leaving them in place creates no usage charge. Choose the option that matches what you want to keep.

Resources you used:

• The local Redis 8.10.2 container managed by Docker Compose through compose.yaml.
• The Compose-managed redis-data named volume that stores BullMQ queue state plus the benchmark's Redis data.
• The node_modules/ directory containing the pinned project dependencies.
• The dist/ directory containing compiled JavaScript.
• The generated benchmark-results.json report.

Keep everything running

No action is needed. Choose this option if you plan to run more dispatch experiments soon.

• The Redis container remains healthy for future benchmark runs.
• The redis-data volume retains its local Redis data.
• The benchmark-results.json report remains available for comparison.
• The node_modules/ directory keeps the pinned dependencies ready.
• The dist/ directory keeps the compiled benchmark ready.

Pause - I'll come back to this later

Stop the Redis container to free local resources while preserving its named volume.

• Stop the local Redis container from the folder containing compose.yaml by running:

docker compose down

What Does This Command Do?

Docker Compose stops the Redis container. It also removes the Compose network.

The named volume remains available. Your Redis data can return with the next container.

• Check Docker Desktop for the Compose application.

You should no longer see its Redis container running. The named volume remains stored locally.

Redis Still Running?

• Confirm the terminal is inside the folder containing compose.yaml.
• Confirm Docker Desktop is running when Compose cannot reach Docker.
• Ask for help with the shutdown command.

• Return to the Redis start command from Step 1 when you want to resume the benchmark.

Delete - I don't want to use this again

This cleanup permanently removes the Redis runtime data plus the generated benchmark artifacts. Your TypeScript source files remain available for a clean rebuild.

• Remove the Redis container plus its named volume from the folder containing compose.yaml by running:

docker compose down --volumes

What Gets Deleted?

Docker Compose removes the local Redis container. It also removes the Compose network.

The command deletes the named volume. This erases its BullMQ queue state plus its benchmark keys.

• Check Docker Desktop for the Compose application.

You should no longer see a running Redis container for this project. Its named volume should also be gone.

Redis Data Still Present?

• Confirm you included the volume option in the cleanup command.
• Confirm the terminal is inside the folder containing compose.yaml.
• Ask for help removing the Compose resources.

• Delete the generated local artifacts from the folder containing package.json by running:

rm -rf node_modules dist benchmark-results.json

What Does This Remove?

The node_modules/ directory holds downloaded dependencies. The dist/ directory holds compiled JavaScript.

The benchmark-results.json file is generated output. Your source files plus configuration files remain in place.

• Check the project file list in VS Code.

You should no longer see node_modules/ or dist/.

You should no longer see benchmark-results.json.

Generated Files Still Listed?

• Confirm the terminal is inside the folder containing package.json.
• Refresh the project file list in VS Code after the command finishes.
• Ask for help cleaning the generated files.

Cleanup complete. Your source files remain ready for a clean dependency install plus another benchmark run.

Nice Work!

Nice Work!

You made it! You built a typed LLM dispatch benchmark that sends simulated jobs through two independent BullMQ workers.

Your final comparison exposes a reproducible overspend in a shared Redis budget. It then proves atomic budget safety while preserving measurable throughput and latency.

You've learned how to:

• Coordinate two independent worker processes through one distributed queue. Enforce a shared concurrency ceiling across both workers. Apply a global rate limit to their combined workload.
• Measure throughput across the workload. Report p50/p95/p99 latency separately for queue wait, simulated processing, and end-to-end time.
• Reproduce a deterministic budget race. Replace the unsafe read-then-write gate with an atomic Lua reservation. Preserve the comparison in benchmark-results.json.
• Complete the Secret Mission by simulating provider throttling. Return the affected job to waiting with Queue.rateLimit() plus RateLimitError without recording an ordinary failure.

Ready to quiz yourself?