Tutorial: Build a Streaming Usage-Log Analyzer in Node.js

Analyze a 150 MB model-usage log in constant memory with Node.js streams, then serve a streaming report with backpressure, disconnect handling, and loop-delay metrics.

Intermediate22 min readUpdated October 18, 2026

On this page· What you will build

Why this matters

Every AI app produces usage logs: one line per model call, with tokens, latency, and status. They grow fast, and someone always asks "what did we spend last month, and which model is slow?". This tutorial answers that with plain Node.js, no dependencies, and puts every idea from the Node.js deep dive to work: streams, backpressure, pipeline(), cancelling work when a client leaves, and measuring the event loop to catch blocking code. You will see the difference in numbers, not just in theory.

What you will build#

A generator that writes a million usage records (about 150 MB) while respecting backpressure.
A report that summarizes the file in one pass and constant memory, compared against the naive version.
An export that filters failed calls into a new file with pipeline().
An HTTP server that streams the report as it runs, cancels when the client disconnects, and exposes event loop delay at /health.
Tests for the summarizer with node:test.

Hover or tap a component to see what it does.

Streaming usage-log analyzer: a large JSONL file flows through readline and a single-pass summarize() in constant memory, and an HTTP server streams the report with backpressure, cancels on disconnect, and exposes event loop delay at /health.

Show text description

Data moves through the program a chunk at a time; nothing holds the whole file, and every stage can slow down or stop the stage before it.

usage.jsonl: One JSON object per model call: model, tokens, latency, and status.
readline: Turns the file stream into an async iterable of lines, so only a few chunks are in memory at once.
summarize(): One pass over the lines; per-model counters and a 120-bucket latency histogram, however large the file.
HTTP server: Streams progress lines, waits for drain when the client is slow, and aborts the read when the client disconnects.
curl -N: Reads the report as it is produced.
/health: Reports the p99 and max event loop delay since the last check, so blocking work shows up as a number.

Report. The file is read, summarized, and streamed to the client as it is processed.

Health check. The health endpoint reports how late the event loop is running, which exposes blocking work.

You need Node.js 22.18 or later, curl, and about 300 MB of free disk space. Read the Node.js deep dive first.

1. Set up the project#

Create the project

mkdir usage-analyzer && cd usage-analyzer
npm init -y
npm pkg set type=module
npm install -D typescript @types/node
mkdir src test

There are no runtime dependencies: everything in this tutorial is built into Node.js.

Add the scripts

npm pkg set scripts.generate="node src/generate.ts" scripts.report="node src/report.ts" scripts.report:naive="node src/report-naive.ts" scripts.export="node src/export-errors.ts" scripts.serve="node --watch src/server.ts" scripts.test="node --test" scripts.typecheck="tsc"

Configure TypeScript
Use the same tsconfig.json as Tutorial 1: strict mode, noEmit, allowImportingTsExtensions, erasableSyntaxOnly, and verbatimModuleSyntax, so Node.js can run the files directly and tsc only checks them.

2. Generate a large log, with backpressure#

A write stream buffers data in memory until the disk accepts it. write() returns false when that buffer is full; if you keep writing anyway, the buffer grows without limit. Waiting for the drain event is what backpressure means in practice.

src/generate.ts

import { once } from "node:events";
import { createWriteStream } from "node:fs";
 
const count = Number(process.argv[2] ?? 1_000_000);
const MODELS = ["claude-opus-5-5", "claude-sonnet-5-5", "claude-haiku-5-5"] as const;
 
// Deterministic pseudo-random numbers (Park–Miller), so everyone generates the same file.
let seed = 42;
const random = () => (seed = (seed * 16_807) % 2_147_483_647) / 2_147_483_647;
const between = (min: number, max: number) => Math.floor(min + random() * (max - min));
 
const out = createWriteStream("usage.jsonl");
const start = Date.parse("2026-10-01T00:00:00Z");
 
for (let i = 0; i < count; i++) {
  const model = MODELS[between(0, MODELS.length)]!;
  const record = {
    ts: new Date(start + i * 1_000).toISOString(),
    user: `u_${between(1, 500)}`,
    model,
    input_tokens: between(200, 8_000),
    output_tokens: between(20, 1_500),
    latency_ms: between(300, model === "claude-opus-5-5" ? 9_000 : 4_000),
    status: random() < 0.02 ? "error" : "ok",
  };
  // write() returns false when the stream's buffer is full: wait for "drain" before writing more.
  if (!out.write(`${JSON.stringify(record)}\n`)) await once(out, "drain");
}
 
out.end();
await once(out, "finish");
console.log(`wrote ${count.toLocaleString("en-US")} records to usage.jsonl`);

npm run generate
# wrote 1,000,000 records to usage.jsonl

The file is about 148 MB, one JSON object per line:

usage.jsonl (first line)

{"ts":"2026-10-01T00:00:00.000Z","user":"u_262","model":"claude-opus-5-5","input_tokens":5936,"output_tokens":409,"latency_ms":3573,"status":"ok"}

3. Summarize in one pass#

The summarizer is the core of the program. Two decisions keep its memory constant however large the file is:

It accepts an iterable of lines and handles one at a time, so it never needs the whole file.
It stores latency as a histogram: 120 counters of 100 ms each, instead of a million numbers. Percentiles come from the histogram, accurate to one bucket.

src/summarize.ts

export interface UsageRecord {
  ts: string;
  user: string;
  model: string;
  input_tokens: number;
  output_tokens: number;
  latency_ms: number;
  status: "ok" | "error";
}
 
const BUCKET_MS = 100;
const BUCKETS = 120; // 0–12 s; anything slower lands in the last bucket
 
interface ModelStats {
  calls: number;
  errors: number;
  inputTokens: number;
  outputTokens: number;
  /** Latency histogram: a fixed number of counters, however many records there are. */
  latency: number[];
}
 
export interface Summary {
  lines: number;
  skipped: number;
  models: Map<string, ModelStats>;
}
 
function isUsageRecord(value: unknown): value is UsageRecord {
  const r = value as Partial<UsageRecord> | null;
  return (
    typeof r === "object" && r !== null &&
    typeof r.model === "string" &&
    typeof r.input_tokens === "number" &&
    typeof r.output_tokens === "number" &&
    typeof r.latency_ms === "number" &&
    (r.status === "ok" || r.status === "error")
  );
}
 
export interface SummarizeOptions {
  signal?: AbortSignal;
  /** Called every `progressEvery` lines; awaited, so a slow consumer slows the reader down. */
  onProgress?: (lines: number) => void | Promise<void>;
  progressEvery?: number;
}
 
/** Aggregate per-model usage from JSONL lines in a single pass, in constant memory. */
export async function summarize(lines: AsyncIterable<string> | Iterable<string>, options: SummarizeOptions = {}): Promise<Summary> {
  const { signal, onProgress, progressEvery = 100_000 } = options;
  const summary: Summary = { lines: 0, skipped: 0, models: new Map() };
 
  for await (const line of lines) {
    if (!line.trim()) continue;
    summary.lines++;
 
    let record: unknown;
    try {
      record = JSON.parse(line);
    } catch {
      record = null;
    }
    if (!isUsageRecord(record)) {
      summary.skipped++;
      continue;
    }
 
    let stats = summary.models.get(record.model);
    if (!stats) {
      stats = { calls: 0, errors: 0, inputTokens: 0, outputTokens: 0, latency: new Array<number>(BUCKETS).fill(0) };
      summary.models.set(record.model, stats);
    }
    stats.calls++;
    if (record.status === "error") stats.errors++;
    stats.inputTokens += record.input_tokens;
    stats.outputTokens += record.output_tokens;
    stats.latency[Math.min(BUCKETS - 1, Math.floor(record.latency_ms / BUCKET_MS))]!++;
 
    if (summary.lines % progressEvery === 0) {
      signal?.throwIfAborted();
      await onProgress?.(summary.lines);
    }
  }
  return summary;
}
 
/** Approximate percentile from a histogram: the upper edge of the bucket that contains it. */
export function percentile(histogram: number[], p: number): number {
  const total = histogram.reduce((a, b) => a + b, 0);
  let seen = 0;
  for (const [i, n] of histogram.entries()) {
    seen += n;
    if (seen >= total * (p / 100)) return (i + 1) * BUCKET_MS;
  }
  return histogram.length * BUCKET_MS;
}
 
export function formatReport(summary: Summary): string[] {
  const rows = [...summary.models].sort(([, a], [, b]) => b.calls - a.calls);
  const fmt = (n: number) => n.toLocaleString("en-US");
  return [
    `${fmt(summary.lines)} lines, ${fmt(summary.skipped)} skipped`,
    "",
    `${"model".padEnd(20)}${"calls".padStart(10)}${"errors".padStart(8)}${"input tok".padStart(16)}${"output tok".padStart(14)}${"p50".padStart(8)}${"p95".padStart(8)}`,
    ...rows.map(
      ([model, s]) =>
        `${model.padEnd(20)}${fmt(s.calls).padStart(10)}${`${((s.errors / s.calls) * 100).toFixed(1)}%`.padStart(8)}` +
        `${fmt(s.inputTokens).padStart(16)}${fmt(s.outputTokens).padStart(14)}` +
        `${`${percentile(s.latency, 50)}ms`.padStart(8)}${`${percentile(s.latency, 95)}ms`.padStart(8)}`,
    ),
  ];
}

A malformed line is counted as skipped instead of crashing the run: in a million-line log written by many processes, a truncated last line is normal. The signal and onProgress options are not used yet; the server needs them in step 6.

4. Stream the file, then compare with the naive version#

readline turns a file stream into an async iterable of lines. The file is read 64 KB at a time, and the next chunk is only read once the loop has consumed the previous lines.

src/report.ts

import { createReadStream } from "node:fs";
import { createInterface } from "node:readline";
import { formatReport, summarize } from "./summarize.ts";
 
const file = process.argv[2] ?? "usage.jsonl";
const started = performance.now();
 
// readline turns the byte stream into an async iterable of lines, reading 64 KB at a time.
const lines = createInterface({ input: createReadStream(file), crlfDelay: Infinity });
const summary = await summarize(lines);
 
console.log(formatReport(summary).join("\n"));
console.log(`\n${Math.round(performance.now() - started)} ms, peak memory ${Math.round(process.resourceUsage().maxRSS / 1024)} MB`);

Now write the version almost everyone writes first: read the whole file, split it, summarize the array.

src/report-naive.ts

import { readFile } from "node:fs/promises";
import { formatReport, summarize } from "./summarize.ts";
 
const file = process.argv[2] ?? "usage.jsonl";
const started = performance.now();
 
// The tempting version: load the whole file, split it, then summarize.
const lines = (await readFile(file, "utf8")).split("\n");
const summary = await summarize(lines);
 
console.log(formatReport(summary).join("\n"));
console.log(`\n${Math.round(performance.now() - started)} ms, peak memory ${Math.round(process.resourceUsage().maxRSS / 1024)} MB`);

Checkpoint. Run both:

npm run report

1,000,000 lines, 0 skipped
 
model                    calls  errors       input tok    output tok     p50     p95
claude-opus-5-5        333,454    2.0%   1,368,428,677   253,175,915  4700ms  8600ms
claude-sonnet-5-5      333,421    2.0%   1,369,240,748   253,438,362  2200ms  3900ms
claude-haiku-5-5       333,125    2.0%   1,364,487,636   252,912,848  2200ms  3900ms
 
1407 ms, peak memory 65 MB

npm run report:naive
# ...the same table...
# 1362 ms, peak memory 393 MB

Same results, about the same time, six times the memory. The naive version holds the whole 148 MB file as a string, then a million separate line strings on top. Run npm run generate -- 3000000 (a 444 MB file) and the gap widens: the streaming version peaks around 81 MB, the naive one above 1 GB. Memory for the naive version grows with the file; for the streaming one it barely moves. Exact numbers vary by machine; the shape does not.

process.resourceUsage().maxRSS is the peak resident memory of the process in kilobytes, which is the number that matters when a container has a memory limit.

5. Export failed calls with pipeline()#

When the report shows errors, the next question is "which ones?". pipeline() connects a source, any number of transform stages, and a destination. It applies backpressure between every stage and, if any stage fails, destroys all of them so no file handle is left open. An async generator is the simplest transform stage.

src/export-errors.ts

import { createReadStream, createWriteStream } from "node:fs";
import { createInterface } from "node:readline";
import { pipeline } from "node:stream/promises";
 
let exported = 0;
 
// A transform stage: lines in, only failed calls out.
async function* onlyErrors(lines: AsyncIterable<string>) {
  for await (const line of lines) {
    if (line.includes('"status":"error"')) {
      exported++;
      yield `${line}\n`;
    }
  }
}
 
await pipeline(
  createInterface({ input: createReadStream("usage.jsonl"), crlfDelay: Infinity }),
  onlyErrors,
  createWriteStream("errors.jsonl"),
);
console.log(`exported ${exported.toLocaleString("en-US")} failed calls to errors.jsonl`);

npm run export
# exported 19,957 failed calls to errors.jsonl

The filter checks the raw string instead of parsing JSON. It is safe here because the generator controls the format, and it is several times faster. On logs from many sources, parse the line instead.

6. Serve the report over HTTP#

The server streams progress while it reads, then the final table. It has three responsibilities beyond the report itself:

Backpressure toward the client. If res.write() returns false, wait for drain. Because onProgress is awaited inside summarize, a slow client slows the file reading down instead of filling memory.
Cancellation. When the client disconnects, the close event aborts the controller. The aborted signal destroys the file stream and stops summarize at its next progress check.
Observability. /health reports how late the event loop has been running since the previous check, which turns "the server feels slow" into a number.

src/server.ts

import { once } from "node:events";
import { createReadStream, readFileSync } from "node:fs";
import { createServer, type ServerResponse } from "node:http";
import { monitorEventLoopDelay } from "node:perf_hooks";
import { createInterface } from "node:readline";
import { formatReport, summarize } from "./summarize.ts";
 
const FILE = "usage.jsonl";
const PORT = 3001;
 
// Samples how late the event loop runs its timers. A healthy server stays in single-digit ms.
const loopDelay = monitorEventLoopDelay({ resolution: 10 });
loopDelay.enable();
 
/** Write a chunk, and wait if the client is reading slower than we produce. */
async function send(res: ServerResponse, chunk: string) {
  if (!res.write(chunk)) await once(res, "drain");
}
 
async function report(url: URL, res: ServerResponse) {
  const controller = new AbortController();
  res.on("close", () => controller.abort()); // fires when the response ends or the client disconnects
 
  res.writeHead(200, { "Content-Type": "text/plain; charset=utf-8", "Cache-Control": "no-cache" });
 
  // ?blocking=1 shows what not to do: read and split the whole file synchronously.
  const lines = url.searchParams.has("blocking")
    ? readFileSync(FILE, "utf8").split("\n")
    : createInterface({ input: createReadStream(FILE, { signal: controller.signal }), crlfDelay: Infinity });
 
  try {
    const summary = await summarize(lines, {
      signal: controller.signal,
      onProgress: (n) => send(res, `processed ${n.toLocaleString("en-US")} lines\n`),
    });
    await send(res, `\n${formatReport(summary).join("\n")}\n`);
    res.end();
  } catch (error) {
    if (controller.signal.aborted) {
      console.log("client disconnected: report cancelled");
      return;
    }
    console.error("report failed", error);
    res.end("\nreport failed\n");
  }
}
 
const server = createServer((req, res) => {
  const url = new URL(req.url ?? "/", `http://localhost:${PORT}`);
 
  if (url.pathname === "/health") {
    const ms = (ns: number) => Math.round(ns / 1e6);
    res.writeHead(200, { "Content-Type": "application/json" });
    res.end(JSON.stringify({ status: "ok", loopDelayMs: { p99: ms(loopDelay.percentile(99)), max: ms(loopDelay.max) } }));
    loopDelay.reset(); // each health check reports the period since the previous one
    return;
  }
  if (url.pathname === "/report") {
    void report(url, res);
    return;
  }
  res.writeHead(404, { "Content-Type": "text/plain" }).end("not found\n");
});
 
server.listen(PORT, () => console.log(`listening on http://localhost:${PORT}  (try: curl -N http://localhost:${PORT}/report)`));

void report(url, res) marks a promise that is deliberately not awaited; report handles its own errors, so nothing is left as an unhandled rejection.

Checkpoint. Start the server with npm run serve, then in a second terminal:

curl -N http://localhost:3001/report

processed 100,000 lines
processed 200,000 lines
...
processed 1,000,000 lines
 
1,000,000 lines, 0 skipped
 
model                    calls  errors       input tok    output tok     p50     p95
claude-opus-5-5        333,454    2.0%   1,368,428,677   253,175,915  4700ms  8600ms
...

The progress lines arrive while the file is being read, not all at once at the end. Press Ctrl+C during a run and the server logs client disconnected: report cancelled: the work stops the moment nobody is waiting for it. With a model stream instead of a file, that is the difference between paying for tokens and not.

7. Catch blocking code in the act#

Now measure what the deep dive described. Run /health in a loop in one terminal:

while true; do curl -s -w "  (%{time_total}s)\n" http://localhost:3001/health; sleep 0.3; done

while ($true) { (Measure-Command { curl.exe -s http://localhost:3001/health | Write-Host }).TotalSeconds; Start-Sleep -Milliseconds 300 }

In another terminal, run the streaming report, then the blocking one:

curl -sN http://localhost:3001/report > /dev/null
curl -sN "http://localhost:3001/report?blocking=1" > /dev/null

During the streaming report, health checks keep answering in milliseconds:

{"status":"ok","loopDelayMs":{"p99":11,"max":11}}  (0.019s)
{"status":"ok","loopDelayMs":{"p99":11,"max":11}}  (0.022s)

During the blocking report, one health check waits more than a second, and the loop delay shows why:

{"status":"ok","loopDelayMs":{"p99":1577,"max":1577}}  (1.207s)

Both versions do the same work. The streaming one hands control back to the event loop after every 64 KB chunk, so other requests run in between. readFileSync and the loop over an in-memory array never give it back: for over a second, every user of this server waits. The baseline of about 10 ms comes from the monitor's own 10 ms sampling resolution and the operating system's timer granularity; it is the spikes that matter.

8. Test the summarizer#

test/summarize.test.ts

import assert from "node:assert/strict";
import { test } from "node:test";
import { percentile, summarize } from "../src/summarize.ts";
 
const line = (model: string, latency_ms: number, status = "ok") =>
  JSON.stringify({ ts: "2026-10-01T00:00:00Z", user: "u_1", model, input_tokens: 100, output_tokens: 10, latency_ms, status });
 
test("aggregates calls, errors, and tokens per model", async () => {
  const summary = await summarize([line("a", 500), line("a", 700, "error"), line("b", 300)]);
  const a = summary.models.get("a")!;
  assert.equal(summary.lines, 3);
  assert.deepEqual([a.calls, a.errors, a.inputTokens, a.outputTokens], [2, 1, 200, 20]);
  assert.equal(summary.models.get("b")!.calls, 1);
});
 
test("skips malformed lines instead of failing the whole run", async () => {
  const summary = await summarize([line("a", 500), "{not json", JSON.stringify({ model: "a" }), ""]);
  assert.equal(summary.lines, 3); // the empty line is ignored entirely
  assert.equal(summary.skipped, 2);
});
 
test("works on an async iterable, the shape readline produces", async () => {
  async function* lines() {
    yield line("a", 100);
    yield line("a", 200);
  }
  assert.equal((await summarize(lines())).models.get("a")!.calls, 2);
});
 
test("percentile returns the upper edge of the matching 100 ms bucket", () => {
  const histogram = [0, 0, 0, 0, 0, 90, 0, 0, 0, 10]; // 90 calls at 500–599 ms, 10 at 900–999 ms
  assert.equal(percentile(histogram, 50), 600);
  assert.equal(percentile(histogram, 95), 1000);
});
 
test("stops when the signal is aborted", async () => {
  const controller = new AbortController();
  controller.abort();
  const lines = Array.from({ length: 10 }, () => line("a", 100));
  await assert.rejects(summarize(lines, { signal: controller.signal, progressEvery: 5 }), { name: "AbortError" });
});

npm test          # 5 passing
npm run typecheck

Add usage.jsonl and errors.jsonl to .gitignore: generated data does not belong in the repository.

Common mistakes#

Ignoring the return value of write(). The stream accepts everything and buffers it in memory. Writing a large file this way can use as much memory as the file itself.
Storing every value to compute percentiles. A million latencies in an array is exactly the memory growth streaming was meant to avoid. Use a histogram (or a sketch such as t-digest) when approximate percentiles are enough.
Not listening for close on the response. The server keeps reading, computing, and, with a model stream, paying, long after the client has gone.
Using readFileSync "because the file is small". Files grow. In a request handler, any synchronous I/O blocks every concurrent request for its full duration.
Measuring blocking with request timings alone. A slow response could be slow I/O. Event loop delay tells you specifically that the main thread was busy.

Try it

Extend the analyzer

Add per-user totals to the summary and a /top-users endpoint that returns the ten users with the most output tokens as JSON. What happens to memory as the number of users grows, and when would that stop being acceptable?
Estimate cost: add a price table per model (dollars per million input and output tokens) and print a cost column.
Accept a ?model=claude-haiku-5-5 query parameter on /report that only counts that model, and validate it against the known model names (return 400 otherwise).
Move the blocking version into a worker_threads worker and confirm /health stays flat while it runs.

Check your understanding

Key takeaways

Process large data as a stream of lines and keep only aggregates, such as counters and histograms, to stay in constant memory.
Respect backpressure: check write()'s return value and wait for drain, and await slow consumers inside processing loops.
pipeline() connects stages with backpressure and cleans every stage up on failure; async generators make good transform stages.
Abort work when the client disconnects; for model streams, that is money saved.
Measure event loop delay to find blocking code; synchronous I/O in a request handler makes every user wait.

Next steps#

React and Next.js

Next article: Server Components, Client Components, Server Actions, and Route Handlers.

Cost engineering and model routing

Turn usage numbers like these into cost decisions.

Finished this article? Track it to see your progress on the roadmap.