---
title: 'Caching, Queues and Asynchronous Processing'
source: 'https://academia.sh/en/courses/asynchronous-processing'
language: en
updated: '2026-08-23T07:00:21+00:00'
license: 'CC BY-SA 4.0'
lessons: 25
outcomes:
    - 'Selects cache layers and invalidation strategies'
    - 'Explains the difference between a message queue and a stream, and where each one is used'
    - 'Builds background jobs that stay durable through retries and a dead-letter queue'
---

# Caching, Queues and Asynchronous Processing

Shortening the request path: cache layers and invalidation strategies, the delivery and ordering guarantees that message queues and streams provide, background workers with backpressure and load shedding, and the routes that carry real-time data from the server to the client.

## Caching

Preventing the same response from being reproduced: cache layers stretching from the client to the database, cache-aside and write strategies, time-based, event-based, and version-based invalidation, key design carrying namespace and tenant separation, preventing the stampede effect with single-flight, HTTP validators, and measuring the hit ratio.

### Cache Layers

Source: https://academia.sh/en/courses/asynchronous-processing/cache-layers

Measuring a read path that produces the same response over and over, and setting up client, edge, server, and database caches as a chain: how many requests each layer stops, the split between shared and private caches, and lowering the rate of requests that reach the origin.

The Data Access Layer and Business Logic course built the full path from a request down
to the data: the repository pattern, transaction boundaries, query budgets, domain rules.
One assumption remained at the end of the course: every request reaches the origin of the
data. That assumption does not always hold. In the library system, a book's detail page is
opened thousands of times a day, and the book's title and author have not changed in
months. If the same query produces the same response, storing the response is cheaper
than repeating the query.

This lesson first establishes a measure of that repetition, then lists the places where a
response can be stored. Each storage location is a **cache** layer, and the layers form a
chain: the earlier a request stops in the chain, the more cheaply it is served.

#### Measured Repetition

The measurements run against the library schema used throughout the course: the branch,
book, member, and loan relations. The block below builds the database from scratch.

```sh
# setup.sh — sets up the library database (same columns as the M16/K04 schema)
rm -f library.db
sqlite3 library.db <<'SQL'
CREATE TABLE branch (branch_id INTEGER PRIMARY KEY, name TEXT NOT NULL, city TEXT NOT NULL);
CREATE TABLE book (book_id INTEGER PRIMARY KEY, title TEXT NOT NULL, author TEXT NOT NULL,
                    publication_year INTEGER, branch_id INTEGER REFERENCES branch(branch_id));
CREATE TABLE member (member_id INTEGER PRIMARY KEY, first_name TEXT NOT NULL, last_name TEXT NOT NULL,
                  email TEXT, registered_at TEXT NOT NULL);
CREATE TABLE loan (loan_id INTEGER PRIMARY KEY,
                    book_id INTEGER NOT NULL REFERENCES book(book_id),
                    member_id INTEGER NOT NULL REFERENCES member(member_id),
                    pickup_date TEXT NOT NULL, return_date TEXT);
INSERT INTO branch VALUES (1,'Central','Ankara'),(2,'Bahcelievler','Ankara'),
  (3,'Kadikoy','Istanbul'),(4,'Konak','Izmir');
INSERT INTO book SELECT n, 'Book ' || n, 'Author ' || (n % 40 + 1), 1950 + n % 70, n % 4 + 1
  FROM (WITH RECURSIVE s(n) AS (SELECT 1 UNION ALL SELECT n+1 FROM s WHERE n < 300) SELECT n FROM s);
INSERT INTO member SELECT n, 'Member' || n, 'Surname' || n, 'member' || n || '@example.test', '2024-01-01'
  FROM (WITH RECURSIVE s(n) AS (SELECT 1 UNION ALL SELECT n+1 FROM s WHERE n < 200) SELECT n FROM s);
INSERT INTO loan SELECT n, n % 300 + 1, n % 200 + 1,
  '2025-' || substr('0' || (n % 12 + 1), -2) || '-' || substr('0' || (n % 28 + 1), -2),
  CASE WHEN n % 5 = 0 THEN NULL ELSE '2025-12-31' END
  FROM (WITH RECURSIVE s(n) AS (SELECT 1 UNION ALL SELECT n+1 FROM s WHERE n < 4000) SELECT n FROM s);
CREATE INDEX loan_member ON loan(member_id);
CREATE INDEX book_branch ON book(branch_id);
SQL
sqlite3 library.db "SELECT 'branch=' || (SELECT count(*) FROM branch) || ' book=' || (SELECT count(*) FROM book) || ' member=' || (SELECT count(*) FROM member) || ' loan=' || (SELECT count(*) FROM loan);"
```

```
branch=4 book=300 member=200 loan=4000
```

The origin exposes three reads: book detail, branch shelf count, and a member's loan
history. Every read is counted — the same approach as the query counter in the
Performance Problems topic.

```js
// origin.mjs — the origin of the library data; every read is counted
import { DatabaseSync } from "node:sqlite";

export function openOrigin(file = "library.db") {
  const db = new DatabaseSync(file);
  const counter = { queries: 0 };
  const single = (sql, ...d) => { counter.queries += 1; return db.prepare(sql).get(...d); };
  return {
    counter,
    bookDetail: (id) => single(
      "SELECT book_id, title, author, publication_year, branch_id FROM book WHERE book_id = ?", id),
    branchStock: (branchId) => single(
      `SELECT ? AS branch_id, count(*) AS on_shelf FROM book b
       WHERE b.branch_id = ? AND NOT EXISTS
         (SELECT 1 FROM loan l WHERE l.book_id = b.book_id AND l.return_date IS NULL)`,
      branchId, branchId),
    memberHistory: (memberId) => single(
      "SELECT ? AS member_id, count(*) AS total FROM loan WHERE member_id = ?", memberId, memberId),
    close: () => db.close(),
  };
}

// Request stream: book detail requests concentrate on a handful of titles (deterministic generator).
export function requestStream(count) {
  let seed = 20250729;
  const random = () => ((seed = (seed * 1103515245 + 12345) % 2147483648) / 2147483648);
  const requests = [];
  for (let i = 0; i < count; i++) {
    const r = random();
    if (r < 0.7) requests.push(["book", Math.floor(random() * 20) + 1]);
    else if (r < 0.9) requests.push(["branch", Math.floor(random() * 4) + 1]);
    else requests.push(["member", Math.floor(random() * 50) + 1]);
  }
  return requests;
}
```

The number that results when a stream of six hundred requests is applied directly to the
origin is the measure of the repetition.

```js
// no-cache.mjs — every request goes to the origin: query count equals request count
import { openOrigin, requestStream } from "./origin.mjs";

const origin = openOrigin();
const requests = requestStream(600);
const reader = { book: origin.bookDetail, branch: origin.branchStock, member: origin.memberHistory };

const t = performance.now();
for (const [kind, id] of requests) reader[kind](id);
const duration = performance.now() - t;

const unique = new Set(requests.map(([kind, id]) => `${kind}:${id}`)).size;
console.log(`requests=${requests.length}  unique_keys=${unique}  origin_queries=${origin.counter.queries}`);
console.log(`repeat queries producing the same response = ${origin.counter.queries - unique}`);
console.log(`duration=${duration.toFixed(1)} ms (machine-dependent)`);
origin.close();
```

{/* mask: (?<=duration=)[0-9.]+(?= ms) */}
```
requests=600  unique_keys=53  origin_queries=600
repeat queries producing the same response = 547
duration=545.3 ms (machine-dependent)
```

Six hundred requests produced only 53 distinct responses. The remaining 547 queries ran
to reproduce a result the database had already produced. The duration field is
machine-dependent and varies on every run; the numbers 600 and 53 do not change, because
they are properties of the workload.

The opportunity here is not speeding up the query. The query is already indexed and its
plan is sound. The opportunity is **not running the query at all**.

#### The Four Layers in the Chain

A response can be stored at four points between the origin and the user.

The **client cache** sits on the user's own device: the browser's HTTP cache or the
application's local storage. Because it never reaches the network, it is the cheapest
layer. In exchange, it serves only a single user, and its content is out of the server's
control — the copy placed there stays until the lifetime it was given at write time runs
out.

The **edge cache** is a shared intermediate point geographically close to the user. It
was introduced for static files in the Static Hosting lesson of the Frontend Quality
course; the same layer also works for API responses. Because it is shared, a copy warmed
by one user is used by others. For the same reason, it cannot carry data private to one
person: a member's history placed there could be served to a different member.

The **server cache** is under the application's own control. It comes in two forms:
in-process memory and a separate store shared by every instance. The in-process form is
the fastest, but each instance keeps its own copy; the shared form costs a round trip but
gives a single point of truth. This layer lives in the application's code, so it can
apply an invalidation decision instantly.

The **database cache** is inside the origin itself: the buffer pool and the query plan
cache. This layer was already running in the measurement above — all 600 queries ran
against warm pages. The result was still 600 queries. The database cache lowers a
query's **cost**, not the **number** of queries. The connection, parsing, plan
selection, and round trip are paid again on every call.

#### Measuring the Chain

Setting up the four layers as a chain and counting how many requests each layer stops
turns the layer discussion into a number. The setup below implements three cache layers;
the fourth is inside the origin. Each layer has a capacity: once it fills, the oldest
entry is evicted.

```js
// layers.mjs — the four-layer chain: where does each request stop?
import { openOrigin } from "./origin.mjs";

// A small cache that evicts the oldest entry once it is at capacity.
function smallCache(capacity) {
  const box = new Map();
  return {
    get: (k) => box.get(k),
    has: (k) => box.has(k),
    set(k, v) {
      box.delete(k); box.set(k, v);
      if (box.size > capacity) box.delete(box.keys().next().value);
    },
  };
}

const origin = openOrigin();
const reader = { book: origin.bookDetail, branch: origin.branchStock, member: origin.memberHistory };
const SHARED = new Set(["book", "branch"]);          // resources that return the same thing to every member

const clients = new Map();                          // one cache per user, 4 entries
const edge = smallCache(10);                         // shared, shared resources only
const server = smallCache(200);                      // shared, everything
const counter = { client: 0, edge: 0, server: 0, origin: 0 };

function read(user, kind, id) {
  const key = `${kind}:${id}`;
  if (!clients.has(user)) clients.set(user, smallCache(4));
  const client = clients.get(user);
  const place = (v) => {
    if (SHARED.has(kind)) edge.set(key, v);
    client.set(key, v);
    return v;
  };
  if (client.has(key)) { counter.client += 1; return client.get(key); }
  if (SHARED.has(kind) && edge.has(key)) { counter.edge += 1; return place(edge.get(key)); }
  const serverKey = SHARED.has(kind) ? key : `${key}@${user}`;
  if (server.has(serverKey)) { counter.server += 1; return place(server.get(serverKey)); }
  counter.origin += 1;
  const value = reader[kind](id);
  server.set(serverKey, value);
  return place(value);
}

let seed = 20250729;
const random = () => ((seed = (seed * 1103515245 + 12345) % 2147483648) / 2147483648);
const REQUESTS = 600;
for (let i = 0; i < REQUESTS; i++) {
  const user = Math.floor(random() * 40) + 1;
  const r = random();
  const kind = r < 0.7 ? "book" : r < 0.9 ? "branch" : "member";
  const id = kind === "book" ? Math.floor(random() * 20) + 1
    : kind === "branch" ? Math.floor(random() * 4) + 1 : user;
  read(user, kind, id);
}

console.log(`requests=${REQUESTS}`);
for (const [name, n] of Object.entries(counter)) {
  console.log(`${name.padEnd(8)} ${String(n).padStart(4)}  (%${((n / REQUESTS) * 100).toFixed(1)})`);
}
console.log(`origin queries = ${origin.counter.queries}`);
origin.close();
```

```
requests=600
client     98  (%16.3)
edge      189  (%31.5)
server    260  (%43.3)
origin     53  (%8.8)
origin queries = 53
```

The number of requests reaching the origin fell from 600 to 53; every distinct key in
the workload hit the origin exactly once. The rest of the chain is the interesting part.
The closest layer stopped only 16% of requests, because its capacity is limited to four
entries and each user's copy is separate. The shared edge layer stopped 31% of them. The
server layer did the most work, at 43%: it is the layer with the largest capacity and
the broadest scope.

The fact that these numbers depend on capacity is the real lesson of this measurement.
If the edge layer's capacity were increased, its share would grow and the server's would
shrink. The layers do not replace one another; each one picks up what the others leave
behind.

#### Which Layer a Piece of Data Belongs In

The layer is chosen using two criteria.

The first is **sharing scope**. If a response returns the same thing to every user, it
can be placed in shared layers. Book detail and branch shelf count are like this. A
member's loan history is not; placing it in the edge cache means breaching the
authorization boundary examined in the Authentication and Authorization course. In the
measurement code, this distinction is made with the `SHARED` set, and records private to
one person are keyed with the user's identity in the server layer.

The second is **staleness tolerance**. The cached copy stays out of date for a while
once the origin changes. For a book's author, that period can be hours; for a branch's
shelf count, seconds. For the loan transaction itself, it is zero — the result of a
write cannot be read from a cache.

A layer's closeness to the user is inversely proportional to how controllable it is. An
entry in the server layer can be deleted by code instantly. Deleting one in the edge
layer takes a separate operation. For the copy on the client, there is no way at all: it
is used until the lifetime given at write time runs out. For that reason, the data
placed in the closest layer is the data given the longest lifetime, and that decision is
made knowing it cannot be reversed.

#### Summary

- A workload of six hundred requests produced only 53 distinct responses; 547 queries
  reproduced a result already produced before. The opportunity in caching is not
  speeding up the query but not running it at all.
- A response can be stored at four points: the client, edge, server, and database
  caches. The first three lower the query count; the database cache only lowers the
  cost of each query.
- When the chain was measured, requests reaching the origin fell from 600 to 53; the
  layers' shares were distributed according to their capacities. The layers do not
  replace one another — they pick up what the others leave behind.
- Only data that returns the same thing to every user belongs in shared layers; data
  private to one person breaches the authorization boundary in a shared cache.
- A layer gets cheaper and less controllable as it moves closer to the user; the copy
  on the client cannot be revoked until the lifetime given at write time runs out.

#### Next Step

In this lesson, the cache was placed by hand inside the read path: look it up, fetch it
from the origin if it is missing, store it. This pattern has a name, and it is not the
only option. A cache that sits beside the read path and one that every read passes
through set up different divisions of responsibility; on the write path, the question of
when the cache gets updated arises — at write time, or later. The next lesson runs these
three strategies against the same workload and compares the number of reads and writes
reaching the origin, along with the inconsistency window.

### Cache Strategies

Source: https://academia.sh/en/courses/asynchronous-processing/cache-strategies

Comparing cache-aside, write-through, and write-behind on the same workload: the number of reads and writes reaching the origin, the hit ratio, the durability trade-off of write-behind, and the race between a miss and a write in cache-aside.

The previous lesson placed the cache by hand inside the read path: look it up, fetch it
from the origin if it is missing, store it. This pattern has a name, and it is not the
only option. A cache can sit beside the read path, or it can be a layer that every read
passes through. On the write path there is a separate question: when the origin is
updated, what happens to the cached copy — is it updated at the same time, is it
deleted, or does the write go to the cache first?

This lesson runs three strategies against the same workload: **cache-aside**,
**write-through**, and **write-behind**. What gets measured is the number of reads and
writes reaching the origin, and the number of records where the origin actually falls
behind.

#### Measurement Setup

The measurement runs on the shelf-copy count: issuing a loan lowers this number, and a
return raises it. It is the most frequently written field in the library data, which
makes it well suited to distinguishing between write strategies. A column holding this
count is added to the book table from the previous lesson.

```sh
# setup.sh — the lesson's database: a shelf-copy count column is added to the book table
rm -f library.db
sqlite3 library.db <<'SQL'
CREATE TABLE branch (branch_id INTEGER PRIMARY KEY, name TEXT NOT NULL, city TEXT NOT NULL);
CREATE TABLE book (book_id INTEGER PRIMARY KEY, title TEXT NOT NULL, author TEXT NOT NULL,
                    branch_id INTEGER REFERENCES branch(branch_id), on_shelf INTEGER NOT NULL);
INSERT INTO branch VALUES (1,'Central','Ankara'),(2,'Bahcelievler','Ankara'),(3,'Kadikoy','Istanbul');
INSERT INTO book SELECT n, 'Book ' || n, 'Author ' || (n % 10 + 1), n % 3 + 1, 3
  FROM (WITH RECURSIVE s(n) AS (SELECT 1 UNION ALL SELECT n+1 FROM s WHERE n < 60) SELECT n FROM s);
SQL
sqlite3 library.db "SELECT 'book=' || count(*) || ' total_on_shelf=' || sum(on_shelf) FROM book;"
```

```
book=60 total_on_shelf=180
```

The workload is 1200 operations: 85% reads, 15% writes, concentrated on twelve popular
books. Each strategy runs against its own copy of the database, so the comparison starts
from the same initial state.

#### Three Strategies

All three setups expose the same interface: `read` and `write`. The difference between
them lies only in the bodies of these two functions.

```js
// strategies.mjs — same workload across three strategies: origin access and inconsistency are measured
import { DatabaseSync } from "node:sqlite";
import { copyFileSync, rmSync } from "node:fs";

// Same workload: 85% reads, 15% writes; requests concentrate on the first 12 books.
function workload(count) {
  let seed = 20250729;
  const random = () => ((seed = (seed * 1103515245 + 12345) % 2147483648) / 2147483648);
  return Array.from({ length: count }, () => {
    const isWrite = random() < 0.15;
    return [isWrite ? "write" : "read", Math.floor(random() * 12) + 1, isWrite ? (random() < 0.5 ? -1 : 1) : 0];
  });
}

function openOrigin(file) {
  rmSync(file, { force: true });
  copyFileSync("library.db", file);
  const db = new DatabaseSync(file);
  const s = { reads: 0, writes: 0 };
  return {
    s,
    read: (id) => { s.reads += 1; return db.prepare("SELECT on_shelf FROM book WHERE book_id = ?").get(id).on_shelf; },
    write: (id, d) => { s.writes += 1; db.prepare("UPDATE book SET on_shelf = ? WHERE book_id = ?").run(d, id); },
    check: (id) => db.prepare("SELECT on_shelf FROM book WHERE book_id = ?").get(id).on_shelf,
    close: () => db.close(),
  };
}

function run(name, setup) {
  const origin = openOrigin(`${name}.db`);
  const cache = new Map();
  const truth = new Map();                       // the workload's logical ground truth
  const m = { hit: 0, miss: 0, stale: 0 };
  const { read, write, finish } = setup(origin, cache, m);
  for (const [kind, id, delta] of workload(1200)) {
    if (kind === "read") {
      const d = read(id);
      if (!truth.has(id)) truth.set(id, origin.check(id));
      if (d !== truth.get(id)) m.stale += 1;
    } else {
      if (!truth.has(id)) truth.set(id, origin.check(id));
      const next = Math.max(0, truth.get(id) + delta);
      truth.set(id, next);
      write(id, next);
    }
  }
  const pending = finish ? finish() : 0;
  const diverged = [...truth.keys()].filter((id) => origin.check(id) !== truth.get(id)).length;
  origin.close();
  rmSync(`${name}.db`, { force: true });
  return { name, ...origin.s, ...m, pending, diverged };
}

const cacheAside = (o, c, m) => ({
  read(id) {
    if (c.has(id)) { m.hit += 1; return c.get(id); }
    m.miss += 1; const d = o.read(id); c.set(id, d); return d;
  },
  write(id, d) { o.write(id, d); c.delete(id); },     // entry is deleted after the write
});

const writeThrough = (o, c, m) => ({
  read(id) {
    if (c.has(id)) { m.hit += 1; return c.get(id); }
    m.miss += 1; const d = o.read(id); c.set(id, d); return d;
  },
  write(id, d) { o.write(id, d); c.set(id, d); },     // origin and cache are updated together
});

const writeBehind = (o, c, m) => {
  const pending = new Map();
  let count = 0;
  return {
    read(id) {
      if (c.has(id)) { m.hit += 1; return c.get(id); }
      m.miss += 1; const d = o.read(id); c.set(id, d); return d;
    },
    write(id, d) {                                  // write goes to the cache first, the origin later
      c.set(id, d); pending.set(id, d);
      if ((count += 1) % 20 === 0) { for (const [i, v] of pending) o.write(i, v); pending.clear(); }
    },
    finish: () => pending.size,                   // writes that would be lost if the process stopped here
  };
};

const headers = ["strategy", "org.reads", "org.writes", "hits", "misses", "stale", "pending", "diverged"];
console.log(headers.map((h, i) => (i === 0 ? h.padEnd(16) : h.padStart(11))).join(""));
for (const [name, setup] of [["cache-aside", cacheAside], ["write-through", writeThrough], ["write-behind", writeBehind]]) {
  const r = run(name.replace(/ /g, "-"), setup);
  console.log(name.padEnd(16) + [r.reads, r.writes, r.hit, r.miss, r.stale, r.pending, r.diverged]
    .map((n) => String(n).padStart(11)).join(""));
}
```

```
strategy          org.reads org.writes       hits     misses      stale    pending   diverged
cache-aside             156        175        869        156          0          0          0
write-through             7        175       1018          7          0          0          0
write-behind              7         74       1018          7          0         10          5
```

#### What the Numbers Say

**Cache-aside** made 156 origin reads. There are twelve distinct keys; the remaining 144
misses came from the entry being deleted after every write. Every deleted entry gets
reloaded on the next read. This is the strategy's cost: as write frequency rises, so
does the miss count.

**Write-through** made 7 origin reads — just one for each key seen for the first time.
Because a write updates the cached entry with the new value instead of deleting it, the
entry never dropped out. The hit count rose from 869 to 1018. In exchange, the cache
starts carrying values that may never be read again: a rarely read book's shelf count
enters the cache on every write and takes up space.

**Write-behind** lowered the origin write count from 175 to 74. Writes went to the cache
first, were flushed to the origin in a batch every twenty writes, and consecutive writes
to the same book merged into a single update. It is the only strategy that cuts the
write load on the origin to less than half.

The last two columns show the price of this gain. When the workload ended, 10 writes
were still pending in the cache, and 5 records at the origin had diverged from their
true value. Had the process stopped at that moment, those writes would have been lost —
the cache rests on memory, the origin on disk. Write-behind pays with **durability** to
lower the write load. In the library example, one copy of the shelf count can be lost;
the same strategy cannot be used for the loan record itself.

It is worth noting that the `stale` column is zero on every row. None of the three
strategies gave the reader a wrong value, because all of them reflected the write in the
cache. The inconsistency did not occur at the reader; it occurred between the origin and
the cache, and it showed up in the `diverged` column.

#### Cache-Aside's Race

In the run above, operations executed one after another. On a real server, two requests
advance at the same time, and a known gap in cache-aside surfaces: a miss load can
collide with a write. The block below builds this ordering step by step.

```js
// cache-aside-race.mjs — cache-aside's classic race: a miss load collides with a write
import { DatabaseSync } from "node:sqlite";
import { copyFileSync, rmSync } from "node:fs";

rmSync("race.db", { force: true });
copyFileSync("library.db", "race.db");
const db = new DatabaseSync("race.db");
const originRead = (id) => db.prepare("SELECT on_shelf FROM book WHERE book_id = ?").get(id).on_shelf;
const originWrite = (id, d) => db.prepare("UPDATE book SET on_shelf = ? WHERE book_id = ?").run(d, id);
const cache = new Map();
const log = (step, text) => console.log(`${String(step).padStart(2)}. ${text.padEnd(54)} ` +
  `cache=${cache.has(7) ? cache.get(7) : "-"} origin=${originRead(7)}`);

// Request A: it is reading, took a miss, and read the origin; it has not stored it in the cache yet.
const aRead = originRead(7);
log(1, "A: miss, read from the origin (value in hand)");

// Request B completes in between: one copy was checked out.
originWrite(7, aRead - 1);
cache.delete(7);
log(2, "B: updated the origin, deleted the cache entry");

// Request A now stores the stale value it is holding in the cache.
cache.set(7, aRead);
log(3, "A: stored the stale value it was holding in the cache");

let stale = 0;
for (let i = 0; i < 50; i++) {
  const d = cache.has(7) ? cache.get(7) : originRead(7);
  if (d !== originRead(7)) stale += 1;
}
log(4, `stale among the next 50 reads: ${stale}`);
db.close();
rmSync("race.db", { force: true });
```

```
 1. A: miss, read from the origin (value in hand)          cache=- origin=3
 2. B: updated the origin, deleted the cache entry         cache=- origin=2
 3. A: stored the stale value it was holding in the cache  cache=3 origin=2
 4. stale among the next 50 reads: 50                      cache=3 origin=2
```

The write rule was followed to the letter — the origin was updated, the entry was
deleted — and the cache still ended up stale. The flaw is in the ordering: a delete
cannot delete an entry that has not been stored yet. Staleness does not correct itself;
the wrong value is served until the entry is written again or its lifetime expires.

This is proof that cache-aside is never sufficient on its own. It is paired with at
least a time limit; when the ordering must be resolved definitively, a version stamp is
added to the entry, or loading the same key is collapsed into a single call. All three
are the subject of later lessons.

#### Selection Criteria

The three strategies are not interchangeable options; each fits a different read–write
ratio.

For read-heavy, rarely changing data, **cache-aside** is enough; the code only needs to
know the read path, and the application keeps running if the cache goes down. Where the
same key is both read and written often, **write-through** avoids misses, in exchange
for filling the cache with values that are never read. Where the write rate exceeds what
the origin can bear and the loss of individual writes is tolerable, **write-behind**
lowers the load; where the loss is not tolerable, this strategy is ruled out.

There is also a fourth option, in which the write never touches the cache at all:
**write-around**. The origin is updated; the cache entry is neither deleted nor written,
so it stays stale until it expires. It is used for data written once and not read again
for a long time, to avoid polluting the cache, but it means the staleness window is
accepted outright.

#### Summary

- Cache-aside deletes the entry on a write; across the 1200-operation workload it
  produced 156 origin reads, 144 of which were reloads of deleted entries.
- Write-through updates the entry instead of deleting it, which brought origin reads
  down to 7 and raised the hit count to 1018; in exchange, it also keeps values in the
  cache that are never read.
- Write-behind lowered origin writes from 175 to 74, because consecutive writes to the
  same key merged in the batch flush; its cost is 10 pending writes that would be lost
  if the process stopped.
- In all three strategies, the reader never saw a stale value; the inconsistency
  occurred between the origin and the cache. Write-behind cannot be used for data that
  requires durability.
- When a miss load collides with a write in cache-aside, the delete has no effect, and
  the stale entry does not correct itself; the strategy is not sufficient on its own.

#### Next Step

The last measurement left a question open: when does a stale entry in the cache go
away? Every strategy in this lesson assumed it caught the moment of the write. That
assumption falls apart when the write comes from another process, when the data is
changed directly in the database, or when an ordering flaw like the one above occurs.
How long an entry should live, which event should evict it, and how a stale entry
becomes unreachable when the key itself changes are a separate set of decisions. The
next lesson compares time-based, event-based, and version-based invalidation on the same
workload and measures the number of stale reads each one produces.

### Invalidation

Source: https://academia.sh/en/courses/asynchronous-processing/invalidation

Measuring time-based, event-based, and version-based invalidation on the same workload: the trade-off a lifetime sets up between origin reads and stale reads, the delay of an event consumer fed by an outbox, and the orphaned-entry cost of putting the version in the key.

The previous lesson's last measurement left behind a stale entry: even though the write
rule was followed, the old value stayed in the cache, and it stayed there for fifty
reads. Without a mechanism to evict that entry, the problem does not resolve itself.

Evicting an entry from the cache is called **invalidation**, and it has three distinct
triggers. An entry can drop on its own after a set period, an event can evict it, or the
key itself can change and make the entry unreachable. This lesson runs all three against
the same workload and measures the number of stale reads each one produces.

#### Measurement Setup

The schema is the one from the previous lesson; the `outbox` table introduced in the
Business Logic Placement topic is added on top of it. The write operation both updates
the shelf count and drops an event record into this table within the same transaction.

```sh
# setup.sh — the previous lesson's schema, plus the outbox table from M16/K04
rm -f library.db
sqlite3 library.db <<'SQL'
CREATE TABLE book (book_id INTEGER PRIMARY KEY, title TEXT NOT NULL,
                    branch_id INTEGER NOT NULL, on_shelf INTEGER NOT NULL);
CREATE TABLE outbox (message_id INTEGER PRIMARY KEY, body TEXT NOT NULL,
                           status TEXT NOT NULL DEFAULT 'pending');
INSERT INTO book SELECT n, 'Book ' || n, n % 3 + 1, 3
  FROM (WITH RECURSIVE s(n) AS (SELECT 1 UNION ALL SELECT n+1 FROM s WHERE n < 60) SELECT n FROM s);
SQL
sqlite3 library.db "SELECT 'book=' || count(*) || ' total_on_shelf=' || sum(on_shelf) FROM book;"
```

```
book=60 total_on_shelf=180
```

The workload is again 1200 operations. The clock is not real time but the **workload
step**: every operation advances the clock by one. This way, the lifetime comparison
produces the same numbers independent of the machine; the unit of time is the step, not
milliseconds.

#### Three Approaches

The three strategies answer the same three questions differently: what is the key, how
long does the entry live, and what happens when a write occurs.

```js
// invalidation.mjs — time-based, event-based, and version-based invalidation on the same workload
import { DatabaseSync } from "node:sqlite";
import { copyFileSync, rmSync } from "node:fs";

function workload(count) {                            // 85% reads, 15% writes, 12 popular books
  let seed = 20250729;
  const random = () => ((seed = (seed * 1103515245 + 12345) % 2147483648) / 2147483648);
  return Array.from({ length: count }, () => {
    const isWrite = random() < 0.15;
    return [isWrite ? "write" : "read", Math.floor(random() * 12) + 1, isWrite ? (random() < 0.5 ? -1 : 1) : 0];
  });
}

function openOrigin(file) {
  rmSync(file, { force: true });
  copyFileSync("library.db", file);
  const db = new DatabaseSync(file);
  const s = { reads: 0 };
  return {
    s, db,
    read: (id) => { s.reads += 1; return db.prepare("SELECT on_shelf FROM book WHERE book_id = ?").get(id).on_shelf; },
    truth: (id) => db.prepare("SELECT on_shelf FROM book WHERE book_id = ?").get(id).on_shelf,
    write(id, delta) {                               // write and event in the same transaction (outbox)
      db.exec("BEGIN");
      const next = Math.max(0, this.truth(id) + delta);
      db.prepare("UPDATE book SET on_shelf = ? WHERE book_id = ?").run(next, id);
      db.prepare("INSERT INTO outbox (body) VALUES (?)").run(JSON.stringify({ bookId: id }));
      db.exec("COMMIT");
    },
    close: () => db.close(),
  };
}

function run(label, setup) {
  const origin = openOrigin("invalidation.db");
  const box = new Map();                          // key -> { value, validUntil }
  const m = { hit: 0, miss: 0, stale: 0 };
  let step = 0;                                    // clock: not real time, the workload step
  const strategy = setup(origin, box, () => step);
  for (const [kind, id, delta] of workload(1200)) {
    step += 1;
    if (kind === "read") {
      const key = strategy.key(id);
      const entry = box.get(key);
      let value;
      if (entry && entry.validUntil > step) { m.hit += 1; value = entry.value; }
      else { m.miss += 1; value = origin.read(id); box.set(key, { value, validUntil: strategy.lifetime(step) }); }
      if (value !== origin.truth(id)) m.stale += 1;
    } else {
      origin.write(id, delta);
      strategy.written(id);
    }
    strategy.tick?.();
  }
  origin.close();
  rmSync("invalidation.db", { force: true });
  return `${label.padEnd(22)}${[origin.s.reads, m.hit, m.miss, m.stale].map((n) => String(n).padStart(10)).join("")}`;
}

// 1) Time-based: a write never touches the cache, the entry drops once its lifetime runs out.
const timeBased = (lifetime) => () => ({
  key: (id) => `book:${id}`,
  lifetime: (step) => step + lifetime,
  written: () => {},
});

// 2) Event-based: a write produces an event in the outbox, the consumer deletes the key.
const eventBased = (delay) => (origin, box) => {
  let count = 0;
  return {
    key: (id) => `book:${id}`,
    lifetime: () => Infinity,
    written: () => {},
    tick() {                                      // the consumer runs once every `delay` steps
      if ((count += 1) % delay !== 0) return;
      const messages = origin.db.prepare(
        "SELECT message_id, body FROM outbox WHERE status = 'pending'").all();
      for (const msg of messages) {
        box.delete(`book:${JSON.parse(msg.body).bookId}`);
        origin.db.prepare("UPDATE outbox SET status = 'processed' WHERE message_id = ?").run(msg.message_id);
      }
    },
  };
};

// 3) Version-based: the key carries a version; a write raises the version, the old key becomes unreachable.
const versionBased = () => {
  const version = new Map();
  return {
    key: (id) => `book:v${version.get(id) ?? 1}:${id}`,
    lifetime: () => Infinity,
    written: (id) => version.set(id, (version.get(id) ?? 1) + 1),
  };
};

console.log(["invalidation", "org.reads", "hits", "misses", "stale"]
  .map((h, i) => (i === 0 ? h.padEnd(22) : h.padStart(10))).join(""));
console.log(run("time (lifetime=10)", timeBased(10)));
console.log(run("time (lifetime=50)", timeBased(50)));
console.log(run("time (lifetime=200)", timeBased(200)));
console.log(run("event (delay=5)", eventBased(5)));
console.log(run("event (delay=25)", eventBased(25)));
console.log(run("version", versionBased));
```

```
invalidation           org.reads      hits    misses     stale
time (lifetime=10)           626       399       626        18
time (lifetime=50)           232       793       232       147
time (lifetime=200)           72       953        72       445
event (delay=5)              155       870       155        14
event (delay=25)             153       872       153        97
version                      156       869       156         0
```

#### What the Numbers Say

The **time-based** rows trace a single curve. As the lifetime rises from ten to two
hundred, origin reads fall from 626 to 72, and stale reads rise from 18 to 445. These two
numbers are the two ends of the same dial: the lifetime is the exchange rate between the
load placed on the origin and the staleness accepted in return. As it approaches zero,
the cache becomes pointless; as it grows, the cache builds its own reality.

The **event-based** rows step off this curve. With a five-step consumer delay, origin
reads are 155 and stale reads are 14. For the same level of staleness, the time-based
approach demanded four times as many origin reads. The difference comes from tying
invalidation to a **cause**: the entry drops not because time passed, but because the
data changed. A key that never changes stays in the cache forever.

The weakness of the event-based approach shows up in the second row. When the consumer
delay rose from five to twenty-five, stale reads climbed from 14 to 97. The staleness
window is now the time between an event being produced and being processed. This is a
delivery problem: the event was produced and recorded, but no one has read it yet.

In the **version-based** row, stale reads are zero. Because the write changes the key,
the old entry is never even looked up; the new key is never found in the cache, so the
first read goes to the origin. Origin reads, at 156, are nearly the same as the
event-based approach, but staleness has been eliminated entirely.

#### When the Version Enters the Key

Version-based invalidation has two consequences: no delete ever happens, and old entries
keep sitting in the cache. The block below shows both. It also shows that a version can
be placed on an entire **namespace** instead of a single record; this form is called a
**generation**.

```js
// version-key.mjs — once the version enters the key, old entries become unreachable, but are not deleted
const box = new Map();
const version = new Map();                        // version per book
let branchVersion = 1;                            // a shared version at the branch level (generation)

const bookKey = (id) => `book:v${version.get(id) ?? 1}:${id}`;
const listKey = (branchId, page) => `branch:${branchId}:g${branchVersion}:list:${page}`;

box.set(bookKey(7), { on_shelf: 3 });
box.set(listKey(2, 1), ["Book 5", "Book 8"]);
box.set(listKey(2, 2), ["Book 11"]);
console.log("starting keys:", [...box.keys()].join("  "));

version.set(7, 2);                                // book 7 was updated
branchVersion += 1;                               // branch lists are invalidated in bulk
console.log("key looked up for book 7         :", bookKey(7),
  "-> in cache:", box.has(bookKey(7)));
console.log("key looked up for branch 2 page 1:", listKey(2, 1),
  "-> in cache:", box.has(listKey(2, 1)));

box.set(bookKey(7), { on_shelf: 2 });
const live = new Set([bookKey(7), listKey(2, 1), listKey(2, 2)]);
const orphaned = [...box.keys()].filter((a) => !live.has(a));
console.log(`entries=${box.size}  orphaned entries=${orphaned.length}  ->`, orphaned.join("  "));
```

```
starting keys: book:v1:7  branch:2:g1:list:1  branch:2:g1:list:2
key looked up for book 7         : book:v2:7 -> in cache: false
key looked up for branch 2 page 1: branch:2:g2:list:1 -> in cache: false
entries=4  orphaned entries=3  -> book:v1:7  branch:2:g1:list:1  branch:2:g1:list:2
```

The value of a generation counter is this: no matter how many list pages a branch has,
incrementing a single number invalidates all of them. The keys never need to be found
and deleted one by one — in a shared cache, searching by key pattern is expensive, and
often dangerous.

The cost is in the last line. Three of the four entries are now dead entries that no
read will ever look up again. These entries take up space and can only be cleared by a
lifetime or an eviction policy. Version-based invalidation is therefore always set up
together with an upper time bound.

The second condition is that the version counter must be **shared**. In the example
above, the counter lives in process memory; when two application instances run, one
raising the version does not get seen by the other, which keeps reading the old key. The
counter itself must be kept in a shared store, so every read pays for one extra access.

#### Where Each Approach Fits

The three approaches do not exclude one another; they are layered on top of each other.

**Time-based** invalidation is always present, because it is the only guarantee left
when the others fail. When an event is lost, or an application instance misses a version
increase, the lifetime is the only thing bounding staleness. In client and edge caches,
it is the only option: there is no way to reach those copies and delete them.

**Event-based** invalidation reacts when the data changes and decouples staleness from
the lifetime. Its correctness depends on the event actually being delivered. The
consumer delay in the measurement was its most optimistic case: the event had been
recorded and was only read late. A lost event leaves a stale entry behind forever.

**Version-based** invalidation moves eviction into the read path: instead of deleting
the wrong entry, it looks up the correct entry under a different name. It is the most
resistant approach to race conditions, because it does not depend on the ordering of a
delete. Its cost is the space that dead entries accumulate.

#### Summary

- The lifetime is the exchange rate between origin load and accepted staleness: as the
  lifetime rose from 10 to 200, origin reads fell from 626 to 72, and stale reads rose
  from 18 to 445.
- Event-based invalidation steps off this curve: with a five-step consumer delay, 155
  origin reads and 14 stale reads — a quarter of the origin load for the same level of
  staleness.
- The staleness window of the event-based approach is the time between an event being
  produced and being processed; when the delay rose from five to twenty-five, stale
  reads climbed from 14 to 97.
- Once the version enters the key, stale reads drop to zero and no delete ever happens;
  with a generation counter, a single increment invalidates an entire namespace.
- The cost of the version-based approach is orphaned entries; it is therefore always
  set up with an upper time bound, and the counter must be shared.

#### Next Step

In this lesson, the key changed three times: first `book:7`, then a form carrying the
version, then a list key carrying a branch's generation counter. Each time, one more
piece of information went into the key, and each time, this was done by hand. When a key
structure is built by hand, two mistakes become unavoidable: two different queries
landing on the same key, and one tenant's data being served to another tenant. The next
lesson treats the key as a design object; it writes a generator built on namespace,
tenant separation, and canonical ordering of criteria, and demonstrates both mistakes by
running them.

### Cache Key Design

Source: https://academia.sh/en/courses/asynchronous-processing/cache-key-design

Demonstrating, by running them, two flaws of a hand-concatenated key — collision and tenant leak — and writing a function that generates a key from namespace, schema version, generation, tenant, and canonical criteria; measuring the effect of key cardinality on the hit ratio and on correctness.

In the previous lesson, the key took three different forms: first `book:7`, then a form
carrying the version, then a list key carrying the branch's generation counter. Each
time, the key was built by hand through string concatenation.

The key is the cache's only addressing mechanism. When it is built wrong, the cache
silently returns the wrong response — it does not error, it does not slow down, it just
serves someone else's data. This lesson demonstrates, by running them, two flaws of hand
concatenation, then writes a generator that treats the key as a design object.

#### Two Flaws

The measurements run against a seven-book catalog. One of the books carries a colon in
its title; this is necessary to bring out the first flaw.

```sh
# setup.sh — book data for search and list queries (M16/K04 seed)
rm -f library.db
sqlite3 library.db <<'SQL'
CREATE TABLE branch (branch_id INTEGER PRIMARY KEY, name TEXT NOT NULL, city TEXT NOT NULL);
CREATE TABLE book (book_id INTEGER PRIMARY KEY, title TEXT NOT NULL, author TEXT NOT NULL,
                    publication_year INTEGER, branch_id INTEGER REFERENCES branch(branch_id));
INSERT INTO branch VALUES (1,'Central','Ankara'),(2,'Bahcelievler','Ankara'),(3,'Kadikoy','Istanbul');
INSERT INTO book VALUES
  (1,'Blindness','José Saramago',1995,1),(2,'The Disconnected','Oğuz Atay',1972,1),
  (3,'The Book of Sand','Jorge Luis Borges',1975,2),(4,'Yaban','Yakup Kadri',1932,2),
  (5,'Silent House','Orhan Pamuk',1983,3),(6,'Motherland Hotel','Yusuf Atılgan',1973,3),
  (7,'Sand: Borges Selection','Anthology',2010,1);
SQL
sqlite3 library.db "SELECT branch_id, count(*) FROM book GROUP BY branch_id;"
```

```
1|3
2|2
3|2
```

The block below runs two scenarios. In the first, the search criteria are concatenated
with a colon; in the second, the branch list key carries only the page number.

```js
// collision.mjs — two flaws of a hand-concatenated key: collision and tenant leak
import { DatabaseSync } from "node:sqlite";

const db = new DatabaseSync("library.db");
const search = (title, author) => db.prepare(
  "SELECT title FROM book WHERE title LIKE ? AND author LIKE ? ORDER BY book_id")
  .all(`%${title}%`, `%${author}%`).map((k) => k.title);
const branchList = (branchId, page) => db.prepare(
  "SELECT title FROM book WHERE branch_id = ? ORDER BY book_id LIMIT 2 OFFSET ?")
  .all(branchId, (page - 1) * 2).map((k) => k.title);

const box = new Map();
const cached = (key, produce) => {
  if (box.has(key)) return { key, result: box.get(key), source: "cache" };
  const result = produce();
  box.set(key, result);
  return { key, result, source: "origin" };
};

console.log("--- collision: two different queries land on the same key");
const a = cached(`search:Sand:Borges`, () => search("Sand", "Borges"));
const b = cached(`search:Sand:Borges`, () => search("Sand: Borges", ""));
console.log(`A title="Sand" author="Borges"   key=${a.key} ${a.source}: ${a.result.join(", ")}`);
console.log(`B title="Sand: Borges" author=""  key=${b.key} ${b.source}: ${b.result.join(", ")}`);
console.log(`The correct result B would get from the origin: ${search("Sand: Borges", "").join(", ")}`);

console.log("--- tenant leak: the key does not carry branch information");
const s2 = cached("list:page:1", () => branchList(2, 1));
const s3 = cached("list:page:1", () => branchList(3, 1));
console.log(`branch 2 page 1 -> ${s2.source}: ${s2.result.join(", ")}`);
console.log(`branch 3 page 1 -> ${s3.source}: ${s3.result.join(", ")}`);
const correct = branchList(3, 1);
const leaked = s3.result.filter((x) => !correct.includes(x));
console.log(`records served to branch 3 that do not belong to branch 3: ${leaked.length} -> ${leaked.join(", ")}`);
db.close();
```

```
--- collision: two different queries land on the same key
A title="Sand" author="Borges"   key=search:Sand:Borges origin: The Book of Sand
B title="Sand: Borges" author=""  key=search:Sand:Borges cache: The Book of Sand
The correct result B would get from the origin: Sand: Borges Selection
--- tenant leak: the key does not carry branch information
branch 2 page 1 -> origin: The Book of Sand, Yaban
branch 3 page 1 -> cache: The Book of Sand, Yaban
records served to branch 3 that do not belong to branch 3: 2 -> The Book of Sand, Yaban
```

The first flaw is a **collision**: the moment the character used as the separator also
occurs inside a value, two different sets of criteria produce the same string. The
second search got the first one's answer from the cache, and no one noticed.

The second flaw is a **tenant leak**. In the library system, a branch is a **tenant**:
each branch sees only its own data. Because the key did not carry the branch, the
Kadikoy branch's list showed Bahcelievler's books. The authorization check set up
server-side in the Authentication and Authorization course is not in play here at all;
the query never ran, the response came from the cache. **The cache is the one layer
that can walk around behind an authorization check.**

#### The Parts of a Key

Both flaws came from building the key by hand. A single function that generates the key
gathers the rules in one place.

```js
// key.mjs — a function that generates a cache key from namespace, tenant, and canonical criteria
import { createHash } from "node:crypto";

const SCHEMA = 1;                                   // key schema version
const escape = (m) => encodeURIComponent(String(m));

export function generateKey({ namespace, operation, tenant, generation = 1, criteria = {} }) {
  const criteriaString = Object.entries(criteria)
    .filter(([, v]) => v !== undefined && v !== null && v !== "")
    .sort(([a], [b]) => (a < b ? -1 : 1))          // order-independent: the same criteria set gives the same key
    .map(([k, v]) => `${escape(k)}=${escape(v)}`)
    .join("&");
  const body = `${escape(namespace)}:s${SCHEMA}:g${generation}:t${escape(tenant)}:${escape(operation)}`;
  const tail = criteriaString.length <= 48 ? criteriaString : `c_${createHash("sha256").update(criteriaString).digest("hex").slice(0, 16)}`;
  return tail ? `${body}:${tail}` : body;
}

if (import.meta.url === `file://${process.argv[1]}`) {
  const A = (o) => generateKey({ namespace: "library", generation: 4, ...o });

  console.log("--- the two colliding queries are now in separate keys");
  console.log(A({ operation: "search", tenant: "branch-2", criteria: { title: "Sand", author: "Borges" } }));
  console.log(A({ operation: "search", tenant: "branch-2", criteria: { title: "Sand: Borges", author: "" } }));

  console.log("--- tenant in the key: same query, different branch");
  console.log(A({ operation: "list", tenant: "branch-2", criteria: { page: 1 } }));
  console.log(A({ operation: "list", tenant: "branch-3", criteria: { page: 1 } }));

  console.log("--- criteria order does not change the key");
  const s1 = A({ operation: "search", tenant: "branch-1", criteria: { author: "Atay", title: "The Disconnected" } });
  const s2 = A({ operation: "search", tenant: "branch-1", criteria: { title: "The Disconnected", author: "Atay" } });
  console.log(`${s1}\nequal: ${s1 === s2}`);

  console.log("--- a long criteria set gets summarized");
  const long = A({ operation: "search", tenant: "branch-1", criteria: {
    title: "The Disconnected", author: "Oğuz Atay", yearFrom: 1970, yearTo: 1980,
    lang: "tr", status: "on_shelf", sort: "title", page: 3 } });
  console.log(`${long}\nlength=${long.length}`);
}
```

```
--- the two colliding queries are now in separate keys
library:s1:g4:tbranch-2:search:author=Borges&title=Sand
library:s1:g4:tbranch-2:search:title=Sand%3A%20Borges
--- tenant in the key: same query, different branch
library:s1:g4:tbranch-2:list:page=1
library:s1:g4:tbranch-3:list:page=1
--- criteria order does not change the key
library:s1:g4:tbranch-1:search:author=Atay&title=The%20Disconnected
equal: true
--- a long criteria set gets summarized
library:s1:g4:tbranch-1:search:c_648d4755bc125b8e
length=49
```

The key has five parts. The **namespace** separates different applications that share
the same cache store. The **schema version** versions the key structure itself: when the
value format changes, this number is incremented, and every entry in the old format
becomes unreachable in a single move. The **generation** carries the bulk invalidation
from the previous lesson. The **tenant** places the authorization boundary inside the
key. The **operation** and the **canonical criteria** are the query's identity.

Canonical form is achieved with three rules: empty criteria are dropped, criteria are
sorted by name, and the name and value are escaped. Sorting keeps the same query from
producing two separate entries when its criteria arrive in a different order. Escaping
makes the collision in the first flaw structurally impossible: the colon inside a value
becomes `%3A` and stops being a separator.

The last part is the length limit. When the criteria set grows long, it gets summarized,
but the key's readable body is preserved. This distinction is valuable at operating
time: which application, which branch, and which operation an entry in the cache
belongs to can be seen just by looking at the key; only which criteria produced it stays
hidden.

#### Extra Fields and Missing Fields

Which fields the key carries is a correctness decision, and it can break in both
directions.

```js
// cardinality.mjs — extra fields in the key break the hit ratio, missing fields break correctness
import { DatabaseSync } from "node:sqlite";
import { generateKey } from "./key.mjs";

const db = new DatabaseSync("library.db");
const list = (branchId, page) => db.prepare(
  "SELECT title FROM book WHERE branch_id = ? ORDER BY book_id LIMIT 2 OFFSET ?")
  .all(branchId, (page - 1) * 2).map((k) => k.title).join(", ");

const requests = [];                               // 400 requests, 3 branches x 2 pages
for (let i = 0; i < 400; i++) requests.push([(i % 3) + 1, (i % 2) + 1, `r-${i}`]);

function run(name, makeKey) {
  const box = new Map();
  const m = { hit: 0, miss: 0, wrong: 0 };
  for (const [branchId, page, requestId] of requests) {
    const key = makeKey(branchId, page, requestId);
    let value;
    if (box.has(key)) { m.hit += 1; value = box.get(key); }
    else { m.miss += 1; value = list(branchId, page); box.set(key, value); }
    if (value !== list(branchId, page)) m.wrong += 1;
  }
  const ratio = ((m.hit / requests.length) * 100).toFixed(1);
  return `${name.padEnd(29)}${String(m.hit).padStart(8)}${String(m.miss).padStart(8)}` +
    `${(ratio + "%").padStart(10)}${String(m.wrong).padStart(9)}${String(box.size).padStart(8)}`;
}

const base = { namespace: "library", operation: "list", generation: 1 };
console.log(["key form", "hits", "misses", "ratio", "wrong", "entries"]
  .map((h, i) => (i === 0 ? h.padEnd(29) : h.padStart(i === 3 ? 10 : i === 4 ? 9 : 8))).join(""));
console.log(run("correct (tenant + criteria)", (b, p) =>
  generateKey({ ...base, tenant: `branch-${b}`, criteria: { page: p } })));
console.log(run("extra field (request id)", (b, p, r) =>
  generateKey({ ...base, tenant: `branch-${b}`, criteria: { page: p, request: r } })));
console.log(run("missing field (no tenant)", (b, p) =>
  generateKey({ ...base, tenant: "shared", criteria: { page: p } })));
db.close();
```

```
key form                         hits  misses     ratio    wrong entries
correct (tenant + criteria)       394       6     98.5%        0       6
extra field (request id)            0     400      0.0%        0     400
missing field (no tenant)         398       2     99.5%      200       2
```

An extra field zeroes out the hit ratio. Because the request id differs on every
request, no key is ever looked up a second time; the cache fills with four hundred
entries and produces not a single hit. A timestamp, a session id, or a header that does
not affect the response produces the same result.

The missing-field row is more dangerous. It is the row with the highest hit ratio —
99.5% — with all but two of the four hundred requests served from two entries. That
same row has two hundred wrong responses. **The highest hit ratio belongs to the wrong
implementation.** The hit ratio alone is not a quality measure; correctness must be
proven separately.

The rule that follows is this: the key must carry every input that changes the
response, and none that does not. Among the ones that change the response, the
authorization context — tenant, role, visibility level — is always present. Among the
ones that do not, there are the request id, tracing headers, and the session token. The
token entering the key is also a security flaw: a secret value ends up written
somewhere that anyone listing cache keys can see.

#### Summary

- A key hand-concatenated with a separator character collides the moment a value itself
  carries that character; two different searches were served from the same entry.
- When the key does not carry the tenant, the cache walks around behind the
  authorization check: one branch's list was served to another branch.
- The key generator carries five parts: namespace, schema version, generation, tenant,
  and canonical criteria. Escaping prevents collision, and sorting prevents the same
  query from splitting into two entries.
- Long criteria sets get summarized, but the readable body is preserved; which
  application and which tenant an entry belongs to stays readable from the key.
- Putting an extra field in the key zeroed out the hit ratio; putting a missing field in
  produced 200 wrong responses at a 99.5% hit ratio. The hit ratio alone is not proof of
  correctness.

#### Next Step

Up to this lesson, every measurement ran in a single sequence of operations: one request
takes a miss, goes to the origin, stores the entry, and the next request gets a hit. On
a real server, hundreds of requests look up the same key at the same time. The moment
that key's entry drops, all of them take a miss at once and all of them go to the origin
at once; the origin the cache is protecting sees its highest load at exactly the moment
the cache goes empty. The next lesson measures this behavior with concurrent requests
and builds the lock that collapses the call reaching the origin for the same key down to
a single production.

### Hot Key and Stampede Effect

Source: https://academia.sh/en/courses/asynchronous-processing/hot-key-and-stampede-effect

Measuring how a hundred concurrent requests pile onto the origin when a hot key's entry drops, and collapsing that to a single production with a single-flight lock; stale-while-revalidate zeroing out the wait, jittering the lifetime to prevent simultaneous expiry, and spreading a hot key across shards.

Every measurement in the previous lessons ran in a single sequence: one request takes a
miss, goes to the origin, stores the entry, and the next request gets a hit. This
sequence assumes requests wait for one another. A real server does not.

The library system's most-read key is the Central branch's stock count. When that entry
drops, every request arriving in the time between the drop and the new entry being
stored takes a miss, and every single one of them goes to the origin. This behavior is
called a **cache stampede**, and its result is this: the origin sees its highest load,
on the key it is protected the most, at exactly the moment the cache goes empty. This
lesson measures the behavior and compares three countermeasures.

#### A Hundred Requests Arriving at Once

In the measurement setup, origin access is an asynchronous call: the query and the
round trip together take 30 milliseconds. What gets counted is not the duration but
**the number of calls reaching the origin**.

```js
// stampede.mjs — the moment a hot key's entry drops, 100 requests arrive at once
import { setTimeout as wait } from "node:timers/promises";

const DELAY = 30;                                  // origin access: query + round trip
const counter = { origin: 0 };
async function produceFromOrigin(key) {            // branch stock count: an expensive aggregate query
  counter.origin += 1;
  await wait(DELAY);
  return { key, on_shelf: 137 };
}

const KEY = "library:s1:g1:tbranch-1:stock";
const REQUESTS = 100;

async function run(name, setup) {
  counter.origin = 0;
  const m = { pending: 0, stale: 0 };
  const read = setup(m);
  const t = performance.now();
  await Promise.all(Array.from({ length: REQUESTS }, () => read(KEY)));
  const duration = performance.now() - t;
  return `${name.padEnd(28)}${String(counter.origin).padStart(15)}${String(m.pending).padStart(20)}` +
    `${String(m.stale).padStart(18)}${(String(Math.round(duration / 10) * 10) + " ms").padStart(12)}`;
}

// 1) Plain cache-aside: no entry exists, everyone goes to the origin.
const plain = (m) => {
  const box = new Map();
  return async (a) => {
    if (box.has(a)) return box.get(a);
    m.pending += 1;
    const d = await produceFromOrigin(a);
    box.set(a, d);
    return d;
  };
};

// 2) Single-flight: one production for the same key; the rest wait on the same promise.
const singleFlight = (m) => {
  const box = new Map();
  const flight = new Map();
  return async (a) => {
    if (box.has(a)) return box.get(a);
    m.pending += 1;
    if (!flight.has(a)) {
      flight.set(a, produceFromOrigin(a).then((d) => { box.set(a, d); flight.delete(a); return d; }));
    }
    return flight.get(a);
  };
};

// 3) Stale-while-revalidate: an expired entry is returned immediately, the refresh starts in the background.
const staleWhileRevalidate = (m) => {
  const box = new Map([[KEY, { value: { key: KEY, on_shelf: 131 }, fresh: false }]]);
  const flight = new Map();
  return async (a) => {
    const entry = box.get(a);
    if (entry && !entry.fresh && !flight.has(a)) {     // start the refresh, do not wait
      flight.set(a, produceFromOrigin(a).then((d) => { box.set(a, { value: d, fresh: true }); flight.delete(a); }));
    }
    if (entry) { if (!entry.fresh) m.stale += 1; return entry.value; }
    m.pending += 1;
    return produceFromOrigin(a);
  };
};

console.log(["strategy", "origin calls", "pending requests", "stale responses", "duration~"]
  .map((h, i) => (i === 0 ? h.padEnd(28) : h.padStart([0, 15, 20, 18, 12][i]))).join(""));
console.log(await run("plain", plain));
console.log(await run("single-flight", singleFlight));
console.log(await run("stale-while-revalidate", staleWhileRevalidate));
```

{/* mask: (?<=plain[ 0-9]{83})[34]0(?= ms)|(?<=single-flight[ 0-9]{75})[34]0(?= ms) */}
```
strategy                       origin calls    pending requests   stale responses   duration~
plain                                   100                 100                 0       30 ms
single-flight                             1                 100                 0       30 ms
stale-while-revalidate                    1                   0               100        0 ms
```

The duration column is rounded to ten milliseconds and is machine-dependent; the first
three columns are not.

In the **plain** setup, a hundred requests produced a hundred origin calls. The code is
not wrong: every request checked the cache, did not find the entry, and went to the
origin. The flaw lies in the gap between the miss and the entry being stored. This gap
lasts as long as the origin access, and every request arriving during that time
restarts the same work. The size of the load is set by how fast requests arrive: on an
endpoint receiving a thousand requests per second, a 30-millisecond gap means thirty
concurrent queries.

The **single-flight** lock brought the origin calls down to one. The setup is simple:
before production starts, the key is written to a flight table, and later requests wait
on the same promise. All hundred requests still waited — after all, the data was not
there yet — but the origin saw a single query. This is the direct fix for the stampede
effect.

**Stale-while-revalidate** trades differently. An expired entry is not deleted; when a
request arrives, the old value is returned immediately and the refresh starts in the
background. The pending-request count dropped to zero, and in exchange, a hundred
responses carried stale data. The same pattern introduced client-side in the Caching
Strategies lesson of the Frontend Quality course makes the same trade on the server
side: staleness instead of waiting.

The choice between the three rows depends on the data. A branch's stock count can be a
few seconds out of date; there, stale-while-revalidate is the best choice. The
eligibility check performed during a loan cannot work with stale data; there,
single-flight is used.

The scope of the flight table is also a separate decision. An in-process table only
protects that one instance; if four application instances are running, four calls
reach the origin instead of one. To bring the origin down to strictly one call, the
lock must be kept in a shared store; this raises correctness while adding one network
round trip to every miss.

#### Expirations Landing on the Same Moment

The second form of the stampede effect does not show up in a single key, but in a set
of keys loaded at the same time. When an application instance restarts, or a generation
counter is incremented, hundreds of keys enter the cache in the same step. If they are
all given the same lifetime, they all drop in the same step.

```js
// hot-key.mjs — simultaneous expiry, and a hot key piling onto a single node
import { createHash } from "node:crypto";

const digest = (m) => parseInt(createHash("sha256").update(m).digest("hex").slice(0, 8), 16);

console.log("--- 200 keys loaded at the same step, lifetime=100");
for (const [name, lifetime] of [["fixed lifetime", () => 100], ["jittered lifetime", (a) => 100 + (digest(a) % 40)]]) {
  const perStep = new Map();
  for (let i = 1; i <= 200; i++) {
    const key = `library:s1:g1:tbranch-1:book:${i}`;
    const step = lifetime(key);
    perStep.set(step, (perStep.get(step) ?? 0) + 1);
  }
  const highest = Math.max(...perStep.values());
  console.log(`${name.padEnd(18)} expiry steps=${perStep.size}  highest miss in a single step=${highest}`);
}

console.log("--- 4-shard shared cache, 60% of requests on a single key");
const HOT = "library:s1:g1:tbranch-1:stock";
const requests = Array.from({ length: 1000 }, (_, i) =>
  i % 10 < 6 ? HOT : `library:s1:g1:tbranch-${(i % 3) + 2}:book:${i % 40}`);

function distribution(transform) {
  const shard = [0, 0, 0, 0];
  for (const [i, a] of requests.entries()) shard[digest(transform(a, i)) % 4] += 1;
  return shard;
}
const print = (name, b) => console.log(`${name.padEnd(22)}${b.map((n) => String(n).padStart(8)).join("")}` +
  `   busiest shard=%${((Math.max(...b) / 1000) * 100).toFixed(0)}`);

print("single key", distribution((a) => a));
print("split into 4 copies", distribution((a, i) => (a === HOT ? `${a}.${i % 4}` : a)));
print("split into 8 copies", distribution((a, i) => (a === HOT ? `${a}.${i % 8}` : a)));
```

```
--- 200 keys loaded at the same step, lifetime=100
fixed lifetime     expiry steps=1  highest miss in a single step=200
jittered lifetime  expiry steps=40  highest miss in a single step=9
--- 4-shard shared cache, 60% of requests on a single key
single key                 102     684     108     106   busiest shard=%68
split into 4 copies        102     234     408     256   busiest shard=%41
split into 8 copies        102     309     333     256   busiest shard=%33
```

With a fixed lifetime, all two hundred keys dropped in a single step. Once a small
deviation derived from the key was added to the lifetime, expirations spread across
forty steps, and the highest miss count in any single step fell to 9. It matters that
the deviation is derived from the key: if a random number were used instead, the same
key would drop at a different time on different instances, and the behavior would
become unreproducible.

#### The Hot Key Itself

The second half of the output measures a separate problem. When a shared cache is
split across multiple nodes, a key lands on a shard according to its digest. If 60% of
requests target a single key, that key's shard sees 68% of requests; the remaining
three shards sit idle. Increasing the number of shards does not fix this distribution,
because the problem is not too few keys — it is **the weight of a single key**.

The fix is to replicate the key: the hot key is split into a fixed number of copies,
and a request is routed to one of them. With four copies, the busiest shard fell from
68% to 41%; with eight, to 33%. The distribution is not even — the digest function does
not spread the copies evenly across shards — but the pile-up is broken.

Replication costs two things. The same value is stored multiple times, and
invalidation must now evict not a single key but every copy. Second, because the
number of copies is fixed, the generation counter from the previous lesson does its job
here too: when the generation is incremented, every copy becomes unreachable in a
single move.

The third and cheapest countermeasure for a hot key is keeping a short-lived local copy
inside the application instance. This is the server link of the layer chain measured in
the first lesson: a local entry lasting a few seconds, sitting in front of the shared
cache, cuts off most of the network round trips going to the hot key.

#### Summary

- When a hot key's entry drops, every request arriving in the gap between the miss and
  the entry being stored goes to the origin: a hundred concurrent requests produced a
  hundred origin calls.
- A single-flight lock brings production for the same key down to one; a hundred
  requests still waited, but the origin saw a single query. If the lock lives
  in-process, the protection weakens by the number of instances.
- Stale-while-revalidate brings the pending-request count to zero and returns stale
  responses in exchange; the choice depends on the data's tolerance for staleness.
- Two hundred keys loaded at the same time dropped in a single step with a fixed
  lifetime; once a deviation derived from the key was added to the lifetime, the
  highest miss count in a single step fell from 200 to 9.
- In a shared cache split into shards, a single hot key loaded one shard with 68% of
  requests; once the key was split into eight copies, the busiest shard fell to 33%.

#### Next Step

Up to this lesson, every cache was set up inside the application, and every decision
was made in code. Once the response is sent to the client, the caching decision is no
longer in the code's hands: the browser, the shared caches in between, and the edge
layer look only at what is written in the response's headers. This is the only way to
talk to the layer called "uncontrollable" in the first lesson, and it has its own
vocabulary — which response can be stored, how fresh it counts as, how to tell whether
a stored copy is still valid without asking the origin. The next lesson builds this
vocabulary against an HTTP server with real requests.

### HTTP Caching

Source: https://academia.sh/en/courses/asynchronous-processing/http-caching

The response carrying its own cache rules: ETag and Last-Modified validators producing a 304 on conditional requests, the body transfer dropping to zero, the validator changing when the resource changes, and choosing the public, private, max-age, and no-store directives, along with stale-while-revalidate, by resource type.

Every cache in the previous lessons lived inside the application: the key was
generated in code, the entry was stored in code, invalidation was triggered in code.
That control ends the moment the response is sent to the client. The layers called
"uncontrollable" in the first lesson — the browser cache, the shared caches in
between, the edge layer — do not know what the code thinks; they only look at what is
written in the response's headers.

This is why HTTP has its own cache vocabulary, and it answers two questions
separately: how long a response counts as **fresh**, and once its freshness runs out,
how to ask whether the stored copy is still valid without downloading the body from
the origin. This lesson answers both questions against real requests.

#### A Server That Generates Validators

The server below serves four endpoints of the library API. A **validator** is
generated from the digest of each response body, and conditional requests are
answered against this value. The date header on responses is held fixed, so the
measurement's output is the same on every run.

```js
// http-caching.mjs — the library API: validators and cache-control directives
import { createServer } from "node:http";
import { createHash } from "node:crypto";

const books = new Map([[7, { bookId: 7, title: "Sand: Borges Selection", author: "Anthology", on_shelf: 3 }]]);
const stock = { branchId: 1, on_shelf: 137 };
const history = { memberId: 4, total: 12 };
const MODIFIED = new Map([[7, new Date("2025-07-01T09:00:00Z")]]);   // last-modified timestamps

const makeEtag = (body) => `"${createHash("sha256").update(body).digest("hex").slice(0, 16)}"`;

function respond(request, response, { body, cacheControl, lastModified = null, vary = null }) {
  const text = JSON.stringify(body);
  const etag = makeEtag(text);
  const headers = { "cache-control": cacheControl, etag };
  if (lastModified) headers["last-modified"] = lastModified.toUTCString();
  if (vary) headers.vary = vary;

  const incomingEtag = request.headers["if-none-match"];
  const incomingDate = request.headers["if-modified-since"];
  const etagMatches = incomingEtag !== undefined && incomingEtag.split(",").map((e) => e.trim()).includes(etag);
  const dateMatches = !etagMatches && incomingDate !== undefined && lastModified !== null &&
    Math.floor(lastModified.getTime() / 1000) <= Math.floor(Date.parse(incomingDate) / 1000);

  if (etagMatches || dateMatches) { response.writeHead(304, headers); return response.end(); }
  response.writeHead(200, { ...headers, "content-type": "application/json; charset=utf-8",
    "content-length": Buffer.byteLength(text) });
  response.end(text);
}

const server = createServer((request, response) => {
  response.sendDate = false;                                        // to keep the output deterministic
  const path = new URL(request.url, "http://local").pathname;

  if (request.method === "POST" && path === "/book/7") {            // the book is updated: the validator changes
    const book = books.get(7);
    book.on_shelf -= 1;
    MODIFIED.set(7, new Date("2025-07-02T14:30:00Z"));
    response.writeHead(204, { "cache-control": "no-store" });
    return response.end();
  }
  if (path === "/book/7") {                                         // a rarely changing shared resource
    return respond(request, response, { body: books.get(7), cacheControl: "public, max-age=60",
      lastModified: MODIFIED.get(7), vary: "Accept-Language" });
  }
  if (path === "/branch/1/stock") {                                 // a frequently changing shared resource
    return respond(request, response, { body: stock, cacheControl: "public, max-age=5, stale-while-revalidate=30" });
  }
  if (path === "/member/4/history") {                               // a resource private to one person
    return respond(request, response, { body: history, cacheControl: "private, max-age=30" });
  }
  if (path === "/loan/approval") {                                  // a response that must never be stored
    return respond(request, response, { body: { approval: "U-4711", remaining: 2 }, cacheControl: "no-store" });
  }
  response.writeHead(404, { "cache-control": "no-store" });
  response.end();
});

server.listen(8547, "127.0.0.1", () => console.log("library API 127.0.0.1:8547"));
```

The block that starts and measures the server is separate. The validator and the
last-modified date are read from the response and placed as a conditional header on
later requests.

```bash
# The responses' cache headers, and the result of conditional requests
node http-caching.mjs & server=$!
sleep 0.5
K=http://127.0.0.1:8547

headers() { curl -sS -D - -o /dev/null "$@" | tr -d '\r' | grep -iE '^(HTTP/|etag|last-modified|cache-control|vary)'; }
measure() { curl -sS -o /dev/null -w "status=%{http_code} downloaded=%{size_download} bytes\n" "$@"; }

echo "== 1. first request: validators and directives"
headers "$K/book/7"

ETAG=$(curl -sS -D - -o /dev/null "$K/book/7" | tr -d '\r' | awk '/^[Ee]tag:/ {print $2}')
DATE=$(curl -sS -D - -o /dev/null "$K/book/7" | tr -d '\r' | sed -n 's/^[Ll]ast-[Mm]odified: //p')
echo "== 2. requesting the same resource conditionally"
measure "$K/book/7"
measure -H "If-None-Match: $ETAG" "$K/book/7"
measure -H "If-Modified-Since: $DATE" "$K/book/7"

echo "== 3. once the resource changes, the old validator no longer matches"
curl -sS -o /dev/null -X POST "$K/book/7"
measure -H "If-None-Match: $ETAG" "$K/book/7"
headers "$K/book/7" | grep -iE '^(etag|last-modified)'

echo "== 4. directives by resource type"
for path in /book/7 /branch/1/stock /member/4/history /loan/approval; do
  printf '%-18s %s\n' "$path" "$(curl -sS -D - -o /dev/null "$K$path" | tr -d '\r' | sed -n 's/^[Cc]ache-[Cc]ontrol: //p')"
done

kill $server
```

```
library API 127.0.0.1:8547
== 1. first request: validators and directives
HTTP/1.1 200 OK
cache-control: public, max-age=60
etag: "0853c3f417fbfc00"
last-modified: Tue, 01 Jul 2025 09:00:00 GMT
vary: Accept-Language
== 2. requesting the same resource conditionally
status=200 downloaded=79 bytes
status=304 downloaded=0 bytes
status=304 downloaded=0 bytes
== 3. once the resource changes, the old validator no longer matches
status=200 downloaded=79 bytes
etag: "5633883a8bff3669"
last-modified: Wed, 02 Jul 2025 14:30:00 GMT
== 4. directives by resource type
/book/7            public, max-age=60
/branch/1/stock    public, max-age=5, stale-while-revalidate=30
/member/4/history  private, max-age=30
/loan/approval     no-store
```

#### Freshness and Validation

The second part of the output separates two different jobs. The unconditional
request returned 200, and 79 bytes were downloaded. When the same resource was
requested with the `If-None-Match` header, the response was 304, and the downloaded
body dropped to zero bytes. The third line got the same result with
`If-Modified-Since`.

What a **304 Not Modified** response saves is the body, not the round trip. The
request still went to the server, and the server still computed the validator. The
saving grows as the body grows: negligible for a seventy-nine-byte book record,
decisive for a fifty-kilobyte list response.

Freshness comes before validation. A response marked `max-age=60` counts as **fresh**
for sixty seconds, and during that time the cache never asks the server at all. Once
that time runs out, the copy is not discarded; it moves to a **stale** state and gets
validated on the next request. If validation returns 304, the copy counts as fresh
again. The two mechanisms work together: `max-age` eliminates the round trip, the
validator eliminates the body.

The third part proves what the validator means. After the book was updated with a
POST request, the conditional request made with the old validator still on hand
returned 200, not 304, and the body was downloaded again. The new response's tag and
last-modified date have both changed. Because the validator is tied to the resource's
content, this behavior is correct by construction.

There is a distinction between the two validators. The last-modified date's
resolution is one second: for a record that changes twice within the same second, the
old copy can be mistaken for valid. The tag generated from the body digest has no
such limit. In exchange, computing the tag requires producing the body; the date, on
the other hand, is usually read straight from the record's own column and can be
answered without running a query.

#### What the Directives Mean

The four lines in the last section are four decisions made for four different
resource types.

`public, max-age=60` — book detail returns the same thing to every user and changes
rarely. `public` says the response can be stored in **shared** caches.

`public, max-age=5, stale-while-revalidate=30` — a branch's stock count changes
often. It counts as fresh for five seconds; for the following thirty seconds, the
stale copy keeps being served while the cache refreshes it in the background. This is
the HTTP counterpart of the behavior set up in-process in the previous lesson; the
difference is that the decision is now carried out by the intermediate cache, not the
application.

`private, max-age=30` — a member's loan history is private to that person. `private`
says the response can be stored only in the end user's own cache, and cannot be
placed in a shared cache. This is the HTTP counterpart of the tenant leak from the
Cache Key Design lesson, and it is prevented with a single header line.

`no-store` — a loan approval must not be stored anywhere. This directive is often
confused with `no-cache`: `no-cache` does not forbid storing the response, it only
requires validation before it is used. For a response that must not be stored, the
correct directive is `no-store`.

A separate lifetime can be given for shared caches: `s-maxage` binds only shared
caches, and it overrides `max-age`. This is the route used when a long lifetime is
wanted at the edge layer and a short one in the browser.

#### Vary and the Shared Key

The `Vary: Accept-Language` header on the book response changes the shared cache's
key. The Cache Key Design lesson measured that a cache key must carry every input
that changes the response; in HTTP, these inputs are request headers, and `Vary`
declares them by name. Without this header, the cache uses only the address as the
key, and a user requesting Turkish could be served an English response.

There are two dangerous values for the `Vary` header. A response marked
`Vary: Cookie` or `Vary: Authorization` is technically correct but practically
uncacheable: because every user's cookie differs, every request lands on a separate
key, and the hit ratio drops to zero. This is exactly the "extra field" row measured
in the Cache Key Design lesson. For responses private to one person, the correct fix
is `private`, not `Vary`.

#### Summary

- HTTP caching is two separate mechanisms: `max-age` sets how long a response counts
  as fresh, and the validator determines whether a stale copy is still valid.
- When the conditional request returned 304, the downloaded body dropped from 79
  bytes to 0; the round trip still happened. The saving is in the body, and it grows
  as the body grows.
- When the resource was updated, the tag generated from the body digest and the
  last-modified date both changed; the request made with the old validator got 200
  instead of 304.
- The last-modified date's resolution is one second; the tag generated from the body
  digest has no such limit, but producing it requires computing the body.
- `public` allows a shared cache, `private` allows only the end user's cache,
  `no-store` allows none. `no-cache` does not forbid storing the response; it forbids
  using it without validation.

#### Next Step

Throughout this topic, every lesson produced a number: the number of queries reaching
the origin, the number of stale reads, hits and misses, downloaded bytes. Each of
these numbers was meaningful on its own, but none of them answers the question "is
the cache working" by itself. In the Cache Key Design lesson, the highest hit ratio
belonged to the wrong implementation; in another setup, a high hit ratio can come
from cheap keys that are never even read. The next lesson ties these numbers into a
set of metrics: how the hit ratio is calculated, what it hides, what it does to
average response time once combined with the miss penalty, and how the hit ratio
behaves as cache capacity changes.

### Cache Metrics

Source: https://academia.sh/en/courses/asynchronous-processing/cache-metrics

Calculating the hit ratio and its effect on average response time: the curve the hit ratio traces as capacity grows, what the eviction count means, the per-namespace breakdown a single total ratio hides, the difference between byte hit ratio and request hit ratio, and the measured gain from removing low-repeat data from the cache.

Throughout this topic, every lesson produced a number: queries reaching the origin,
stale reads, hits and misses, downloaded bytes. Each of these numbers was meaningful on
its own, but none of them answers the question "is the cache working" by itself. In the
Cache Key Design lesson, the highest hit ratio belonged to the wrong implementation;
this lesson shows what a single ratio hides even in correct implementations.

The metric's definition is simple. The **hit ratio** is the share of requests served
from the cache:

$$h = \frac{\text{hits}}{\text{hits} + \text{misses}}$$

Average response time is made up of two components: every request checks the cache, and
the ones that miss also go to the origin. If cache access is $t_c$ and origin access is
$t_o$:

$$T = t_c + (1 - h)\,t_o$$

This form explains why an improvement in the hit ratio pays off disproportionately.
When the ratio rises from 90% to 95%, the improvement is five points, but because the
$(1-h)$ factor drops from 0.10 to 0.05, the load placed on the origin is cut in
**half**.

#### Measurement Setup

```sh
# setup.sh — book and loan data for the measurement
rm -f library.db
sqlite3 library.db <<'SQL'
CREATE TABLE book (book_id INTEGER PRIMARY KEY, title TEXT NOT NULL, author TEXT NOT NULL,
                    publication_year INTEGER, branch_id INTEGER NOT NULL);
CREATE TABLE loan (loan_id INTEGER PRIMARY KEY, book_id INTEGER NOT NULL,
                    member_id INTEGER NOT NULL, return_date TEXT);
INSERT INTO book SELECT n, 'Book ' || n, 'Author ' || (n % 40 + 1), 1950 + n % 70, n % 4 + 1
  FROM (WITH RECURSIVE s(n) AS (SELECT 1 UNION ALL SELECT n+1 FROM s WHERE n < 300) SELECT n FROM s);
INSERT INTO loan SELECT n, n % 300 + 1, n % 200 + 1, CASE WHEN n % 5 = 0 THEN NULL ELSE '2025-12-31' END
  FROM (WITH RECURSIVE s(n) AS (SELECT 1 UNION ALL SELECT n+1 FROM s WHERE n < 4000) SELECT n FROM s);
CREATE INDEX loan_member ON loan(member_id);
SQL
sqlite3 library.db "SELECT 'book=' || (SELECT count(*) FROM book) || ' loan=' || (SELECT count(*) FROM loan);"
```

```
book=300 loan=4000
```

The measurement needs a cache that gathers, in one place, the behaviors built piece by
piece throughout this topic: a lifetime, **least-recently-used eviction** once capacity
fills, and counters.

```js
// cache.mjs — a cache with expiry, least-recently-used eviction, and counters
export function openCache({ capacity = 128, lifetime = Infinity, clock = () => Date.now() } = {}) {
  const box = new Map();
  const counter = { hit: 0, miss: 0, eviction: 0, expiry: 0, hitBytes: 0, missBytes: 0 };
  const sizeOf = (d) => Buffer.byteLength(JSON.stringify(d));
  return {
    counter,
    get(key) {
      const entry = box.get(key);
      if (entry === undefined) { counter.miss += 1; return undefined; }
      if (entry.validUntil <= clock()) {
        box.delete(key); counter.expiry += 1; counter.miss += 1; return undefined;
      }
      box.delete(key); box.set(key, entry);   // the most recently used entry moves to the end
      counter.hit += 1; counter.hitBytes += entry.size;
      return entry.value;
    },
    set(key, value) {
      const size = sizeOf(value);
      counter.missBytes += size;
      box.delete(key);
      box.set(key, { value, size, validUntil: clock() + lifetime });
      while (box.size > capacity) { box.delete(box.keys().next().value); counter.eviction += 1; }
    },
    get entryCount() { return box.size; },
    hitRatio: () => counter.hit / (counter.hit + counter.miss),
    byteHitRatio: () => counter.hitBytes / (counter.hitBytes + counter.missBytes),
  };
}
```

The eviction order comes from the `Map`'s insertion order: a read entry is deleted and
re-inserted, moving it to the end; once capacity is exceeded, the entry at the front —
the one unused for the longest — drops.

#### Capacity and Hit Ratio

The first measurement turns a single dial: cache capacity. The workload is 2000
requests and covers three operation types; different key counts produce different
repeat rates.

```js
// capacity.mjs — how the hit ratio and origin queries change as capacity grows
import { DatabaseSync } from "node:sqlite";
import { openCache } from "./cache.mjs";

const db = new DatabaseSync("library.db");
const query = { count: 0 };
const origin = {
  book: (id) => { query.count += 1; return db.prepare(
    "SELECT book_id, title, author, publication_year FROM book WHERE book_id = ?").get(id); },
  stock: (id) => { query.count += 1; return db.prepare(
    `SELECT ? AS branch_id, count(*) AS on_shelf FROM book k WHERE k.branch_id = ? AND NOT EXISTS
     (SELECT 1 FROM loan o WHERE o.book_id = k.book_id AND o.return_date IS NULL)`).get(id, id); },
  list: (id) => { query.count += 1; return db.prepare(
    "SELECT book_id, title FROM book WHERE branch_id = ? ORDER BY book_id").all(id); },
};

function workload(count) {                          // book requests concentrate on 60 titles
  let seed = 20250729;
  const random = () => ((seed = (seed * 1103515245 + 12345) % 2147483648) / 2147483648);
  return Array.from({ length: count }, () => {
    const r = random();
    if (r < 0.75) return ["book", Math.floor(random() * 60) + 1];
    if (r < 0.9) return ["stock", Math.floor(random() * 4) + 1];
    return ["list", Math.floor(random() * 4) + 1];
  });
}

const requests = workload(2000);
console.log(["capacity", "hits", "misses", "hit ratio", "evictions", "origin queries", "duration~"]
  .map((h, i) => (i === 0 ? h.padEnd(10) : h.padStart([0, 8, 8, 14, 11, 16, 11][i]))).join(""));

for (const capacity of [4, 8, 16, 32, 64, 128]) {
  query.count = 0;
  const cache = openCache({ capacity });
  const t = performance.now();
  for (const [kind, id] of requests) {
    const key = `library:s1:g1:t${kind === "book" ? "shared" : `branch-${id}`}:${kind}:${id}`;
    let value = cache.get(key);
    if (value === undefined) { value = origin[kind](id); cache.set(key, value); }
  }
  const duration = Math.round((performance.now() - t) / 10) * 10;
  const c = cache.counter;
  console.log(String(capacity).padEnd(10) + [String(c.hit).padStart(8), String(c.miss).padStart(8),
    (`%` + (cache.hitRatio() * 100).toFixed(1)).padStart(14), String(c.eviction).padStart(11),
    String(query.count).padStart(16), (duration + " ms").padStart(11)].join(""));
}
db.close();
```

{/* mask: (?<=[0-9])[ ]+[0-9]{2,4}(?= ms$) */}
```
capacity      hits  misses     hit ratio  evictions  origin queries  duration~
4              133    1867          %6.7       1863            1867     910 ms
8              289    1711         %14.4       1703            1711     760 ms
16             588    1412         %29.4       1396            1412     460 ms
32            1053     947         %52.6        915             947     200 ms
64            1851     149         %92.5         85             149      10 ms
128           1932      68         %96.6          0              68      10 ms
```

The duration column is rounded to ten milliseconds and is machine-dependent; the first
five columns are not.

The shape of the curve tells the story. When capacity rises from 32 to 64, the hit
ratio jumps from 52.6% to 92.5%; going from 64 to 128, it gains only four points. The
location of the jump gives the workload's **working set**: 60 book keys, 4 stock keys,
and 4 list keys, 68 distinct keys in total. While capacity sits below this number,
every new key evicts another one, and the eviction count comes out nearly equal to the
miss count — at capacity 4, 1867 misses against 1863 evictions. Once capacity passes
the working set, evictions drop to zero, and the remaining 68 misses are just the first
loads.

This gives the right question for a capacity decision: not "how much memory should we
allocate," but "how many keys are in the working set." The eviction counter answers
this question directly. If evictions are close to the miss count, the cache is feeding
its own entries to each other; if they are close to zero, capacity is sufficient.

The duration column shows what the formula predicts. At capacity 4, the hit ratio was
6.7% and the workload took 910 milliseconds; at 92.5%, it fell to 10 milliseconds. The
gap comes from the miss penalty being large in this workload — the stock query is more
expensive than the others.

#### What the Total Ratio Hides

The table above had a single ratio because every part of the workload was
cache-friendly. Real applications are not like that. The workload below adds two
hundred different members' loan histories: a read with many keys and a low repeat
rate.

```js
// breakdown.mjs — what a single hit ratio hides: broken down by namespace and by byte
import { DatabaseSync } from "node:sqlite";
import { openCache } from "./cache.mjs";

const db = new DatabaseSync("library.db");
const origin = {
  book: (id) => db.prepare("SELECT book_id, title, author FROM book WHERE book_id = ?").get(id),
  stock: (id) => db.prepare(
    `SELECT ? AS branch_id, count(*) AS on_shelf FROM book k WHERE k.branch_id = ? AND NOT EXISTS
     (SELECT 1 FROM loan o WHERE o.book_id = k.book_id AND o.return_date IS NULL)`).get(id, id),
  list: (id) => db.prepare("SELECT book_id, title FROM book WHERE branch_id = ? ORDER BY book_id").all(id),
  history: (id) => db.prepare(
    "SELECT loan_id, book_id FROM loan WHERE member_id = ? ORDER BY loan_id").all(id),
};

let seed = 20250729;
const random = () => ((seed = (seed * 1103515245 + 12345) % 2147483648) / 2147483648);
const requests = Array.from({ length: 2000 }, () => {
  const r = random();
  if (r < 0.70) return ["book", Math.floor(random() * 60) + 1];
  if (r < 0.80) return ["stock", Math.floor(random() * 4) + 1];
  if (r < 0.85) return ["list", Math.floor(random() * 4) + 1];
  return ["history", Math.floor(random() * 200) + 1];    // 200 members, low repeat rate
});

function run(cacheable) {
  const cache = openCache({ capacity: 64 });
  const perKind = { book: [0, 0], stock: [0, 0], list: [0, 0], history: [0, 0] };  // [hits, misses]
  let originQueries = 0;
  for (const [kind, id] of requests) {
    if (!cacheable.has(kind)) { origin[kind](id); originQueries += 1; perKind[kind][1] += 1; continue; }
    const key = `library:s1:g1:t${kind === "book" ? "shared" : `branch-${(id % 4) + 1}`}:${kind}:${id}`;
    const previousHits = cache.counter.hit;
    let value = cache.get(key);
    if (value === undefined) { value = origin[kind](id); originQueries += 1; cache.set(key, value); }
    perKind[kind][cache.counter.hit > previousHits ? 0 : 1] += 1;
  }
  return { cache, perKind, originQueries };
}

const { cache, perKind, originQueries } = run(new Set(["book", "stock", "list", "history"]));

console.log(["namespace", "requests", "hits", "misses", "hit ratio"]
  .map((h, i) => (i === 0 ? h.padEnd(10) : h.padStart([0, 8, 8, 8, 14][i]))).join(""));
for (const [name, [i, k]] of Object.entries(perKind)) {
  console.log(name.padEnd(10) + [String(i + k).padStart(8), String(i).padStart(8), String(k).padStart(8),
    ("%" + ((i / (i + k)) * 100).toFixed(1)).padStart(14)].join(""));
}
const c = cache.counter;
console.log(`\ntotal request hit ratio : %${(cache.hitRatio() * 100).toFixed(1)}`);
console.log(`byte hit ratio          : %${(cache.byteHitRatio() * 100).toFixed(1)}`);
console.log(`bytes returned from hits: ${c.hitBytes}   bytes from the origin: ${c.missBytes}`);
console.log(`origin queries          : ${originQueries}`);

const second = run(new Set(["book", "stock", "list"]));    // history is never cached at all
console.log(`\nwith history not cached:`);
console.log(`  book hit ratio        : %${((second.perKind.book[0] / (second.perKind.book[0] + second.perKind.book[1])) * 100).toFixed(1)}`);
console.log(`  evictions             : ${second.cache.counter.eviction} (previously: ${c.eviction})`);
console.log(`  origin queries        : ${second.originQueries}`);
db.close();
```

```
namespace requests    hits  misses     hit ratio
book          1380     965     415         %69.9
stock          202     189      13         %93.6
list           110      84      26         %76.4
history        308      19     289          %6.2

total request hit ratio : %62.8
byte hit ratio          : %51.7
bytes returned from hits: 283673   bytes from the origin: 264714
origin queries          : 743

with history not cached:
  book hit ratio        : %90.7
  evictions             : 77 (previously: 679)
  origin queries        : 449
```

The total ratio is 62.8%, and reported on its own, it gives the impression of a
"moderately effective cache." The breakdown says something else: the loan-history
namespace's hit ratio is 6.2%. Two hundred distinct keys and a low repeat rate make
this data a poor fit for caching. Worse, this namespace is actively harmful — every
entry it places takes up capacity and gets evicted without ever being read again. This
is what drags the book namespace's hit ratio down to 69.9%.

The last section proves this. When loan history was not cached at all, the book
namespace's hit ratio rose from 69.9% to 90.7%, the eviction count fell from 679 to 77,
and the total queries reaching the origin dropped from 743 to 449. **Removing** a
namespace from the cache lowered the load on the origin. This result could only be
found with a broken-down measurement; the total ratio would have looked similar under
both conditions.

The operating rule that follows is this: the hit ratio is measured per namespace. This
is the second job of the namespace segment placed in the key in the Cache Key Design
lesson — making the metric decomposable.

#### Byte Hit Ratio

The second pair of lines in the output shows a separate distinction. The **request hit
ratio** is 62.8%, the **byte hit ratio** is 51.7%. The difference comes from the values
not being equal in size: list and history responses are much larger than a book
record.

The two ratios answer different questions. The request hit ratio says how many calls
did not go to the origin; this number sets the round-trip and query cost. The byte hit
ratio says how much data did not cross the network; this number sets bandwidth and
transfer time. These two metrics, defined for the edge cache in the Static Hosting
lesson of the Frontend Quality course, carry the same distinction in the server cache.

Which one to track depends on the bottleneck. If the origin's query count is under
strain, the request hit ratio is decisive; if the network is under strain, the byte
hit ratio is.

#### Summary

- The hit ratio is the share of requests served from the cache; average time behaves
  as $T = t_c + (1-h)t_o$, which is why rising from 90% to 95% cuts the origin load in
  half.
- While capacity sits below the working set, the eviction count approaches the miss
  count; in a 68-key working set, raising capacity from 32 to 64 made the hit ratio
  jump from 52.6% to 92.5%.
- The eviction counter is the direct measure for a capacity decision: close to the
  miss count means the cache is eating its own entries; close to zero means capacity
  is sufficient.
- A single total ratio hides differences between namespaces: inside a 62.8% total
  there was a namespace at 6.2%, and once that namespace was removed from the cache,
  origin queries fell from 743 to 449.
- The request hit ratio measures the number of calls reaching the origin, and the
  byte hit ratio measures the amount of data that does not cross the network; in the
  same run, they came out to 62.8% and 51.7%.

#### Next Step

The work done throughout this topic can be summed up in a single sentence: reproducing
the same response was prevented. The 1867 origin queries in the capacity measurement
fell to 68; the read path genuinely got shorter.

One path remains that did not get shorter. A cache only helps **reads**; a write always
goes to the origin and adds to the request's duration. Work attached alongside a write
is the same way: a notification sent when a loan is issued, an overdue report produced
at the end of the day, processing an uploaded cover image. None of this is necessary
for the user's response, but all of it delays the response from arriving. The second
way to shorten the request path is not to speed the work up, but to **take the work out
of the request**. The next lesson opens this path: it takes up when running a job
asynchronously is the right call, what drops out of the request's duration, and what
new responsibilities arise in exchange.

## Messaging

Removing work from the request and carrying it between two ends as a message: the distinction between queue, publish–subscribe, and stream models, measuring delivery semantics, the trade-off between ordering and partitioning, and dead-letter queue and retry strategies.

### Rationale for Asynchronous Processing

Source: https://academia.sh/en/courses/asynchronous-processing/rationale-for-asynchronous-processing

Shortening the request path by removing work from it: measuring step costs and success probability, separating work that must stay in the request from work that can be deferred, the changed meaning of the response, and the durability difference between handing off work in memory and writing it to the outbox.

The previous lesson measured the cache hit ratio and showed with a number how much the
read path had shortened. The same lesson also marked a limit: a cache only helps with
reads. A loan request is a write, and on top of that it drags along other work — an
overdue notification to the member, a recalculation of the branch stock summary, a line
added to the monthly report. None of this can be cached, and all of it adds to the
request's duration.

The second way to shorten the request path is not to speed the work up, but to
**remove** it from the request. This lesson answers three questions in order: what must
stay inside the request, what can be removed, and what removing it costs.

#### The Work Inside the Request

The loan request has five steps. Each step's cost is stated in the model below; these
values are not measured durations, they are model values that represent the steps' size
relative to one another. What actually matters to us is how much each step adds to the
total time.

```js
// request-path.mjs — the steps of a loan request and each step's modeled cost
const STEPS = [
  { name: "authorization check",                    ms: 3,   determines_response: true  },
  { name: "writing the loan record",                ms: 9,   determines_response: true  },
  { name: "overdue notification",                   ms: 140, determines_response: false },
  { name: "branch stock summary recalculation",     ms: 260, determines_response: false },
  { name: "monthly report line",                    ms: 55,  determines_response: false },
];

let accumulated = 0;
for (const s of STEPS) {
  accumulated += s.ms;
  const marker = s.determines_response ? "required " : "deferrable";
  console.log(`${s.name.padEnd(40)} ${String(s.ms).padStart(4)} ms  ${marker}  accumulated=${accumulated} ms`);
}

const total = STEPS.reduce((t, s) => t + s.ms, 0);
const required = STEPS.filter((s) => s.determines_response).reduce((t, s) => t + s.ms, 0);
console.log(`\nrequest duration (all in request)      = ${total} ms`);
console.log(`request duration (required work only)  = ${required} ms`);
console.log(`shortening ratio                         = ${(1 - required / total).toFixed(3)}`);

// How many requests can a single worker finish per second?
console.log(`single worker throughput: ${(1000 / total).toFixed(1)} -> ${(1000 / required).toFixed(1)} req/s`);

// Each step is an external dependency; the request succeeds only if all of them run together.
const p = 0.999;
const all = p ** STEPS.length, few = p ** STEPS.filter((s) => s.determines_response).length;
console.log(`request success probability: ${all.toFixed(5)} -> ${few.toFixed(5)}`);
```

```sh
node request-path.mjs
```

```
authorization check                         3 ms  required   accumulated=3 ms
writing the loan record                     9 ms  required   accumulated=12 ms
overdue notification                      140 ms  deferrable  accumulated=152 ms
branch stock summary recalculation        260 ms  deferrable  accumulated=412 ms
monthly report line                        55 ms  deferrable  accumulated=467 ms

request duration (all in request)      = 467 ms
request duration (required work only)  = 12 ms
shortening ratio                         = 0.974
single worker throughput: 2.1 -> 83.3 req/s
request success probability: 0.99501 -> 0.99800
```

Three numbers are worth reading. First, duration: the two steps that do the request's own
work take 12 milliseconds; the remaining 455 milliseconds is dead time the member spends
waiting for a response. Second, throughput: when the same worker no longer carries that
dead time, it finishes forty times more requests per unit of time. Third, success
probability: each step is a separate external dependency, and the request succeeds only
if all of them run together. When five dependencies shrink to two, the failure rate drops
from five in a thousand to two in a thousand. The report server being unreachable no
longer blocks a loan from being issued.

#### Required Work versus Deferrable Work

The criterion for the split is not speed. Even if the overdue notification were fast, that
would not require it to stay inside the request; the real question is: **would the
response be wrong if this step were not done?**

Work stays inside the request in three cases: a step determines the response's content —
the loan ID returned in it — so writing the record cannot be deferred. A step can cause
the request to be rejected — the authorization check or the stock constraint cannot run
afterward, since the member would already have been told "issued." A step protects an
invariant — the rule that a book is never lent to two members at once has to stay inside
the same transaction boundary as the loan record; that is the rule established in the
Transaction Boundaries lesson.

The remaining work can be deferred. What deferrable work has in common is that doing it
**eventually** is enough: the notification is still correct if it goes out two seconds
later, the stock summary still shows a true report if it arrives a minute behind. The
Business Logic Placement topic sets the same criterion for a domain event's listener: the
work handed to the listener is work whose absence still leaves the loan record valid.

#### The Cost of the Handoff: The Meaning of the Response Changes

Once work is removed, the response no longer says "done" — it says "accepted." The
pattern established in the Synchronous and Asynchronous APIs lesson of the Web API Design
course finds its match here: the server returns `202` and reports the resource where the
work can be tracked. This shift has three consequences.

The client cannot learn from the response that the work is finished; it has to ask or be
notified. An error no longer returns to the client; when deferred work fails, there is no
one waiting on a response, so seeing the failure requires a separate mechanism. And a
**message** is born between the two ends: a record one side writes and the other side
later reads.

Where that message gets written is the subject of this lesson's final measurement.

#### The Handoff Must Be Durable

The shortest way to remove work from a request is to leave it for later in the same
process: put it on an array, return the response, process it afterward. The measurement
below shows what this loses. The setup that follows builds the same tables that The Data
Access Layer and Business Logic course used.

```sh
# setup.sh — builds the library database from scratch
rm -f library.db
sqlite3 library.db <<'SQL'
CREATE TABLE branch (branch_id INTEGER PRIMARY KEY, name TEXT NOT NULL);
CREATE TABLE member (member_id INTEGER PRIMARY KEY, name TEXT NOT NULL);
CREATE TABLE book (book_id INTEGER PRIMARY KEY, title TEXT NOT NULL, branch_id INTEGER NOT NULL);
CREATE TABLE loan (loan_id INTEGER PRIMARY KEY, book_id INTEGER NOT NULL,
                    member_id INTEGER NOT NULL, pickup_date TEXT NOT NULL, return_date TEXT);
CREATE TABLE outbox (message_id INTEGER PRIMARY KEY, type TEXT NOT NULL,
                           body TEXT NOT NULL, status TEXT NOT NULL DEFAULT 'pending');
INSERT INTO branch VALUES (1,'Central'),(2,'Bahcelievler');
INSERT INTO member VALUES (1,'Alice Kane'),(2,'Ben Ortiz'),(3,'Clara Diaz');
INSERT INTO book VALUES (1,'Blindness',1),(2,'The Book of Sand',1),(3,'Puslu Kitalar',2);
SQL
```

The first version leaves the side jobs in an in-memory array. The process terminating
before the array is flushed is represented with `process.exit`; its real-world
counterpart is a deployment rolling out a new version or the process crashing.

```js
// in-memory.mjs — side jobs are left in an in-memory array; the process exits without flushing it
import { DatabaseSync } from "node:sqlite";
const db = new DatabaseSync("library.db");
const sideJobs = [];

function issueLoan(memberId, bookId) {
  db.exec("BEGIN IMMEDIATE");
  const loanId = Number(db.prepare(
    "INSERT INTO loan (book_id, member_id, pickup_date) VALUES (?,?,'2025-07-20')")
    .run(bookId, memberId).lastInsertRowid);
  db.exec("COMMIT");
  for (const type of ["overdue_notification", "stock_summary", "report_line"])
    sideJobs.push({ type, loanId });
  return loanId;
}

for (const [memberId, bookId] of [[1, 1], [2, 2], [3, 3]]) issueLoan(memberId, bookId);
console.log(`request done; side jobs pending in memory = ${sideJobs.length}`);
process.exit(0);   // process restarted: array emptied
```

The second version writes the same side jobs to the outbox **in the same transaction** as
the loan record. The outbox pattern was introduced in the Domain Events lesson; its use
here is the same, except what gets written is not a domain event but work to be done.

```js
// with-outbox.mjs — side jobs are written to the outbox in the same transaction as the loan record
import { DatabaseSync } from "node:sqlite";
const db = new DatabaseSync("library.db");

function issueLoan(memberId, bookId) {
  db.exec("BEGIN IMMEDIATE");
  const loanId = Number(db.prepare(
    "INSERT INTO loan (book_id, member_id, pickup_date) VALUES (?,?,'2025-07-20')")
    .run(bookId, memberId).lastInsertRowid);
  for (const type of ["overdue_notification", "stock_summary", "report_line"])
    db.prepare("INSERT INTO outbox (type, body) VALUES (?,?)")
      .run(type, JSON.stringify({ loanId, memberId, bookId }));
  db.exec("COMMIT");
  return loanId;
}

for (const [memberId, bookId] of [[1, 1], [2, 2], [3, 3]]) issueLoan(memberId, bookId);
console.log("request done; side jobs in the outbox");
process.exit(0);   // process restarted
```

The restarted process counts what is left behind.

```js
// count.mjs — restarted process: what is left
import { DatabaseSync } from "node:sqlite";
const db = new DatabaseSync("library.db");
const count = (s) => db.prepare(s).get().n;
console.log(`loan rows = ${count("SELECT count(*) AS n FROM loan")}, ` +
            `recoverable side jobs = ${count("SELECT count(*) AS n FROM outbox WHERE status = 'pending'")}`);
```

```sh
sh setup.sh
node in-memory.mjs
node count.mjs
sh setup.sh
node with-outbox.mjs
node count.mjs
```

```
request done; side jobs pending in memory = 9
loan rows = 3, recoverable side jobs = 0
request done; side jobs in the outbox
loan rows = 3, recoverable side jobs = 9
```

In both versions, three loan records were written and all three members got a fast
response. The difference is in the number of recoverable side jobs: the nine jobs
accumulated in memory vanished along with the process, while the nine jobs written to the
outbox were found by the restarted process. Three members never received a notification,
and no one noticed — because the response returned to the request had already succeeded.

The rule that follows from this is the basic condition of asynchronous processing: **if
work is removed from the request, the place it is removed to must be durable.** The
handoff of the work must share the same atomicity as the transaction it belongs to;
otherwise the loss is silent.

#### When Not to Remove It

There is a cost in the other direction too. Removed work becomes invisible to whoever
made the request: the member sees that they got the book, but does not know whether the
notification arrived. If the state of the work is not tracked with a separate resource,
the failure accumulates somewhere no one sees.

The second cost is delay. Side work gets done "eventually," but if that "eventually" is
not measured, it is unbounded. Three criteria justify keeping work inside the request:
the response's content depends on that work, the work protects an invariant, or the
work's delay is indistinguishable from an error to the user.

The third is complexity. A handoff done with two lines in memory, once it becomes
durable, requires a table, a reader loop, a retry rule, and failure isolation. Everything
after this lesson is about how those four pieces get built.

#### Summary

- Two of the five steps in the request path determined the response; once the remaining
  three were removed, the modeled request duration fell from 467 milliseconds to 12
  milliseconds, and the number of external dependencies fell from five to two.
- The same worker's throughput rose from 2.1 req/s to 83.3 req/s, and the request success
  probability rose from 0.99501 to 0.99800.
- A step stays inside the request if it determines the content of the response, can lead
  to the request being rejected, or protects an invariant; the remaining work can be
  deferred.
- Once work is removed, the meaning of the response shifts from "done" to "accepted";
  because an error no longer returns to the client, its visibility has to be built
  separately.
- When side jobs were accumulated in memory, all nine were lost when the process ended;
  when written to the outbox in the same transaction, all nine were found by the
  restarted process.

#### Next Step

The outbox made the work durable, but on its own it is not a transport mechanism. Who
reads the rows? What happens if two readers take the same row at the same time? If a
reader takes a row and crashes before processing it, how does that work come back? These
are all questions for a single structure: the **queue** that carries messages between two
ends. The next lesson builds the queue in its smallest form — producer, consumer,
acknowledgement, and visibility timeout — and measures what a delivery guarantee actually
means by counting how many times each message gets delivered.

### Message Queues

Source: https://academia.sh/en/courses/asynchronous-processing/message-queues

Building the queue in its smallest form: write, receive, acknowledge and visibility timeout, measuring the redelivery of an unacknowledged message, competing consumers giving a single message to a single side, and how queue depth behaves according to production and consumption rate.

The previous lesson wrote the side jobs to the outbox in the same transaction as the loan
record and showed that all nine jobs were still there after the process restarted. The
questions we left open were these: who reads these rows, what happens if two readers take
the same row at the same time, and if a reader takes a row and crashes before processing
it, how does that work come back?

All three are answered by a single structure. A **message queue** is a durable list that
one side writes to and another side reads from, where the reader decides when what it
read gets deleted. This lesson builds the queue in its smallest form and counts how many
times each message gets delivered.

#### The Queue's Four Operations

The writing side is called the **producer**; the reading side is called the
**consumer**. The contract between them consists of four operations. The producer drops
a message into the queue with `write`. The consumer asks for a message with `receive`.
When the work is done, it tells the queue to delete the message with `acknowledge`. The
fourth is not an operation but a duration: a received message is not given to another
consumer until it is acknowledged, but this protection does not last forever — once the
**visibility timeout** expires, the message becomes visible again.

The model below holds all four. Time is read not from the real clock but from a virtual
clock that is advanced explicitly, so the measurements stay independent of the machine.

```js
// queue.mjs — the smallest queue model, with acknowledgement and a visibility timeout
export class Queue {
  constructor(visibility = 30) {
    this.visibility = visibility;     // how many ms a received message is withheld from others
    this.messages = [];               // {id, body, visibleAt, delivered}
    this.nextId = 1;
    this.now = 0;                     // virtual clock: independent of real time
    this.totalDelivered = 0;
  }
  write(body) {                       // producer side
    this.messages.push({ id: this.nextId++, body, visibleAt: 0, delivered: 0 });
    return this.messages.at(-1).id;
  }
  receive() {                         // consumer side
    const m = this.messages.find((m) => m.visibleAt <= this.now);
    if (m === undefined) return null;
    m.visibleAt = this.now + this.visibility;
    m.delivered += 1;
    this.totalDelivered += 1;
    return { id: m.id, body: m.body, delivered: m.delivered };
  }
  acknowledge(id) {                   // work is done: the message is deleted from the queue
    const before = this.messages.length;
    this.messages = this.messages.filter((m) => m.id !== id);
    return before !== this.messages.length;
  }
  advance(ms) { this.now += ms; }
  get depth() { return this.messages.length; }
  get visible() { return this.messages.filter((m) => m.visibleAt <= this.now).length; }
}
```

In the expected case, the producer writes five jobs and a single consumer receives and
acknowledges all of them.

```js
// happy-path.mjs — five jobs are written, a single consumer receives and acknowledges all of them
import { Queue } from "./queue.mjs";
const q = new Queue();
const JOBS = [
  "overdue_notification loan=1", "overdue_notification loan=2", "stock_summary branch=1",
  "reservation_advance book=3", "report_line loan=1",
];
for (const job of JOBS) q.write(job);
console.log(`produced=${JOBS.length}  queue depth=${q.depth}`);

let processed = 0;
for (let message = q.receive(); message !== null; message = q.receive()) {
  processed += 1;
  q.acknowledge(message.id);
}
console.log(`processed=${processed}  total delivered=${q.totalDelivered}  remaining depth=${q.depth}`);
```

```sh
node happy-path.mjs
```

```
produced=5  queue depth=5
processed=5  total delivered=5  remaining depth=0
```

Five jobs, five deliveries, zero remaining. There is no direct call between producer and
consumer: the producer does not know whether the consumer is running, the consumer does
not know who produced the job. The queue separates the two **in time** — a second kind of
separation layered on top of the loose coupling in domain events.

#### Why Acknowledgement Is a Separate Step

`receive` does not delete the message. If it did, a consumer crashing after taking the
job would mean the job is lost. Instead the message is made temporarily invisible, and
the decision to delete it is left to acknowledgement. The measurement below runs the case
where a consumer terminates without acknowledging its second job.

```js
// unacknowledged.mjs — an unacknowledged message is redelivered once the visibility timeout expires
import { Queue } from "./queue.mjs";
const q = new Queue(30);
for (const job of ["overdue_notification loan=1", "stock_summary branch=1", "report_line loan=1"])
  q.write(job);

// First consumer: it terminates while receiving the second job, without acknowledging it.
const willCrash = new Set(["stock_summary branch=1"]);
for (let message = q.receive(); message !== null; message = q.receive()) {
  const status = willCrash.has(message.body) ? "left unacknowledged" : "acknowledged";
  if (status === "acknowledged") q.acknowledge(message.id);
  console.log(`t=${q.now} consumer-1 message=${message.id} delivered=${message.delivered} ${status}`);
}
console.log(`t=${q.now} depth=${q.depth} visible=${q.visible}`);

q.advance(30);                       // visibility timeout expired
console.log(`t=${q.now} depth=${q.depth} visible=${q.visible}`);

// Second consumer finds the same job.
for (let message = q.receive(); message !== null; message = q.receive()) {
  q.acknowledge(message.id);
  console.log(`t=${q.now} consumer-2 message=${message.id} delivered=${message.delivered} acknowledged`);
}
console.log(`total delivered=${q.totalDelivered}  messages written=3  remaining depth=${q.depth}`);
```

```sh
node unacknowledged.mjs
```

```
t=0 consumer-1 message=1 delivered=1 acknowledged
t=0 consumer-1 message=2 delivered=1 left unacknowledged
t=0 consumer-1 message=3 delivered=1 acknowledged
t=0 depth=1 visible=0
t=30 depth=1 visible=1
t=30 consumer-2 message=2 delivered=2 acknowledged
total delivered=4  messages written=3  remaining depth=0
```

Three messages were written, four deliveries were made. The second message was delivered
twice because its first delivery was left unacknowledged. The measurement's most
important lines are the two in the middle: the unacknowledged message was still sitting
in the queue (`depth=1`) but was not visible (`visible=0`); once the timeout expired, it
became visible again. The visibility timeout is the expiration date on the assumption
that "this job is currently being done by someone else."

Choosing the duration is a trade-off. If it is set shorter than the job's longest
possible duration, the message gets handed to a second consumer while the job is still
running, and the same job gets done twice. If it is set too long, work left behind by a
consumer that really did crash takes a long time to reach anyone. The right fix for
long-running jobs is not to lengthen the timeout, but to have the consumer extend it
while it is still working.

From this comes the queue's basic guarantee: **unacknowledged work is not lost, but not
losing it comes at the price of it possibly being repeated.** The delivery count is not
the number of times a message was processed; it is only the number of times the queue
handed it out.

#### Competing Consumers

The queue's second basic property is that multiple consumers reading from the same queue
**share** the work. This arrangement is called competing consumers: every message goes
to exactly one consumer, because a received message becomes invisible to the others.

```js
// competing-consumers.mjs — three consumers read from the same queue; each message goes to one consumer
import { Queue } from "./queue.mjs";
const q = new Queue();
for (let i = 1; i <= 12; i++) q.write(`overdue_notification loan=${i}`);

const takenBy = new Map();                    // message -> the consumer that handled it
const count = { "consumer-1": 0, "consumer-2": 0, "consumer-3": 0 };
const names = Object.keys(count);
for (let turn = 0; q.depth > 0; turn++) {
  const name = names[turn % names.length];
  const message = q.receive();
  if (message === null) break;
  takenBy.set(message.id, name);
  count[name] += 1;
  q.acknowledge(message.id);
}
console.log(`distribution: ${Object.entries(count).map(([a, n]) => `${a}=${n}`).join("  ")}`);
console.log(`distinct messages=${takenBy.size}  total delivered=${q.totalDelivered}  remaining=${q.depth}`);
```

```sh
node competing-consumers.mjs
```

```
distribution: consumer-1=4  consumer-2=4  consumer-3=4
distinct messages=12  total delivered=12  remaining=0
```

Twelve jobs were split four each across three consumers. The distinct message count
matching the total delivery count says that no message went to more than one consumer.
Because the overdue-notification job can be split three ways, adding a consumer is
directly adding capacity; the loan request itself has no such divisibility.

#### Queue Depth

The number of messages waiting in the queue is called **queue depth**, and it measures
exactly one thing: the difference between production rate and consumption rate. The run
below produces five jobs per round; each consumer finishes three jobs per round.

```js
// depth.mjs — queue depth grows once the production rate exceeds the consumption rate
import { Queue } from "./queue.mjs";

function run(consumerCount, rounds = 6, production = 5, consumptionRate = 3) {
  const q = new Queue();
  const depths = [];
  for (let t = 1; t <= rounds; t++) {
    for (let i = 0; i < production; i++) q.write(`job t=${t} n=${i}`);
    for (let i = 0; i < consumerCount * consumptionRate; i++) {
      const message = q.receive();
      if (message === null) break;
      q.acknowledge(message.id);
    }
    q.advance(1000);
    depths.push(q.depth);
  }
  return depths;
}

for (const n of [1, 2, 3]) {
  const d = run(n);
  console.log(`consumers=${n}  end-of-round depths: ${d.join(" ")}`);
}
```

```sh
node depth.mjs
```

```
consumers=1  end-of-round depths: 2 4 6 8 10 12
consumers=2  end-of-round depths: 0 0 0 0 0 0
consumers=3  end-of-round depths: 0 0 0 0 0 0
```

With a single consumer, depth grows by two every round; the queue is not swallowing the
work, only postponing it. With two consumers, capacity rises to six and depth stays at
zero; a third consumer adds nothing. A queue is not a rate regulator; it smooths out a
burst, it does not close a sustained gap. Depth rising over time is a direct signal of a
consumer shortage, and it is the first metric worth watching.

#### What the Queue Gives and What It Does Not

The queue gives three things: durability of the message, divisibility of the work, and
separation of the parties in time. What it does not give is equally decisive.

It does not give single delivery. As we saw in the measurement, an unacknowledged message
is redelivered; making processing idempotent under repetition is the consumer's
responsibility. The `processed_message` guard built in the Idempotent Transactions lesson
exists exactly for this condition.

It does not give ordering. The moment competing consumers share the work, two messages'
finish order can differ from their write order. If the same book's two events land on two
different consumers, the second one can finish first.

It does not give broadcast. A message goes to exactly one consumer. If the same event
needs to be seen by the notification service, the report, and the search index all at
once, the queue by itself is not enough.

#### Summary

- The queue is defined by four operations: the producer writing, the consumer receiving,
  acknowledging once the work is done, and a received message not being given to anyone
  else during the visibility timeout.
- An unacknowledged message stays in the queue but becomes invisible; once the timeout
  expired it became visible again, and a total of four deliveries were measured for the
  three messages written.
- If the visibility timeout is set shorter than the job's longest duration, the same job
  gets done twice; if set too long, a crashed consumer's work waits a long time.
- With competing consumers, every message goes to a single side: twelve jobs split four
  each across three consumers, and the distinct message count matched the total delivery
  count.
- Queue depth measures the difference between production and consumption rate; with a
  single consumer it grew by two every round, it stayed at zero once a second consumer
  was added, and a third one changed nothing.

#### Next Step

The last of the things the queue does not give is the subject of the next lesson. The
loan-issued event concerns not just the notification service, but also the branch stock
summary and the monthly report. Writing three copies to three separate queues to reach
all three parties ties the producer back to the list of consumers — exactly what we
wanted to avoid. The next lesson builds the publish–subscribe model that breaks that tie,
counts how many deliveries a single message turns into depending on the number of
subscribers, and measures whether a slow subscriber affects the others.

### Publish–Subscribe

Source: https://academia.sh/en/courses/asynchronous-processing/publish-subscribe

Delivering a single message to more than one interested party: the distinction between topic and subscription, counting how many deliveries a single publish turns into depending on subscriber count, isolating a slow subscriber's backlog from the others, measuring copies dropped by a filter, and a subscription created after publication being unable to see the past.

The previous lesson gathered what the queue does not give into three points; the last was
broadcast. A message goes to exactly one consumer, because a received message becomes
invisible to the others. But the loan-issued event concerns three separate parties: the
one that sends the overdue notification to the member, the one that recalculates the
branch stock summary, and the one that adds a line to the monthly report. The same event
needs to reach all three.

This lesson's question is where that "all three" gets written to. The answer is not just
a distribution mechanism, but a separation that keeps the producer from knowing who is
listening.

#### The Producer's Three-Destination Problem

The shortest fix is for the producer to write three copies to three queues. It has a
measurable cost: the producer now knows the names of three destinations. When a fourth
interested party — say, the search index — is added, the loan-issuing code has to change.
When a fifth is added, it changes again. The queue separated producer and consumer in
time, but the producer still carries the **list** of consumers.

This is the coupling problem domain events solve in the Business Logic Placement topic:
the publishing side does not count listeners, it only announces what happened.
Publish–subscribe carries the same separation across the process boundary. The producer
writes to a single **topic**; the topic knows who reads it.

#### Topic and Subscription

The separation has two names, and they must not be confused. A **topic** is the named
channel that messages are written to. A **subscription** is an interested party's own
delivery queue on that topic. Every published message is dropped as one copy into every
subscription that exists at the moment of publication; subscriptions do not affect each
other's progress.

The model below holds both of these. Time is read not from the real clock but from a
budget spent explicitly for each subscription, so the measurements stay independent of
the machine.

```js
// topic.mjs — topic and subscriptions: a published message drops one copy into every subscription
export class Topic {
  constructor() {
    this.subscriptions = new Map();  // name -> {queue, duration, filter, processed, filteredOut}
    this.published = 0;              // number of messages published
    this.copies = 0;                 // number of copies dropped into subscriptions
    this.delivered = 0;              // number of copies processed by subscribers
  }
  subscribe(name, { duration = 1, filter = () => true } = {}) {
    this.subscriptions.set(name, { name, queue: [], duration, filter, processed: 0, filteredOut: 0 });
  }
  publish(type, body) {                        // producer side: a single call
    const message = { id: ++this.published, type, body };
    for (const s of this.subscriptions.values()) {
      if (!s.filter(type)) { s.filteredOut += 1; continue; }
      s.queue.push(message);
      this.copies += 1;
    }
    return message.id;
  }
  advance(budget) {                            // each subscription advances at its own rate
    for (const s of this.subscriptions.values()) {
      let remaining = budget;
      while (s.queue.length > 0 && remaining >= s.duration) {
        s.queue.shift(); s.processed += 1; this.delivered += 1; remaining -= s.duration;
      }
    }
  }
  pending(name) { return this.subscriptions.get(name).queue.length; }
  processed(name) { return this.subscriptions.get(name).processed; }
}
```

The producer's number of calls and the system's number of deliveries are two separate
numbers. The run below raises the subscriber count from one to four while never changing
the producer side at all.

```js
// single-publish.mjs — how many deliveries a single publish call turns into by subscriber count
import { Topic } from "./topic.mjs";

for (const subscribers of [["notification"], ["notification", "stock-summary"],
                           ["notification", "stock-summary", "report"],
                           ["notification", "stock-summary", "report", "search-index"]]) {
  const t = new Topic();
  for (const name of subscribers) t.subscribe(name);
  for (const n of [1, 2, 3, 4]) t.publish("loan_issued", { loanId: n });
  t.advance(1000);
  console.log(`subscribers=${subscribers.length}  publish calls=${t.published}  ` +
              `copies dropped=${t.copies}  delivered=${t.delivered}`);
}
```

```sh
node single-publish.mjs
```

```
subscribers=1  publish calls=4  copies dropped=4  delivered=4
subscribers=2  publish calls=4  copies dropped=8  delivered=8
subscribers=3  publish calls=4  copies dropped=12  delivered=12
subscribers=4  publish calls=4  copies dropped=16  delivered=16
```

In all four rows, the publish call count is four; the number of copies dropped is
multiplied by the subscriber count. That multiplication is both the power and the cost of
the publish–subscribe model. The power: the producer still writes a single line even with
four subscribers, and adding an interested party never touches the producer. The cost:
the work the system carries grows as publish count times subscriber count. The fan-out
multiplier determines the cost of carrying the message, not of producing it.

A subscription having its **own** delivery queue does not invalidate the
competing-consumer arrangement from the previous lesson; it combines with it. Multiple
workers can run inside a single subscription and share the copies that land in it among
themselves. The distinction between the two arrangements fits in one sentence: there is
**copying** between subscriptions, and **sharing** within a subscription.

#### Does a Slow Subscriber Hold Up the Others

This is the real test of a separate delivery queue. Let the monthly-report subscriber run
slower than the others: let it take three time units to process each message, one for
the others. Four events are published every round, and each subscription can spend six
time units per round.

```js
// slow-subscriber.mjs — the report subscriber is three times slower; backlog is measured per subscription
import { Topic } from "./topic.mjs";
const t = new Topic();
t.subscribe("notification", { duration: 1 });
t.subscribe("stock-summary", { duration: 1 });
t.subscribe("report", { duration: 3 });

const row = (...v) => v.map((x) => String(x).padStart(14)).join("");
console.log(row("round", "notification", "stock-summary", "report"), "  (processed/pending)");
for (let round = 1; round <= 6; round++) {
  for (let i = 0; i < 4; i++) t.publish("loan_issued", { round, i });
  t.advance(6);                       // each subscription can spend 6 time units per round
  const g = (name) => `${t.processed(name)}/${t.pending(name)}`;
  console.log(row(round, g("notification"), g("stock-summary"), g("report")));
}
```

```sh
node slow-subscriber.mjs
```

```
         round  notification stock-summary        report   (processed/pending)
             1           4/0           4/0           2/2
             2           8/0           8/0           4/4
             3          12/0          12/0           6/6
             4          16/0          16/0           8/8
             5          20/0          20/0         10/10
             6          24/0          24/0         12/12
```

By the end of six rounds, notification and stock-summary have each finished twenty-four
messages, with none pending. The report subscriber is stuck at twelve messages and has
backlogged twelve. The backlog sits only in the slow subscription; the fast ones'
progress never slowed at all.

What provides this isolation is a separate queue per subscription. If the three
subscribers shared a single queue, the slowest one would sit in front of the others and
set the pace for the whole topic. Separate queues prevent that, but they do not eliminate
the problem: a slow subscriber's backlog occupies real space in memory or on disk, and
its growth is unbounded. The previous lesson's queue-depth metric is tracked here per
subscription; which subscriber is falling behind cannot be seen by looking at the total
count.

#### Not Every Subscriber Wants Every Message

The loan-issued event is not the only thing written to the topic. Returns, member
registrations, and branches being added pass through the same channel too. The
notification subscriber has no interest in member registrations, and the report
subscriber only writes loan-issued lines. Receiving and discarding a message it has no
interest in means wasting the write and read cost of every copy dropped into the
subscription.

A subscription is given a **filter**: a condition that decides, by looking at the
message's type, whether the copy gets dropped in at all. It is decisive that the filter
runs on the publish side; a filtered-out message is never written to the subscription in
the first place.

```js
// filter.mjs — a subscription receives only the types it cares about; filtered messages are not copied
import { Topic } from "./topic.mjs";
const TYPES = ["loan_issued", "return_received", "member_registered", "loan_issued",
               "branch_added", "return_received", "member_registered", "loan_issued"];

function run(filtered) {
  const t = new Topic();
  const f = (set) => filtered ? (type) => set.has(type) : () => true;
  t.subscribe("notification", { filter: f(new Set(["loan_issued", "return_received"])) });
  t.subscribe("stock-summary", { filter: f(new Set(["loan_issued", "return_received", "branch_added"])) });
  t.subscribe("report", { filter: f(new Set(["loan_issued"])) });
  for (const type of TYPES) t.publish(type, {});
  t.advance(1000);
  return t;
}

for (const filtered of [false, true]) {
  const t = run(filtered);
  console.log(`filter=${filtered ? "on" : "off"}  published=${t.published}  copies dropped=${t.copies}  ` +
              `notification=${t.processed("notification")} stock-summary=${t.processed("stock-summary")} ` +
              `report=${t.processed("report")}`);
}
```

```sh
node filter.mjs
```

```
filter=off  published=8  copies dropped=24  notification=8 stock-summary=8 report=8
filter=on  published=8  copies dropped=14  notification=5 stock-summary=6 report=3
```

Eight publishes produce twenty-four copies without a filter; fourteen with one. The
report subscriber processes three messages instead of eight, and never even sees the five
it would have discarded. The filter has a hidden benefit too: what a subscription wants
lives in the subscription's definition, not inside the code. Which events the
notification subscriber cares about can be seen without reading the consumer code.

The filter's limit depends on what is in the message's **header** at the time it decides.
A filter that looks at the type is cheap. A filter that looks at a field inside the body —
"only loans from the Central branch" — requires the message to be parsed, and its cost is
multiplied by the number of subscriptions. This is why the split criterion is usually
placed not on the message body, but on the type or topic given at publish time.

#### The Limit of Publish–Subscribe: There Is No History

The model's most decisive property shows up in a measurement. A copy is dropped only into
the subscriptions that exist **at the moment of publication**. For a subscription that
does not exist at that moment, the message was never written at all.

```js
// late-subscriber.mjs — if a subscription is created after publication, earlier messages never reach it
import { Topic } from "./topic.mjs";
const t = new Topic();
t.subscribe("notification");
t.subscribe("stock-summary");

for (const n of [1, 2, 3]) t.publish("loan_issued", { loanId: n });
t.advance(1000);
console.log(`after three publishes : notification=${t.processed("notification")} stock-summary=${t.processed("stock-summary")}`);

t.subscribe("search-index");          // new interested party joins later
for (const n of [4, 5]) t.publish("loan_issued", { loanId: n });
t.advance(1000);
console.log(`after five publishes: notification=${t.processed("notification")} stock-summary=${t.processed("stock-summary")} ` +
            `search-index=${t.processed("search-index")}`);
console.log(`total published=${t.published}  missed by search-index=${t.published - t.processed("search-index")}`);
```

```sh
node late-subscriber.mjs
```

```
after three publishes : notification=3 stock-summary=3
after five publishes: notification=5 stock-summary=5 search-index=2
total published=5  missed by search-index=3
```

The search index saw two of the five loan-issued events. The three it missed are not
waiting somewhere; those copies never existed at all. The same gap shows up in a second
form: a processed copy is removed from the subscription's queue, so if the search index
loses its own data to a failure, it cannot reread even the messages it already saw.

This is not a flaw in the model, it is its definition. Publish–subscribe **distributes**
the message, it does not store it. A subscriber that wants to keep it has to accumulate
what it receives on its own side. If a new interested party needs to build itself up by
looking at the past, or a subscriber needs to go back and reread, what is needed is not
distribution — it is a **durable record**.

#### Summary

- A topic is the named channel that messages are written to; a subscription is an
  interested party's own delivery queue on that topic. There is copying between
  subscriptions, sharing within a subscription.
- Four publish calls turned into sixteen copies instead of four once subscriber count
  rose from one to four; the producer's code never changed. The fan-out multiplier
  determines the carrying cost.
- Separate delivery queues isolate a slow subscriber: by the end of six rounds the report
  subscriber had backlogged twelve messages while notification and stock-summary had each
  finished twenty-four and were left with zero pending.
- The filter runs on the publish side: eight publishes produced twenty-four copies
  without a filter and fourteen with one, and the report subscriber never even saw the
  five messages it had no interest in.
- A copy is dropped only into subscriptions that exist at the moment of publication; the
  search index, joining later, never saw three of five events. The model distributes the
  message, it does not store it.

#### Next Step

Both the queue and the topic share the same assumption: a delivered message has been
consumed, and a consumed message gets deleted. The cost of that assumption showed up in
the final measurement — a party that connects later cannot see the past, and after a
failure no consumer can go back. The next lesson builds the second transport model, the
one that separates consuming from deleting: a durable, ordered log where messages are
only appended and stay in place once read. The same workload is run on both the queue and
the log; a number shows why replaying is impossible on one and as cheap as rolling back an
offset value on the other.

### Streams versus Queues

Source: https://academia.sh/en/courses/asynchronous-processing/streams-vs-queues

Separating consuming from deleting: an append-only log, a per-consumer offset, comparing a single copy against per-subscription copies in bytes, measuring why replaying is impossible on a queue and cheap on a stream, and what retention means for a consumer that has fallen behind.

The previous lesson's final measurement left a gap: the search index, connecting later,
never saw three of the five loan-issued events, because a copy is only dropped into
subscriptions that exist at the moment of publication. The same gap had a second face —
because a processed copy gets deleted, no consumer could go back and reread it.

Both shortcomings come from a single assumption: **a delivered message has been
consumed, and a consumed message gets deleted.** This lesson builds the second transport
model, the one that drops that assumption, and compares both models on the same
workload.

#### Append and Offset

A **stream** is an ordered record where messages are only appended at the end and stay in
place once read. The structure itself is not new: in the Relational Database
Administration course, the write-ahead log worked by the same rule — records are
appended, no record is modified in place. A stream uses the same structure to carry
messages.

It differs from a queue in two respects. First, reading does not delete: a `read` call
does not change the log. Second, how far a consumer has reached is tracked not by the
queue but by the **consumer**. This value is called the **offset**, and it marks a
record's position in the log.

```js
// log.mjs — an append-only log: reading does not delete a record, the consumer holds the offset
import { appendFileSync, readFileSync, writeFileSync, existsSync } from "node:fs";

export function reset(file) { writeFileSync(file, ""); }

export function append(file, record) {              // writing only ever happens at the end
  appendFileSync(file, JSON.stringify(record) + "\n");
}

export function records(file) {
  if (!existsSync(file)) return [];
  const t = readFileSync(file, "utf8").trimEnd();
  return t === "" ? [] : t.split("\n").map((s) => JSON.parse(s));
}

// returns up to `count` records starting at offset; reading does not change the log
export function read(file, offset, count) {
  return records(file).slice(offset, offset + count);
}
```

That `append`'s only write form is `appendFileSync` is not a convenience, it is the
model's rule. Appending never touches the file's earlier bytes; that is why the data
underneath a consumer reading at the same time never changes, and reads and writes never
lock each other out.

#### One Copy, Three Offsets

In the previous lesson we measured the fan-out multiplier: as subscriber count rose, the
number of copies dropped was multiplied along with it. On a stream, the copy count is
one; what gets multiplied is not the copy but the offset value held. The run below stores
the same six events in both forms and compares the space they take on disk.

```js
// three-consumers.mjs — a single log, an offset per consumer; offsets are held instead of copies
import { statSync } from "node:fs";
import { reset, append, records, read } from "./log.mjs";

const EVENTS = [
  { type: "loan_issued", loanId: 1, bookId: 1, memberId: 1 },
  { type: "loan_issued", loanId: 2, bookId: 2, memberId: 2 },
  { type: "return_received", loanId: 1, bookId: 1, memberId: 1 },
  { type: "loan_issued", loanId: 3, bookId: 3, memberId: 3 },
  { type: "reservation_opened", bookId: 1, memberId: 2 },
  { type: "return_received", loanId: 2, bookId: 2, memberId: 2 },
];

reset("events.log");
for (const e of EVENTS) append("events.log", e);

const NAMES = ["notification", "stock-summary", "report"];
for (const name of NAMES) {                      // the per-subscription-copy measure
  reset(`copy-${name}.log`);
  for (const e of EVENTS) append(`copy-${name}.log`, e);
}
const copyBytes = NAMES.reduce((t, a) => t + statSync(`copy-${a}.log`).size, 0);
console.log(`single log       : ${records("events.log").length} records  ${statSync("events.log").size} bytes`);
console.log(`subscription copy: ${NAMES.length} files  ${copyBytes} bytes`);

const offset = { notification: 0, "stock-summary": 0, report: 0 };
const rate = { notification: 6, "stock-summary": 4, report: 1 };   // records readable per round
for (const name of NAMES) offset[name] += read("events.log", offset[name], rate[name]).length;
console.log(`offsets after one round: ${NAMES.map((a) => `${a}=${offset[a]}`).join("  ")}`);
console.log(`log length after reading = ${records("events.log").length}`);
console.log(`records report is behind by = ${records("events.log").length - offset.report}`);
```

```sh
node three-consumers.mjs
```

```
single log       : 6 records  352 bytes
subscription copy: 3 files  1056 bytes
offsets after one round: notification=6  stock-summary=4  report=1
log length after reading = 6
records report is behind by = 5
```

Three numbers are decisive. Data held on disk: 352 bytes against 1056 bytes — the
copy-holding model takes space directly proportional to the number of interested
parties, while the log carries a single copy. The three consumers sit at different
offsets — 6, 4, and 1 — meaning they are not waiting on each other; isolation comes from
the offset, not the copy. And the most important part is the last two lines: all six
records are still in the log after reading, and how far behind the report is — five
records — is computed with a single subtraction. This gap is a lag metric that can be
measured directly on a stream.

The log file holds one record per line, and its content is readable.

```sh
head -3 events.log
```

```
{"type":"loan_issued","loanId":1,"bookId":1,"memberId":1}
{"type":"loan_issued","loanId":2,"bookId":2,"memberId":2}
{"type":"return_received","loanId":1,"bookId":1,"memberId":1}
```

#### Replaying

This is where the two models part ways. The run below puts six loan-issued events first
into a queue that deletes on acknowledgement, then into a log; in both, one consumer
processes all of them, and afterward a **second** consumer tries to ask for the history.

```js
// replay.mjs — the same workload first on a queue, then on a log; the history is replayed
import { reset, append, records, read } from "./log.mjs";

const EVENTS = [1, 2, 3, 4, 5, 6].map((n) => ({ type: "loan_issued", loanId: n }));

// Queue that deletes on acknowledgement: the essence of the behavior built in lesson 02
class Queue {
  constructor() { this.messages = []; }
  write(b) { this.messages.push(b); }
  receive() { return this.messages.length === 0 ? null : this.messages[0]; }
  acknowledge() { this.messages.shift(); }        // acknowledge: the message is deleted from the queue
  get depth() { return this.messages.length; }
}

const q = new Queue();
for (const e of EVENTS) q.write(e);
let first = 0;
for (let m = q.receive(); m !== null; m = q.receive()) { first += 1; q.acknowledge(); }
let second = 0;
for (let m = q.receive(); m !== null; m = q.receive()) { second += 1; q.acknowledge(); }
console.log(`queue: written=${EVENTS.length} first consumer=${first} ` +
            `depth=${q.depth} consumer joining later=${second}`);

reset("events.log");
for (const e of EVENTS) append("events.log", e);
let offset = 0;
for (const record of read("events.log", offset, 100)) offset += 1;
const next = read("events.log", 0, 100).length;      // second consumer, resetting its offset to zero
console.log(`stream: written=${EVENTS.length} first consumer=${offset} ` +
            `log length=${records("events.log").length} consumer joining later=${next}`);

// Can the same consumer fix its own bug and reprocess starting at the fourth record?
const rewound = read("events.log", 3, 100);
console.log(`stream: offset rewound to 3 -> records reread=${rewound.length} ` +
            `first loan=${rewound[0].loanId}`);
```

```sh
node replay.mjs
```

```
queue: written=6 first consumer=6 depth=0 consumer joining later=0
stream: written=6 first consumer=6 log length=6 consumer joining later=6
stream: offset rewound to 3 -> records reread=3 first loan=4
```

On the queue, the second consumer could read zero records. This is not a configuration
gap: because acknowledgement deletes the message, there was nothing left to read. On the
stream, the second consumer read all six records, and nothing in the log changed to make
that happen.

The third line shows **replay**'s real use. If the report consumer notices it has been
computing incorrectly starting at the fourth event, it only needs to rewind its offset to
3 with the fixed code; the log gives it the same records in the same order. The queue has
no counterpart to this, because there is no offset to rewind — the only state a queue
holds is which message is still there.

The cost sits in the same place. Because the offset lives with the consumer, **saving it
is also the consumer's job**. On a queue, the queue itself holds the "processed"
information; on a stream, unless the consumer writes the offset to its own durable
storage, it has no way to know where to resume when it restarts. When the offset gets
written — before the work or after — is the subject of the next lesson.

#### Retention

A queue empties as it is consumed. A log does not empty; it only grows. This is why, on a
stream, the decision to delete is tied not to consumption but to a rule: **retention**.
The rule can be based on time, size, or record count; what they share is that none of
them look at where the consumers are.

```js
// retention.mjs — records whose retention has expired are trimmed; what a lagging consumer loses
import { writeFileSync } from "node:fs";
import { reset, append, records, read } from "./log.mjs";

let earliest = 0;                             // offset of the oldest record still in the log
function trim(file, keep) {                   // keeps only the newest `keep` records
  const all = records(file);
  const dropped = Math.max(0, all.length - keep);
  writeFileSync(file, all.slice(dropped).map((r) => JSON.stringify(r) + "\n").join(""));
  earliest += dropped;
  return dropped;
}
function readFromOffset(file, offset, count) {  // absolute offset, into the trimmed log
  if (offset < earliest) return null;           // the requested record no longer exists
  return read(file, offset - earliest, count);
}

reset("events.log");
for (let n = 1; n <= 10; n++) append("events.log", { type: "loan_issued", loanId: n });

const offset = { notification: 10, report: 2 };  // report is eight records behind
console.log(`before trim: log=${records("events.log").length} records  earliest offset=${earliest}`);
console.log(`  report offset=${offset.report} -> readable=${readFromOffset("events.log", offset.report, 100).length}`);

const dropped = trim("events.log", 4);
console.log(`after trim: dropped=${dropped}  log=${records("events.log").length} records  earliest offset=${earliest}`);
const reportRead = readFromOffset("events.log", offset.report, 100);
console.log(`  notification offset=${offset.notification} -> readable=${readFromOffset("events.log", offset.notification, 100).length}`);
console.log(`  report offset=${offset.report} -> ${reportRead === null ? "offset trimmed, record cannot be recovered" : reportRead.length}`);
console.log(`records report lost = ${earliest - offset.report}`);
```

```sh
node retention.mjs
```

```
before trim: log=10 records  earliest offset=0
  report offset=2 -> readable=8
after trim: dropped=6  log=4 records  earliest offset=6
  notification offset=10 -> readable=0
  report offset=2 -> offset trimmed, record cannot be recovered
records report lost = 4
```

Before trimming, report could read eight records. Once only the newest four records were
kept, the log's earliest offset became 6, and report's offset was left behind that value;
there is nothing left for it to read. The number of records it lost is the difference
between the two offsets: four.

The notification consumer was not affected by the same trim at all, because its offset
was at the end of the log and it had no new record to read anyway. The metric worth
watching for the rule follows from this: the gap between the most-behind consumer's
offset and the log's earliest offset. As that gap approaches zero, either retention gets
extended or the lagging consumer gets sped up.

#### Which Model Solves What

The two models are not interchangeable; they answer different questions.

A queue is built for **dividing** work. Every message goes to one consumer once, work is
shared among consumers, and processed work is deleted to free up space. Sending an
overdue notification works this way: who sends it does not matter, and once sent, keeping
the message has no value.

A stream is built for **recording** an event. Every consumer reads all the records
starting from its own offset, more than one party can see the same record, and because
the past is not deleted, it can be reread. The branch stock summary derived from loan
events works this way: if the calculation changes, the summary is rebuilt from the log.

A single question is enough to draw the line: **is reading the same record a second time
meaningful?** If the answer is no, use a queue; if yes, use a stream. This distinction is
not free — a stream needs storage for retention, offset management, and a trim rule; a
queue needs none of these.

#### Summary

- A stream is an ordered log where messages are only appended and reading does not
  delete a record; how far a consumer has reached lives not in the queue but in the
  consumer's own offset value.
- Six events took 352 bytes in a single log, versus 1056 bytes across three files when
  copied per subscription; the three consumers advanced at offsets 6, 4, and 1 without
  waiting on each other.
- On the same workload, a consumer joining later could read zero records on the queue
  and six on the stream; a consumer that rewound its offset to 3 reread three records.
- Because the offset lives with the consumer, saving it durably is also the consumer's
  job; on a queue, the queue itself holds that information.
- A log does not empty with consumption, it gets trimmed by a retention rule: once only
  the newest four records were kept, the earliest offset became 6, and the consumer at
  offset 2 was left unable to recover four records.

#### Next Step

In both models, one question was left unanswered: doing the work and writing the "done"
information are two separate steps. On a queue, is the acknowledgement sent before the
work or after? On a stream, is the offset recorded before the work or after? Ordering has
two options, and both lose something — one counts work as processed before it is done,
the other makes the same work happen twice. The next lesson runs the same workload in
three modes: at-most-once, at-least-once, and deduplication with an idempotency key. It
measures the loss count and the repeated-delivery count, and shows why "exactly-once" end
to end is not a mode but a result.

### Delivery Semantics

Source: https://academia.sh/en/courses/asynchronous-processing/delivery-semantics

The three modes determined by the order of acknowledgement and effect: measuring the loss and repeated-effect counts across at-most-once, at-least-once, and deduplicated runs of the same workload, a deduplication guard bringing the effect count rather than the delivery count down to one, and no ordering preventing both when the effect sits outside the database.

The previous lesson compared the two transport models and left the same question
unanswered in both: doing the work and writing the "done" information are two separate
steps. On a queue, is the acknowledgement sent before the work or after? On a stream, is
the offset recorded before the work or after?

The order of the two steps is not an implementation detail; it is the decision that
determines the system's **delivery semantics**. This lesson runs the same workload in two
orderings and a third arrangement, and counts the loss and repeated-effect numbers.

#### What the Three Names Mean

**At-most-once**: every message is processed zero or one times; loss is possible,
repetition is not. Writing the acknowledgement **before** the effect produces this mode —
the work is done after the message is already counted as processed, so if something
happens in between, the work never gets done at all.

**At-least-once**: every message is processed one or more times; repetition is possible,
loss is not. Writing the acknowledgement **after** the effect produces this mode — the
message is counted as processed after the work is done, so if something happens in
between, the message gets redelivered.

**Exactly-once** is not a third ordering. As long as the acknowledgement and the effect
are two separate steps, a gap remains between them, and the process can die in that gap.
Exactly-once is not a delivery mode, it is the **result** of deduplication layered on top
of at-least-once. The measurement will show this with two numbers.

The measurement setup builds the outbox table from the The Data Access Layer and
Business Logic course, an effects table, and the `processed_message` guard introduced in
the Idempotent Transactions lesson. The effect is sending the overdue notification; every
send writes a row to the `notification` relation.

```sh
# setup.sh — outbox, effect table, and processed-message guard for the delivery measurement
rm -f delivery.db
sqlite3 delivery.db <<'SQL'
CREATE TABLE outbox (message_id INTEGER PRIMARY KEY, type TEXT NOT NULL,
                           body TEXT NOT NULL, status TEXT NOT NULL DEFAULT 'pending');
CREATE TABLE notification (notification_id INTEGER PRIMARY KEY, message_id INTEGER NOT NULL,
                       member_id INTEGER NOT NULL);
CREATE TABLE processed_message (message_id INTEGER PRIMARY KEY, processed_at TEXT NOT NULL);
SQL
echo "tables: $(sqlite3 delivery.db "SELECT group_concat(name, ' ') FROM sqlite_master WHERE type='table'")"
```

```
tables: outbox notification processed_message
```

#### Same Workload, Three Modes

Twelve overdue notifications will be processed. In three of them — messages 3, 7, and 11
— the process will die in the exact middle of the first attempt. Death is represented
with an exception; uncommitted work is rolled back, the process restarts, and it retries
whatever messages still appear pending.

```js
// modes.mjs — the same workload in three modes: at-most-once, at-least-once, deduplication
import { DatabaseSync } from "node:sqlite";

const CRASHES = new Set([3, 7, 11]);        // these messages have their first attempt cut in half

function reset(db) {
  for (const t of ["outbox", "notification", "processed_message"]) db.exec(`DELETE FROM ${t}`);
  for (let n = 1; n <= 12; n++)
    db.prepare("INSERT INTO outbox (message_id, type, body) VALUES (?,?,?)")
      .run(n, "overdue_notification", JSON.stringify({ memberId: (n % 3) + 1 }));
}

const send = (db, m) => db.prepare("INSERT INTO notification (message_id, member_id) VALUES (?,?)")
  .run(m.message_id, JSON.parse(m.body).memberId);
const acknowledge = (db, m) => db.prepare("UPDATE outbox SET status='processed' WHERE message_id=?")
  .run(m.message_id);

function process(db, m, mode, crash) {
  if (mode === "at-most-once") {
    acknowledge(db, m);                               // acknowledge first
    if (crash) throw new Error("crash");
    send(db, m);                                       // effect after
  } else if (mode === "at-least-once") {
    send(db, m);                                       // effect first
    if (crash) throw new Error("crash");
    acknowledge(db, m);                                 // acknowledge after
  } else {
    db.exec("BEGIN IMMEDIATE");                      // effect and guard in the same transaction
    if (db.prepare("SELECT 1 AS v FROM processed_message WHERE message_id=?").get(m.message_id) === undefined) {
      send(db, m);
      db.prepare("INSERT INTO processed_message (message_id, processed_at) VALUES (?, '2025-07-20')").run(m.message_id);
    }
    db.exec("COMMIT");
    if (crash) throw new Error("crash");
    acknowledge(db, m);
  }
}

function run(db, mode) {
  reset(db);
  let delivered = 0;
  for (const pass of [1, 2]) {                        // 1: pass with crashes, 2: restart
    for (const m of db.prepare("SELECT message_id, body FROM outbox WHERE status='pending' ORDER BY message_id").all()) {
      delivered += 1;
      try {
        process(db, m, mode, pass === 1 && CRASHES.has(m.message_id));
      } catch {
        if (db.isTransaction) db.exec("ROLLBACK");   // process died: uncommitted work rolled back
      }
    }
  }
  const count = (s) => db.prepare(s).get().n;
  return {
    delivered,
    effect: count("SELECT count(*) AS n FROM notification"),
    lost: 12 - count("SELECT count(DISTINCT message_id) AS n FROM notification"),
    repeated: count("SELECT count(*) AS n FROM notification") - count("SELECT count(DISTINCT message_id) AS n FROM notification"),
  };
}

const db = new DatabaseSync("delivery.db");
console.log("mode".padEnd(16) + "delivered  effect  lost  repeated");
for (const mode of ["at-most-once", "at-least-once", "deduplication"]) {
  const s = run(db, mode);
  console.log(mode.padEnd(16) +
    `${String(s.delivered).padStart(9)}${String(s.effect).padStart(8)}${String(s.lost).padStart(6)}${String(s.repeated).padStart(10)}`);
}
```

```sh
node modes.mjs
```

```
mode            delivered  effect  lost  repeated
at-most-once           12       9     3         0
at-least-once          15      15     0         3
deduplication          15      12     0         0
```

The first row shows the cost of at-most-once. Twelve deliveries were made, nine
notifications were sent; three members never got a notification, and no one noticed,
because the message had already been marked processed. Loss is silent: since no row is
left pending in the outbox, the system sees itself as clean.

The second row shows the opposite direction. No message was lost, but the delivery count
rose from twelve to fifteen and fifteen notifications were sent — three members received
the same notification twice. In this mode, the measure of repetition is directly how many
redeliveries happened.

The third row is the real point, and it needs a careful read. **The delivery count is
still fifteen.** Deduplication did not deduplicate the delivery; the queue still handed
out three messages twice. What changed is the effect: twelve notifications were sent, no
fewer and no more. The guard saw on the second delivery that the message had already been
processed, and skipped the effect.

#### The Condition Under Which Deduplication Works

The third mode giving the correct result depends on a placement detail: writing the
effect and writing the guard row are inside the **same transaction boundary**. The two
writes between `BEGIN IMMEDIATE` and `COMMIT` either commit together or roll back
together. If they were separated, two new flaws would appear: if the guard is written
first and the process dies before the effect is written, the message counts as processed
and the notification never goes out at all — the loss from the first row comes back. If
the effect is written first and the process dies before the guard, the effect gets
reapplied on the second delivery — the repetition from the second row comes back.

The rule established in the Transaction Boundaries lesson shows up here once again: two
writes that must be correct together have to share the same atomic boundary. The guard's
job is to make the fact "this message's effect has been applied" durable **at the same
instant** as the effect itself.

Here, the guard's key was the message's identity. If a message has no such identity, or
different messages can carry the same job, the key is derived from the meaning of the
work; the criterion set in the Idempotency Keys lesson of the Web API Design course is
the same one — the key defines the criterion by which a repeat counts as "the same job."

#### When the Effect Is Outside the Database

In the previous measurement, the effect was a table row, meaning it fit inside the same
transaction as the guard. A real overdue notification goes not to a table but to an
outside service, and that call cannot be wrapped inside a database transaction. The run
below measures this case: the external effect is written to a file, the acknowledgement
is kept in the database, and the process dies exactly between the two.

```js
// external-effect.mjs — when the effect sits outside the database, which order loses what
import { appendFileSync, readFileSync, writeFileSync } from "node:fs";
import { DatabaseSync } from "node:sqlite";

const SENT = "sent.log";               // notifications the notification service actually sent
const RECEIVED = "receiver-keys.log";  // idempotency keys the notification service has seen

function serviceSend(key, hasGuard) {
  if (hasGuard) {
    const seen = new Set(readFileSync(RECEIVED, "utf8").split("\n").filter(Boolean));
    if (seen.has(key)) return false;              // the same key is not processed a second time
    appendFileSync(RECEIVED, key + "\n");
  }
  appendFileSync(SENT, `notification ${key}\n`);
  return true;
}

function run(db, order, hasGuard) {
  writeFileSync(SENT, ""); writeFileSync(RECEIVED, "");
  db.exec("DELETE FROM outbox");
  db.prepare("INSERT INTO outbox (message_id, type, body) VALUES (1,'overdue_notification','{}')").run();
  let calls = 0;

  for (const pass of [1, 2]) {          // 1: pass with a crash, 2: restart
    for (const m of db.prepare("SELECT message_id FROM outbox WHERE status='pending'").all()) {
      const key = `message=${m.message_id}`;
      const acknowledge = () => db.prepare("UPDATE outbox SET status='processed' WHERE message_id=?").run(m.message_id);
      try {
        if (order === "effect first") { calls += 1; serviceSend(key, hasGuard); }
        else acknowledge();
        if (pass === 1) throw new Error("crash");    // process died between the two steps
        if (order === "effect first") acknowledge();
        else { calls += 1; serviceSend(key, hasGuard); }
      } catch { /* process died: retried on the next pass */ }
    }
  }
  const effect = readFileSync(SENT, "utf8").split("\n").filter(Boolean).length;
  return { calls, effect, lost: effect === 0 ? 1 : 0, repeated: Math.max(0, effect - 1) };
}

const db = new DatabaseSync("delivery.db");
console.log("order".padEnd(34) + "calls  effect  lost  repeated");
for (const [name, order, guard] of [["effect first, ack after", "effect first", false],
                                    ["ack first, effect after", "ack first", false],
                                    ["effect first + dedup at receiver", "effect first", true]]) {
  const s = run(db, order, guard);
  console.log(name.padEnd(34) +
    `${String(s.calls).padStart(5)}${String(s.effect).padStart(8)}${String(s.lost).padStart(7)}${String(s.repeated).padStart(11)}`);
}
```

```sh
node external-effect.mjs
```

```
order                             calls  effect  lost  repeated
effect first, ack after               2       2      0          1
ack first, effect after               0       0      1          0
effect first + dedup at receiver      2       1      0          0
```

The first two rows are a closed dilemma. If the send happens first, the notification goes
out twice; if the acknowledgement is written first, the notification never goes out at
all. There is no third ordering in between, because two different systems — the database
and the notification service — share no common transaction boundary. This is exactly the
difficulty established in the Distributed Transaction Problem lesson.

The third row shows where the fix lives: **the call was made twice, the effect happened
once.** The difference is that the guard sits not on the sending side but on the
receiving side. Once the notification service sees the idempotency key and disregards
the second call, the sending side staying in at-least-once mode stops being a flaw.

From this comes the lesson's conclusion: **exactly-once delivery** cannot be built end to
end; what can be built is **exactly-once effect**, and its condition is that a guard
which makes the repeat idempotent sits at the same boundary as the effect. If the effect
is external, that boundary is inside the receiver.

#### Which Mode, Where

At-most-once is not a flaw, it is a deliberate choice; its condition is that the cost of
loss is smaller than the cost of processing. In a usage-metrics stream producing
thousands of rows per second, losing a single row does not change the outcome, and
writing an acknowledgement for every row would multiply the carrying cost.

In the loan system, the criterion reverses. An overdue notification never going out leads
to the member paying a penalty; a line missing from the monthly report shows a wrong
number; a returned book the branch stock summary misses shows as still on the shelf. All
three call for at-least-once. The cost of repetition, though, varies from job to job: if
the same stock summary is calculated twice, the result is the same, but if the same
notification is sent twice, the member gets two messages. The guard is mandatory for the
jobs in that second group.

#### Summary

- Writing the acknowledgement before the effect produces at-most-once; writing it after
  produces at-least-once. Exactly-once is not a separate ordering, it is the result of
  deduplication layered on top of at-least-once.
- On the same twelve-message workload, at-most-once produced 12 deliveries with 9
  effects and 3 losses; at-least-once produced 15 deliveries with 15 effects and 3
  repeated notifications.
- In the deduplicated run, the delivery count stayed at 15, but the effect dropped to
  12: the guard deduplicates the effect, not the delivery.
- The guard working correctly depends on the effect and the guard row being written
  inside the same transaction boundary; when they are separated, loss or repetition
  comes back.
- When the effect sits outside the database, no ordering prevents both: sending first
  meant the notification went out twice, writing the acknowledgement first meant it
  never went out at all; with an idempotency key on the receiving side, two calls
  reduced to a single effect.

#### Next Step

Delivery count and effect count came apart, but neither measurement looked at the
delivery's **order** at all. There are places in the loan system where order matters: if
the same book's "issued" and "returned" events are processed out of order, the stock
summary shows the book as both on the shelf and with the member. In earlier lessons, the
way to speed work up was adding consumers; adding consumers breaks order. The next lesson
measures this tension: where partitioning by key preserves order and where it drops it,
and how concurrency and order violations change as the number of partitions grows.

### Ordering and Partitioning

Source: https://academia.sh/en/courses/asynchronous-processing/ordering-and-partitioning

The trade-off between ordering guarantees and concurrency: measuring, as an inversion count, competing consumers finishing the same key's events out of order; partitioning by key preserving in-key order while giving up global order; the effect of partition count on concurrency; and a hot key unbalancing partition load.

The previous lesson separated the delivery count from the effect count, but neither
measurement looked at the delivery's order at all. There are places in the loan system
where order is decisive. If the same book's "issued" and "returned" events are processed
out of order, the branch stock summary shows the book as both on the shelf and with the
member; the book's final state forms according to the processing order, not the events'
real time.

This lesson measures a tension. So far, the way to speed work up was adding consumers;
adding consumers breaks order. To see the numerical face of the problem, an order
violation first has to be turned into something measurable.

#### How an Ordering Violation Is Counted

Events are given a number that reflects the order they were produced in. After
processing, events are arranged by finish time. If two events land in this arrangement in
the reverse of their production order, an **inversion** occurs — the same count
Algorithms uses to measure sorting performance. Two separate counts are kept: the
**global** inversion count over all pairs, and the **in-key** inversion count only
between events of the same book.

The distinction matters, because the loan system does not need global order. Which order
two different books' events finish in changes nothing. What produces a wrong result when
broken is the order of **the same book's** events.

```js
// partition.mjs — a fixed-seed event sequence, the partitioning run, and the inversion counter
export function generator(seed) {              // fixed-seed linear congruential generator
  let x = seed;
  return () => (x = (x * 1103515245 + 12345) % 2147483648) / 2147483648;
}

export function events(count = 24, seed = 20250720) {
  const rand = generator(seed);
  const TYPES = ["issued", "return_received", "reserved"];
  const list = [];
  for (let order = 0; order < count; order++) {
    const bookId = 1 + Math.floor(rand() * 6);            // book 1..6
    list.push({ order, bookId, type: TYPES[Math.floor(rand() * 3)], duration: 1 + Math.floor(rand() * 3) });
  }
  return list;
}

const sortByFinish = (a) => [...a].sort((x, y) => x.finish - y.finish || x.slot - y.slot);

// Each partition is processed by a single worker, in arrival order.
export function partitioned(list, partitionCount) {
  const finish = new Array(partitionCount).fill(0);
  const result = list.map((e) => {
    const p = e.bookId % partitionCount;                  // partition key: book id
    finish[p] += e.duration;
    return { ...e, slot: p, finish: finish[p] };
  });
  return { result: sortByFinish(result), duration: Math.max(...finish), load: finish };
}

// No partitioning: every message goes to whichever worker is free at that moment.
export function competing(list, workerCount) {
  const available = new Array(workerCount).fill(0);
  const result = list.map((e) => {
    let k = 0;
    for (let i = 1; i < workerCount; i++) if (available[i] < available[k]) k = i;
    available[k] += e.duration;
    return { ...e, slot: k, finish: available[k] };
  });
  return { result: sortByFinish(result), duration: Math.max(...available), load: available };
}

// Number of inversions among events arranged by finish order.
export function violations(byFinish) {
  let global = 0, inKey = 0;
  for (let i = 0; i < byFinish.length; i++)
    for (let j = i + 1; j < byFinish.length; j++)
      if (byFinish[i].order > byFinish[j].order) {
        global += 1;
        if (byFinish[i].bookId === byFinish[j].bookId) inKey += 1;
      }
  return { global, inKey };
}
```

#### Adding Consumers Breaks Order

The first measurement runs the competing-consumer arrangement: every message goes to
whichever worker is free at that moment, with no partitioning. Twenty-four events are
spread across six different books, and their durations are generated from a fixed seed.

```js
// competing-order.mjs — adding workers with no partitioning: duration drops, in-key order breaks
import { events, competing, violations } from "./partition.mjs";
const list = events();

console.log(`events=${list.length}  distinct books=${new Set(list.map((e) => e.bookId)).size}`);
console.log("workers  duration  global inversions  in-key inversions");
for (const n of [1, 2, 3, 4]) {
  const r = competing(list, n);
  const v = violations(r.result);
  console.log(`${String(n).padStart(7)}${String(r.duration).padStart(10)}` +
              `${String(v.global).padStart(19)}${String(v.inKey).padStart(19)}`);
}
```

```sh
node competing-order.mjs
```

```
events=24  distinct books=6
workers  duration  global inversions  in-key inversions
      1        44                  0                  0
      2        22                  4                  1
      3        15                  9                  1
      4        12                 17                  3
```

With a single worker, both numbers are zero: the ordering guarantee is not free, it comes
with the condition of a single consumer. When worker count rises to four, total duration
drops from 44 to 12, but the in-key inversion count rises to three. Those three pairs are
cases where two of the same book's events finish in reverse order — exactly where the
stock summary produces a wrong result.

The result is a trade-off, and neither extreme is acceptable. A single consumer preserves
order but throughput drops to a third. Four consumers give throughput but break
correctness. What is needed sits somewhere in the middle: preserving order **only where
it is needed**.

#### Partitioning by Key

Where it is needed is clear: the same book's events. So let messages be distributed not
to workers, but to **keys**. Let every message fall into a **partition** derived from a
**partition key**, and let every partition be processed by a single worker, in arrival
order.

The rule gives two conditions at once. All of the same key's events fall into the same
partition, so the order between them is preserved. Because different partitions run
independently of one another, concurrency equal to the partition count is achieved.

```js
// partitioned-order.mjs — the same workload keeps its in-key order once partitioned by key
import { events, partitioned, violations } from "./partition.mjs";
const list = events();

console.log("partitions  duration  global inversions  in-key inversions  partition loads");
for (const p of [1, 2, 3, 4, 6]) {
  const r = partitioned(list, p);
  const v = violations(r.result);
  console.log(`${String(p).padStart(10)}${String(r.duration).padStart(10)}` +
              `${String(v.global).padStart(19)}${String(v.inKey).padStart(19)}   ${r.load.join(" ")}`);
}
```

```sh
node partitioned-order.mjs
```

```
partitions  duration  global inversions  in-key inversions  partition loads
         1        44                  0                  0   44
         2        24                 20                  0   24 20
         3        20                 24                  0   20 13 11
         4        21                 25                  0   3 14 21 6
         6        14                 44                  0   14 10 7 6 3 4
```

The in-key inversion column is zero regardless of partition count. The global inversion
count, though, rises together with partition count, reaching 44 at six partitions.
Reading the two columns together is the essence of the lesson: **partitioning buys in-key
order by giving up global order.** What is lost never had value in the first place; what
is preserved determines correctness.

The duration column tells a second truth. Duration drops as partition count rises, but
not evenly: 21 at four partitions, 20 at three. The partition-loads column shows why — at
four partitions the loads are 3, 14, 21, and 6; the other three sit idle until the
fullest partition finishes its work. Raising the partition count raises concurrency
**capacity**, not concurrency actually achieved. What determines real concurrency is how
evenly the load lands across the partitions.

##### Choosing the Partition Key

Choosing the key is not a performance tuning setting, it is a correctness decision: the
ordering guarantee holds exactly within the set the key defines, and not at all outside
it.

If the book's id is chosen as the key, the same book's issue and return events stay
ordered, but the same member's events across two different books do not. If the member's
id is chosen, the opposite holds. The question that decides the criterion is: **which
pair of events, processed out of order, produces a wrong result?** For the stock summary
the answer is the book, because whether a book is on the shelf depends only on that
book's events. For a counter that tracks a member's open loan count, the answer is the
member.

If both are needed, a single key is not enough. In that case, either a coarser key is
chosen — the branch, say — and concurrency drops, or an effect is designed that does not
depend on order. The second is usually cheaper: a version number is attached to the
event, and the consumer applies only a version greater than the one it has already seen.
An effect built this way silently ignores a stale event that arrives late, and needs no
ordering guarantee at all.

#### Hot Key

Partition imbalance is not random; it usually comes from a single key producing far more
events than the others. A heavily requested book can generate more than half of all
transactions on its own. A key like this is called a **hot key**.

```js
// hot-key.mjs — one book's load piles into a single partition; adding partitions does not split it
import { generator, partitioned } from "./partition.mjs";

function hotEvents(count = 30, seed = 7) {
  const rand = generator(seed);
  const list = [];
  for (let order = 0; order < count; order++) {
    const bookId = rand() < 0.6 ? 3 : 1 + Math.floor(rand() * 6);   // book 3 is heavily requested
    list.push({ order, bookId, duration: 1 + Math.floor(rand() * 3) });
  }
  return list;
}

const list = hotEvents();
const total = list.reduce((t, e) => t + e.duration, 0);
const hotLoad = list.filter((e) => e.bookId === 3).reduce((t, e) => t + e.duration, 0);
console.log(`events=${list.length}  total duration=${total}  book 3 events=${list.filter((e) => e.bookId === 3).length} load=${hotLoad}`);
console.log("partitions  duration  partition loads");
for (const p of [1, 2, 4, 8]) {
  const r = partitioned(list, p);
  console.log(`${String(p).padStart(10)}${String(r.duration).padStart(10)}   ${r.load.join(" ")}`);
}

const p = 2;
const hotPartition = 3 % p;
const waiting = partitioned(list, p).result.map((e) => ({ ...e, wait: e.finish - e.duration }));
const avg = (d) => d.length === 0 ? "-" : (d.reduce((t, e) => t + e.wait, 0) / d.length).toFixed(1);
const blocked = waiting.filter((e) => e.slot === hotPartition && e.bookId !== 3);
let t = 0;                                  // if the same events sat in a partition with no hot key
const alone = [...blocked].sort((a, b) => a.order - b.order).map((e) => { const b = t; t += e.duration; return b; });
console.log(`partitions=2  average wait of ${blocked.length} non-book-3 events in the hot partition = ${avg(blocked)}`);
console.log(`              same events in a partition with no hot key                               = ` +
            `${(alone.reduce((a, b) => a + b, 0) / alone.length).toFixed(1)}`);
```

```sh
node hot-key.mjs
```

```
events=30  total duration=59  book 3 events=18 load=28
partitions  duration  partition loads
         1        59   59
         2        41   18 41
         4        28   9 13 9 28
         8        28   0 10 3 28 9 3 6 0
partitions=2  average wait of 5 non-book-3 events in the hot partition = 11.0
              same events in a partition with no hot key                               = 5.0
```

Raising partition count from four to eight changed the duration not at all: 28 in both.
At eight partitions, two partitions sit completely empty, and one carries a load of 28
units on its own. The number is not a coincidence — it equals book 3's total load.
**Because the ordering guarantee makes a key indivisible, the hot key's load is the
system's lower bound**; adding partitions does not lower that bound.

The last two lines show a second cost. Five events that land in the same partition and
have nothing to do with the hot key wait an average of 11 units; on their own in a
partition, they would wait 5. The hot key's work delays the unrelated work sitting behind
it too. This behavior is called **head-of-line blocking** — the queueing-layer
counterpart of the same phenomenon encountered in network layers.

The fix runs through changing the key. If the ordering guarantee is genuinely needed for
the hot key, a single partition is unavoidable, and the only remedy is making that
partition's work cheaper. If it is not needed, the key is refined: a composite key like
"book and event type" instead of just the book spreads the hot book's events across more
than one partition — but that means giving up the order between issue and return too. The
decision comes back to the same question: which pair's order determines correctness.

#### Summary

- An ordering violation is measured as an inversion count; what matters in the loan
  system is not the global inversion count but the in-key inversion count between the
  same book's events.
- Without partitioning, in the competing-consumer arrangement, duration dropped from 44
  to 12 as worker count rose to four, but the in-key inversion count rose from zero to
  three.
- With partitioning by key, the in-key inversion count stayed at zero regardless of
  partition count; the global inversion count rose to 44 at six partitions. Partitioning
  buys in-key order by giving up global order.
- Raising partition count raises concurrency capacity, not concurrency itself: at four
  partitions the loads landed as 3, 14, 21, and 6, so the duration came out worse than
  at three partitions.
- Because a hot key cannot be split, it sets the system's lower bound — duration stayed
  at the same 28 at four and eight partitions — and it raised the average wait of
  unrelated work in the same partition from 5 units to 11.

#### Next Step

Every measurement so far quietly carried one assumption: a retried job eventually
succeeds. The crash was transient, the second attempt finished the job. In reality, some
messages fail on the second attempt too, and the tenth — their body is malformed, the
member they reference has been deleted, a business rule will never accept them. In
at-least-once mode, a message like this never leaves the queue at all, and in a
partitioned arrangement it halts every job behind it. The next lesson builds the
structure that removes these messages from the queue and isolates them: an attempt
counter, a threshold, and a dead-letter queue.

### Dead-Letter Queues

Source: https://academia.sh/en/courses/asynchronous-processing/dead-letter-queues

Isolating messages that will never succeed: unbounded retry letting the wasted-attempt count grow without limit, an attempt counter and threshold draining the queue, the fields a dead-letter record must carry, redriving messages once their cause has been fixed, and the trade-off between the threshold value and wrongly isolating a transient failure.

Every measurement so far quietly carried one assumption: a retried job eventually
succeeds. The crash was transient, the second attempt finished the job. In reality, some
messages fail on the second attempt too, and the tenth — their body cannot be parsed, the
member record they reference has been deleted, the business rule they carry will never
accept them.

In at-least-once mode, a message like this never leaves the queue at all. This lesson
measures how much damage that message causes, and builds the structure that removes it
from the queue and isolates it.

#### Permanent Failure versus Transient Failure

The distinction rests on whether retrying is meaningful. A **transient failure** is one
that disappears once conditions change: a database connection dropping, an outside
service being briefly unreachable, two writes colliding. Retrying is the correct response
to these failures.

A **permanent failure** comes not from conditions but from the message itself. The same
message, run through the same code, produces the same result on every attempt; retrying
only repeats the failure. When both failure types share the same path, the only way to
separate out the permanent one is to look at the attempt count.

The measurement setup builds two relations: the job queue holding messages and attempt
counters, and the dead-letter queue holding isolated messages.

```sh
# setup.sh — job queue and dead-letter queue relations
rm -f queue.db
sqlite3 queue.db <<'SQL'
CREATE TABLE job_queue (message_id INTEGER PRIMARY KEY, type TEXT NOT NULL, body TEXT NOT NULL,
                         attempt INTEGER NOT NULL DEFAULT 0,
                         status TEXT NOT NULL DEFAULT 'pending');
CREATE TABLE dead_letter (dead_letter_id INTEGER PRIMARY KEY, message_id INTEGER NOT NULL, type TEXT NOT NULL,
                         body TEXT NOT NULL, attempt INTEGER NOT NULL, last_error TEXT NOT NULL,
                         source TEXT NOT NULL);
SQL
sqlite3 queue.db "SELECT name FROM sqlite_master WHERE type='table' ORDER BY name"
```

```
dead_letter
job_queue
```

#### The Cost of Unbounded Retry

Twenty overdue notifications will be processed; two of them — messages 7 and 13 —
reference a member record that does not exist, so they will fail on every attempt. The
run below runs the same workload under three settings: eight rounds with no threshold,
twenty-four rounds with no threshold, and eight rounds with a three-attempt threshold.

```js
// threshold.mjs — comparing unbounded retry with threshold-based isolation on the same workload
import { DatabaseSync } from "node:sqlite";

const PERMANENT = new Set([7, 13]);       // messages with a malformed body: fail on every attempt
const COUNT = 20;

function reset(db) {
  db.exec("DELETE FROM job_queue; DELETE FROM dead_letter");
  for (let n = 1; n <= COUNT; n++)
    db.prepare("INSERT INTO job_queue (message_id, type, body) VALUES (?,?,?)")
      .run(n, "overdue_notification", JSON.stringify({ loanId: n }));
}

function process(m) {
  if (PERMANENT.has(m.message_id)) throw new Error(`member not found (loan ${m.message_id})`);
}

function run(db, threshold, roundCount) {     // threshold = 0 -> no isolation
  reset(db);
  let delivered = 0, succeeded = 0, wasted = 0;
  for (let round = 1; round <= roundCount; round++) {
    for (const m of db.prepare("SELECT * FROM job_queue WHERE status='pending' ORDER BY message_id").all()) {
      delivered += 1;
      const attempt = m.attempt + 1;
      db.prepare("UPDATE job_queue SET attempt=? WHERE message_id=?").run(attempt, m.message_id);
      try {
        process(m);
        db.prepare("UPDATE job_queue SET status='processed' WHERE message_id=?").run(m.message_id);
        succeeded += 1;
      } catch (e) {
        wasted += 1;
        if (threshold > 0 && attempt >= threshold) {
          db.prepare(`INSERT INTO dead_letter (message_id, type, body, attempt, last_error, source)
                      VALUES (?,?,?,?,?,'job_queue')`)
            .run(m.message_id, m.type, m.body, attempt, e.message);
          db.prepare("UPDATE job_queue SET status='dead' WHERE message_id=?").run(m.message_id);
        }
      }
    }
  }
  const count = (s) => db.prepare(s).get().n;
  return { delivered, succeeded, wasted,
           depth: count("SELECT count(*) AS n FROM job_queue WHERE status='pending'"),
           dead: count("SELECT count(*) AS n FROM dead_letter") };
}

const db = new DatabaseSync("queue.db");
console.log("mode".padEnd(13) + "rounds  delivered  succeeded  wasted  remaining depth  dead letters");
for (const [name, threshold, rounds] of [["unbounded", 0, 8], ["unbounded", 0, 24], ["threshold=3", 3, 8]]) {
  const s = run(db, threshold, rounds);
  console.log(name.padEnd(13) +
    `${String(rounds).padStart(6)}${String(s.delivered).padStart(11)}${String(s.succeeded).padStart(11)}` +
    `${String(s.wasted).padStart(8)}${String(s.depth).padStart(17)}${String(s.dead).padStart(14)}`);
}
```

```sh
node threshold.mjs
```

```
mode         rounds  delivered  succeeded  wasted  remaining depth  dead letters
unbounded         8         34         18      16                2             0
unbounded        24         66         18      48                2             0
threshold=3       8         24         18       6                0             2
```

The first two rows are snapshots of the same workload at two different durations, and
their only difference is the wasted-attempt count: 16 over eight rounds, 48 over
twenty-four. Successful work stays fixed at 18 in both, remaining depth stays at 2 in
both. Because the two malformed messages never finish, they get redelivered every round;
the capacity spent grows directly proportional to round count, with no upper bound.

The third row shows what the threshold does. Wasted attempts stopped at 6 — the two
messages were tried three times each and removed from the queue. Total delivery is 24
instead of 34. The gain is not only in the count: remaining depth dropped from two to
zero, meaning the queue is **finished**. The two malformed messages have not been lost —
they have been moved somewhere separate.

In the previous lesson's partitioned arrangement, the same message costs even more. If a
malformed message sits at the head of a partition that preserves order, every message
behind it waits in that partition; the ordering guarantee does not permit skipping the
failed message. The threshold is the only mechanism that gets the partition flowing
again.

#### What a Dead-Letter Record Must Carry

The destination a message goes to once it crosses the threshold is called the
**dead-letter queue**. The reason it is a queue is that what sits there is not a log
entry but still a processable **message**. This is why the record carries at least five
fields: the message's body as-is, its type, how many times it was tried, the last
error's text, and which queue it came from. Without the body, the message cannot be
redriven; without the error text, investigating the cause requires repeating the attempt.

```js
// redrive.mjs — the dead-letter record's contents, and redriving messages once the cause is fixed
import { DatabaseSync } from "node:sqlite";
const db = new DatabaseSync("queue.db");

let broken = new Set([7, 13]);
const process = (m) => { if (broken.has(m.message_id)) throw new Error(`member not found (loan ${m.message_id})`); };

db.exec("DELETE FROM job_queue; DELETE FROM dead_letter");
for (let n = 1; n <= 20; n++)
  db.prepare("INSERT INTO job_queue (message_id, type, body) VALUES (?,?,?)")
    .run(n, "overdue_notification", JSON.stringify({ loanId: n }));

function drain(threshold = 3) {
  let processed = 0;
  for (let round = 1; round <= threshold; round++)
    for (const m of db.prepare("SELECT * FROM job_queue WHERE status='pending' ORDER BY message_id").all()) {
      const attempt = m.attempt + 1;
      db.prepare("UPDATE job_queue SET attempt=? WHERE message_id=?").run(attempt, m.message_id);
      try {
        process(m);
        db.prepare("UPDATE job_queue SET status='processed' WHERE message_id=?").run(m.message_id);
        processed += 1;
      } catch (e) {
        if (attempt >= threshold) {                  // threshold crossed: message is isolated
          db.prepare(`INSERT INTO dead_letter (message_id, type, body, attempt, last_error, source)
                      VALUES (?,?,?,?,?,'job_queue')`)
            .run(m.message_id, m.type, m.body, attempt, e.message);
          db.prepare("UPDATE job_queue SET status='dead' WHERE message_id=?").run(m.message_id);
        }
      }
    }
  return processed;
}

const count = (s) => db.prepare(`SELECT count(*) AS n FROM ${s}`).get().n;
console.log(`first pass: processed=${drain()}  pending=${count("job_queue WHERE status='pending'")}`);
for (const r of db.prepare("SELECT * FROM dead_letter ORDER BY message_id").all())
  console.log(`  message=${r.message_id} type=${r.type} attempt=${r.attempt} source=${r.source} ` +
              `body=${r.body} last_error=${r.last_error}`);

broken = new Set();                                  // cause fixed: missing member records were completed
const redriven = db.prepare("SELECT * FROM dead_letter ORDER BY message_id").all();
for (const r of redriven) {
  db.prepare("UPDATE job_queue SET status='pending', attempt=0 WHERE message_id=?").run(r.message_id);
  db.prepare("DELETE FROM dead_letter WHERE dead_letter_id=?").run(r.dead_letter_id);
}
console.log(`redriven=${redriven.length}  second pass: processed=${drain()}`);
console.log(`result: processed=${count("job_queue WHERE status='processed'")} ` +
            `pending=${count("job_queue WHERE status='pending'")} dead letters=${count("dead_letter")}`);
```

```sh
node redrive.mjs
```

```
first pass: processed=18  pending=0
  message=7 type=overdue_notification attempt=3 source=job_queue body={"loanId":7} last_error=member not found (loan 7)
  message=13 type=overdue_notification attempt=3 source=job_queue body={"loanId":13} last_error=member not found (loan 13)
redriven=2  second pass: processed=2
result: processed=20 pending=0 dead letters=0
```

The dead-letter record itself gives the diagnosis of the failure: two messages of the
same type, the same error text, three attempts each. After the missing member records
were completed, the two messages were redriven back to the job queue, their attempt
counters were reset, and they were processed on the second pass. All twenty of the
twenty messages ended up processed.

**Redrive** has two conditions. The attempt counter must be reset, or else the message,
having already crossed the threshold, gets isolated again on its first failure. And the
cause must have genuinely been fixed; a redrive done without fixing it sends the same
messages back to the dead-letter queue with the same number of attempts. The second
condition makes it mandatory that redriving be tied to a diagnosis, not a button.

#### Choosing the Threshold Value

The threshold cannot tell two failures apart; it only looks at the attempt count. Its
value is therefore a choice between two flaws: a small threshold mistakes a transient
failure that would clear on the fourth attempt for a permanent one and isolates it; a
large threshold tries a permanently failing message more times than it needs to.

The run below sets up two of twenty messages as permanently broken, five as clearing on
the third attempt, and four as clearing on the fifth, and varies the threshold from one
to eight.

```js
// threshold-selection.mjs — the threshold value: choosing between wrongful isolation and wasted attempts
const MESSAGES = [];                       // needed = the attempt number that succeeds, 0 = never
for (let n = 1; n <= 20; n++) {
  const needed = n === 7 || n === 13 ? 0 : (n % 5 === 0 ? 5 : (n % 3 === 0 ? 3 : 1));
  MESSAGES.push({ message_id: n, needed });
}

function run(threshold) {
  let wasted = 0, wronglyIsolated = 0, permanentlyIsolated = 0, processed = 0;
  for (const m of MESSAGES) {
    for (let attempt = 1; attempt <= threshold; attempt++) {
      if (m.needed !== 0 && attempt >= m.needed) { processed += 1; break; }
      wasted += 1;
      if (attempt === threshold) {         // threshold reached: message goes to the dead-letter queue
        if (m.needed === 0) permanentlyIsolated += 1; else wronglyIsolated += 1;
      }
    }
  }
  return { processed, wasted, wronglyIsolated, permanentlyIsolated };
}

console.log(`messages=${MESSAGES.length}  permanently broken=${MESSAGES.filter((m) => m.needed === 0).length}  ` +
            `clears on attempt 3=${MESSAGES.filter((m) => m.needed === 3).length}  ` +
            `clears on attempt 5=${MESSAGES.filter((m) => m.needed === 5).length}`);
console.log("threshold  processed  wasted attempts  wrongly isolated  permanently isolated");
for (const threshold of [1, 2, 3, 5, 8]) {
  const s = run(threshold);
  console.log(`${String(threshold).padStart(9)}${String(s.processed).padStart(11)}${String(s.wasted).padStart(17)}` +
              `${String(s.wronglyIsolated).padStart(18)}${String(s.permanentlyIsolated).padStart(21)}`);
}
```

```sh
node threshold-selection.mjs
```

```
messages=20  permanently broken=2  clears on attempt 3=5  clears on attempt 5=4
threshold  processed  wasted attempts  wrongly isolated  permanently isolated
        1          9               11                 9                    2
        2          9               22                 9                    2
        3         14               28                 4                    2
        5         18               36                 0                    2
        8         18               42                 0                    2
```

The table moves monotonically in both directions. As the threshold grows, the wrongly
isolated message count drops from 9 to 0, and the wasted-attempt count rises from 11 to
42. Processed work saturates at a threshold of five: raising it to eight gains not a
single message, it only tries the two permanently broken messages three more times each.

This also tells us how to choose the threshold. The right value sits a little above the
highest attempt count the work genuinely needs; anything past that is pure waste. The
"attempt count genuinely needed" is not guessed, it is measured — the distribution of
attempt counts across successful jobs gives this value directly.

#### Isolating Is Not Solving

The danger of a dead-letter queue is that it makes the problem invisible. The messages
are there, the queue looks clean, the metrics look healthy — and three members never
received their notification. Isolation only stops the loss from being silent if it is
monitored.

Three rules provide that. Dead-letter queue depth is a monitored metric, and any value
other than zero produces an alert; unlike ordinary queue depth, there is no "normal"
baseline here. Every record has an owner: whichever team produces the message also
examines the isolated one. And records have an age — a dead letter sitting unexamined for
months means it is never going to be examined.

**Clustering** the isolated messages is also a diagnostic tool. A large number of
records carrying the same error text points to a single root cause and does not need to
be examined one by one; scattered errors say each one is its own case. This is the real
payoff of storing the error text in the dead-letter record.

#### Summary

- A transient failure disappears once conditions change and responds to retrying; a
  permanent failure comes from the message itself and produces the same result on every
  attempt.
- With unbounded retry, the two malformed messages produced 16 wasted attempts over
  eight rounds and 48 over twenty-four; successful work stayed at 18 in both, and the
  capacity spent had no upper bound.
- With a three-attempt threshold, wasted attempts stopped at 6, the queue emptied, and
  two messages moved to the dead-letter queue; in a partitioned arrangement, the
  threshold gets the partition a malformed message halted flowing again.
- A dead-letter record carries the body, the type, the attempt count, the last error
  text, and the source queue; once the cause was fixed, two redriven messages with reset
  attempt counters were processed.
- As the threshold grew, wrongly isolated messages dropped from 9 to 0 and wasted
  attempts rose from 11 to 42; the right value sits a little above the highest attempt
  count successful jobs actually need.

#### Next Step

The threshold said how many times to retry; it did not say when to retry. In earlier
measurements, attempts were made back to back, with no wait in between. Most transient
failures come from a resource being overloaded, and retrying without waiting loads that
resource even more; worse, hundreds of consumers failing at the same moment and retrying
at the same moment pile the load onto a single point. The next lesson establishes how
wait durations are calculated: exponential backoff, an upper bound, and the jitter that
spreads out the wave of retries.

### Poison Messages and Retry

Source: https://academia.sh/en/courses/asynchronous-processing/poison-messages-and-retry

Isolating the message that brings down the consumer, and calculating wait durations: the necessity of incrementing the attempt counter at delivery time, comparing immediate, fixed, and exponential schedules by total wait and retry budget, jitter lowering the peak of the retry wave, and reconciling backoff with the visibility timeout.

The previous lesson noted that the threshold said how many times to retry, but not when.
This lesson's two topics come out of that gap. First, a failure type the threshold does
not handle: a message that does not leave the consumer processing it standing. Second,
how the wait between attempts gets calculated.

Both will be measured on the same job: overdue notifications taken from the outbox.

#### Poison Message

The malformed messages in the previous lesson behaved politely: they threw an error, the
consumer caught it, incremented the attempt counter, and moved on to the next message. A
**poison message** does not do this. Processing it terminates the consumer process — a
body that exhausts memory, a parse that enters an infinite loop, a failure that cannot be
caught. Because the process dies, the error-handling code never runs.

The result is a measurable vicious cycle. The process restarts, receives the same message
from the queue, and dies again. If the attempt counter is incremented when the error is
caught, it never gets incremented at all; the threshold is never reached; the dead-letter
queue never kicks in.

The run below makes the third message in a five-message queue lethal, and compares two
different positions for the counter: one that increments when the error is caught, and
one that increments at delivery time, before the work begins.

```js
// poison-counter.mjs — the message that kills the process: which counter position makes isolation work
const LETHAL = new Set([3]);              // this message terminates the consumer process
const THRESHOLD = 3;

// A single process lifetime: it dies on the lethal message, the code behind it never runs.
function processLifetime(queue, counterAtDelivery, dead) {
  let delivered = 0;
  for (const m of queue.filter((m) => m.status === "pending")) {
    delivered += 1;
    if (counterAtDelivery) {              // counter increments at delivery time, before the work
      m.attempt += 1;
      if (m.attempt >= THRESHOLD) { m.status = "dead"; dead.push(m.id); continue; }
    }
    if (LETHAL.has(m.id)) return { delivered, died: true };
    m.status = "processed";               // counter would have incremented on catch; it does not when the process dies
  }
  return { delivered, died: false };
}

function run(counterAtDelivery, maxRestarts = 12) {
  const queue = [1, 2, 3, 4, 5].map((id) => ({ id, attempt: 0, status: "pending" }));
  const dead = [];
  let delivered = 0, deaths = 0, restarts = 0;
  for (let b = 1; b <= maxRestarts; b++) {
    restarts = b;
    const r = processLifetime(queue, counterAtDelivery, dead);
    delivered += r.delivered;
    if (r.died) deaths += 1;
    if (queue.every((m) => m.status !== "pending")) break;
  }
  return { restarts, delivered, deaths, deadLetters: dead.length,
           processed: queue.filter((m) => m.status === "processed").length,
           remaining: queue.filter((m) => m.status === "pending").length };
}

console.log("counter position".padEnd(22) + "restarts  delivered  process deaths  processed  dead letters  remaining");
for (const [name, atDelivery] of [["on catch", false], ["at delivery", true]]) {
  const s = run(atDelivery);
  console.log(name.padEnd(22) +
    `${String(s.restarts).padStart(8)}${String(s.delivered).padStart(11)}${String(s.deaths).padStart(16)}` +
    `${String(s.processed).padStart(11)}${String(s.deadLetters).padStart(15)}${String(s.remaining).padStart(11)}`);
}
```

```sh
node poison-counter.mjs
```

```
counter position      restarts  delivered  process deaths  processed  dead letters  remaining
on catch                    12         14              12          2              0          3
at delivery                  3          7               2          4              1          0
```

In the first row, all twelve of twelve restarts ended in death, only two messages were
processed, and three stayed in the queue. The number twelve is not a result, it is the
run's upper bound: the loop stopped because it was stopped there, not on its own. The
dead-letter column is zero — a threshold existed, but because the counter never
incremented, it was never reached.

In the second row, three restarts were enough. The poison message was delivered twice;
on the third delivery, because the counter had reached the threshold, it was isolated
before the work even started. The two messages behind it were both processed. The queue
emptied.

The rule fits in one sentence: **the attempt counter must be incremented at delivery
time, independent of the work's outcome.** Placing the counter on the error path only
protects against cases where the error path actually runs. This is why the `attempt`
field built in the previous lesson was written before the work; the same reasoning holds
for the queue's own delivery count too — the delivery count we measured in lesson 02
incremented on the `receive` call, not when the work finished.

Isolation bounds the poison message's cost but does not zero it out. Two process deaths
happened, and whatever other work those processes were holding at the time was left
half-done. This is why a second defense is used for jobs carrying poison-message risk:
validating the body before it is processed. The criterion set in the Validation Layers
lesson applies here too — a malformed body never enters the work at all, it is taken
straight to the dead-letter queue.

#### Calculating the Wait Duration

In the runs so far, attempts were made back to back, with no wait between them. Most
transient failures come from a resource being overloaded, and retrying without waiting
loads that resource even more. The rule that gives the wait duration is called a
**backoff schedule**.

```js
// backoff.mjs — wait schedules, and how many attempts fit in a retry budget
const BASE = 500, CAP = 30_000;                 // ms
const SCHEDULES = {
  "immediate": () => 0,
  "fixed 2 s": () => 2000,
  "exponential": (n) => BASE * 2 ** (n - 1),
  "exponential + cap": (n) => Math.min(CAP, BASE * 2 ** (n - 1)),
};
const NAMES = Object.keys(SCHEDULES);

console.log("attempt" + NAMES.map((a) => a.padStart(19)).join(""));
for (let n = 1; n <= 8; n++)
  console.log(String(n).padStart(7) + NAMES.map((a) => String(SCHEDULES[a](n)).padStart(19)).join(""));

const total = (name, count) => Array.from({ length: count }, (_, i) => SCHEDULES[name](i + 1))
  .reduce((t, x) => t + x, 0);
console.log("total" + NAMES.map((a) => String(total(a, 8)).padStart(19)).join(""));

const BUDGET = 5 * 60 * 1000;                       // 5-minute time budget
console.log("\nschedule".padEnd(19) + "attempts fit in 5 min  time elapsed when budget runs out");
for (const name of NAMES) {
  let elapsed = 0, attempt = 0;
  while (elapsed + SCHEDULES[name](attempt + 1) <= BUDGET && attempt < 1000) {
    attempt += 1; elapsed += SCHEDULES[name](attempt);
  }
  console.log(name.padEnd(19) + String(attempt === 1000 ? "unlimited" : attempt).padStart(23) +
              String(elapsed).padStart(28) + " ms");
}
```

```sh
node backoff.mjs
```

```
attempt          immediate          fixed 2 s        exponential  exponential + cap
      1                  0               2000                500                500
      2                  0               2000               1000               1000
      3                  0               2000               2000               2000
      4                  0               2000               4000               4000
      5                  0               2000               8000               8000
      6                  0               2000              16000              16000
      7                  0               2000              32000              30000
      8                  0               2000              64000              30000
total                  0              16000             127500              91500

schedule          attempts fit in 5 min  time elapsed when budget runs out
immediate                        unlimited                           0 ms
fixed 2 s                              150                      300000 ms
exponential                              9                      255500 ms
exponential + cap                       14                      271500 ms
```

Four schedules produce four separate behaviors. Immediate retry never waits: the time
budget never runs out, so only the threshold can put a limit on attempts. Far from
reducing the resource's load, it concentrates the load at the moment of failure.

A fixed interval is predictable, but wrong at both ends: waiting two seconds for a
failure that would clear in one second is unnecessary, and retrying every two seconds
during an outage that lasts a minute wears the resource out with a hundred and fifty
pointless attempts.

**Exponential backoff** solves both by doubling the wait duration on every attempt: if
the failure is short, early attempts catch it; if it is long, attempts spread thin. Its
cost shows up in the seventh row — the wait has climbed to 32 seconds. This is why a
**cap** is imposed; in the capped version, total wait drops from 127.5 seconds to 91.5
seconds, and fourteen attempts fit into the five-minute budget instead of nine. The cap
keeps the thinning-out from becoming unbounded.

The same schedules were built for the client side in the Retry and Backoff lesson of the
Application Architecture: Routing, State and Data course. The difference on the consumer
side is that the party waiting is not a user but a message: queue depth keeps rising
during the wait, and the delay lands not just on that message but on all the work behind
it.

#### Jitter

Even when a backoff schedule is correct for a single consumer, it creates a new problem
once a large number of consumers fail at the same time. Because they all use the same
schedule, they all retry at the same time; the resource, trying to recover, meets wave
after wave arriving together.

The fix is adding randomness to the wait duration. This is called **jitter**. **Full
jitter** picks the wait randomly between zero and the calculated duration;
**decorrelated jitter** picks it, at every step, from a wider range that depends on the
previous wait.

The run below tracks two hundred consumers that fail at the same instant, over four
attempts, and buckets the retry moments into hundred-millisecond slots. The generator is
fixed-seed; the numbers are the same on every run.

```js
// jitter.mjs — the retry wave of 200 consumers that fail at the same instant
const generator = (seed) => { let x = seed; return () => (x = (x * 1103515245 + 12345) % 2147483648) / 2147483648; };
const CONSUMERS = 200, ATTEMPTS = 4, BUCKET = 100;      // bucket width 100 ms
const BASE = 500, CAP = 30_000;

function wave(mode, seed = 20250720) {
  const rand = generator(seed);
  const moments = [];
  for (let c = 0; c < CONSUMERS; c++) {
    let moment = 0, previous = BASE;
    for (let n = 1; n <= ATTEMPTS; n++) {
      const base = Math.min(CAP, BASE * 2 ** (n - 1));
      if (mode === "no jitter") moment += base;
      else if (mode === "full jitter") moment += Math.round(rand() * base);          // [0, base]
      else {                                                                       // decorrelated
        previous = Math.min(CAP, Math.round(BASE + rand() * (previous * 3 - BASE)));
        moment += previous;
      }
      moments.push(moment);
    }
  }
  const buckets = new Map();
  for (const m of moments) { const b = Math.floor(m / BUCKET); buckets.set(b, (buckets.get(b) ?? 0) + 1); }
  return { requests: moments.length, peak: Math.max(...buckets.values()), filledBuckets: buckets.size,
           lastBucket: Math.max(...buckets.keys()) };
}

console.log("schedule".padEnd(20) + "total requests  peak per 100 ms  filled buckets  last bucket");
for (const mode of ["no jitter", "full jitter", "decorrelated jitter"]) {
  const s = wave(mode);
  console.log(mode.padEnd(20) + String(s.requests).padStart(15) + String(s.peak).padStart(17) +
              String(s.filledBuckets).padStart(17) + String(s.lastBucket).padStart(14));
}
```

```sh
node jitter.mjs
```

```
schedule            total requests  peak per 100 ms  filled buckets  last bucket
no jitter                       800              200                4            75
full jitter                     800               69               65            67
decorrelated jitter             800               28              163           389
```

Across all three schedules, the total request count is the same: eight hundred. The only
thing that changes is how those requests spread out over time. In the no-jitter
schedule, eight hundred requests fit into just four buckets, with two hundred requests
per bucket — all two hundred consumers knock on the door in the same hundred
milliseconds. Full jitter brings the peak down to 69 and spreads the requests over
sixty-five buckets. Decorrelated jitter brings the peak down to 28, but the last bucket
is 389 — meaning the spread stretches out to thirty-nine seconds.

This is where the trade-off sits. Jitter lowers the burst load the resource sees; in
exchange, some messages' retries run late. If the peak of the load is large enough to
bring the resource down again, jitter is mandatory; if the job is highly sensitive to
delay, the spread is kept narrow.

#### Retry Budget and Visibility Timeout

A retry rule is defined together with two limits. The **retry budget** looks at count:
the threshold. The time budget looks at duration: how long after the first delivery a
message gets taken to the dead-letter queue. Both are needed together, because the
threshold alone, combined with backoff, makes the duration unpredictable — in the capped
exponential schedule, fourteen attempts took longer than four and a half minutes;
uncapped, nine attempts reached the same duration.

The second reconciliation is with the queue itself. The visibility timeout built in
lesson 02 said how long a received message stays withheld from others. If the consumer
holds onto the message and waits through the backoff duration, the moment that wait
exceeds the visibility timeout, the message becomes visible to another consumer and the
same job runs concurrently. The correct placement is for the consumer not to hold the
message while waiting: the message is put back into the queue, and the moment it becomes
visible again is pushed forward by the backoff duration. This way the wait is kept on the
queue's own timeline, and the consumer process does other work in the meantime.

A third limit concerns the failure rate. When a resource becomes wholly unreachable,
every message gets the same error, and retries by themselves turn into a source of load.
In this case, what needs to stop is not the schedule of individual messages but the whole
consumer; this is the problem the **circuit breaker** pattern, introduced in the Retry
and Backoff lesson mentioned above, solves.

#### Summary

- A poison message is not merely a message that fails; because it terminates the
  consumer process, it never runs the error-handling code at all.
- With the counter incremented on catch, all twelve of twelve restarts ended in death,
  two messages were processed, and three stayed in the queue; incremented at delivery
  time, three restarts were enough, the message was isolated on its third delivery, and
  the queue emptied.
- Capped exponential backoff brought total wait over eight attempts down from 127.5
  seconds to 91.5 seconds, and fit fourteen attempts into the five-minute budget instead
  of nine.
- Two hundred consumers' eight hundred retries piled into four buckets under the
  no-jitter schedule with a peak of 200; full jitter brought the peak down to 69,
  decorrelated jitter to 28, stretching the spread out to thirty-nine seconds.
- The retry budget and the time budget are defined together; the backoff duration must
  not exceed the visibility timeout, and the wait must be kept on the queue's timeline,
  not the consumer's.

#### Next Step

The message is now durable. It is written to a persistent place, reaches more than one
interested party, keeps its order where order is needed, has its repetition made
idempotent, has what it cannot process isolated, and has its retries tied to a measured
schedule. Across all these measurements, one side was always taken as a given: the
consumer processing the message. How many worker processes are there, how many messages
does one finish per second, what is gained by raising that number, what happens when the
production rate exceeds their capacity? The next topic treats the consumer as a unit of
scaling, and it starts with the Worker Processes lesson.

## Background Jobs

The worker process as a unit of scaling and supervision of a crashed worker, the recurring task definition and the lock that prevents overlap, progress reporting and cancellation in a long-running job, backpressure where production rate exceeds consumption rate, and load shedding that enables controlled rejection under overload.

### Worker Processes

Source: https://academia.sh/en/courses/asynchronous-processing/worker-processes

Treating the job as a unit of scaling separate from the message: setting up a worker pool, measuring how worker count affects throughput, replacing a crashed worker under supervision, and graceful shutdown.

The previous lesson finished making the message itself durable: attempt counter,
exponential backoff, jitter, and the dead-letter queue. A message no longer gets lost;
it is retried when it fails, and isolated when it does not recover.

All of those guarantees concerned the message itself. The party that **processes** the
message is still standing in as a nameless "consumer." Yet that party is a process: it
holds memory, it crashes, it restarts, it gets shut down. How many of it there are is a
decision, and that decision determines how many jobs the service finishes per hour. This
lesson treats the consumer as a **unit of scaling** and teaches how to measure how many
of that unit there should be.

#### What Is a Worker

A **worker** is an independent unit of execution that pulls a job from the queue and
processes it. It can be a separate process, a separate thread, or a separate container;
what they share is that each **can crash on its own and be replicated on its own**.

Three properties separate a worker from a server handling requests. A worker has no
client; nobody is waiting for a response, so the job can take as long as it needs. A
worker's source of work is not a network listener but a queue; it cannot refuse to
accept load, only consume it slowly. And a worker scales **horizontally**: since a
second worker feeds from the same queue, adding one requires no coordination.

The jobs moved to the background in the library loan service are well known: sending
overdue notifications, generating the monthly loan report, processing an uploaded cover
image, recomputing the branch inventory summary. This lesson takes the first as its
example.

#### The Job Itself

A worker process carries a single responsibility: run an incoming job definition and
report the result. It does not know where the queue stands, how many times the job has
been attempted, or how many workers exist.

```js
// worker.mjs — takes a job, does it, reports the result; pool.mjs forks it
if (!process.send) {
  console.log("this file does not run standalone; pool.mjs forks it");
  process.exit(0);
}

const INDEX = Number(process.argv[2]);            // sequence number the pool assigns to the worker
const wait = (ms) => new Promise((c) => setTimeout(c, ms));

// Sending the overdue notification: an outbound call that takes 100 ms.
async function sendNotification(job) {
  if (job.broken) process.exit(1);                // the job crashes the process itself
  await wait(100);
  return `member ${job.memberId} notified`;
}

process.on("message", async (job) => {
  const result = await sendNotification(job);
  process.send({ type: "done", jobId: job.jobId, index: INDEX, result });
});

process.on("SIGTERM", () => process.exit(0));    // the job is not left half-done when the pool shuts down
```

```
this file does not run standalone; pool.mjs forks it
```

When the file runs directly, `process.send` is undefined, and the worker says so and
exits. The only case where it does anything meaningful is when a parent process forks
it.

The 100-millisecond wait inside `sendNotification` stands in for an outbound call. It
marks the job as **I/O-bound**: the duration passes not by using the processor but by
waiting for a response. That distinction will shortly decide the worker-count question.

#### The Pool and Pull-Based Dispatch

There are two ways to split jobs among workers. In **push-based** dispatch, the parent
process divides the jobs up front: twelve jobs, four workers, three jobs each. This
works efficiently only when the jobs take equal time; if one runs long, that worker
stays busy while the others sit idle. In **pull-based** dispatch, a worker asks for the
next job as soon as it finishes one. The load balances itself.

The pool below is pull-based: every time `dispatch` is called it sends the job at the
head of the queue, and closes the worker once the queue is empty.

```js
// pool.mjs — forks a fixed number of workers, distributes the job queue to them, measures duration
import { fork } from "node:child_process";

const JOBS = Array.from({ length: 12 }, (_, i) => ({
  jobId: i + 1, loanId: 500 + i, memberId: 100 + i,
}));

function runPool(workerCount, jobs) {
  return new Promise((resolve) => {
    const queue = [...jobs];
    const completed = [];
    const workers = [];
    const start = performance.now();

    const dispatch = (worker) => {
      const job = queue.shift();
      if (job) worker.send(job);
      else worker.kill("SIGTERM");
    };

    for (let s = 1; s <= workerCount; s++) {
      const worker = fork(new URL("worker.mjs", import.meta.url), [String(s)]);
      worker.on("message", (message) => {
        completed.push(message);
        if (completed.length === jobs.length) {
          const duration = performance.now() - start;
          for (const w of workers) w.kill("SIGTERM");
          resolve({ duration, completed });
        } else dispatch(worker);
      });
      workers.push(worker);
      dispatch(worker);
    }
  });
}

for (const n of [1, 2, 4]) {
  const { duration, completed } = await runPool(n, JOBS);
  const slice = Math.floor(duration / 100);   // each job takes 100 ms; slice = 100 ms
  console.log(`worker=${n}  job=${completed.length}  duration=${slice} slices`);
}
```

```
worker=1  job=12  duration=12 slices
worker=2  job=12  duration=6 slices
worker=4  job=12  duration=3 slices
```

The duration is printed in units of **slices** of 100 milliseconds, the length of a
single job; this makes the result readable independently of the machine's speed. The
slice count is rounded down, so process-startup overhead does not show up in the
number. On a slower machine the slice counts may shift up by one.

What matters is the ratio: twelve jobs take twelve slices with one worker, six with two
workers, three with four workers. Doubling the worker count halves the duration. This
is the direct consequence of the job being **divisible**: sending a notification to one
member does not wait on sending a notification to another.

##### The Limit of Scaling

There are two points where the ratio stops holding forever.

The first is the type of job. The job above was I/O-bound: the worker spent most of its
time waiting for a response, so the payoff continued even past the number of cores. If
the job were **CPU-bound** — resizing the cover image, for instance — the payoff would
stop at the core count, and adding workers past that point would only produce context-
switching overhead.

The second is a shared resource. If twelve workers draw from the same database
connection pool, the bottleneck is the connection count, not the worker count. If
sending notifications goes to a single external service, the bottleneck is the
concurrency that service accepts. In both cases, adding workers does not shorten the
total duration; it only spreads the waiting across more workers.

This is why the worker count is not a guess but a measurement result: the count is
raised, the total duration is measured, and the process stops where the duration stops
falling.

#### A Crashed Worker and Supervision

A worker can crash because it is an independent process: memory runs out, the operating
system kills it, the job it is processing brings the process itself down. The pool has
two responsibilities in response. **Replacing** the crashed worker, and **not losing**
the job it was holding.

The second is the critical one. A worker reports nothing while it crashes; the pool
only sees the `exit` event. If which job the worker was carrying is not known at that
point, the job silently disappears. That is why the pool keeps every worker's current
job in a separate map.

```js
// resilient.mjs — when a worker crashes it is replaced, and the job it held is put back in the queue
import { fork } from "node:child_process";

const JOBS = [
  { jobId: 1, loanId: 500, memberId: 100 },
  { jobId: 2, loanId: 501, memberId: 101 },
  { jobId: 3, loanId: 502, memberId: 102, broken: true },
  { jobId: 4, loanId: 503, memberId: 103 },
  { jobId: 5, loanId: 504, memberId: 104 },
];
const MAX_ATTEMPTS = 2;

const queue = [...JOBS];
const attempt = new Map();                       // jobId -> how many times attempted
const completed = [], deadLetter = [];
const inFlight = new Map();                      // worker -> the job it is currently processing
let open = 0, nextIndex = 1;

function spawnWorker() {
  const worker = fork(new URL("worker.mjs", import.meta.url), [String(nextIndex++)]);
  open++;
  worker.on("message", (m) => { inFlight.delete(worker); completed.push(m.jobId); dispatch(worker); });
  worker.on("exit", () => {
    open--;
    const job = inFlight.get(worker);             // set: the worker crashed while working the job
    if (job) {
      inFlight.delete(worker);
      if (attempt.get(job.jobId) < MAX_ATTEMPTS) queue.push(job);
      else deadLetter.push(job.jobId);
      spawnWorker();                              // replaces the crashed worker
    }
    if (open === 0) report();
  });
  dispatch(worker);
}

function dispatch(worker) {
  const job = queue.shift();
  if (!job) return worker.kill("SIGTERM");
  attempt.set(job.jobId, (attempt.get(job.jobId) ?? 0) + 1);
  inFlight.set(worker, job);
  worker.send(job);
}

function report() {
  console.log("completed jobs    :", completed.sort((a, b) => a - b).join(", "));
  console.log("dead-letter queue :", deadLetter.join(", ") || "-");
  console.log("workers spawned   :", nextIndex - 1);
  console.log("job 3 attempts    :", attempt.get(3));
}

spawnWorker();
spawnWorker();
```

```
completed jobs    : 1, 2, 4, 5
dead-letter queue : 3
workers spawned   : 4
job 3 attempts    : 2
```

Job three crashes its worker. On the `exit` event the pool puts the job back in the
queue and spawns a replacement for the crashed worker. The second attempt gives the
same result, so the job crosses the attempt limit and goes to the dead-letter queue
introduced in the previous lesson. The other four jobs are unaffected; four workers are
spawned in total, two at the start and two replacing the crashed ones.

The name for this pattern is **supervision**: the unit doing the work is separated from
the unit managing its lifetime. A worker does not try to repair its own failure, it
crashes; the pool repairs it. This separation noticeably simplifies the worker's code.

Supervision has a hidden cost. A job put back in the queue may have been **partially
done** at the moment it crashed: the notification may have been sent but not recorded
as sent. This is why every retried job must be protected by the idempotency key
introduced in the Web API Design curriculum; otherwise supervision buys durability at
the cost of sending the member two notifications.

#### Graceful Shutdown

The pool closes a worker with `SIGTERM`. The worker catches this signal and exits. If
the signal goes unhandled, the process terminates immediately under the default
behavior, and the job it was holding is left half-done.

**Graceful shutdown** is three steps: stop taking new jobs, finish the one in hand,
then exit. In practice the parent process first sends `SIGTERM`, grants the worker a
bounded window to finish, and forces the issue with `SIGKILL` if the window runs out.
This second signal cannot be caught by the worker; that is why the window granted must
be larger than the longest job.

This is also the order deployment tools follow when stopping worker containers. A
worker that does not catch the signal produces a half-done job on every redeployment.

#### Summary

- A worker is an independent unit of execution that pulls a job from the queue and processes it; being able to crash on its own and be replicated on its own makes it a unit of scaling.
- In pull-based dispatch, a worker asks for the next job as soon as it finishes one; the load balances itself even when job durations are unequal.
- The payoff from worker count is found by measurement: it can go past the core count for an I/O-bound job, it stops there for a CPU-bound job, and it stops much earlier when a shared resource is involved.
- Supervision replaces a crashed worker and puts the job it was holding back in the queue; the cost is that the job may have been partially done, which is why a retried job is protected with an idempotency key.
- Graceful shutdown means stop taking new jobs, finish the one in hand, then exit; if the window granted is shorter than the longest job, every shutdown produces a half-done job.

#### Next Step

In this lesson, jobs came from outside: the pool distributed a list it already held.
Most background jobs, however, do not come from outside — they come from the
**calendar**. The overdue notification runs every morning at six, the monthly loan
report on the first of the month, the branch inventory summary every fifteen minutes.
Binding a job to a time rule looks like a one-line definition, but underneath it are
two questions: how is recurrence defined, and what happens when two runs of the same
task overlap. The next lesson builds the recurring task definition and prevents
overlap by measuring it.

### Scheduled Tasks

Source: https://academia.sh/en/courses/asynchronous-processing/scheduled-tasks

The definition and hazards of calendar-triggered jobs: computing the recurrence rule, the end-of-month and time-zone traps, the fixed-rate versus fixed-delay distinction, and a measured solution that prevents two runs from overlapping with a lease lock.

The previous lesson took jobs from outside: the pool held a list and distributed it to
workers. An important share of background jobs, though, do not come from outside — they
come from the **calendar**. The overdue notification runs every morning at six, the
branch inventory summary every fifteen minutes, the monthly loan report on the first of
every month.

Summing this up as "set a timer and call the job" is misleading. A scheduled task has
two hard questions. First, how recurrence is defined: what happens when "the 31st of
every month" lands on a date that does not exist, does a run get skipped when the local
clock changes. Second, overlap: if one run does not finish before the next trigger, two
runs execute at once and produce the same report twice. This lesson builds both, and
solves the second by measuring it.

#### Recurrence Definition

A recurring task has two parts: **what** gets done and **when** it gets done. The
second is a data structure, and it can be tested with a single function: "what is the
next run instant from right now."

```js
// schedule.mjs — computes the upcoming run instants from a recurring task definition (all UTC)
const DEFINITIONS = {
  overdueNotification: { type: "daily", hour: 6, minute: 0 },
  inventorySummary: { type: "interval", minute: 15 },
  monthlyReport: { type: "monthly", day: 1, hour: 2, minute: 30 },
  endOfMonth: { type: "monthly", day: 31, hour: 2, minute: 30 },
};

const MIN = 60_000;

function nextRun(definition, now) {
  const d = new Date(now);
  if (definition.type === "interval") {
    const step = definition.minute * MIN;
    return new Date(Math.floor(now / step) * step + step);
  }
  if (definition.type === "daily") {
    const today = Date.UTC(d.getUTCFullYear(), d.getUTCMonth(), d.getUTCDate(),
                           definition.hour, definition.minute);
    return new Date(today > now ? today : today + 24 * 60 * MIN);
  }
  // monthly: if this month's day has passed, the same day next month
  const thisMonth = Date.UTC(d.getUTCFullYear(), d.getUTCMonth(), definition.day,
                        definition.hour, definition.minute);
  return new Date(thisMonth > now ? thisMonth
    : Date.UTC(d.getUTCFullYear(), d.getUTCMonth() + 1, definition.day, definition.hour, definition.minute));
}

const START = Date.parse("2026-01-31T23:47:00Z");   // fixed starting instant for the measurement
for (const [name, definition] of Object.entries(DEFINITIONS)) {
  let at = START;
  const moments = [];
  for (let i = 0; i < 3; i++) {
    at = nextRun(definition, at).getTime();
    moments.push(new Date(at).toISOString().slice(0, 16).replace("T", " "));
  }
  console.log(name.padEnd(20), moments.join("  "));
}
```

```
overdueNotification  2026-02-01 06:00  2026-02-02 06:00  2026-02-03 06:00
inventorySummary     2026-02-01 00:00  2026-02-01 00:15  2026-02-01 00:30
monthlyReport        2026-02-01 02:30  2026-03-01 02:30  2026-04-01 02:30
endOfMonth           2026-03-03 02:30  2026-03-31 02:30  2026-05-01 02:30
```

Because the starting instant is written as a fixed constant, the output is the same on
every run; this turns the scheduling rule into a testable function. Supplying "now"
from outside the function is the one decision that makes scheduling testable.

The first three lines are as expected. The fourth line is **not**.

##### The End-of-Month Trap

The `endOfMonth` definition says "the 31st of every month at 02:30." The instants
produced are March 3, March 31, and May 1. February has no 31st; `Date.UTC(2026, 1,
31)` carries the overflowing day into the next month and produces March 3. April has no
31st either, so it slides to May 1.

This is not a bug in the computation, it is a **gap in the definition**. For a task tied
to the last day of the month, a choice has to be made between two policies: **skip** the
nonexistent day (no run that month) or **clamp** to the last actual day of the month
(the 28th, 29th, or 30th). Which one is correct depends on the job — a monthly report
should clamp, while a subscription renewal ending on the 31st should not be skipped. A
policy that is never written down is a policy handed over to the runtime's overflow
behavior.

##### Time Zone

The whole computation above ran on UTC. A task tied to local time produces two further
problems. On the night the clock jumps forward for daylight saving time, local 02:30
**never happens**; the run for that night is skipped. On the night it jumps back, local
02:30 **happens twice**; the run fires twice.

This is why the scheduling decision splits into two layers. The task's trigger instant
is kept and stored in UTC; local time is used only when displaying it to a user. If a
local calendar boundary is genuinely required (for example, "after the branch closes"),
the time zone identity becomes part of the definition and the conversion is redone on
every run.

#### Fixed Rate and Fixed Delay

The second detail of recurrence is what the next run is scheduled against.

In **fixed-rate** scheduling, runs are pinned to the calendar: 02:30, 03:30, 04:30. No
matter how long a run takes, the next one's instant does not move. In **fixed-delay**
scheduling, the wait is counted from the **end** of the previous run: if a run took 40
minutes and the delay is one hour, the next run starts at 04:10.

The difference between them is drift. An hourly task that takes forty minutes runs
about 14 times a day under fixed delay instead of 24, and its run instants creep
forward every day; this creeping is called **drift**. Fixed rate has no drift, but in
exchange it makes overlap between runs possible: if a run takes longer than an hour, the
next trigger arrives while the previous one is still running.

So the choice between fixed rate and fixed delay is a choice between drift and overlap.
If drift cannot be tolerated, overlap has to be prevented separately.

#### Overlap

Overlap does not only arise from a long run. A second, and more commonly encountered,
source is the application being deployed as **multiple instances**. If the scheduler
runs inside the application, a service deployed across three instances sends the
overdue notification three times every morning at six. A third source is redeployment:
if the new instance comes up before the old one shuts down, the trigger doubles.

The fix for all three is the same: a task must **acquire the right to run** before it
runs.

#### Lease Lock

The structure that grants this right is called a **lock**. The form used for
background tasks is the **lease lock**: the lock is not granted indefinitely but for a
fixed duration. The duration is necessary, because if the process holding the lock
crashes, there is no one left to release it; the lease lets the lock drop on its own in
that case.

The correctness of the lock depends on a single condition: acquisition must be
**atomic**. If two runs both ask "is there a lock" and both get the answer "no," the
lock is useless. This is why the check and the write are combined into a single
statement.

```js
// task.mjs — the recurring task that generates the monthly loan report
//   node task.mjs <label>           run without a lock
//   node task.mjs <label> --lock    run with a lease lock
import { DatabaseSync } from "node:sqlite";
import { appendFileSync } from "node:fs";

const LABEL = process.argv[2];
const LOCKED = process.argv.includes("--lock");
const LEASE_MS = 5000;                             // the lock's lease duration

const db = new DatabaseSync("schedule.db");
db.exec("PRAGMA busy_timeout = 5000");
db.exec(`CREATE TABLE IF NOT EXISTS lock (
           name TEXT PRIMARY KEY, owner TEXT NOT NULL, expires INTEGER NOT NULL)`);

// Acquiring the lock is a single atomic write: the row is created if absent, or taken
// over if present but its lease has expired. If two runs try at once, only one changes the row.
const acquireLock = (name, owner) => {
  const now = Date.now();
  return db.prepare(
    `INSERT INTO lock (name, owner, expires) VALUES (?, ?, ?)
       ON CONFLICT(name) DO UPDATE SET owner = excluded.owner, expires = excluded.expires
       WHERE lock.expires < ?`).run(name, owner, now + LEASE_MS, now).changes === 1;
};
const releaseLock = (name, owner) =>
  db.prepare("DELETE FROM lock WHERE name = ? AND owner = ?").run(name, owner);

function generateReport() {
  appendFileSync("trace.log", `started ${LABEL}\n`);
  const deadline = Date.now() + 300;                // report generation takes 300 ms
  while (Date.now() < deadline);
  appendFileSync("trace.log", `done ${LABEL}\n`);
}

if (!LOCKED) generateReport();
else if (acquireLock("monthly-report", LABEL)) {
  try { generateReport(); } finally { releaseLock("monthly-report", LABEL); }
} else {
  appendFileSync("trace.log", `skipped ${LABEL}\n`);
}
```

Three details are what make the lock correct. The lock's **name** is the task's name,
not the process name; different instances use the same name. Releasing the lock also
filters on the `owner` column, so a run whose lock was taken over because its lease
expired cannot delete someone else's lock. And the release sits inside `finally`, so it
runs on failure too.

#### Measurement

Whether the lock works is not asserted, it is measured. The driver below starts the
same task twice at the exact same time and computes the **peak concurrent run count**
from the started/done markers in `trace.log`.

```js
// driver.mjs — starts the same task twice at once and parses the trace
import { spawn } from "node:child_process";
import { writeFileSync, readFileSync, rmSync } from "node:fs";

const spawnRun = (argv) =>
  new Promise((c) => spawn("node", argv, { stdio: "ignore" }).on("exit", c));

function parseTrace() {
  let open = 0, peak = 0, runs = 0, skipped = 0;
  for (const line of readFileSync("trace.log", "utf8").trim().split("\n")) {
    if (line.startsWith("started")) { open++; runs++; peak = Math.max(peak, open); }
    else if (line.startsWith("done")) open--;
    else skipped++;
  }
  return { runs, skipped, peak };
}

for (const mode of ["unlocked", "locked"]) {
  writeFileSync("trace.log", "");
  rmSync("schedule.db", { force: true });
  const extra = mode === "locked" ? ["--lock"] : [];
  await Promise.all([
    spawnRun(["task.mjs", "A", ...extra]),
    spawnRun(["task.mjs", "B", ...extra]),
  ]);
  const { runs, skipped, peak } = parseTrace();
  console.log(`${mode.padEnd(9)} runs=${runs} skipped=${skipped} concurrent-peak=${peak}`);
}
```

```
unlocked  runs=2 skipped=0 concurrent-peak=2
locked    runs=1 skipped=1 concurrent-peak=1
```

In unlocked mode both runs generate the report and the peak comes out 2: the monthly
loan report has been computed twice. In locked mode, runs is 1, skipped is 1, and peak
is 1. The second run exits without doing the work because it cannot acquire the lock.

Notice that a skipped run is **not counted as an error**. Failing to acquire the lock
is an expected outcome for scheduled tasks; it is logged but raises no alert. What does
raise an alert is the same task going a long time without ever running.

#### Missed Run

The lock prevents overlap, but it leaves a question open: if a run could not happen at
all because of the lock or because the service was down, does it get made up for later?

There are two policies. Under the **no-catch-up** policy, a missed run is forgotten; the
next trigger performs a normal run. Tasks that compute a final state, like the branch
inventory summary, fit this, because the new run already covers the result of the
missed one. Under the **catch-up** policy, a separate run is produced for every missed
interval. The monthly loan report fits this, because each month's report is a separate
output and nothing replaces it if one is skipped.

The danger of the catch-up policy is that every run accumulated during a long outage
fires all at once. This is why catch-up is given an upper bound: at most the last N
intervals are made up, the rest fall into the log.

#### Summary

- The recurring task definition reduces to a "next run instant from right now" function; taking "now" as an input from outside makes the schedule testable.
- A definition tied to a nonexistent day of the month overflows into the next month; a skip or clamp-to-last-day policy must be written down explicitly.
- Fixed rate prevents drift but is open to overlap; fixed delay prevents overlap but shifts the run instants.
- Overlap arises from a long run, a multi-instance deployment, and redeployment; the fix for all three is a lease lock acquired before the run starts.
- The lock's correctness depends on the atomicity of acquisition, and its durability on the lease expiring on its own; the measurement brought the peak concurrent run count down from 2 to 1.

#### Next Step

The tasks in this lesson were short: a report that finishes in three hundred
milliseconds. The lease duration could be chosen accordingly. But a share of background
jobs run for minutes — a monthly report scanning thirty thousand loan records,
regenerating an uploaded cover image at several sizes. Once a job like that starts, two
questions follow: how does the party that started it learn how far it has gotten, and
where does the job stop when it is abandoned. The next lesson builds progress
reporting and cancellation for long-running jobs.

### Long-Running Jobs

Source: https://academia.sh/en/courses/asynchronous-processing/long-running-jobs

Making a background job that runs for minutes manageable: breaking the job into chunks, writing progress to a durable status record, building cooperative cancellation with an abort signal, and resuming from where cancellation left off.

The previous lesson built recurring tasks and prevented two runs from overlapping with
a lease lock. The task there was short: a report that finishes in three hundred
milliseconds. The lease duration could be chosen accordingly.

A share of background jobs, though, run for minutes. A monthly report scanning thirty
thousand loan records, regenerating an uploaded cover image at four different sizes,
recomputing the branch inventory summary from scratch. As soon as a job like that
starts, two questions follow. How does the party that started it learn **how far it has
gotten**. And when the job is abandoned — the user closed the tab, the service is
shutting down, the lease expired — **where does it stop** and what does it leave behind
there.

#### Breaking the Job into Chunks

The first decision that makes a long job manageable is turning it from a single block
into a sequence of **chunks**. A job that has not been chunked cannot do three things
at once: it cannot report progress, it cannot be cancelled, and if it is left half-done
it leaves no option but starting over.

The rule for chunking is simple: the job turns into a loop that processes a fixed
number of units per round, and **two things** happen at the end of every round. State
is written as a checkpoint, and control returns to the event loop. The second is
mandatory in a single-threaded runtime: without returning, the process can do nothing
else until that job finishes — it cannot even answer a health check.

#### Progress Reporting and Cancellation

Progress is written to two places at once. A counter kept in memory is not enough,
because that counter goes away when the process crashes. This is why progress is
written to a **durable status record** keyed by the job's identity — here, the
`job_status` table.

Cancellation, meanwhile, is **cooperative**: it is not possible to forcibly break an
outside process's loop. The outside party only raises a signal; the party doing the
work reads that signal at points it chooses and stops there. In JavaScript this signal
is produced with `AbortController`; the `signal.throwIfAborted()` call throws if the
signal has been raised.

```js
// long-job.mjs — monthly loan report: advances chunk by chunk, reports progress, honors cancellation
import { DatabaseSync } from "node:sqlite";

const db = new DatabaseSync(":memory:");
db.exec(`CREATE TABLE loan (loan_id INTEGER PRIMARY KEY, member_id INTEGER NOT NULL,
                             days INTEGER NOT NULL);
         CREATE TABLE job_status (job_id TEXT PRIMARY KEY, status TEXT NOT NULL,
                                 processed INTEGER NOT NULL, total INTEGER NOT NULL)`);
const insert = db.prepare("INSERT INTO loan (loan_id, member_id, days) VALUES (?, ?, ?)");
for (let i = 1; i <= 30_000; i++) insert.run(i, 100 + (i % 500), i % 30);

const RECORD_COUNT = db.prepare("SELECT count(*) AS n FROM loan").get().n;
const CHUNK = 3000;                                  // records processed per round

const readStatus = (jobId) =>
  db.prepare("SELECT processed, total FROM job_status WHERE job_id = ?").get(jobId)
  ?? { processed: 0, total: 0 };
const writeStatus = (jobId, status, processed, total) =>
  db.prepare(`INSERT INTO job_status (job_id, status, processed, total) VALUES (?, ?, ?, ?)
                ON CONFLICT(job_id) DO UPDATE
                SET status = excluded.status, processed = excluded.processed,
                    total = excluded.total`).run(jobId, status, processed, total);

async function generateReport(jobId, signal, onProgress) {
  let { processed, total } = readStatus(jobId);      // from zero or from where it was left
  try {
    while (processed < RECORD_COUNT) {
      signal.throwIfAborted();                       // cancellation is only seen at a chunk boundary
      for (const s of db.prepare(
        "SELECT days FROM loan ORDER BY loan_id LIMIT ? OFFSET ?").all(CHUNK, processed))
        total += s.days;
      processed += CHUNK;
      writeStatus(jobId, "running", processed, total); // write first, then report
      onProgress(Math.round((processed / RECORD_COUNT) * 100), processed);
      await new Promise((c) => setImmediate(c));     // hand-off point back to the event loop
    }
    writeStatus(jobId, "done", processed, total);
    return total;
  } catch (h) {
    writeStatus(jobId, "cancelled", processed, total); // the point it stopped at is recorded
    throw h;
  }
}

const JOB = "report-2026-01";
const printProgress = (percent, processed) => console.log(`  progress %${percent}  processed=${processed}`);

console.log("--- first run: cancelled at %40 ---");
const controller = new AbortController();
try {
  await generateReport(JOB, controller.signal, (percent, processed) => {
    printProgress(percent, processed);
    if (percent >= 40) controller.abort(new Error("user cancelled"));
  });
} catch (h) {
  const d = readStatus(JOB);
  console.log(`cancelled: ${h.message}  processed=${d.processed}/${RECORD_COUNT}  total=${d.total}`);
}

console.log("--- second run: continues from where it was left ---");
const total = await generateReport(JOB, new AbortController().signal, printProgress);
console.log(`done: total=${total}  average days=${(total / RECORD_COUNT).toFixed(2)}`);
```

```
--- first run: cancelled at %40 ---
  progress %10  processed=3000
  progress %20  processed=6000
  progress %30  processed=9000
  progress %40  processed=12000
cancelled: user cancelled  processed=12000/30000  total=174000
--- second run: continues from where it was left ---
  progress %50  processed=15000
  progress %60  processed=18000
  progress %70  processed=21000
  progress %80  processed=24000
  progress %90  processed=27000
  progress %100  processed=30000
done: total=435000  average days=14.50
```

Cancellation fires once progress reaches forty percent; so the measurement can be
repeated, the cancellation point is tied to a progress threshold rather than a
duration. The output shows three things.

The job **stopped at twelve thousand records**, not thirty thousand. The stopping
point was written to the `job_status` table: the `status` column is `cancelled`, the
`processed` column is 12000. And the second run resumed not from zero but from the
twelve-thousandth record — the forty-percent point already reached; the overall total
came out 435000, meaning no
record was counted twice.

The third point matters more than the first two. The cancelled job was left at a
**resumable** point. This is the only meaningful definition of cancellation for a
long-running job: cancellation does not undo the work, it freezes it at a consistent
point.

##### Write Order

The `writeStatus` call happening **before** the `onProgress` call in the code is a
deliberate choice. If the order were reversed, a process crash after the progress
report but before the write would leave progress reported but not saved; resuming
would then reprocess records that had actually already been handled.

By the same reasoning, `writeStatus` and processing the chunk must sit inside a single
transaction boundary. This condition is satisfied here because the report total is
kept in memory and written together with the status. If the chunk wrote a row to a
table instead, that write and the progress update would need to be handled together
inside the unit of work introduced in the Data Access Layer and Business Logic
curriculum.

#### Chunk Size

Chunk size looks like a single number, but it sets two magnitudes at once. A small
chunk speeds up cancellation, because the signal is only read at a chunk boundary; a
large chunk cuts down on progress writes, because every round means one write.

| Chunk size | Progress writes | Cancellation delay (worst case) |
|---|---|---|
| 100 | 300 | 100 records |
| 500 | 60 | 500 records |
| 3000 | 10 | 3000 records |
| 10000 | 3 | 10000 records |

For a thirty-thousand-record job the numbers fall out of division: the write count is
`30000 / chunk`, and the cancellation delay in the worst case is one chunk. A chunk of
one hundred shortens how long the user waits after pressing cancel, but it makes three
hundred writes to the status table; a chunk of ten thousand cuts writes to three, but
cancellation takes as long as processing ten thousand records.

The choice is made according to the nature of the job. For a job the user started and
is waiting on, cancellation delay dominates. For a job running overnight that nobody is
watching, write count dominates; the progress record there is kept not for
cancellation but for resuming from where it was left.

#### Sources of Cancellation

The cancellation signal does not come from a single place. It has four sources, and all
four can be tied to the same signal.

**User cancellation** is the most visible one: a stop request from the report screen
turns into a call carrying the job's identity, and that call raises the signal.
**Time limit** is the second source; `AbortSignal.timeout(ms)` produces a signal that
raises itself once the given duration elapses, and it keeps the job from running for an
unacceptable length of time. **Shutdown** is the third: a process that receives
`SIGTERM` during the graceful shutdown sequence built in the previous lesson raises the
signal for its running jobs and waits for them to finish. **Lease expiry** is the
fourth; if the job's lock is about to be taken over, the old run must cancel itself, or
two runs end up continuing the same job.

When all four are wired to the same structure, a single checkpoint remains in the job's
code. `AbortSignal.any([...])` folds several signals into one; the job reads only that
one signal and does not need to know who cancelled it.

#### Where the Job's Result Goes

Progress reporting has one gap: here it was written with `console.log`. In a real
service, the party that needs to read progress is a different process — the client
requesting the report. This is why the `job_status` table serves not only for resuming
but also as a **queryable status record**: a client can read this record using the
job's identity.

How that read happens — whether the client asks over and over, or the server sends
changes on its own — is the question this curriculum's last topic answers.

#### Summary

- A long job is turned into a loop that processes fixed-size chunks; state is written at the end of every round and control returns to the event loop.
- Progress is written not to memory but to a durable status record keyed by the job's identity; the record serves both as queryable progress and as the resume point.
- Cancellation is cooperative: the outside party only raises a signal, and the party doing the work reads that signal at chunk boundaries and stops at a consistent point.
- The status write happens before the report; the reverse order causes the same records to be processed twice because of progress that was reported but not saved.
- Chunk size is the trade-off between cancellation delay and progress-write count; delay dominates for a job the user is waiting on, write count dominates for a job running overnight.

#### Next Step

In this lesson there was a single job, and chunking it was enough. In a real
background, jobs flow continuously: every loan transaction spawns a notification job,
every cover upload spawns an image-processing job. The **arrival rate** of jobs and the
**completion rate** of workers are independent of each other, so they are not obligated
to be equal. When the arrival rate exceeds the completion rate the queue grows; if the
growth continues, memory, disk, or the latency budget runs out. The next lesson names
this imbalance and compares the options for a bounded queue with numbers.

### Backpressure

Source: https://academia.sh/en/courses/asynchronous-processing/backpressure

The consequences of production rate exceeding consumption rate: unbounded growth of wait time in an unbounded queue, a numerical simulation comparing the bounded queue's three policies, and backpressure measured through streams and propagated to the producer.

The previous lesson dealt with a single long job: breaking it into chunks, recording
its progress, and making it cancellable was enough. In a real background, jobs do not
arrive one at a time, they **flow**: every loan transaction spawns a notification job,
every cover upload spawns an image-processing job.

The rate jobs arrive at and the rate workers finish them are two independent
magnitudes. Nothing keeps them equal; the arrival rate is set by request traffic, the
completion rate by worker count and job duration. When the arrival rate exceeds the
completion rate, the queue grows. This lesson measures what that growth produces and
builds how the system communicates it to the producer.

#### Rate Mismatch

If a queue receives $\lambda$ jobs per second and drains $\mu$ jobs per second, the
queue length changes at a rate of $\lambda - \mu$. As long as $\lambda < \mu$, the
queue stays empty and wait time approaches the job's own duration. The moment
$\lambda > \mu$, the queue lengthens by $\lambda - \mu$ every second, and this growth
does not stop on its own.

The direct consequence of growth is wait time. Queueing theory's basic relation says
that in a stable queue, average length $L$, arrival rate $\lambda$, and average wait
time $W$ satisfy:

$$L = \lambda W$$

Read backward, the relation delivers the real warning: $W = L / \lambda$. If queue
length is unbounded, **wait time is unbounded too**. Long before memory fills up, a job
still sitting in the queue loses its meaning even though it will still be processed — an
overdue notification delayed four hours is no different from one that was never sent.

This is why queue length is not a resource problem, it is a **latency budget** problem.

#### The Bounded Queue and Three Policies

The fix is bounding the queue. Bounding it immediately raises a question: what happens
to an incoming job when the queue is full. There are three answers, and each one gives
something up.

**Blocking**: the producer is held until room opens up in the queue. No job is lost,
but the slowdown passes to the producer — this, in the narrow sense, is
**backpressure**. **Tail drop**: an incoming job is rejected while the queue is full.
**Head drop**: the oldest job at the head of the queue is dropped, and the new job is
appended.

The simulation below runs all four under the same load: a hundred jobs produced per
second, sixty consumed, a ten-second window, the bounded queue's capacity a hundred.

```js
// simulation.mjs — how the queue behaves under four policies once production rate exceeds consumption rate
const WINDOW = 10_000;             // simulation window (ms)
const PRODUCE = 100;               // jobs produced per second
const CONSUME = 60;                // jobs consumed per second
const CAPACITY = 100;              // the bounded queue's capacity

function run(policy) {
  const queue = [];                // each entry: the instant it was produced (ms)
  let produced = 0, delivered = 0, dropped = 0, blocked = 0;
  let peak = 0, longestWait = 0;
  let produceAcc = 0, consumeAcc = 0;   // accumulators as integers: rise by rate every ms

  for (let t = 1; t <= WINDOW; t++) {
    produceAcc += PRODUCE;
    while (produceAcc >= 1000) {
      produceAcc -= 1000;
      produced++;
      if (policy === "unbounded" || queue.length < CAPACITY) queue.push(t);
      else if (policy === "tail-drop") dropped++;
      else if (policy === "head-drop") { queue.shift(); dropped++; queue.push(t); }
      else { produced--; produceAcc += 1000; blocked++; break; }      // block: the producer waits
    }
    consumeAcc += CONSUME;
    while (consumeAcc >= 1000 && queue.length) {
      consumeAcc -= 1000;
      longestWait = Math.max(longestWait, t - queue.shift());
      delivered++;
    }
    peak = Math.max(peak, queue.length);
  }
  return { policy, produced, delivered, dropped, blocked, peak,
           remaining: queue.length, longestWait };
}

const col = (n) => String(n).padStart(6);
console.log("policy      produced delivered dropped remaining  peak  longest-wait  producer-wait");
for (const p of ["unbounded", "tail-drop", "head-drop", "block"]) {
  const s = run(p);
  console.log(`${s.policy.padEnd(11)}${col(s.produced)}${col(s.delivered)}${col(s.dropped)}` +
    `${col(s.remaining)}${col(s.peak)}${String(s.longestWait).padStart(13)} ms` +
    `${String(s.blocked).padStart(14)} ms`);
}
```

```
policy      produced delivered dropped remaining  peak  longest-wait  producer-wait
unbounded    1000   600     0   400   400         4000 ms             0 ms
tail-drop    1000   600   301    99   100         1664 ms             0 ms
head-drop    1000   600   301    99   100          997 ms             0 ms
block         699   600     0    99   100         1666 ms          7490 ms
```

The simulation does not use real time; it advances a virtual clock millisecond by
millisecond and applies the rates through integer accumulators. This is why the numbers
come out the same regardless of the machine.

#### Reading the Numbers

The four rows show four different trade-offs.

Under **unbounded**, no job was lost, but the queue climbed to four hundred in ten
seconds and the longest wait reached four seconds. What matters is not the number
itself but its direction: with a twenty-second window, the queue would reach eight
hundred and the wait eight seconds. This policy **leaves latency unbounded** in
exchange for not losing work.

**Tail drop** and **head drop** delivered the same number of jobs (600) and dropped the
same number (301). The only difference between them is in wait time: 1664 ms versus
997 ms. The difference comes from the **age** of the jobs delivered. Under tail drop
the oldest hundred jobs always sit in the queue; under head drop the queue keeps
refreshing and younger jobs reach the consumer. At the same loss rate, this makes head
drop superior for jobs where freshness matters, like notifications; for jobs where
**order must be preserved**, like a loan request standing in line, dropping the oldest
is unacceptable.

**Blocking** dropped no job, but it could only produce 699 jobs: the producer spent
7490 of the ten seconds' 10000 milliseconds waiting. No job was lost because none was
ever produced. This is the definition of backpressure — the bottleneck reflects its
consequence not onto the party downstream from it, but onto the party feeding it.

The shared lesson of the four rows: the rate mismatch cannot be eliminated, only
**where it gets written down** can be chosen. To memory (unbounded queue), to loss
(dropping), or to the producer's own rate (blocking).

#### How Backpressure Propagates

Blocking sounds abstract, but it is already built into the stream abstraction.
`node:stream`'s writable stream `write` call returns a boolean: `false` reports that
the buffer has crossed its threshold and production should stop. Once room opens up,
the stream emits the `drain` event.

The measurement below writes the same sixty-four chunks twice. In the first, the
return value is listened to; in the second, it is ignored.

```js
// stream.mjs — node:stream backpressure: listening to the write() return value versus ignoring it
import { Writable } from "node:stream";

const CHUNK = Buffer.alloc(1024, 0x61);          // a 1 KiB job body
const COUNT = 64;

const slowSink = () => new Writable({
  highWaterMark: 16 * 1024,                      // our buffer limit: 16 KiB
  write(_chunk, _enc, done) { setTimeout(done, 5); },  // writing each chunk takes 5 ms
});

async function listening() {
  const sink = slowSink();
  let fullSignals = 0, drain = 0, peak = 0;
  for (let i = 0; i < COUNT; i++) {
    const hasRoom = sink.write(CHUNK);
    peak = Math.max(peak, sink.writableLength);
    if (!hasRoom) {                               // buffer full: stop producing
      fullSignals++;
      await new Promise((c) => sink.once("drain", () => { drain++; c(); }));
    }
  }
  await new Promise((c) => sink.end(c));
  return { fullSignals, drain, peak };
}

async function ignoring() {
  const sink = slowSink();
  let fullSignals = 0, peak = 0;
  for (let i = 0; i < COUNT; i++) {
    if (!sink.write(CHUNK)) fullSignals++;         // return value read but not honored
    peak = Math.max(peak, sink.writableLength);
  }
  await new Promise((c) => sink.end(c));
  return { fullSignals, drain: 0, peak };
}

for (const [name, run] of [["listening", listening], ["ignoring", ignoring]]) {
  const s = await run();
  console.log(`${name.padEnd(12)} full-signals=${String(s.fullSignals).padStart(2)}` +
    ` drain=${String(s.drain).padStart(2)} buffer-peak=${s.peak} B`);
}
```

```
listening    full-signals= 4 drain= 4 buffer-peak=16384 B
ignoring     full-signals=49 drain= 0 buffer-peak=65536 B
```

When the return value is listened to, the buffer peak is exactly the threshold itself:
16384 bytes. When it is ignored, all sixty-four chunks pile up in the buffer: 65536
bytes. The ratio across sixty-four chunks is fourfold; if the producer keeps writing
without bound, the ratio is unbounded too, because ignoring a `false` return turns the
buffer into a queue, and moreover an **unbounded** one.

The lesson here is not specific to streams. Every layer that writes a job into a queue
faces the same question: does it stop when a write is refused, or does it accumulate
its own queue behind it. Any layer that does not stop renders the limit in front of it
useless.

#### How Far Backpressure Propagates

The last link in the chain is the user, and backpressure cannot be carried all the way
there. A server handling requests cannot hold an HTTP request indefinitely because the
background queue is full; holding it exhausts connections and moves the problem from
the queue to the network.

This is why backpressure is applied in the system's **internal** links: between a
worker and the database, between a queue writer and the queue, between a stream
producer and its receiver. At the outer boundary, a different mechanism is needed —
**explicitly rejecting** the incoming request instead of holding it.

#### Summary

- When production rate exceeds consumption rate, the queue lengthens every second by the gap between them; the growth does not stop on its own.
- Per $L = \lambda W$, unbounded queue length means unbounded wait time; the queue bound is a latency-budget decision, not a memory decision.
- A bounded queue has three policies: blocking loses no job but slows the producer, and tail drop and head drop produce the same loss but deliver jobs of different ages.
- Backpressure is the bottleneck reflecting its consequence onto the party feeding it; `node:stream` applies this through the write return value and the drain event, and keeps the buffer peak at the threshold.
- A layer that ignores the write return value defeats every limit in front of it; backpressure is not truly applied anywhere in the chain unless it is applied throughout it.

#### Next Step

Backpressure works in the internal links, but the client sitting at the outer end of
the chain cannot be held. Once the queue is full, the workers cannot keep up, and the
latency budget is exhausted, the server is left with a single option: refuse the
incoming job. Doing this randomly destabilizes the system; which request gets
rejected, how the rejection is communicated to the client, and what the latency
difference is between the rejected and the accepted are separate decisions. The next
lesson builds controlled rejection under overload and measures the effect of rejection
on latency.

### Load Shedding

Source: https://academia.sh/en/courses/asynchronous-processing/load-shedding

Controlled rejection under overload: admission control based on a concurrency threshold, measuring the latency gap between accepted and rejected requests, the HTTP shape of a rejection, and preventing a retry storm.

The previous lesson built backpressure and ended on a limit: backpressure works in the
system's internal links, it cannot be carried to the client sitting at the outer end. A
server handling requests cannot hold a request indefinitely because the background
queue is full.

So what does it do under overload. The answer is often skipped because it is hard to
accept: **reject the request**. Rejecting is not a malfunction, it is a design
decision. A server that does not reject keeps accepting requests, every one of their
latencies grows, and past a certain point no request is served on time. This state is
called **congestion collapse**, and it is far worse than rejecting: a rejected request
was at least rejected quickly. This lesson builds how rejection is done in a
controlled way and compares the latency of the two states by measuring it.

#### Admission Control

**Load shedding** is the system refusing work that exceeds its capacity instead of
accepting it. The structure that makes this decision is called **admission control**,
and it runs **before** a job starts.

Admission control differs from rate limiting on two points. Rate limiting bounds how
many requests a specific client can make in a specific time; its criterion is the
client, and its purpose is fair sharing. Admission control's criterion is **the
system's current state**; its purpose is protecting the timely completion of jobs
already accepted. The same client is accepted when the system is idle and rejected when
it is saturated.

##### What the Threshold Is Set On

Setting the threshold on requests per second is common but misleading. Requests per
second does not account for job duration: the same rate is comfortable if a job takes
twenty milliseconds and overwhelming if it takes three hundred.

The right criterion is **concurrency**: the number of jobs being processed at once. The
$L = \lambda W$ relation used in the previous lesson gives this directly: if the number
of jobs in the system, $L$, is fixed, the average duration $W$ is fixed as well. Fixing
concurrency fixes latency.

The server below carries two thresholds: the number of jobs that can be processed at
once, and the number of jobs that may wait in the queue. Their sum is the largest load
the system accepts.

```js
// admission.mjs — job server; runs in two modes
//   node admission.mjs --unbounded    every incoming request is queued
//   node admission.mjs --controlled   once the threshold is crossed the request is turned back with 503
import { createServer } from "node:http";

const CONTROLLED = process.argv.includes("--controlled");
const MAX_CONCURRENT = 2;          // number of jobs processed at once
const MAX_QUEUE = 2;               // number of jobs that may wait in the queue
const JOB_DURATION = 300;          // duration of a job (ms)

let running = 0;
const queue = [];

const doWork = () => new Promise((c) => setTimeout(c, JOB_DURATION));

async function start(job) {
  running++;
  await doWork();
  running--;
  job.response.writeHead(200, { "content-type": "text/plain" });
  job.response.end(`${job.status}\n`);
  const next = queue.shift();
  if (next) start(next);
}

createServer((request, response) => {
  response.sendDate = false;
  if (running < MAX_CONCURRENT) return start({ response, status: "accepted" });
  if (!CONTROLLED || queue.length < MAX_QUEUE)
    return queue.push({ response, status: "queued" });
  // Threshold crossed: the request is not held; when to retry is communicated instead.
  response.writeHead(503, { "content-type": "text/plain", "retry-after": "1" });
  response.end("shed\n");
}).listen(8352, "127.0.0.1", () => console.log("listening: 127.0.0.1:8352"));
```

#### Measurement

The script below brings the server up in both modes, sends ten requests at the exact
same time in each, and groups the results by status code and hundred-millisecond
latency slices. Port 8352 is arbitrary and must be free.

```bash
#!/usr/bin/env bash
# Starts admission.mjs in two modes, sends ten requests at once in each mode, summarizes the result.
A=http://127.0.0.1:8352

measure() {
  node admission.mjs "$1" > /dev/null & server=$!
  sleep 1
  pid=()
  for i in $(seq 1 10); do
    curl -sS -o /dev/null -w '%{http_code} %{time_total}\n' "$A/job" & pid+=($!)
  done > raw.txt 2>&1
  wait "${pid[@]}"
  echo "--- $2 ---"
  awk '{ printf "status=%s  latency=%2d slice\n", $1, int($2 * 10) }' raw.txt | LC_ALL=C sort | uniq -c | sed 's/^ *//'
  kill "$server"
  sleep 0.5
}

measure --unbounded  "no admission control: every request is queued"
measure --controlled "admission control on: 503 once the threshold is crossed"
```

```
--- no admission control: every request is queued ---
2 status=200  latency= 3 slice
2 status=200  latency= 6 slice
2 status=200  latency= 9 slice
2 status=200  latency=12 slice
2 status=200  latency=15 slice
--- admission control on: 503 once the threshold is crossed ---
2 status=200  latency= 3 slice
2 status=200  latency= 6 slice
6 status=503  latency= 0 slice
```

The number at the start of each line is how many requests produced that line. The lines are
ordered by `LC_ALL=C sort`, so the order is the same in every environment. The slice count is
a hundred-millisecond unit; because a job takes three hundred milliseconds, the first pair
gets a response in three slices, the second pair in six. On a slower machine the slice counts
may shift up by one.

Both modes do the same job, but their latency distributions are opposite.

In the uncontrolled mode **no request was rejected**, and nobody was satisfied: the
last two requests waited a second and a half. If the load had been a hundred requests
instead of ten, the last request's latency would have been fifteen seconds. Every
accepted request carries the latency of everyone accepted before it.

In the controlled mode four requests were served, six were rejected in **slice zero**,
meaning without being held at all. The accepted requests' latency stayed at three and
six slices; the system kept the promise it made to the requests it accepted.

The difference in one sentence: admission control turns **an uncertain wait into a
certain rejection**. This is an improvement for the client, because it has a behavior
in response to rejection — retrying, deferring the job, informing the user. It has no
behavior in response to an uncertain wait.

#### The Shape of the Rejection

How the rejection is communicated matters as much as the decision to reject. The
script below drives the server to saturation and shows the raw response to one
request.

```bash
#!/usr/bin/env bash
# Shows the headers of the rejection response from a server that has reached saturation.
node admission.mjs --controlled > /dev/null & server=$!
sleep 1
A=http://127.0.0.1:8352

pid=()
for i in 1 2 3 4; do curl -sS -o /dev/null "$A/job" & pid+=($!); done   # 2 running + 2 queued
sleep 0.1
echo "--- response to a request arriving at saturation ---"
curl -sS -D - "$A/job"
wait "${pid[@]}"
kill "$server"
```

```
--- response to a request arriving at saturation ---
HTTP/1.1 503 Service Unavailable
content-type: text/plain
retry-after: 1
Connection: keep-alive
Keep-Alive: timeout=5
Transfer-Encoding: chunked

shed
```

Three details are deliberate choices. The status code is **503**, not 500: 500 says the
server hit an unexpected fault and says nothing about whether retrying would help; 503
says the condition is **temporary**, that it will clear if the client waits. The
difference shows up in the client's behavior: a client receiving 500 does not know what
it is waiting for, one receiving 503 does. The `Retry-After` header states, in seconds,
when that retry should happen; the client does not need to guess. And the body is short: a saturated
server producing a long error body is exactly the work it is trying to avoid.

The connection headers depend on the runtime's default settings.

#### What to Shed

Up to here, the request rejected was the one crossing the threshold. This is the
simplest policy, and it has one flaw: it treats all jobs as equal. Yet in the library
service a loan return and a monthly report request are not equal.

Two criteria give this distinction. In shedding **by priority**, every job is given a
class, and at saturation the lowest class is rejected first; a concurrency share
reserved for the critical class is always kept free. In shedding **by deadline**, every
job arrives with a latency budget, and a job whose budget expires while waiting in the
queue is dropped, unprocessed, when its turn comes. The second is especially valuable
because it catches the case where the job at the head of the queue has already become
meaningless — the client may have disconnected.

In both policies, the shedding decision must be fed by a **measured indicator**, not a
fixed threshold: time spent in the queue, worker saturation, the latency of an external
dependency. A fixed threshold sheds either too early or too late once the shape of the
load changes.

#### Retry Storm

Shedding itself produces a new hazard. Six clients rejected at the same time come back
at the same time once their `Retry-After` duration expires. This produces a loop that
keeps the server saturated, and it is called a **retry storm**.

The antidote was already built earlier in this curriculum: the exponential backoff and
jitter introduced in the messaging topic. The `Retry-After` value is a base duration;
the client adds a random jitter to that base and grows the duration on every failed
attempt. A client-side upper attempt limit is also required, or the rejected job comes
back forever.

There is a countermeasure on the server side too: logging a rejected request must be
**cheaper** than logging an accepted one. Logging every rejection in detail under
saturation spends the capacity load shedding just gained on the logging work itself.

#### Summary

- Load shedding is refusing work that exceeds capacity instead of accepting it; admission control, which makes this decision, runs before a job starts.
- The threshold is set on concurrency, not requests per second; per $L = \lambda W$, fixing the number of jobs in the system fixes latency.
- The measurement showed latency spreading from 3 slices to 15 slices in the uncontrolled server, and staying at 3–6 slices for accepted jobs in the controlled server, with the rest rejected in slice zero.
- Admission control turns an uncertain wait into a certain rejection; 503 with `Retry-After` tells the client when to retry, while 500 only reports an unexpected fault and says nothing about timing.
- Shedding produces a retry storm; the antidote is exponential backoff with jitter on the client side and keeping rejection cheap on the server side.

#### Next Step

Everything built up to here concerned the job's **server-side** durability: a job falls
to a worker, fires on time, has its progress recorded, does not overflow the queue, and
is rejected in a controlled way under overload. One end of this chain is still open.
The client that started the job learned it was accepted but does not know **when the
result will be ready**. The progress percentage of a long report, the change in
position in a reservation queue, the update of a branch inventory counter — all of it
is ready on the server side and invisible on the client side. The next topic covers how
this information reaches the client, starting with the simplest route: the client
asking over and over.

## Real-Time Data

The request cost of short and long polling, server-sent events and reconnection, the WebSocket server built with its own handshake and connection lifecycle, channel distribution in a multi-instance deployment, and responses produced piece by piece.

### Polling Approaches

Source: https://academia.sh/en/courses/asynchronous-processing/polling-approaches

The simplest way for a client to learn about a change on the server: the request and empty-response cost of short polling, the connection-holding trade-off of long polling, measuring both against the same event, and closing the event gap with a version number.

The previous topic completed the job's resilience on the server side: a job lands on a
worker, fires on time, has its progress recorded, does not overflow the queue, and is
rejected in a controlled way under overload. One end of the chain is still open. The
client that started the job knows it was accepted, but does not know when the result
will be ready.

The progress percentage of a long-running report, a change in position in a reservation
queue, an update to a branch stock counter — all of these are ready on the server side
but invisible on the client side. HTTP's default direction makes this harder: the client
opens the connection, the client sends the request. The server has no way to speak on
its own.

This topic builds four ways for information to reach the client. We start with the
simplest: the client asking again and again.

#### Polling

**Polling** is the client sending requests to the server at regular intervals to learn
its state. It requires no additional protocol; no extra state is kept on the server
side; every intermediary, firewall, and cache sees it as an ordinary HTTP request. This
is why polling is the default answer to the real-time data problem, and in many cases it
is the right answer.

It has two forms, and the difference between them comes down to a single decision:
**what does the server do when it has nothing to give**.

In **short polling**, the server responds immediately. If there is no change, it says
"no change" and the connection closes; the client waits a while and asks again.

In **long polling**, the server delays the response. It keeps the connection open until
a change occurs; if a change arrives, it writes the response at that moment. If nothing
happens for a set duration, it sends an empty response and the client reconnects.

#### Setting Up Both Approaches on One Server

The server below serves the progress recorded in the `job_status` table from the
previous topic. The status has a **version number**; the client attaches the version it
last saw to the request.

```js
// polling.mjs — serves the status of the report job in two forms: short polling and long polling
import { createServer } from "node:http";

const HOLD_LIMIT = 1000;           // long polling's connection-hold duration (ms)
let version = 0, progress = 0;     // status of the report-2026-01 job
const counter = { status_request: 0, wait_request: 0, empty_response: 0 };
const pending = new Set();         // long polling responses kept open

const body = () => JSON.stringify({ version, progress }) + "\n";
const write = (response, text, code = 200) => {
  response.writeHead(code, { "content-type": "application/json; charset=utf-8" });
  response.end(text);
};
const empty = (response) => { counter.empty_response++; response.writeHead(204).end(); };

createServer((request, response) => {
  response.sendDate = false;
  const url = new URL(request.url, "http://local");
  const seen = Number(url.searchParams.get("version") ?? 0);

  if (url.pathname === "/status") {               // short polling: respond immediately
    counter.status_request++;
    return version > seen ? write(response, body()) : empty(response);
  }

  if (url.pathname === "/wait") {                 // long polling: hold until it changes
    counter.wait_request++;
    if (version > seen) return write(response, body());
    const record = { response };
    record.timer = setTimeout(() => { pending.delete(record); empty(response); },
                               HOLD_LIMIT);
    return void pending.add(record);
  }

  if (url.pathname === "/advance") {              // worker reported progress
    version++; progress += 10;
    for (const p of pending) { clearTimeout(p.timer); write(p.response, body()); }
    pending.clear();
    return write(response, body());
  }

  if (url.pathname === "/metrics")
    return write(response, JSON.stringify(counter) + "\n");

  write(response, JSON.stringify({ error: "route not found" }) + "\n", 404);
}).listen(8361, "127.0.0.1", () => console.log("listening: 127.0.0.1:8361"));
```

The only structural difference between the two endpoints is the `pending` set. In short
polling, the server handles the request and forgets it; in long polling, it stores the
response object and writes to those stored objects when `/advance` arrives. In other
words, long polling adds **state** to the server.

#### Measurement

The script below meets and counts the same single event — the report advancing one
step — with both approaches. Port 8361 is arbitrary and must be free.

```bash
#!/usr/bin/env bash
# Starts polling.mjs, measures the same single event with both approaches, stops it.
A=http://127.0.0.1:8361
B='%{http_code} %{size_download} B\n'

node polling.mjs > /dev/null & server=$!
sleep 1
echo "--- short polling: six requests ---"
for i in 1 2 3 4 5; do curl -sS -w "$B" "$A/status?version=0"; done
curl -sS -o /dev/null -X POST "$A/advance"          # the job just advanced
curl -sS -w "$B" "$A/status?version=0"
curl -sS "$A/metrics"
kill "$server"; sleep 0.3

node polling.mjs > /dev/null & server=$!
sleep 1
echo "--- long polling: two requests ---"
curl -sS -w "$B" "$A/wait?version=0" & waiting=$!
sleep 0.3
curl -sS -o /dev/null -X POST "$A/advance"          # the job just advanced
wait $waiting
curl -sS -w "$B" "$A/wait?version=1"                # no change: time expires
curl -sS "$A/metrics"
kill "$server"
```

```
--- short polling: six requests ---
204 0 B
204 0 B
204 0 B
204 0 B
204 0 B
{"version":1,"progress":10}
200 28 B
{"status_request":6,"wait_request":0,"empty_response":5}
--- long polling: two requests ---
{"version":1,"progress":10}
200 28 B
204 0 B
{"status_request":0,"wait_request":2,"empty_response":1}
```

The same event was met with six requests in short polling and two requests in long
polling. Five responses in short polling were empty; since the body of an empty response
is zero bytes, the data transferred is small, but the **request count** stayed the same.
Every request means a connection setup, a routing decision, an authentication check, and
a log line; an empty body does not make any of these cheaper.

In long polling, the second request stayed open for one second and closed with a 204
because nothing changed. This is long polling's unavoidable cost: a connection cannot be
held forever, because intermediaries and load balancers have their own timeouts.

#### The Cost, Computed

The cost of the two approaches can be computed from the observation window and the event
frequency. The table below is derived assuming a single event within a sixty-second
window.

| Approach | Requests | Empty responses | Worst-case latency |
|---|---|---|---|
| Short polling, 1 s interval | 60 | 59 | 1 s |
| Short polling, 5 s interval | 12 | 11 | 5 s |
| Long polling, 30 s hold | 3 | 1 | network latency |

In short polling, the request count is set by `window / interval`, and the worst-case
latency is the interval itself. The two are inversely proportional, and this is short
polling's one real trade-off: halving the latency doubles the request count.

In long polling, the request count is set by `window / hold time + event count`, and the
latency does not depend on the interval. This is why long polling is markedly cheaper than
short polling for events that are infrequent but latency-sensitive.

The direction of this computation reverses as event frequency rises. For a source that
produces more than one event per second, long polling opens a connection for every
event, while short polling carries several changes in a single request. Data that
changes often and where intermediate values do not matter — a branch stock counter, for
instance — is served more cheaply with short polling.

#### The Server-Side Price

Long polling adds state to the server, and that state has a limit. Every client held
open keeps a socket, a file descriptor, and a response object. A thousand clients means
a thousand open connections; even if they are doing no work, they consume resources.

Three consequences follow. The server's concurrent-connection limit is now a different
quantity from its concurrent-**request** limit and must be sized accordingly. The
admission control from the previous lesson must not count these connections; a pending
long-poll does not use the processor, so counting it as "running work" makes the server
look needlessly saturated. And the hold time must be chosen **shorter** than the
intermediaries' timeout; otherwise a layer in between, not the server, cuts the
connection, and the client cannot tell why.

#### The Event Gap

Long polling's overlooked flaw is this: in the interval between the response being
written and the client reconnecting, the server is **not listening**. An event that
occurs during that gap cannot be written to any open connection.

In the code, the version number is what closes this gap. The client attaches the version
it last saw to the request; the server checks the condition `version > seen` before
holding the connection, and answers a client that has fallen behind immediately. This is
why the second long poll in the measurement was sent with `version=1`.

This is a shared requirement of every approach that serves real-time data: the client
must be able to say where it left off. A version number, an event ID, or a timestamp —
the name changes, the function does not. The next lesson shows the same job done by the
`Last-Event-ID` header.

#### Summary

- Polling is the client sending requests at regular intervals to learn its state; it is the default approach because it requires no additional protocol.
- In short polling, the server responds immediately and produces empty responses; the request count grows with `window / interval`, and the worst-case latency is the interval itself.
- In long polling, the server holds the connection until a change occurs; in the measurement, the same event was met with two requests instead of six, and the latency stopped depending on the interval.
- Long polling's price is the state it adds to the server: every open connection consumes resources, and the hold time must be chosen shorter than intermediaries' timeouts.
- The event gap between two polls is closed by the client attaching the version it last saw to the request.

#### Next Step

Long polling closes the connection after holding it and writing a single response.
Opening a new connection for every event turns into an absurd arrangement once events
come frequently: ten separate connections for ten events that should be written to the
same client one after another. Yet a connection that is already open can have more than
one event written onto it. The next lesson builds the standardized form of this: a
one-way stream where the server writes events one after another onto a single open
response body, and the browser side takes care of reconnecting on its own.

### Server-Sent Events

Source: https://academia.sh/en/courses/asynchronous-processing/server-sent-events

Events written one after another onto a single open response body: the line format of the event stream, the header conventions of a streaming response, reconnection by event ID, and the function of the heartbeat line against intermediaries.

Long polling held the connection and closed it after writing a single response. A
reservation queue position changing five times meant five separate connections: for
every event the client reconnected, the server re-authorized, and the intermediaries
re-routed.

Yet the first connection was already open. An HTTP response body does not have to be
written all at once; it can be kept open and written to in pieces. **Server-Sent
Events** (SSE) standardizes this: a one-way, text-based stream from server to client
that carries more than one event.

#### Format

The stream's body is line-based and marked with the `text/event-stream` content type.
An event consists of field lines and ends with **a blank line**. Four fields are
defined, plus a comment line that is not a field.

`data:` is the event's body and the only required field; multiple `data:` lines are
joined with line breaks. `event:` gives the event's type; the client can attach
different handlers for different types. `id:` is the event's identifier and marks where
the stream left off. `retry:` tells the client how many milliseconds to wait after a
disconnect. A line starting with a colon is a comment; it produces no event and is used
as a **heartbeat**.

The entire format is plain text, so the stream can be read directly with `curl`. This is
SSE's most practical aspect for debugging.

#### Server

The server below publishes a member's position in a reservation queue: five steps of
progress and a closing event announcing that the book is ready.

```js
// sse.mjs — publishes reservation queue progress via server-sent events
import { createServer } from "node:http";

// The member's position in the reservation queue; the last event announces that the request has been fulfilled.
const EVENTS = [
  { id: 1, type: "queue", data: { position: 5, eta_min: 12 } },
  { id: 2, type: "queue", data: { position: 4, eta_min: 9 } },
  { id: 3, type: "queue", data: { position: 3, eta_min: 7 } },
  { id: 4, type: "queue", data: { position: 2, eta_min: 4 } },
  { id: 5, type: "queue", data: { position: 1, eta_min: 2 } },
  { id: 6, type: "ready", data: { book: "The Disconnected", branch: 3 } },
];

const counter = { connection: 0, event: 0 };
const wait = (ms) => new Promise((c) => setTimeout(c, ms));

createServer(async (request, response) => {
  response.sendDate = false;
  const url = new URL(request.url, "http://local");

  if (url.pathname === "/metrics") {
    response.writeHead(200, { "content-type": "application/json; charset=utf-8" });
    return response.end(JSON.stringify(counter) + "\n");
  }
  if (url.pathname !== "/stream") return response.writeHead(404).end();

  counter.connection++;
  // Streaming response: the body never ends, so no length is reported and
  // intermediaries' response buffering is disabled.
  response.writeHead(200, {
    "content-type": "text/event-stream; charset=utf-8",
    "cache-control": "no-store",
    "x-accel-buffering": "no",
  });

  // The client reports where it left off via the Last-Event-ID header; otherwise it starts from the beginning.
  const lastEventId = Number(request.headers["last-event-id"] ?? 0);
  const cut = Number(url.searchParams.get("cut") ?? 0);

  response.write("retry: 2000\n\n");                // reconnection interval (ms)
  response.write(": heartbeat\n\n");                 // comment line: keeps the connection alive

  let sent = 0;
  for (const e of EVENTS.filter((e) => e.id > lastEventId)) {
    if (cut && sent === cut) break;                 // simulates a connection drop
    await wait(50);
    if (response.writableEnded) return;              // client disconnected
    response.write(`id: ${e.id}\nevent: ${e.type}\ndata: ${JSON.stringify(e.data)}\n\n`);
    counter.event++; sent++;
  }
  response.end();
}).listen(8371, "127.0.0.1", () => console.log("listening: 127.0.0.1:8371"));
```

#### Reading the Stream

The `curl -N` option disables output buffering and shows the stream as it arrives. The
script below first cuts the stream off after three events, then resumes it from where it
left off. Port 8371 is arbitrary and must be free.

```bash
#!/usr/bin/env bash
# Starts sse.mjs, reads the stream with an interruption, resumes with Last-Event-ID, stops it.
node sse.mjs > /dev/null & server=$!
sleep 1
A=http://127.0.0.1:8371

echo "--- first connection: server cuts off after three events ---"
curl -sN "$A/stream?cut=3"
echo "--- reconnection: resuming from where it left off with Last-Event-ID ---"
curl -sN -H 'Last-Event-ID: 3' "$A/stream"
echo "--- metrics ---"
curl -sS "$A/metrics"
kill "$server"
```

```
--- first connection: server cuts off after three events ---
retry: 2000

: heartbeat

id: 1
event: queue
data: {"position":5,"eta_min":12}

id: 2
event: queue
data: {"position":4,"eta_min":9}

id: 3
event: queue
data: {"position":3,"eta_min":7}

--- reconnection: resuming from where it left off with Last-Event-ID ---
retry: 2000

: heartbeat

id: 4
event: queue
data: {"position":2,"eta_min":4}

id: 5
event: queue
data: {"position":1,"eta_min":2}

id: 6
event: ready
data: {"book":"The Disconnected","branch":3}

--- metrics ---
{"connection":2,"event":6}
```

The metrics line shows the actual gain: **six events, two connections**. Without the
interruption, a single connection would have sufficed. Carrying the same six events with
long polling would have required six connections, six authorizations, and six routing
decisions.

#### Reconnection

The stream breaks sooner or later: the server is redeployed, an intermediary times out,
the network drops. What distinguishes SSE from long polling is that a disconnect is **a
condition the protocol expects**.

Two fields govern this behavior. The duration announced with `retry:` sets how long the
client waits after a disconnect; the server can vary this value with the state of the
load, and so manage a reconnection storm. The identifier announced with `id:` is stored
on the client and sent back as the `Last-Event-ID` header on the reconnection request.
The server reads this header and resumes the stream from there.

The browser-side `EventSource` interface does both on its own: it detects the
disconnect, waits for the `retry` duration, and puts the last ID it saw into the header.
The server's only responsibility is making sure that ID is **a meaningful resume
point**.

This creates a requirement on the server side. Resuming by identifier requires events to
be kept for a while; if the server knows only the "current state," events produced
during a disconnect are lost. Two solutions exist in practice: a ring buffer holding the
last N events, or sending **the entire state** instead of an event. The second makes the
identifier unnecessary, because every event is sufficient on its own; in exchange, every
event is larger.

#### Rules of a Streaming Response

A streaming response differs from an ordinary response in four points, and three of
these differences show up in the headers.

The body's length is **not reported**, because it is not known; `Content-Length` cannot be
written. The response must not be cached; `Cache-Control: no-store` says this, otherwise
an intermediate cache would try to buffer the stream and deliver it as a single piece.
Intermediaries' response buffering must be disabled; the code does this with the
`X-Accel-Buffering` header, which is not a standard header but is a widely recognized
directive. The fourth difference is in the code: before every write, whether the client
has disconnected is checked, otherwise the server keeps writing to a closed socket.

This is also where the heartbeat line's function comes from. Intermediaries can treat a
connection with no data flowing for a long time as dead and close it. A comment line
sent at regular intervals — `: heartbeat`, which produces no event — keeps the
connection alive and has no side effect on the client side.

#### Limits

SSE is one-way. The client cannot send data to the server; it must open a separate HTTP
request for that. This is sufficient for tracking a reservation queue, but not for an
arrangement where both sides are constantly talking.

The second limit is in the transport layer. Over HTTP/1.1, the number of concurrent
connections browsers open to the same origin is limited, and a stream that stays open
permanently holds on to one of that share; a user with several tabs open can exhaust the
limit. Over HTTP/2, streams are multiplexed onto a single connection, so this limit
disappears.

The third limit is the data format: the stream is text, binary data cannot be carried. If
binary data is needed, it must be encoded, and that inflates the size.

#### Summary

- Server-sent events form a one-way, text-based stream from server to client, written one after another onto a single HTTP response body that is kept open.
- An event consists of the `data`, `event`, `id`, and `retry` fields and ends with a blank line; a comment line starting with a colon is used as a heartbeat.
- In the measurement, six events were carried over two connections; the same events would have required six connections with long polling.
- A disconnect is a condition the protocol expects: the client waits for the `retry` duration and sends the last ID it saw with the `Last-Event-ID` header, and the server resumes the stream from there.
- A streaming response does not report its length, must not be cached, must have intermediary buffering disabled, and must check whether the client is still connected before every write.

#### Next Step

As long as the stream is one-way, it is sufficient for a reservation queue. But the
library service has cases where both sides talk: a clerk correcting a branch stock
counter, a member withdrawing from the queue, several clerks editing the same loan
record at the same time. Opening a separate HTTP request for each of these gives back
the latency the stream gained. The next lesson covers how a two-way connection is set
up: a connection that starts over HTTP, switches to its own frame format after a
handshake, and carries messages in both directions.

### WebSocket Server

Source: https://academia.sh/en/courses/asynchronous-processing/websocket-server

Building a two-way connection from scratch: the upgrade handshake over HTTP and generating the accept key, writing and parsing the frame format, the ping–pong exchange and the closing handshake, and how connection state affects horizontal scaling.

Server-sent events were one-way: the server talks, the client listens. The library
service has cases where both sides talk — a clerk correcting a branch stock counter, a
member withdrawing from the reservation queue. Opening a separate HTTP request for each
of these gives back the latency the stream gained.

**WebSocket** is the transport that answers this need: a single connection that starts
over HTTP, switches to its own frame format after a handshake, and carries messages in
both directions. This lesson builds it without using a ready-made package, because the
one thing that cannot be learned without building it is **what the connection holds onto**
and why that causes trouble at scale.

#### The Upgrade Handshake

The connection starts with an ordinary HTTP request. The client sends the
`Upgrade: websocket` and `Connection: Upgrade` headers, along with a randomly generated
sixteen-byte value encoded in base64, in the `Sec-WebSocket-Key` header.

The server concatenates this key with a GUID fixed in the specification, takes its
SHA-1 digest, encodes it in base64, and returns the result in the
`Sec-WebSocket-Accept` header. The response's status code is **101 Switching
Protocols**, and from this point on there is no HTTP on the connection.

```js
// accept-key.mjs — generation and verification of the accept value in the handshake
import { createHash } from "node:crypto";

const GUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11";  // value fixed in the protocol
const key = "dGhlIHNhbXBsZSBub25jZQ==";               // sample client key from the specification
const accept = createHash("sha1").update(key + GUID).digest("base64");

console.log("Sec-WebSocket-Key   :", key);
console.log("Sec-WebSocket-Accept:", accept);
console.log("specification value :", accept === "s3pPLMBiTxaQ9kYGzzhZRbK+xOo=");
```

```
Sec-WebSocket-Key   : dGhlIHNhbXBsZSBub25jZQ==
Sec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=
specification value : true
```

This computation **has no security function**. SHA-1 here is not a signature but a
check: it shows that the other side genuinely understood the request as a WebSocket
upgrade — that a server which does not speak WebSocket has not accepted the connection
by mistake. Authentication must be done
separately, in the upgrade request itself — in a cookie, a header, or the first message.

#### Frame Format

After the handshake, **frames** are carried on the connection. A frame's header is at
least two bytes. The high bit of the first byte is `FIN`, which says whether the message
ends with this frame; the low four bits are the **opcode**: text, binary, close, ping,
pong.

The high bit of the second byte is the **mask bit**, and the remaining seven bits are
the length. A length up to 125 is written directly; the value 126 announces that the
next two bytes carry the real length, and 127 that the next eight bytes do.

The masking rule is one-directional: every frame going from client to server **must**
be masked, and none going from server to client should be. Masking provides no
confidentiality — the key is inside the frame. Its purpose is to stop an old
intermediary from mistaking the frame body for a valid HTTP request and poisoning its
cache.

```js
// frame.mjs — writes and parses the WebSocket frame format
import { randomBytes } from "node:crypto";

export const OPCODE = { text: 0x1, binary: 0x2, close: 0x8, ping: 0x9, pong: 0xa };

// Frame: [FIN + opcode][mask bit + length][extended length?]
//        [mask key?][body].  Every frame coming from the client must be masked.
export function writeFrame(opcode, body, masked = false) {
  const data = Buffer.isBuffer(body) ? body : Buffer.from(String(body), "utf8");
  const n = data.length;
  const ext = n < 126 ? 0 : n < 65536 ? 2 : 8;
  const header = Buffer.alloc(2 + ext + (masked ? 4 : 0));
  header[0] = 0x80 | opcode;                       // FIN = 1: frame is a single piece
  header[1] = (masked ? 0x80 : 0) | (ext === 0 ? n : ext === 2 ? 126 : 127);
  if (ext === 2) header.writeUInt16BE(n, 2);
  if (ext === 8) header.writeBigUInt64BE(BigInt(n), 2);
  if (!masked) return Buffer.concat([header, data]);
  const key = randomBytes(4);
  key.copy(header, 2 + ext);
  const maskedBody = Buffer.from(data);
  for (let i = 0; i < n; i++) maskedBody[i] ^= key[i % 4];
  return Buffer.concat([header, maskedBody]);
}

// Parses one full frame from the buffer; returns null if the frame is incomplete.
export function parseFrame(buffer) {
  if (buffer.length < 2) return null;
  const opcode = buffer[0] & 0x0f, fin = (buffer[0] & 0x80) !== 0;
  const masked = (buffer[1] & 0x80) !== 0;
  let n = buffer[1] & 0x7f, offset = 2;
  if (n === 126) { if (buffer.length < 4) return null; n = buffer.readUInt16BE(2); offset = 4; }
  else if (n === 127) {
    if (buffer.length < 10) return null;
    n = Number(buffer.readBigUInt64BE(2)); offset = 10;
  }
  const key = masked ? buffer.subarray(offset, offset + 4) : null;
  if (masked) offset += 4;
  if (buffer.length < offset + n) return null;
  const body = Buffer.from(buffer.subarray(offset, offset + n));
  if (key) for (let i = 0; i < n; i++) body[i] ^= key[i % 4];
  return { opcode, fin, masked, body, consumed: offset + n };
}
```

The fact that `parseFrame` returns `null` on an incomplete frame is not a detail, it is a
necessity. The layer underneath is TCP, and TCP is a **byte stream**: a frame that was
sent can arrive in two pieces, and two frames can arrive in one piece. This is why the
reading side accumulates incoming bytes and pulls as many frames as it can parse from
the buffer.

#### Server

The server attaches to the HTTP server's `upgrade` event. This event hands over the raw
socket when an upgrade request arrives; the server itself writes the response line and
headers.

```js
// ws-server.mjs — a WebSocket server that does its own handshake and frame parsing
import { createServer } from "node:http";
import { createHash } from "node:crypto";
import { OPCODE, writeFrame, parseFrame } from "./frame.mjs";

const GUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11";
const generateAccept = (key) => createHash("sha1").update(key + GUID).digest("base64");

const clients = new Set();
const log = (...s) => console.log(...s);

const server = createServer((request, response) => {
  response.sendDate = false;
  response.writeHead(426, { "content-type": "text/plain", upgrade: "websocket" });
  response.end("this endpoint only accepts WebSocket connections\n");
});

server.on("upgrade", (request, socket) => {
  const key = request.headers["sec-websocket-key"];
  if (request.headers.upgrade?.toLowerCase() !== "websocket" || !key)
    return socket.destroy();

  socket.write("HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\n" +
               `Connection: Upgrade\r\nSec-WebSocket-Accept: ${generateAccept(key)}\r\n\r\n`);
  clients.add(socket);
  log(`connection opened (open=${clients.size})`);

  const send = (opcode, body) => socket.write(writeFrame(opcode, body));  // server does not mask
  let buffer = Buffer.alloc(0);

  socket.on("data", (chunk) => {
    buffer = Buffer.concat([buffer, chunk]);
    let frame;
    while ((frame = parseFrame(buffer))) {
      buffer = buffer.subarray(frame.consumed);
      log(`frame received    opcode=0x${frame.opcode.toString(16)}` +
          ` masked=${frame.masked} length=${frame.body.length}`);

      if (frame.opcode === OPCODE.text) {
        const message = JSON.parse(frame.body.toString());
        send(OPCODE.text, JSON.stringify({ type: "ack", branch: message.branch, newStock: 41 }));
      } else if (frame.opcode === OPCODE.ping) {
        send(OPCODE.pong, frame.body);              // answers a ping with a pong using the same body
      } else if (frame.opcode === OPCODE.close) {
        const code = frame.body.length >= 2 ? frame.body.readUInt16BE(0) : 1005;
        log(`close request     code=${code} reason=${frame.body.subarray(2).toString()}`);
        send(OPCODE.close, frame.body);              // closing handshake: reply with the same code
        socket.end();
      }
    }
  });

  socket.on("close", () => {
    clients.delete(socket);
    log(`connection closed (open=${clients.size})`);
  });
});

server.listen(8381, "127.0.0.1", () => log("listening: 127.0.0.1:8381"));
```

#### Client

The client uses the same format, with two differences: it writes the handshake request
itself, and it masks every frame it sends.

```js
// ws-client.mjs — a small WebSocket client that hand-shakes and sends frames over raw TCP
import { connect } from "node:net";
import { createHash, randomBytes } from "node:crypto";
import { OPCODE, writeFrame, parseFrame } from "./frame.mjs";

const GUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11";
const key = randomBytes(16).toString("base64");
const expectedAccept = createHash("sha1").update(key + GUID).digest("base64");

const queue = [], waiting = [];
const receiveFrame = () => queue.length
  ? Promise.resolve(queue.shift())
  : new Promise((c) => waiting.push(c));
const frameArrived = (f) => (waiting.length ? waiting.shift()(f) : queue.push(f));

let buffer = Buffer.alloc(0), handshakeDone = false;
const socket = connect(8381, "127.0.0.1", () => {
  socket.write("GET /channel HTTP/1.1\r\nHost: 127.0.0.1:8381\r\n" +
               "Upgrade: websocket\r\nConnection: Upgrade\r\n" +
               `Sec-WebSocket-Key: ${key}\r\nSec-WebSocket-Version: 13\r\n\r\n`);
});

socket.on("error", (err) => {                      // server is not up: exit quietly
  console.log("could not connect to server:", err.code);
  process.exit(0);
});

socket.on("data", (chunk) => {
  buffer = Buffer.concat([buffer, chunk]);
  if (!handshakeDone) {
    const end = buffer.indexOf("\r\n\r\n");
    if (end < 0) return;
    const headers = buffer.subarray(0, end).toString();
    buffer = buffer.subarray(end + 4);
    handshakeDone = true;
    console.log("status line       :", headers.split("\r\n")[0]);
    console.log("accept key        :",
      /sec-websocket-accept: (.+)/i.exec(headers)?.[1] === expectedAccept ? "correct" : "incorrect");
    talk();
  }
  let f;
  while ((f = parseFrame(buffer))) { buffer = buffer.subarray(f.consumed); frameArrived(f); }
});

async function talk() {
  const send = (opcode, body) => socket.write(writeFrame(opcode, body, true));

  send(OPCODE.text, JSON.stringify({ type: "stock", branch: 3, delta: -1 }));
  const res1 = await receiveFrame();
  console.log("text reply        :", res1.body.toString(), `(masked=${res1.masked})`);

  send(OPCODE.ping, "is-alive");
  const res2 = await receiveFrame();
  console.log("ping reply        : opcode=0x" + res2.opcode.toString(16), res2.body.toString());

  const reason = Buffer.concat([Buffer.from([0x03, 0xe8]), Buffer.from("job done")]);
  send(OPCODE.close, reason);                       // 0x03e8 = 1000, normal closure
  const res3 = await receiveFrame();
  console.log("close reply       : opcode=0x" + res3.opcode.toString(16),
              "code=" + res3.body.readUInt16BE(0));
  socket.end();
}
```

#### Lifecycle

The script below first sends an HTTP request without upgrading, then runs the client.
Port 8381 is arbitrary and must be free.

```bash
#!/usr/bin/env bash
# Starts ws-server.mjs, first sends a plain HTTP request without upgrading, then runs the client, stops it.
node ws-server.mjs > server.log & server=$!
sleep 1

echo "--- plain HTTP request without upgrading ---"
curl -sS -i http://127.0.0.1:8381/channel | head -3
echo "--- what the client sees ---"
node ws-client.mjs
sleep 0.3
echo "--- the server's log ---"
cat server.log
kill "$server"
```

```
--- plain HTTP request without upgrading ---
HTTP/1.1 426 Upgrade Required
content-type: text/plain
upgrade: websocket
--- what the client sees ---
status line       : HTTP/1.1 101 Switching Protocols
accept key        : correct
text reply        : {"type":"ack","branch":3,"newStock":41} (masked=false)
ping reply        : opcode=0xa is-alive
close reply       : opcode=0x8 code=1000
--- the server's log ---
listening: 127.0.0.1:8381
connection opened (open=1)
frame received    opcode=0x1 masked=true length=38
frame received    opcode=0x9 masked=true length=8
frame received    opcode=0x8 masked=true length=10
close request     code=1000 reason=job done
connection closed (open=0)
```

The log shows the lifecycle's four phases. **Opening**: the upgrade is accepted, the
connection counter increases. **Data**: the text frame from the client is masked, the
reply that comes back is not. **Control**: the ping (0x9) frame is answered with a pong
(0xa) using the same body. **Closing**: the 0x8 frame is a closing **request**, not a
command; the socket closes after the other side sends its own close frame.

The ping–pong pair is indispensable in real deployments. A TCP connection can appear
open for a long time when the other side has silently vanished; a ping sent at regular
intervals not being answered with a pong is the only reliable sign that the connection
is dead. The same traffic also stops intermediaries from closing an idle connection —
it is the form here that corresponds to the heartbeat line in server-sent events.

#### Scaling

The real difference between WebSocket and HTTP is not in the headers, it is in the
**state**. An ordinary HTTP request is stateless; it gives the same result no matter
which server instance it lands on. A WebSocket connection, on the other hand, **binds**
to an instance and stays there for the connection's duration.

This has three consequences. The unit of scaling is now the connection, not the
request; the server's capacity is measured not by requests per second but by how many
sockets and how much memory it can hold at once. Redeployment breaks every open
connection; reconnecting and refreshing state on the client side becomes mandatory. And
most importantly: two clients listening to the same channel can end up on **different
instances**. A message arriving at one instance does not reach the other's client on its
own.

This third consequence is the fundamental problem of every multi-instance real-time
deployment.

#### Summary

- A WebSocket connection starts with an HTTP request; the server concatenates the client's key with a fixed GUID, returns its SHA-1 digest as base64, and switches the protocol with 101.
- The accept key is not a security measure; it is a check that the request was understood as WebSocket and that an old response has not been replayed.
- The frame header carries the FIN bit, the opcode, the mask bit, and the length; every frame from the client is masked, and every frame from the server is unmasked.
- Because the layer underneath is a byte stream, the parser must wait on an incomplete frame and pull as many frames as it can parse from the buffer.
- The lifecycle is opening, data, ping–pong, and a two-sided closing handshake; a connection binding to one instance is what makes the unit of scaling the connection rather than the request.

#### Next Step

This lesson had a single server instance and a single client; where the message would go
was clear. In a real deployment, clients listening to the same channel are spread across
different instances: of three clerks watching a branch stock counter, two might be
connected to the first instance and one to the second. An update arriving at the first
instance does not reach the clerk on the second instance — because nobody forwards that
message to them. The next lesson builds the structure that closes this gap and shows, by
running it, that a message from one instance actually reaches another instance's
subscriber.

### Broadcast and Channel Management

Source: https://academia.sh/en/courses/asynchronous-processing/broadcast-and-channel-management

Getting a message to reach every subscriber in a multi-instance deployment: the channel concept, measuring how local delivery falls short, building an inter-instance broadcast channel, the cost of fan-out, and subscription accounting.

The previous lesson built the two-way connection and ended with a warning: a WebSocket
connection binds to a server instance and stays there for the connection's duration. The
same holds for server-sent events; every stream kept open sits in the memory of a
specific instance.

In a single-instance deployment, this is invisible. In a service spread across two
instances, the consequence is this: if two of three clerks watching a branch stock
counter are connected to the first instance and one to the second, an update arriving at
the first instance **does not reach** the third clerk. This lesson measures that gap and
closes it.

#### Channel

First, naming. The set of messages a client is interested in is called a **channel**;
in the library service, something like `branch-3`, `loan-1204`, or `report-2026-01`. A
channel differs from the queue introduced in the messaging topic in two respects.

A channel **has no memory**: a message does not reach a subscriber who is not listening
at broadcast time, and it does not accumulate anywhere. And a channel is
**multi-subscriber**: a copy of the same message goes to every subscriber, and the
first one to receive it does not consume it. In other words, a channel is the
connection-level counterpart of the topic and the publish–subscribe model from the
messaging topic.

These two properties are generally true for real-time data — nobody cares about a
branch stock counter's value from three seconds ago. If persistence is needed, a
channel is not enough; the message must also be written to a durable queue or table.

#### Application Instance

The instance below offers two endpoints. `/subscribe` opens an event stream to the
client and adds it to the channel set; `/publish` posts a message to a channel. It runs
in two modes: `--local` mode, which delivers the broadcast only to its own subscribers,
and `--relay` mode, which sends the broadcast to the relay.

```js
// instance.mjs — application instance: gives clients channel subscriptions, produces broadcasts
//   node instance.mjs <port> --local   the broadcast goes only to its own subscribers
//   node instance.mjs <port> --relay   the broadcast goes to the relay, the relay delivers to everyone
import { createServer } from "node:http";

const PORT = Number(process.argv[2] ?? 8392);    // default: first instance
const WITH_RELAY = process.argv.includes("--relay");
const RELAY = "http://127.0.0.1:8391";

const channels = new Map();                        // channel -> Set<subscriber response>
const counter = { subscription: 0, local_delivery: 0, from_relay: 0 };

const readBody = (request) => new Promise((c) => {
  let v = ""; request.on("data", (p) => (v += p)); request.on("end", () => c(v));
});

function localDeliver(channel, body) {
  for (const r of channels.get(channel) ?? []) {
    r.write(`data: ${body}\n\n`);
    counter.local_delivery++;
  }
}

// The side listening to the relay: delivers every incoming broadcast to its own local subscribers.
async function listenToRelay() {
  const response = await fetch(`${RELAY}/listen`);
  let buffer = "";
  for await (const chunk of response.body) {
    buffer += Buffer.from(chunk).toString();
    let k;
    while ((k = buffer.indexOf("\n\n")) >= 0) {
      const block = buffer.slice(0, k); buffer = buffer.slice(k + 2);
      if (!block.startsWith("data: ")) continue;   // comment line: ignore
      const message = JSON.parse(block.slice(6));
      counter.from_relay++;
      localDeliver(message.channel, message.data);
    }
  }
}

createServer(async (request, response) => {
  response.sendDate = false;
  const url = new URL(request.url, "http://local");
  const channel = url.searchParams.get("channel");

  if (url.pathname === "/subscribe") {
    response.writeHead(200, { "content-type": "text/event-stream; charset=utf-8",
                              "cache-control": "no-store" });
    response.write(`: subscribed channel=${channel} instance=${PORT}\n\n`);
    if (!channels.has(channel)) channels.set(channel, new Set());
    channels.get(channel).add(response);
    counter.subscription++;
    return void request.on("close", () => channels.get(channel).delete(response));
  }

  if (url.pathname === "/publish") {
    const data = await readBody(request);
    if (WITH_RELAY) await fetch(`${RELAY}/broadcast`, {
      method: "POST", body: JSON.stringify({ channel, data }),
    });
    else localDeliver(channel, data);
    return response.writeHead(202).end();
  }

  if (url.pathname === "/metrics") {
    response.writeHead(200, { "content-type": "application/json; charset=utf-8" });
    return response.end(JSON.stringify({ instance: PORT, ...counter }) + "\n");
  }
  response.writeHead(404).end();
}).listen(PORT, "127.0.0.1", () => {
  if (WITH_RELAY) listenToRelay();
  console.log(`instance: 127.0.0.1:${PORT}`);
});
```

The subscription being **removed** when the connection closes is a critical detail.
Without the `request.on("close")` handler, the channel set keeps holding closed
response objects; attempting to write to them produces an error and the set grows
without bound. This is the most common leak in any structure that holds long-lived
connections.

#### Inter-Instance Broadcast Channel

The structure the instance talks to in `--relay` mode is a middle layer whose only job
is getting the message to every instance. Each instance connects to it with a stream; a
broadcast from one instance is written to **all** connected instances — including the
instance that sent the broadcast.

```js
// relay.mjs — inter-instance broadcast channel; keeps one stream open per application instance
import { createServer } from "node:http";

const instances = new Set();                        // instance responses listening to the relay
const counter = { broadcast: 0, delivery: 0 };

const readBody = (request) => new Promise((c) => {
  let v = ""; request.on("data", (p) => (v += p)); request.on("end", () => c(v));
});

createServer(async (request, response) => {
  response.sendDate = false;
  const url = new URL(request.url, "http://local");

  if (url.pathname === "/listen") {                 // instance connects to the relay
    response.writeHead(200, { "content-type": "text/event-stream; charset=utf-8",
                              "cache-control": "no-store" });
    response.write(": connected to relay\n\n");
    instances.add(response);
    return void request.on("close", () => instances.delete(response));
  }

  if (url.pathname === "/broadcast") {               // a broadcast from one instance: to everyone
    const body = await readBody(request);
    counter.broadcast++;
    for (const i of instances) { i.write(`data: ${body}\n\n`); counter.delivery++; }
    return response.writeHead(202).end();
  }

  if (url.pathname === "/metrics") {
    response.writeHead(200, { "content-type": "application/json; charset=utf-8" });
    return response.end(JSON.stringify({ ...counter, instance: instances.size }) + "\n");
  }
  response.writeHead(404).end();
}).listen(8391, "127.0.0.1", () => console.log("relay: 127.0.0.1:8391"));
```

Sending the broadcast back to the instance that sent it is not a bug, it is a
simplification: local delivery happens in exactly one place, along the path coming back
from the relay. Otherwise the message would be delivered twice on its own instance.

#### Measurement

The script below runs two application instances and one relay. The subscriber connects
to the **second** instance, while the broadcast arrives at the **first** instance. Ports
8391–8393 are arbitrary and must be free.

```bash
#!/usr/bin/env bash
# Two application instances plus one relay; the subscriber is on the second instance, the broadcast arrives at the first instance.
measure() {
  node relay.mjs > /dev/null & relay=$!
  node instance.mjs 8392 "$1" > /dev/null & i1=$!
  node instance.mjs 8393 "$1" > /dev/null & i2=$!
  sleep 1

  curl -sN --max-time 1.2 "http://127.0.0.1:8393/subscribe?channel=branch-3" > subscriber.txt &
  listener=$!
  sleep 0.4
  curl -sS -o /dev/null -X POST -d '{"branch":3,"stock":41}' \
       "http://127.0.0.1:8392/publish?channel=branch-3"
  wait $listener

  echo "--- $2 ---"
  echo "what the subscriber on the second instance saw:"
  sed '/^$/d; s/^/  /' subscriber.txt
  echo "counters:"
  curl -sS http://127.0.0.1:8392/metrics | sed 's/^/  /'
  curl -sS http://127.0.0.1:8393/metrics | sed 's/^/  /'
  curl -sS http://127.0.0.1:8391/metrics | sed 's/^/  relay /'
  kill $relay $i1 $i2 2>/dev/null
  sleep 0.5
}

measure --local "no relay"
measure --relay "with relay"
```

```
--- no relay ---
what the subscriber on the second instance saw:
  : subscribed channel=branch-3 instance=8393
counters:
  {"instance":8392,"subscription":0,"local_delivery":0,"from_relay":0}
  {"instance":8393,"subscription":1,"local_delivery":0,"from_relay":0}
  relay {"broadcast":0,"delivery":0,"instance":0}
--- with relay ---
what the subscriber on the second instance saw:
  : subscribed channel=branch-3 instance=8393
  data: {"branch":3,"stock":41}
counters:
  {"instance":8392,"subscription":0,"local_delivery":0,"from_relay":1}
  {"instance":8393,"subscription":1,"local_delivery":1,"from_relay":1}
  relay {"broadcast":1,"delivery":2,"instance":2}
```

In the first run, the subscriber saw only the subscription confirmation; the broadcast
went nowhere. The first instance's `local_delivery` counter is zero, because there is no
subscriber on that instance. The message was not lost, and it did not produce an error —
**nothing happened, silently**. This is the dangerous side of this flaw in
multi-instance deployments: it gives no sign while developing with a single instance.

In the second run, the subscriber received the message. The counters show the path step
by step: one broadcast arrived at the relay, it was delivered to two instances, both
instances received one message each, but only the instance with a subscriber made a
local delivery.

#### The Cost of Fan-Out

The relay's `delivery` counter is a count of copies, not broadcasts. One broadcast turns
into as many copies as there are instances. For $N$ instances and $M$ broadcasts per
second, the number of messages the relay writes is $N \times M$, and for instances with
no subscriber, this is work that is entirely wasted. In the measurement, the first
instance received the message and delivered it nowhere.

There are two ways to reduce this. Keeping a **subscription registry**, the relay knows
which channels each instance is listening to and sends the broadcast only to the
relevant instances. The gain is significant when the number of channels is larger than
the number of instances; the cost is that a registration message goes to the relay
every time a subscriber opens or closes. With **channel partitioning**, channels are
assigned to instances in advance, and the client is routed to the instance hosting the
channel it is interested in; distribution disappears entirely, but so does flexibility,
because a client listening to several channels must open several connections.

As scale grows, a third limit appears: the relay itself is a single point. In real
deployments, this job is handed off to the publish–subscribe infrastructure introduced
in this course's messaging topic; the relay becomes a thin adaptation layer in front of
that infrastructure.

#### Is Sticky Routing Enough

A frequently used shortcut is always routing the same client to the same instance; this
is called **sticky routing**. It is useful because it gets the client back to the same
instance when a connection drops, but it does **not solve** the problem here: two
different clients listening to the same channel can still land on different instances.

Stickiness becomes a solution only when it is channel-based — that is, when the routing
decision looks at the channel rather than the client. This is exactly the channel
partitioning approach above, and it requires the load balancer to know about
application concepts.

#### Summary

- A channel is the set of messages a client is interested in; it has no memory and is multi-subscriber, so data that needs persistence must be written separately.
- Because an open connection sits in the memory of a specific instance, local delivery does not get a message to other instances' subscribers in a multi-instance deployment, and this flaw produces no error.
- An inter-instance broadcast channel closes this gap; the sending instance also getting the message back keeps local delivery on a single path.
- If a subscription is not removed when the connection closes, the channel set keeps holding closed responses and grows without bound.
- The cost of fan-out is multiplied by the number of instances; a subscription registry or channel partitioning reduces it, and sticky routing alone does not solve it.

#### Next Step

All four approaches in this topic carried an **event** from server to client: a status
changed, a queue advanced, stock was updated. One case remains. Sometimes what needs to
be carried is not an event, it is **the response itself** — the output of a report
scanning thirty thousand loan records can be streamed to the client at the rate it is
produced. This requires no new protocol; it is done by writing the HTTP response body in
pieces. The next lesson builds this final form and closes the course.

### Streaming Responses

Source: https://academia.sh/en/courses/asynchronous-processing/streaming-responses

Producing the response itself in pieces: the framing of chunked transfer encoding, measuring the time to first byte, the limits of reporting an error after headers have been sent, and backpressure in a streaming response.

All four approaches in this topic carried an **event** from server to client: a status
changed, a queue advanced, stock was updated. One case remains. Sometimes what needs to
be carried is not an event, it is **the response itself**.

A monthly report scanning thirty thousand loan records can be sent line by line before
the result is complete. This requires no new protocol; it is done by writing the HTTP
response body in pieces, and in exchange, the time the client waits to see the first
data shrinks significantly.

#### Two Forms

A response body can be sent in two forms. If its length is known, it is announced in the
`Content-Length` header, and the receiver knows from the start how many bytes to
expect. If its length is not known, **chunked transfer encoding** is used: the body is
split into chunks that each carry their own length, and the last chunk announces the
end of the stream with a length of zero.

The server below serves the same report in three ways: `/full`, which accumulates the
whole thing before sending it; `/streaming`, which writes each line the moment it is
produced; and `/cut`, which abandons production partway through.

```js
// chunked.mjs — serves the monthly loan report in two forms
//   /full        all lines are accumulated in memory, sent as a single piece
//   /streaming   each line is written the moment it is produced: chunked transfer
//   /cut         production fails after the third line
import { createServer } from "node:http";

const LINES = 10;
const wait = (ms) => new Promise((c) => setTimeout(c, ms));
const makeLine = (i) =>
  `${String(i).padStart(2, "0")};branch-${1 + (i % 3)};loan=${40 + i * 3}\n`;

createServer(async (request, response) => {
  response.sendDate = false;
  const path = new URL(request.url, "http://local").pathname;

  if (path === "/full") {                          // length known: single piece
    const lines = [];
    for (let i = 1; i <= LINES; i++) { await wait(100); lines.push(makeLine(i)); }
    const body = lines.join("");
    response.writeHead(200, { "content-type": "text/csv; charset=utf-8",
                              "content-length": Buffer.byteLength(body) });
    return response.end(body);
  }

  if (path === "/streaming" || path === "/cut") {  // length unknown: chunked transfer
    response.writeHead(200, { "content-type": "text/csv; charset=utf-8" });
    for (let i = 1; i <= LINES; i++) {
      await wait(100);
      if (response.writableEnded) return;          // client disconnected
      if (path === "/cut" && i === 4) {
        // Headers have already been sent: the status code can no longer be changed.
        response.end("#error;report cut off midway;line=3\n");
        return;
      }
      if (!response.write(makeLine(i))) await new Promise((c) => response.once("drain", c));
    }
    return response.end();
  }
  response.writeHead(404).end();
}).listen(8401, "127.0.0.1", () => console.log("listening: 127.0.0.1:8401"));
```

Chunked transfer does not need to be requested explicitly. When `Content-Length` is not
written and the body is not finished in a single call, the runtime does the framing
itself. The only thing written is the line itself.

The streaming response checking the `writableEnded` state before every write is the
same rule as in server-sent events: continuing production after the client has
disconnected means wasted processor time and an error writing to a closed socket.

#### Measurement

The script below compares the two forms in terms of headers, latency, and chunk arrival
times. The measuring script `timer.mjs` writes each chunk's arrival moment in
hundred-millisecond slices.

```js
// timer.mjs — writes the arrival moment of each chunk of the streaming response as a 100 ms slice
const start = performance.now();
let response;
try {
  response = await fetch("http://127.0.0.1:8401/streaming");
} catch {
  console.log("could not connect to server");
  process.exit(0);
}
for await (const chunk of response.body) {
  const slice = Math.round((performance.now() - start) / 100);
  process.stdout.write(`${String(slice).padStart(2)} slice  ${Buffer.from(chunk)}`);
}
```

Port 8401 is arbitrary and must be free.

```bash
#!/usr/bin/env bash
# Starts chunked.mjs; measures headers, latencies, and chunk arrivals.
node chunked.mjs > /dev/null & server=$!
sleep 1
A=http://127.0.0.1:8401

echo "--- response headers ---"
for path in full streaming; do
  printf '%-10s ' "$path"
  curl -sS -o /dev/null -D - "$A/$path" | grep -i -E 'content-length|transfer-encoding' | tr -d '\r' | paste -sd' ' -
done

echo "--- first byte and total duration (100 ms slices) ---"
for path in full streaming; do
  curl -sS -o /dev/null -w "$path %{time_starttransfer} %{time_total}\n" "$A/$path"
done | awk '{ printf "%-10s first-byte=%2d slice  total=%2d slice\n", $1, int($2*10), int($3*10) }'

echo "--- arrival times of the streaming response's chunks ---"
node timer.mjs

echo "--- raw chunked framing (first three chunks) ---"
curl -sS --raw -N "$A/streaming" 2>/dev/null | head -c 63 | od -c | head -4

echo "--- if production is cut off midway ---"
curl -sS -w 'status=%{http_code}\n' "$A/cut"
kill "$server"
```

{/* mask: ^[ 0-9]{2}(?= slice  (07|08|09|10);) */}
```
--- response headers ---
full       content-length: 200
streaming  Transfer-Encoding: chunked
--- first byte and total duration (100 ms slices) ---
full       first-byte=10 slice  total=10 slice
streaming  first-byte= 1 slice  total=10 slice
--- arrival times of the streaming response's chunks ---
 1 slice  01;branch-2;loan=43
 2 slice  02;branch-3;loan=46
 3 slice  03;branch-1;loan=49
 4 slice  04;branch-2;loan=52
 5 slice  05;branch-3;loan=55
 6 slice  06;branch-1;loan=58
 7 slice  07;branch-2;loan=61
 8 slice  08;branch-3;loan=64
 9 slice  09;branch-1;loan=67
10 slice  10;branch-2;loan=70
--- raw chunked framing (first three chunks) ---
0000000   1   4  \r  \n   0   1   ;   b   r   a   n   c   h   -   2   ;
0000020   l   o   a   n   =   4   3  \n  \r  \n   1   4  \r  \n   0   2
0000040   ;   b   r   a   n   c   h   -   3   ;   l   o   a   n   =   4
0000060   6  \n  \r  \n   1   4  \r  \n   0   3   ;   b   r   a   n
--- if production is cut off midway ---
01;branch-2;loan=43
02;branch-3;loan=46
03;branch-1;loan=49
#error;report cut off midway;line=3
status=200
```

Three results stand out.

**The headers** give the format away: `/full` announces a length, `/streaming` uses
chunked transfer. The two cannot coexist.

**The total duration is the same in both forms**: ten slices. Streaming does not speed up
the work, because the report is still produced in the same amount of time. The only
thing that changes is the **time to first byte**: it drops from ten slices to one. The
user sees the report's first lines without waiting for the whole thing to finish. This is
the form the shared gain across this entire real-time data topic takes here.

**The chunk arrivals** confirm the first result: the lines arrive one slice apart, not
all together at the end.

#### Framing

The raw output shows the format of chunked transfer. Each chunk begins with a line that
writes its length in **hexadecimal**: `14` hex, i.e. twenty bytes — the length of one
report line. Then comes `\r\n`, then the data itself, then `\r\n` again.

This framing solves the question of how the receiver knows where the body ends when the
length is not known up front: the last chunk is written with a length of zero, and the
stream ends there. It is also technically possible to end a body without announcing a
length by closing the connection, but then the receiver cannot tell whether the response
finished or was cut off.

#### After Headers Have Been Sent

The last measurement section shows the most important limit of streaming responses. At
the `/cut` endpoint, report production fails at the fourth line, but the status code
stays **200**. Because the headers were already sent with the first chunk, and in HTTP,
headers cannot be taken back.

This has three consequences. The error must now be reported **inside** the body; the
output format must have a place for this — here, a notification line starting with `#`.
Second, the receiving side must read this line; if it does not, it will mistake the
incomplete report for a complete one. Third, writing a **completion marker** that shows
the report finished is safer than writing an error marker: if the connection drops, the
error line cannot be written either, but the absence of the completion marker is noticed
in every case.

HTTP also defines trailers for this; they allow a header to be sent after the body. In
practice, intermediary support is inconsistent, so an in-body marker is the more common
and safer choice.

#### Backpressure in a Streaming Response

Checking the return value of `response.write(...)` in the code is this lesson's
counterpart to the rule measured in the backpressure lesson. If the network client is
slow, the lines being written pile up in the server's buffer; if the return value is
ignored, the buffer grows without bound.

When the report is ten lines long, this is invisible. But a server streaming a
thirty-thousand-line report to a slow connection, if it does not listen to the return
value, ends up holding the entire report in memory — losing the very memory gain that
was the point of streaming. The value of a streaming response lies in production speed
being **coupled** to consumption speed.

#### When to Stream

Streaming is not right for every response. The gain materializes when production takes a
long time and the first piece of the result is meaningful on its own: report lines,
search results, log entries. For responses that are small and meaningful only as a
single piece — the detail of one loan record — streaming only adds framing overhead.

There are two practical obstacles. Intermediaries can buffer the response and turn the
stream back into a single piece; in that case the measured time to first byte shows no
gain. And compression can delay streaming, because it produces no output until its
buffer fills; for streaming responses, compression must be configured to flush at chunk
boundaries.

#### Summary

- A response body of unknown length is sent with chunked transfer encoding; each chunk writes its length in hexadecimal, and a zero-length chunk ends the stream.
- Streaming does not shorten the total duration; in the measurement, both forms took ten slices, but the time to first byte dropped from ten slices to one.
- Because headers are sent with the first chunk, the status code cannot be changed afterward; the error is reported inside the body, and writing a completion marker is safer than writing an error marker.
- If the write's return value is not listened to in a streaming response, the entire body piles up in the server's buffer, and streaming's memory gain disappears.
- Streaming's gain depends on a long production time and the first piece being meaningful on its own; intermediary buffering and compression can make that gain invisible.

#### Course Wrap-Up

This course built four separate layers for shortening the path a request is answered
on, and for making the work moved outside that path reliable.

**Caching** covered reusing a computed result: the client, edge, server, and database
layers; the cache-aside, write-through, and write-behind strategies; time-based,
event-based, and version-based invalidation; key design with namespace and tenant
separation; preventing hot keys and cache stampedes; HTTP validators and cache-control
directives; and measuring hit ratio. Its shared lesson: a cache is not an accelerator,
it is the controlled acceptance of staleness.

**Messaging** built taking work out of the request path: the case for asynchronous
processing, the producer–consumer queue, publish–subscribe, the difference between a
stream and a queue, at-most-once, at-least-once, and exactly-once delivery semantics,
the cost of ordering and partitioning, the dead-letter queue, poison messages, and
backoff. Its shared lesson: delivery guarantees are not free; every guarantee trades away
something from ordering, volume, or latency.

**Background Jobs** treated the side that processes the message as a unit of scaling:
worker processes and supervision, recurring task definitions with a lease lock
preventing overlap, progress reporting and cooperative cancellation for long-running
jobs, backpressure where production speed exceeds consumption speed, and load shedding
to enable controlled rejection under overload. Its shared lesson: a rate mismatch cannot
be eliminated, only where it gets written can be chosen — to memory, to loss, or to the
producer's speed.

**Real-Time Data** built the result reaching the client: the request cost of short and
long polling, server-sent events and reconnection by event ID, the lifecycle of a
WebSocket connection built with its own handshake, channel distribution in a
multi-instance deployment, and responses produced piece by piece. Its shared lesson: an
open connection is a state; the price of the latency gain it carries is managing that
state.

There was an assumption held silently throughout the course. The cache, the queue, the
worker, and the stream were always treated as parts of a **single application**; all of
them inside the same deployment unit, in the same codebase, under the same team's
responsibility. This assumption was necessary for learning the concepts, but it does not
last long in real systems.

The next course, **Service Architectures**, deals with splitting exactly this unit:
making a criteria-based choice among monolithic, modular, and distributed
architectures, drawing service boundaries according to the domain model, and applying
inter-service communication and consistency patterns. None of the concepts measured in
this course become invalid there; all of them cross to the other side of a boundary and
become more expensive there. A delivery guarantee is no longer a queue's contract but
two teams'. An idempotency key must also handle duplicates arriving over the network.
Backpressure spreads not within a process but between services. Channel distribution
happens not between instances but between services with different ownership. Every
number measured in this course becomes a tool for reading the larger-scale counterpart
of the same questions there.