Phần 3 · Chương 3.16
Bạn sẽ tạo ra: FR-RTM-09 đóng lại ở cả hai nửa: một cái trần năm kết nối mà không instance gateway nào tự tính được, giữ dưới dạng năm khoá slot giành bằng `SET NX PX` và gia hạn bằng `SET IFEQ PX`; một mã đóng thứ sáu, vì cả năm mã cũ đều đẩy client tới phương thuốc sai còn đây là lời từ chối duy nhất trong bộ mà cách xử lý đúng không phải là thử lại; một lời từ chối hoàn tất cái bắt tay chỉ để đóng nó, vì trình duyệt không đọc được thân phản hồi của một lần upgrade thất bại; và một cái trần hỏng theo hướng mở một cách ồn ào, nơi dòng log là thứ duy nhất phân biệt `unenforced` với `dưới mức trần` · khoảng 70 phút, bao gồm bài tập
Tài liệu gốc: SAD — Tài liệu kiến trúc phần mềm (tiếng Anh)
Bản dịch đang được chuẩn bị. Phần diễn giải của chương này chưa được dịch sang tiếng Việt. Các khối mã bên dưới là bản gốc tiếng Anh và giống hệt bản tiếng Anh của chương — bạn có thể gõ theo chúng ngay bây giờ. Bản dịch đầy đủ sẽ thay thế trang này.
import { Redis } from "ioredis";
import type { Logger } from "@relay/service-kit";
const DEFAULT_REDIS_URL = "redis://localhost:6379";
/** FR-RTM-09's five, and **FR-002 says it is stated exactly once**. The
* requirement is about drift, not about the value: a second literal five is how
* two figures come apart, which is what `services/api/src/limits/policy.ts` did
* when it derived `connect: 3_000` from "ten thousand divided by five" and shipped
* a third number. `connections.test.ts` asserts this constant is the only one. */
export const MAX_CONNECTIONS_PER_USER = 5;
/** How long after a connection's last successful renewal its place stops counting.
* `docs/05-sad.md:574` already published 60 s for this key and the figure is kept.
*
* IT IS ALSO HOW LONG A CRASHED TAB HOLDS A SLOT, which the spec's Q1 names as the
* accepted cost of refusing the newest rather than evicting the oldest. */
export const DEFAULT_BOUND_MS = 60_000;
/** Three renewals per bound, so two consecutive misses do not free a live
* connection's place.
*
* **NOT `PING_INTERVAL_MS`, which is also a number in this file's neighbourhood.**
* `session.ts:48` sets the protocol keepalive to 30_000, and the presence chapter paid for
* conflating three 30-second numbers that turned out to be three quantities — a TTL
* equal to its own refresh interval expires a connected user. Tying a Redis TTL to
* a client-visible keepalive means changing the ping starts expiring slots. Separate
* timer, separate number, and the tests assert the RATIO rather than the values. */
export const DEFAULT_HEARTBEAT_MS = 20_000;
/** Written into a slot key to retire it. Any value that cannot be a connection id
* does, and `randomUUID` never produces this one.
*
* **IT MEANS FREE, NOT BUSY**, and reading it the other way was a defect. See
* `walk` below: a claim takes a tombstoned slot rather than stepping over it. */
const TOMBSTONE = "-";
/** How long a tombstone lingers. One millisecond is enough to be a conditional
* delete and short enough to be invisible — but the number is named and injectable
* because **a test that has to fit inside one millisecond is a test that races**,
* and the arithmetic below only worked by accident when it was a literal. */
export const DEFAULT_TOMBSTONE_MS = 1;
/** What a claim attempt resolved to. `held` is for FR-015's log line and is
* discovered by the walk rather than read: five keys have no cheap `ZCARD`, and
* `research.md` R3 records that as the design's stated cost. */
export type ClaimOutcome =
| { readonly kind: "claimed"; readonly slot: number; readonly held: number }
| { readonly kind: "full"; readonly held: number }
| { readonly kind: "unenforced" };
/** What a renewal resolved to. **`reclaimed` carries a slot that may differ from
* the one handed in** — FR-011b's common case is that the outage cost nothing and
* some other slot was free, so the caller must store the new number. */
export type RenewOutcome =
| { readonly kind: "renewed" }
| { readonly kind: "reclaimed"; readonly slot: number }
| { readonly kind: "full"; readonly held: number }
| { readonly kind: "unenforced" };
export interface Connections {
claim(
environmentId: string,
user: string,
connectionId: string,
): Promise<ClaimOutcome>;
renew(
environmentId: string,
user: string,
connectionId: string,
slot: number,
): Promise<RenewOutcome>;
release(
environmentId: string,
user: string,
connectionId: string,
slot: number,
): Promise<void>;
/** Every place this instance holds, freed at once. FR-011a: a deployment must
* cost no more than one reconnection cycle (NFR-REL-03), and `wss.close()` does
* not close established sockets — so without this a deploy holds five slots for
* a full bound and the next connection is refused. */
releaseAll(
held: readonly {
readonly environmentId: string;
readonly user: string;
readonly connectionId: string;
readonly slot: number;
}[],
): Promise<void>;
close(): Promise<void>;
}
export interface ConnectionsOptions {
url?: string;
logger: Logger;
boundMs?: number;
/** Widened by one test, to hold the window open long enough to assert what
* happens inside it. */
tombstoneMs?: number;
}
/** `conn:{env}:{user}:{slot}`, one key per place, and the shape is an argument
* against a published row.
*
* `docs/05-sad.md:574` prescribes a sorted set pruned with `ZREMRANGEBYSCORE` on
* read. That needs Lua for FR-013's atomic check-and-insert, Constitution VII
* requires "a superseding ADR with profiling evidence" for a second language, and
* this lane cannot produce it — its largest fixture holds five channels while
* NFR-SCL-01 asks about ten thousand connections. **Making each member its own key
* means the TTL is per member by construction rather than worked around**, which
* was the defect that row recorded in the first place. ADR-23 carries the drivers
* and the reversal condition. */
const key = (environmentId: string, user: string, slot: number): string =>
`conn:${environmentId}:${user}:${slot}`;
export function createConnections({
url = process.env["RELAY_REDIS_URL"] ?? DEFAULT_REDIS_URL,
logger,
boundMs = DEFAULT_BOUND_MS,
tombstoneMs = DEFAULT_TOMBSTONE_MS,
}: ConnectionsOptions): Connections {
// THE SAME THREE OPTIONS `limits.ts:95` AND `presence.ts:133` USE, and the cap
// needs them more than either. With ioredis' defaults an unreachable Redis does
// not fail — it queues the command and retries, so `claim` HANGS. Measured at
// 20 s in `connections.test.ts` before these were added, against NFR-PRF-04's
// p95 < 1 s from handshake to `connection.ack`. **A cap that fails open must
// fail open QUICKLY**, or FR-016's "accept the connection" is indistinguishable
// from refusing it.
const client = new Redis(url, {
lazyConnect: true,
maxRetriesPerRequest: 0,
connectTimeout: 1_000,
});
// One `error` listener, because a client without one turns a connection error
// into an unhandled rejection that takes the process down. the fan-out chapter's R10
// found `createFanout` without one while both rate limiters had one and
// explained why.
client.on("error", (error: Error) => {
logger.log("error", "connections.redis", { detail: error.message });
});
/** `SET … IFEQ … PX` THROUGH `call`, AND THE REASON IS THE TYPED CLIENT RATHER
* THAN THE SERVER. Redis 8.10.0 accepts the combination — verified by hand
* against the lane before this module existed. ioredis 6.0.0 declares only
* `set(key, value, 'IFEQ', cmp)` and `set(key, value, 'IFEQ', cmp, 'GET')`;
* **there is no overload pairing `IFEQ` with `PX`**, so the typed API cannot
* express it and `call` is the documented escape hatch.
*
* The premise check in Phase 1 said `IFEQ` was "present in
* `RedisCommander.d.ts`, so no cast" — and the TOKEN's presence is not an
* overload. Same class as everything else this chapter has corrected: a grep
* for a shape rather than for the set.
*
* The alternatives are worse. `SET … IFEQ` then `PEXPIRE` is two commands, and
* a failure between them leaves a slot with no TTL — leaked for ever, which is
* the original defect rather than a variation on it. `KEEPTTL` preserves the
* ORIGINAL expiry, so a renewal would never extend anything and every slot
* would die 60 s after its claim. */
const setIfEq = async (
k: string,
value: string,
comparison: string,
px: number,
): Promise<string | null> =>
(await client.call(
"SET",
k,
value,
"IFEQ",
comparison,
"PX",
String(px),
)) as string | null;
/** COULD NOT ASK is a distinct OUTCOME, not a distinct value — and the first
* version of this function got that wrong in a way its own comment denied.
*
* It returned `T | null` and read `null` as failure. But `SET … NX` returns
* **nil on a miss**, which is the ordinary case when a slot is taken, so every
* claim past the first was reported as `unenforced`: six of fourteen unit tests
* red, all with `expected { kind: 'unenforced' }`. **Arm 1 and arm 5 of the arms
* list produce the same wire value**, and listing them separately did not stop
* me writing code that could not tell them apart.
*
* A wrapper makes the two unmistakable. `presence.ts:232` gets away with `T |
* null` because its `SET` uses `NX`/`XX` where nil and failure lead to the same
* branch; this module has to distinguish them. */
type Asked<T> = { readonly asked: true; readonly reply: T } | { readonly asked: false };
async function failable<T>(op: string, work: () => Promise<T>): Promise<Asked<T>> {
try {
return { asked: true, reply: await work() };
} catch (error) {
// `String(error)` RATHER THAN A TERNARY ON `instanceof Error`, which is what
// `presence.ts:241` does and what the coverage ratchet asked for: the
// non-Error arm is unreachable from any test and a branch nothing can take is
// a branch to delete. `String` on an Error yields "Error: …", one word longer
// than `.message` and never absent.
logger.log("error", "connections.failed", { op, detail: String(error) });
return { asked: false };
}
}
/** Walk the slots, claiming the first free one.
*
* **`SET NX` IS THE ATOMICITY AND THAT IS THE WHOLE ARGUMENT** (FR-013). Two
* connections racing for one slot cannot both win it: the loser's `NX` returns
* nil and it walks to the next. When all five are held both correctly refuse.
* The race is settled by the command rather than by a check-then-act, which is
* why no Lua is needed here.
*
* **TWO COMMANDS PER CONTESTED SLOT, AND THE SECOND ONE IS A BUG FIX.** A
* tombstone means the previous holder let the place go, so it is free — and the
* first version of this walk stepped over it, because `NX` refuses any key that
* exists. One slot briefly skipped is harmless and the release's comment said so.
* `releaseAll` tombstones ALL FIVE AT ONCE, and then the walk found nothing free
* and refused a connection with `connection_limit_reached` — a close code whose
* documented remedy is "close one of the connections you already hold", which a
* client reconnecting after a deploy cannot do: they went with the old instance.
*
* Found as a 2-in-6 flake in the clean-shutdown test, whose first message —
* `no connection.ack within 5s` — was true of a socket that had been refused 4004
* half a second earlier. Widening the tombstone to 500 ms turns it from a flake
* into a test.
*
* `IFEQ TOMBSTONE` keeps the race settled inside the command: two connections
* both finding the tombstone both attempt it, the first wins, the second's
* comparison now fails against the winner's id and it walks on. No connection id
* can equal `-`, so this can never take a live place. */
async function walk(
environmentId: string,
user: string,
connectionId: string,
): Promise<ClaimOutcome> {
for (let slot = 0; slot < MAX_CONNECTIONS_PER_USER; slot += 1) {
const k = key(environmentId, user, slot);
// BOTH COMMANDS UNDER ONE `failable`, and that is the coverage ratchet's
// doing. Two wrappers meant two "could not ask" arms, and the second was
// unreachable: for it to fire, Redis would have to die between a command that
// answered and the next one. One wrapper has one failure path, and the arm
// that nothing could take is gone rather than covered.
const got = await failable("claim", async () => {
const free = await client.set(k, connectionId, "PX", boundMs, "NX");
// `null` HERE MEANS THE SLOT IS TAKEN, which is arm 1 and not arm 5.
if (free === "OK") return free;
return await setIfEq(k, connectionId, TOMBSTONE, boundMs);
});
if (!got.asked) return { kind: "unenforced" };
if (got.reply === "OK") return { kind: "claimed", slot, held: slot };
}
return { kind: "full", held: MAX_CONNECTIONS_PER_USER };
}
return {
claim: walk,
/** **`IFEQ`, NEVER `XX`, and this was a corrected decision.** `XX` tests that
* the key exists and nothing more — measured on Redis 8.10.0, `SET k B XX`
* against a key holding `A` returns OK and the value becomes B. So a connection
* whose slot expired during an outage would come back, find the slot re-claimed,
* and silently take it: six connections against a count of five, FR-001 and
* FR-011 both violated. `IFEQ` compares before writing and is refused both when
* the key holds another id and when the key is gone. */
renew: async (environmentId, user, connectionId, slot) => {
const asked = await failable("renew", () =>
setIfEq(key(environmentId, user, slot), connectionId, connectionId, boundMs),
);
if (!asked.asked) return { kind: "unenforced" };
if (asked.reply === "OK") return { kind: "renewed" };
// Refused: the key is gone, or somebody else holds it. Either way this
// connection has lost its place and FR-011b says what happens next — one
// more attempt to claim, because after a brief outage nothing else took the
// slot and closing the connection would cost a user their place for Redis's
// downtime.
const again = await walk(environmentId, user, connectionId);
if (again.kind === "claimed") return { kind: "reclaimed", slot: again.slot };
// RETURNED WHOLE, because `full` and `unenforced` mean here exactly what they
// mean there. Re-wrapping them cost two branches and one of them was
// unreachable — the walk can only answer `unenforced` if Redis stopped
// answering between this renewal and it. `full` now carries `held` for the
// same reason a claim's does: the caller logs the number.
return again;
},
/** A one-millisecond tombstone, written only if the slot is still ours.
*
* **NOT `DEL`, which has no ownership check** — the same hole `IFEQ` closed on
* the renewal, on the path that fix introduced. A connection whose slot expired
* and was re-claimed would `DEL` the new owner's key and free a place that is
* in use. `SET key - IFEQ id PX 1` refuses that.
*
* THIS COMMENT USED TO SAY the millisecond fails in the safe direction — a
* claim arriving inside it finds the key present, its `NX` fails, and it walks
* to the next slot; one slot briefly skipped, never an over-admit. True of one
* slot and false of five, which is what `releaseAll` writes. The walk above now
* claims a tombstoned slot instead of stepping over it, so the window is not a
* window any more. `GETDEL` exists and is unconditional, so it is no use
* here. */
release: async (environmentId, user, connectionId, slot) => {
await failable("release", () =>
setIfEq(key(environmentId, user, slot), TOMBSTONE, connectionId, tombstoneMs),
);
},
releaseAll: async (held) => {
for (const one of held) {
await failable("releaseAll", () =>
setIfEq(
key(one.environmentId, one.user, one.slot),
TOMBSTONE,
one.connectionId,
tombstoneMs,
),
);
}
},
close: async () => {
await client.quit();
},
};
}@@ -16,12 +16,17 @@ import {
} from "@relay/protocol";
import type { Logger } from "@relay/service-kit";
import { WebSocketServer, type WebSocket } from "ws";
import { ApiError, type ApiClient } from "./api-client.js";
import { authenticate, type Identity } from "./auth.js";
+import {
+ DEFAULT_HEARTBEAT_MS,
+ MAX_CONNECTIONS_PER_USER,
+ type Connections,
+} from "./connections.js";
import type { Fanout } from "./fanout.js";
import { type Membership } from "./membership.js";
import { type Presence } from "./presence.js";
import { Registry, type Connection } from "./registry.js";
import { type Typing } from "./typing.js";
import {
@@ -161,12 +166,32 @@ export interface SessionServerOptions {
/** Injectable for the reason the membership chapter's `rereadIntervalMs` is:
* **a test that waits out two real seconds pays them in the package that paces
* the lane**, which has about four seconds of headroom in the whole budget.
* That chapter's itest builds with 40 to test a
* sixty-second backstop; this one builds with 40 and with 0. */
renewalIntervalMs?: number;
+ /** FR-RTM-09's five-connection cap.
+ *
+ * **OPTIONAL, LIKE THE OTHER FIVE, AND THAT IS A DECISION RATHER THAN A
+ * DEFAULT.** For a typing indicator "optional" means no typing; for a cap it
+ * means NO CAP. Optional is also what let the typing chapter's module go unpassed
+ * from `main.ts` and still compile, leaving the feature inert while 1,174
+ * coverage tests were green.
+ *
+ * Required would make that impossible — the compiler would catch a fixture
+ * that forgot — and it would break three fixtures, two of them fenced by
+ * earlier chapters, for a cap those fixtures do not test. The bet is that
+ * `main.ts`'s two edits and the sealed outsider test are enough, and it is
+ * named here so the next chapter to add a module can decide otherwise and pay
+ * the fixture edits once. */
+ connections?: Connections;
+ /** Injectable for the reason `meterIntervalMs` and `renewalIntervalMs` are: a
+ * test that waits out a real minute pays it in the package that paces the lane.
+ * Defaults to `DEFAULT_HEARTBEAT_MS`, and the tests assert the RATIO to the
+ * bound rather than either value. */
+ heartbeatMs?: number;
}
// THE FOUR PRESENCE TIMINGS ARE NOT HERE, and an earlier draft of this chapter put
// them here. `fanout` and `presence` are INJECTED already built, and an injected thing
// carries its own configuration: a test that wants a hundred-millisecond grace period
// constructs `createPresence({ graceMs: 100, … })` and injects that, the way the
@@ -185,14 +210,40 @@ export function attachSessions({
pingIntervalMs = PING_INTERVAL_MS,
resumeDeadlineMs = SUBSCRIBE_DEADLINE_MS,
presence,
membership,
typing,
renewalIntervalMs = DEFAULT_RENEWAL_INTERVAL_MS,
-}: SessionServerOptions): { registry: Registry; close: () => void } {
+ connections,
+ heartbeatMs = DEFAULT_HEARTBEAT_MS,
+}: SessionServerOptions): {
+ registry: Registry;
+ /** ASYNC AS OF THIS CHAPTER, and `releaseAll` below is the reason. Freeing the
+ * places this instance holds is a round trip to Redis that has to COMPLETE before
+ * `wss.close()`, or the deploy case the method exists for is a race it can lose. */
+ close: () => Promise<void>;
+} {
const registry = new Registry();
+ /** FR-011a. What this instance holds, so a shutdown can free it
+ * all at once.
+ *
+ * A MAP IN THE CLOSURE RATHER THAN A FIELD ON `Connection`. The typing chapter put
+ * one on the connection and corrected it to this; the shared type in
+ * `registry.ts` belongs to every chapter and this is one chapter's concern.
+ *
+ * WITHOUT THIS A DEPLOY HOLDS FIVE SLOTS FOR A FULL BOUND. `sessions.close()`
+ * calls `wss.close()`, which stops the server accepting connections and does
+ * NOT close established sockets — so whether each connection's own close
+ * handler runs is a race with process exit. NFR-REL-03 allows a deployment no
+ * more than one reconnection cycle, and a bound's worth of refusals after every
+ * deploy is more than one. Found by building `traceability.md` during planning:
+ * the release was implemented and untested. */
+ const heldPlaces = new Map<
+ string,
+ { environmentId: string; user: string; connectionId: string; slot: number }
+ >();
/** A frame arriving from the fabric — born on this instance or another,
* indistinguishable by design — becomes message.created for every local
* member of its channel. */
function deliver(channelId: string, message: Message): void {
for (const connection of registry.subscribersOf(channelId)) {
@@ -551,13 +602,81 @@ export function attachSessions({
const token = url.searchParams.get("token");
void (async () => {
// The api verifies, and answers with the identity AND the
// memberships. This is the same one call the connect path already made —
// it just asks a better question than "what may this user hear".
const result = await authenticate(api, token);
+ // T037. THE CAP IS CHECKED HERE, after `authenticate` and
+ // after the establishment limiter, and both orderings are reasons rather
+ // than habits. After authenticate because the environment and the user are
+ // not known before it — the comment above says the same of the rate limit.
+ // After the limiter because a client hammering the door should meet the
+ // cheaper check first; the limiter is one INCR and this is a walk of up to
+ // five.
+ //
+ // CLAIMED BEFORE THE HANDSHAKE AND REFUSED INSIDE IT (T039). Only a
+ // connection that is going to be accepted claims a place, so a bad token
+ // — already known here — leaks nothing. But the REFUSAL completes the
+ // handshake in order to close it, which is the shape `:715` below
+ // established for the quota: a browser cannot read the body of a FAILED
+ // upgrade, so refusing at the seam gives it a bare connection failure with
+ // no code and no message.
+ let claimed: number | undefined;
+ let capFull = false;
+ let pendingId: string | undefined;
+ if (result.outcome === "ok" && connections !== undefined) {
+ const outcome = await connections.claim(
+ result.identity.environmentId,
+ result.identity.userExternalId,
+ // The id the connection will carry. Minted here rather than below so
+ // the claim and the connection agree on it, which FR-011 requires: a
+ // connection occupies exactly one place for its lifetime.
+ (pendingId = randomUUID()),
+ );
+ if (outcome.kind === "claimed") claimed = outcome.slot;
+ else if (outcome.kind === "full") capFull = true;
+ else {
+ // FR-016 and FR-016a. The connection is accepted with the cap
+ // UNENFORCED, and this line is the only externally visible evidence of
+ // that — from outside, an accepted connection looks identical whether
+ // the cap was checked and satisfied or not checked at all. The fan-out
+ // chapter's lesson: the assertion that carries the requirement is the
+ // log line.
+ logger.log("error", "connection.cap_unenforced", {
+ connection_id: pendingId,
+ environment_id: result.identity.environmentId,
+ user: result.identity.userExternalId,
+ });
+ }
+ }
wss.handleUpgrade(req, socket, head, (ws) => {
+ if (capFull) {
+ // T038. An error frame first, because a close reason is a short string
+ // and what a client needs is the limit and the count. Then close 4004,
+ // whose correct handling is the only one in the set that is not a retry
+ // of some kind: close one of the connections you already hold.
+ sendError(
+ ws,
+ "connection_limit_reached",
+ `already holding ${String(MAX_CONNECTIONS_PER_USER)} of ${String(
+ MAX_CONNECTIONS_PER_USER,
+ )} connections; close one and reconnect`,
+ );
+ ws.close(4004, CLOSE_CODES[4004]);
+ // FR-015. The user, the environment and the observed count — and NOT the
+ // credential presented, which is how a live secret reaches a support
+ // ticket (NFR-SEC-06).
+ logger.log("info", "connection.rejected", {
+ reason: "connection_limit_reached",
+ environment_id:
+ result.outcome === "ok" ? result.identity.environmentId : undefined,
+ user: result.outcome === "ok" ? result.identity.userExternalId : undefined,
+ held: MAX_CONNECTIONS_PER_USER,
+ });
+ return;
+ }
if (result.outcome === "refused") {
// 4001: "invalid or expired token" (EIR-WS-05). The close code is
// the protocol package's, not a number invented here.
ws.close(4001, CLOSE_CODES[4001]);
logger.log("info", "connection.rejected", { reason: "bad_token" });
return;
@@ -589,28 +708,41 @@ export function attachSessions({
logger.log("info", "connection.rejected", { reason: "user_banned" });
return;
}
// NO SEND LIMIT ARGUMENT YET. `authenticate` returns the limits with the
// session in movement VII, where the limiter is written; `open` takes four
// parameters until then rather than a fifth nothing can supply.
- void open(ws, result.identity, result.channelIds, req.url ?? "/");
+ void open(
+ ws,
+ result.identity,
+ result.channelIds,
+ req.url ?? "/",
+ pendingId,
+ claimed,
+ );
});
})();
});
async function open(
socket: WebSocket,
identity: Identity,
channelIds: string[],
url: string,
+ /** The id the cap claimed a place with, so the connection and
+ * its slot agree — FR-011's "exactly one place for its lifetime". Absent when
+ * no `connections` module is wired, which is every fixture that does not opt
+ * in and the reason the cap is not enforced there. */
+ claimedId?: string,
+ claimedSlot?: number,
): Promise<void> {
// Cursors are read BEFORE anything else, because their presence decides
// whether this connection is born buffering or born live.
const presented = parseCursors(url);
const connection: Connection = {
- id: randomUUID(),
+ id: claimedId ?? randomUUID(),
identity,
socket,
// Memberships arrived with the identity, from the session
// call at the door. There is no second lookup to fail here — the api is
// still the only source of membership (ADR-05), it just answers both
// questions at once, and a failure now closes the socket before it opens.
@@ -671,29 +803,140 @@ export function attachSessions({
// and gets it three lines apart — see the note there.
void presence?.connected(
identity.environmentId,
identity.userExternalId,
connection.channelIds,
);
+ // FR-016a, AND IT IS A POSITIVE STATEMENT RATHER THAN AN ABSENCE. An accepted
+ // connection looks identical from outside whether the cap was checked and
+ // satisfied or could not be checked at all; the requirement is that the two be
+ // told apart from the logs. Reading `connection.cap_unenforced` and inferring
+ // the other case from its absence needs both lines and a connection id to join
+ // them on — so the accept line carries the fact itself.
+ //
+ // ABSENT, NOT `true`, WHEN NO MODULE IS WIRED. Every gateway module is an
+ // optional parameter and most fixtures pass none; saying `cap_enforced: true`
+ // there would claim a check that never happened.
logger.log("info", "connection.opened", {
connection_id: connection.id,
user: identity.userExternalId,
channels: connection.channelIds.size,
resuming: presented !== undefined,
+ ...(connections === undefined ? {} : { cap_enforced: claimedSlot !== undefined }),
});
// Listeners go on BEFORE the resume, not after the ack. A resume takes
// a round trip to the api, and a socket that dies inside that window
// must still be removed from the registry and release its subscriptions
// — otherwise a client that reconnects impatiently leaks an instance's
// worth of state per attempt.
socket.on("pong", () => {
connection.missedPings = 0;
});
socket.on("message", (raw) => void handle(connection, raw.toString()));
+ // T040 and T040a. ONE RENEWAL TIMER PER CONNECTION, and it is
+ // cleared on close below — an interval that outlives its connection refreshes
+ // a place nobody holds, which the `IFEQ` guard cannot fix on its own because
+ // the value would still be this connection's id.
+ //
+ // FR-011b's THREE BRANCHES, and the first is the common one. A brief Redis
+ // outage expires the slot while nothing else takes it, so the re-claim
+ // succeeds on a possibly different slot number and the connection carries on:
+ // closing here would cost a user their place for the registry's downtime. All
+ // five held by other connections means the cap is genuinely exceeded and this
+ // connection is the one that must go — the alternative is six open against a
+ // count of five. An unreachable registry is FR-016 again: keep it, log it.
+ //
+ // FR-005 IS NOT IN TENSION WITH THIS, and a reader who finds them adjacent
+ // needs the sentence. FR-005 governs a REFUSAL — opening a sixth must not cost
+ // the five. Here the connection has already lost its place to a competitor,
+ // and the choice is between closing it and running over the cap.
+ let slot = claimedSlot;
+ if (slot !== undefined) {
+ heldPlaces.set(connection.id, {
+ environmentId: identity.environmentId,
+ user: identity.userExternalId,
+ connectionId: connection.id,
+ slot,
+ });
+ }
+ const renewal =
+ connections === undefined || slot === undefined
+ ? undefined
+ : setInterval(() => {
+ void (async () => {
+ if (slot === undefined) return;
+ const outcome = await connections.renew(
+ identity.environmentId,
+ identity.userExternalId,
+ connection.id,
+ slot,
+ );
+ if (outcome.kind === "renewed") return;
+ if (outcome.kind === "reclaimed") {
+ slot = outcome.slot;
+ heldPlaces.set(connection.id, {
+ environmentId: identity.environmentId,
+ user: identity.userExternalId,
+ connectionId: connection.id,
+ slot: outcome.slot,
+ });
+ logger.log("info", "connection.reclaimed", {
+ connection_id: connection.id,
+ slot: outcome.slot,
+ });
+ return;
+ }
+ if (outcome.kind === "unenforced") {
+ logger.log("error", "connection.cap_unenforced", {
+ connection_id: connection.id,
+ environment_id: identity.environmentId,
+ user: identity.userExternalId,
+ });
+ return;
+ }
+ // `full`: the place is gone and there is no other. Same code and
+ // message a refused sixth connection gets, because that is what it
+ // means.
+ slot = undefined;
+ sendError(
+ connection.socket,
+ "connection_limit_reached",
+ `already holding ${String(MAX_CONNECTIONS_PER_USER)} of ${String(
+ MAX_CONNECTIONS_PER_USER,
+ )} connections; close one and reconnect`,
+ );
+ connection.socket.close(4004, CLOSE_CODES[4004]);
+ logger.log("info", "connection.rejected", {
+ reason: "connection_limit_reached",
+ environment_id: identity.environmentId,
+ user: identity.userExternalId,
+ held: MAX_CONNECTIONS_PER_USER,
+ });
+ })();
+ }, heartbeatMs);
+ if (renewal !== undefined) renewal.unref();
+
socket.on("close", (code) => {
+ // T041. The timer first, then the place. An interval that
+ // outlives its connection renews a slot nobody holds.
+ //
+ // NO 4009 DRAIN PATH IS ADDED HERE, and that is deliberate:
+ // `session.test.ts:965` says "4009 IS STILL EMITTED BY NOTHING" and `:982`
+ // asserts the gateway never sends it. The deploy case is `releaseAll` in
+ // this module's `close`, not a close code.
+ if (renewal !== undefined) clearInterval(renewal);
+ heldPlaces.delete(connection.id);
+ if (connections !== undefined && slot !== undefined) {
+ void connections.release(
+ identity.environmentId,
+ identity.userExternalId,
+ connection.id,
+ slot,
+ );
+ }
registry.remove(connection.id);
// THIS HANDLER NOW CARRIES TWO ORDERING CONSTRAINTS, not none. Presence is told
// AFTER `registry.remove`, because it asks whether this was the user's last
// connection on this instance and must not count the one that is leaving. The
// unsubscribes come last.
//
@@ -1132,12 +1375,21 @@ export function attachSessions({
connection.socket.ping();
}
}, pingIntervalMs);
return {
registry,
- close: () => {
+ close: async () => {
clearInterval(heartbeat);
+ // FR-011a. Before `wss.close()`, because that call does not
+ // close established sockets — so their own close handlers may never run and
+ // this is the last chance to free their places. SC-013: after a deployment a
+ // user whose connections were on the replaced instance reconnects
+ // immediately rather than waiting out the bound.
+ if (connections !== undefined && heldPlaces.size > 0) {
+ await connections.releaseAll([...heldPlaces.values()]);
+ heldPlaces.clear();
+ }
wss.close();
},
};
}@@ -16,12 +16,33 @@ export const CLOSE_CODES = {
// "a client that cannot tell them apart retries the wrong one for ever".
//
// EIR-WS-06 names four classes to distinguish — authentication, quota, shutdown,
// protocol violation — and a ban is none of them. Numbered here, the way chapter 1.3
// numbered 4002 and 4008.
4003: "banned in this environment",
+ // FR-RTM-09. A SIXTH CODE, AND EVERY REUSE FAILS THIS FILE'S OWN TEST.
+ //
+ // The remedy for this refusal is unlike every other one here: close one of the
+ // connections you already hold, and reconnect immediately. No waiting, no new
+ // credential, no client fix. That is why none of the five above can carry it —
+ // "a client that cannot tell them apart retries the wrong one for ever":
+ //
+ // 4001 would send a client to re-authenticate. The token is valid; minting a
+ // new one succeeds and connecting fails again, which is the channel-control chapter's
+ // infinite loop against a wall.
+ // 4002 would blame a client that did nothing wrong.
+ // 4003 would tell a person they are barred while four of their connections
+ // are working.
+ // 4008 would tell a client to wait for a quota window. Waiting never helps,
+ // and the slot it needs may free a second later.
+ //
+ // EIR-WS-06 names four classes to distinguish — authentication, quota, shutdown,
+ // protocol violation — and a concurrency cap is none of them, exactly as a ban
+ // was none of them. Numbered here, in the space chapter 1.3 and the
+ // channel-control chapter drew from.
+ 4004: "connection limit reached",
4008: "quota exhausted",
4009: "server shutdown (drain)",
} as const;
export type CloseCode = keyof typeof CLOSE_CODES;
@@ -124,12 +145,27 @@ export const ERROR_CODES = {
// NOT `quota_exceeded`. That is a monthly, billable, resets-on-a-date refusal whose
// message promises a resume date; this is a structural limit on one channel that no
// amount of waiting changes. Same status, different fact, and a client that retries on
// the wrong one waits for ever.
channel_member_limit_exceeded:
"this channel already holds the maximum number of members; remove one before adding another",
+ // FR-RTM-09, and the socket's half of the connection cap: an error frame carrying
+ // the limit and the count, sent immediately before close 4004.
+ //
+ // THE FIGURES GO IN THE MESSAGE, not in payload fields. `errorFrameSchema` is a
+ // `z.strictObject` of `code`, `message`, `docs_url` and an optional `field`, so two
+ // new fields would mean widening a shape every error frame shares for one code's
+ // benefit — and the reference's Status line is where the close code is recorded.
+ //
+ // NOT `rate_limited`, which is one word away in the register and means the opposite
+ // thing. That code throttles a tenant's establishments per window and its own
+ // message reads "too many connections; retry after the window resets". Retrying is
+ // exactly what this code must not suggest: five connections are already open and
+ // the remedy is to close one, which no amount of waiting does.
+ connection_limit_reached:
+ "the user already holds the maximum concurrent connections; the message names the limit and the count, and the remedy is to close one and reconnect",
} as const;
export type ErrorCode = keyof typeof ERROR_CODES;
/** Whether a string the api sent is a code this registry defines.
*@@ -12,15 +12,15 @@ describe("close codes cover EIR-WS-06's four classes", () => {
// user is refused anyway. Reusing 4001 would tell a client to re-authenticate, which
// succeeds at minting a token and fails again at connect.
//
// THIS ASSERTION IS WHY THE NUMBER IS DELIBERATE. It failed on the build that added
// 4003 — an exact-set assertion is the only kind that makes a new close code a decision
// rather than an accident, and updating it is the act of making that decision.
- it("contains exactly 4001, 4002, 4003, 4008, 4009", () => {
+ it("contains exactly 4001, 4002, 4003, 4004, 4008, 4009", () => {
expect(Object.keys(CLOSE_CODES).map(Number).sort()).toEqual([
- 4001, 4002, 4003, 4008, 4009,
+ 4001, 4002, 4003, 4004, 4008, 4009,
]);
});
it("gives every code a distinct, non-empty meaning", () => {
const meanings = Object.values(CLOSE_CODES);
expect(new Set(meanings).size).toBe(meanings.length);
@@ -131,6 +131,32 @@ describe("the docs URL is built in one place, with the code as the anchor", () =
it("gives every code a distinct URL", () => {
const urls = (Object.keys(ERROR_CODES) as ErrorCode[]).map(docsUrl);
expect(new Set(urls).size).toBe(urls.length);
});
});
+
+describe("the refusal this chapter's cap adds", () => {
+ // NAMED, NOT COUNTED, for the reason the channel block above gives.
+ //
+ // What matters about this one is that it is **not** `rate_limited`. The two sit one
+ // word apart in the register and mean opposite things: `rate_limited` throttles
+ // frames and says "slow down and retry", which is exactly what a client at the
+ // connection cap must not do — five sockets are already open and no amount of
+ // waiting closes one. The remedy is a client action, not a delay.
+ it("registers connection_limit_reached with a description a client can act on", () => {
+ expect(ERROR_CODES).toHaveProperty("connection_limit_reached");
+ expect(ERROR_CODES.connection_limit_reached).not.toBe("");
+ });
+
+ it("keeps it distinct from rate_limited, which is what it exists instead of", () => {
+ expect(ERROR_CODES.connection_limit_reached).not.toBe(ERROR_CODES.rate_limited);
+ });
+
+ it("names a remedy the client can perform rather than a delay to wait out", () => {
+ // THE ONE ASSERTION ABOUT THE WORDS, and it is the whole reason for a separate
+ // code. A message telling a capped client to retry sends it into a loop against a
+ // wall, which is the failure `codes.ts` has now argued against five times.
+ expect(ERROR_CODES.connection_limit_reached).toContain("close one");
+ expect(ERROR_CODES.rate_limited).toContain("retry");
+ });
+});{"msg":"connection.cap_unenforced","connection_id":"…","user":"tuan"}
{"msg":"connection.opened","connection_id":"…","cap_enforced":false}
{"msg":"connection.opened","connection_id":"…","cap_enforced":true}RELAY_REDIS_URL=redis://127.0.0.1:6399 vitest run src/connections.test.ts
12 failed | 5 passedimport { randomUUID } from "node:crypto";
import { readFileSync } from "node:fs";
import { dirname, join } from "node:path";
import { fileURLToPath } from "node:url";
import { describe, expect, it } from "vitest";
import {
createConnections,
DEFAULT_BOUND_MS,
DEFAULT_HEARTBEAT_MS,
MAX_CONNECTIONS_PER_USER,
} from "./connections.js";
// The slot registry — THE HALF THAT NEEDS NO BROKER (FR-006, FR-006a, FR-024, FR-024a).
//
// This file held all seventeen of the registry's tests and twelve of them talk to a real
// Redis. It is a `.test.ts`, so it runs in the lane chapter 2.1 built specifically to
// need no containers — the lane whose whole point is that `pnpm test` is honest on a
// laptop with nothing running. With the stack down it reported twelve failures that were
// correct behaviour.
//
// **WHICH FIVE STAY WAS MEASURED, NOT ARGUED.** Run the original against a dead broker
// and it reports `12 failed | 5 passed`:
//
// RELAY_REDIS_URL=redis://127.0.0.1:6399 vitest run src/connections.test.ts
//
// Reading the file predicted two and the measurement found five. The heartbeat test was
// filed under "asserts registry behaviour" on the strength of its title; it asserts a
// ratio between two constants and never reaches the broker. **A title is not an inventory
// of what a test touches** — the same defect as a task id in a test title, one category
// over.
//
// AND THE SHARED `beforeEach` IS GONE. The describe these came from built a registry
// against `REDIS` for every test in it, including the two that provably need none. It
// did not break them — `createConnections` connects lazily, which one command settled
// after a reading of the code said otherwise — but a container-free lane holding a Redis
// client it never uses is a lane that will grow one that matters.
//
// The twelve that need a broker are in `connections.itest.ts`, unchanged in behaviour.
const REDIS = process.env["RELAY_REDIS_URL"] ?? "redis://localhost:6379";
const silent = { log: () => {} };
/** A per-run environment, so nothing here collides with the two integration files
* that share a constant `"env-1"` and both lean on the user name "tuan". */
const ENV = `env-${randomUUID()}`;
describe("the slot registry, without a broker", () => {
/** A registry and a user per test, not a shared hook. The two below need the OBJECT
* and not the broker: `release` on a slot never held and `releaseAll([])` both settle
* before any command is sent. */
const registryFor = () => createConnections({ url: REDIS, logger: silent });
const userFor = () => `u-${randomUUID()}`;
it("does not throw for a slot the connection never held", async () => {
// ARM 7, AND THE TITLE SAYS ONLY WHAT THE ASSERTION PROVES. It used to read
// "is a no-op", which claims more: a no-op is a statement about the key, and
// `resolves.toBeUndefined()` is a statement about the promise. The stronger
// property is not observable through this module's own surface — a claim walks
// from slot 0, so whatever an unconditional release did to slot 3 cannot be
// seen from here — and the ownership half of it is the test below.
// The membership-revocation chapter's rule: a claim about an observable
// difference needs falsifying before the test is written.
await expect(
registryFor().release(ENV, userFor(), randomUUID(), 3),
).resolves.toBeUndefined();
});
it("does not throw when it holds nothing", async () => {
// ARM 8: the empty loop, which is the shutdown path of an instance that never
// had a connection. Renamed for the same reason as the test above — "releases
// nothing" describes the keys and the assertion describes the promise.
await expect(registryFor().releaseAll([])).resolves.toBeUndefined();
});
// ---- ARM 5 and ARM 11: the registry cannot be reached -----------------
it("returns unenforced rather than zero when Redis is unreachable (FR-016)", async () => {
// ARM 5 and ARM 11. A port nothing listens on, so every command rejects.
//
// `null` MEANS COULD NOT ASK, and the distinction is the requirement: FR-016
// accepts the connection and logs that the cap was not enforced, which is a
// different fact from a user being under the limit. Conflating the two is what
// the fan-out chapter found in its publisher — "the send returned 201 while
// Redis was down" is true of a publisher that does nothing at all.
const lines: Record<string, unknown>[] = [];
const gone = createConnections({
url: "redis://127.0.0.1:6399",
logger: {
log: (_level: string, msg: string, fields?: Record<string, unknown>) => {
lines.push({ msg, ...fields });
},
},
boundMs: 60,
});
expect(await gone.claim(ENV, userFor(), randomUUID())).toEqual({
kind: "unenforced",
});
expect(lines.some((l) => l["msg"] === "connections.failed")).toBe(true);
await gone.close().catch(() => {});
}, 20_000);
// ---- FR-009 and FR-002: the numbers, and where they live --------------
it("keeps the heartbeat strictly inside the bound, three to one (FR-009)", async () => {
// THE RATIO, NOT THE VALUES. A test pinning 20_000 and 60_000 goes red on a
// deliberate re-derivation and says nothing about the property. What FR-009
// requires is that two consecutive missed renewals cannot free a live
// connection's place, and three-to-one is what delivers it.
expect(DEFAULT_HEARTBEAT_MS).toBeLessThan(DEFAULT_BOUND_MS);
expect(DEFAULT_BOUND_MS / DEFAULT_HEARTBEAT_MS).toBeGreaterThanOrEqual(3);
// And it is NOT the protocol keepalive, which the presence chapter paid for
// conflating.
expect(DEFAULT_HEARTBEAT_MS).not.toBe(30_000);
});
it("states the maximum in exactly one place (FR-002)", async () => {
// The requirement is about DRIFT, not about the value. `policy.ts` derived
// `connect: 3_000` from "ten thousand divided by five" and shipped a third
// number; a second literal five in this module is how the same thing starts.
//
// Read from disk rather than reasoned about: the module's own source is the
// only thing that can answer "how many fives are in it".
const here = dirname(fileURLToPath(import.meta.url));
const source = readFileSync(join(here, "connections.ts"), "utf8");
const body = source
.split("\n")
.filter((line) => !line.trimStart().startsWith("*"))
.filter((line) => !line.trimStart().startsWith("//"))
.filter((line) => !line.trimStart().startsWith("/*"))
.join("\n");
const fives = body.match(/(?<![\w.])5(?![\w.])/g) ?? [];
expect(fives, `bare 5 outside comments: ${fives.length}`).toHaveLength(1);
expect(MAX_CONNECTIONS_PER_USER).toBe(5);
});
});import { randomUUID } from "node:crypto";
import type { Server } from "node:http";
import type { AddressInfo } from "node:net";
import { docsUrl, subjectForChannelMembership } from "@relay/protocol";
import { serve } from "@relay/service-kit";
import { Redis } from "ioredis";
import { afterAll, afterEach, beforeEach, describe, expect, it } from "vitest";
import { WebSocket } from "ws";
import type { ApiClient } from "./api-client.js";
import { createFanout } from "./fanout.js";
import { createConnections, MAX_CONNECTIONS_PER_USER, type Connections } from "./connections.js";
import { createMembership } from "./membership.js";
import { attachSessions } from "./session.js";
// FR-RTM-09's five-connection cap.
//
// PHASE 2 IS US3 AND IT RUNS AGAINST UNCHANGED CODE, deliberately. FR-RTM-09's
// second clause — "each shall receive all events independently" — is a property
// of code that already ships: delivery walks connections, never users. Writing
// these tests BEFORE the cap exists makes them a regression guard for it.
// Written afterwards they would prove nothing, because nobody would know they
// had ever passed.
//
// A PER-RUN ENVIRONMENT, and not the constant the neighbours use. `typing.itest.ts`
// defaults to "env-1" and `resume.itest.ts` hardcodes it; both lean on the user
// name "tuan"; and the gateway's integration config sets no `fileParallelism`, so
// all three files run at once. Once Phase 5 lands, slot keys are namespaced by
// environment, and a constant here would leak this file's tests into each other.
const ENVIRONMENT = `env-${randomUUID()}`;
const REDIS = process.env["RELAY_REDIS_URL"] ?? "redis://localhost:6379";
const silent = { log: () => {} };
/** Every environment this run has booted an instance in — the one above, plus the
* second one FR-012's test needs. Populated by `boot`, and read only by the sweep
* below. */
const environments = new Set<string>();
/** T050c. **A SLOT'S TTL OUTLIVES THE PACKAGE'S RUN**: 60,000 ms of bound against
* a gateway integration lane of about forty-five seconds. So a slot leaked by a
* close handler that never ran survives into the next file and into the next
* battery run — and this is the only file that deliberately fills all five.
*
* Scoped to this run's own environment prefixes, which are UUIDs, so it cannot
* touch another suite's keys. `limits.itest.ts:297` deletes its three rate-limit
* keys by name for the same reason and is the only other Redis cleanup in the
* gateway's integration files; a name list is not available here because the
* production code chooses the user and the slot. */
const raw = new Redis(REDIS);
afterEach(async () => {
for (const environment of environments) {
const keys = await raw.keys(`conn:${environment}:*`);
if (keys.length > 0) await raw.del(...keys);
}
});
afterAll(async () => {
await raw.quit();
});
interface Instance {
url: string;
/** FR-011a's path ON ITS OWN, with the sockets left open — which is the only way
* to see it. `sessions.close()` calls `releaseAll()` and then `wss.close()`, and
* `wss.close()` does not close established sockets; the fixture's own `close()`
* below waits on `server.close()`, which does not return while a socket is open.
* So a test that wants to observe the deploy path calls this and leaves the rest
* to teardown. */
shutdown: () => Promise<void>;
/** FR-007. The instance stops being able to reach the registry, and nothing
* closes its sockets.
*
* **WHAT A `kill -9` LOOKS LIKE FROM REDIS'S SIDE, which is the only side that
* can see it**: no release lands, no renewal lands, and the five keys sit there
* until their bound elapses. In process there is no way to make the server's own
* `close` handlers not run — destroying the socket is what triggers them — so the
* simulation is at the registry rather than at the socket. It does not reproduce
* a half-written command or a torn connection, and it is not claimed to. */
crash: () => Promise<void>;
close: () => Promise<void>;
}
/** One gateway, in process, with a stubbed api — the shape `resume.itest.ts` and
* `typing.itest.ts` use. No api is spawned, so this file claims no port range and
* adds nothing to the seven spawning files in the lane. */
async function boot(options: {
user: string;
channels: string[];
/** FR-012's test needs a second one, and every other test wants this run's own.
* `typing.itest.ts:94` defaults to `"env-1"` and `resume.itest.ts` hardcodes it
* in seven places; both lean on the user name "tuan". Those two claim no slots,
* so the reason here is hygiene within this file rather than a collision with
* them — but this file fills all five places on purpose, and a shared identity
* would leak one test's slots into the next. */
environment?: string;
/** T050a. **ONE MODULE PER INSTANCE, BUILT HERE**, the way
* `typing.itest.ts:101` calls `createTyping(...)` inside `boot()`.
*
* `releaseAll()` is what makes this correctness rather than style: it frees the
* places *this instance* holds, so a fixture sharing one module across two
* gateways would have the crashed instance release the surviving one's slots —
* and T052 would pass for the wrong reason. Two instances, two modules, two
* client pairs, both closed in teardown.
*
* Absent means the cap is not enforced at all, which is what US3's tests want:
* every gateway module is an optional `attachSessions` parameter, so a fixture
* opts in. */
cap?: { boundMs?: number; heartbeatMs?: number; url?: string };
/** FR-015's log line is the assertion that carries the requirement, so a test
* needs the lines. The fan-out chapter's finding: a publisher that does nothing
* satisfies "the send returned 201". */
lines?: Record<string, unknown>[];
}): Promise<Instance> {
const environment = options.environment ?? ENVIRONMENT;
environments.add(environment);
const fanout = createFanout({ url: REDIS, logger: silent });
const membership = createMembership({ url: REDIS, logger: silent });
const logger =
options.lines === undefined
? silent
: {
log: (_level: string, msg: string, fields?: Record<string, unknown>) => {
options.lines?.push({ msg, ...fields });
},
};
const server: Server = serve({
service: "gateway",
health: () => ({}),
logger: silent,
notFoundDocsUrl: docsUrl("not_found"),
});
const api: ApiClient = {
session: async () => ({
environment_id: environment,
user: options.user,
banned: false,
channel_ids: options.channels,
limits: { connect: 3_000, send: 600 },
}),
memberships: async () => options.channels,
backfill: async () => ({}) as never,
sendMessage: async () => {
throw new Error("not used");
},
};
const registry: Connections | undefined =
options.cap === undefined
? undefined
: createConnections({
url: options.cap.url ?? REDIS,
logger,
...(options.cap.boundMs === undefined ? {} : { boundMs: options.cap.boundMs }),
});
const sessions = attachSessions({
server,
api,
logger,
fanout,
membership,
...(registry === undefined ? {} : { connections: registry }),
...(options.cap?.heartbeatMs === undefined
? {}
: { heartbeatMs: options.cap.heartbeatMs }),
});
await new Promise<void>((resolve) => server.listen(0, resolve));
const { port } = server.address() as AddressInfo;
let stopped = false;
const stop = async (): Promise<void> => {
if (stopped) return;
stopped = true;
await sessions.close();
};
return {
url: `ws://127.0.0.1:${port}/v1/ws`,
shutdown: stop,
// Only the client goes. Teardown then runs the ordinary path, where every
// release and every `releaseAll` throws inside the module's own `failable` and
// is logged rather than thrown — so the timers still get cleared and the slots
// still stay held.
crash: async () => {
await registry?.close().catch(() => {});
},
close: async () => {
await stop();
await fanout.close();
await membership.close();
await registry?.close().catch(() => {});
await new Promise<void>((resolve) => server.close(() => resolve()));
},
};
}
interface Recorded {
socket: WebSocket;
frames: { type: string; payload: Record<string, unknown> }[];
/** The close code once it arrives, captured by `record` rather than by whoever
* happens to be waiting. A listener attached later can miss a close that has
* already happened, which is how "no ack within 5s" hid a 4004 for two hours. */
closed?: number;
}
/** Every frame kept, not just the first matching one. **That distinction is the
* whole point of T014**: the fan-out chapter already asserts both of a user's sockets
* receive a message, using a `waitFor` that resolves on the first match — so a
* DUPLICATE passes it unnoticed. Story 3 scenario 1 says "both receive it, and
* each receives it once", and only a count can say the second half. */
function record(socket: WebSocket): Recorded {
const r: Recorded = { socket, frames: [] };
socket.on("message", (raw) => {
r.frames.push(JSON.parse(String(raw)) as Recorded["frames"][number]);
});
socket.on("close", (code: number) => {
r.closed = code;
});
return r;
}
/** **THE FAILURE CARRIES WHAT THE SOCKET ACTUALLY GOT**, and the first version of
* this message did not. `no connection.ack within 5s` was true of a socket that
* had been refused 4004 half a second earlier, and it took a falsification run and
* six repeats to find that out — the typing chapter's rule one level down: a check that
* throws away the evidence costs more than the defect. */
async function untilAcked(r: Recorded): Promise<void> {
const deadline = Date.now() + 5_000;
while (Date.now() < deadline) {
if (r.frames.some((f) => f.type === "connection.ack")) return;
await new Promise((resolve) => setTimeout(resolve, 25));
}
const seen = r.frames.map((f) => f.type).join(", ") || "nothing";
const codes = r.frames
.filter((f) => f.type === "error")
.map((f) => String(f.payload["code"]))
.join(", ");
throw new Error(
`no connection.ack within 5s — frames: ${seen}${
codes === "" ? "" : ` (${codes})`
}; close: ${r.closed === undefined ? "still open" : String(r.closed)}`,
);
}
/** Polls rather than waits on a single frame. A connection is acked before its
* SUBSCRIBE has necessarily landed, which cost the typing chapter a flake at 315 ms —
* the fix there was polling helpers, not a re-run. */
async function untilCount(
r: Recorded,
type: string,
atLeast: number,
): Promise<void> {
const deadline = Date.now() + 5_000;
while (Date.now() < deadline) {
if (r.frames.filter((f) => f.type === type).length >= atLeast) return;
await new Promise((resolve) => setTimeout(resolve, 25));
}
throw new Error(
`only ${r.frames.filter((f) => f.type === type).length} ${type} within 5s`,
);
}
const count = (r: Recorded, type: string): number =>
r.frames.filter((f) => f.type === type).length;
describe("every connection a person holds is a first-class recipient (US3)", () => {
const open: Instance[] = [];
const sockets: WebSocket[] = [];
/** A FRESH USER PER TEST, and the first version of this file had one per FILE.
* Slot keys are `conn:{env}:{user}:{slot}`, so tests sharing a user share five
* places — and the cap tests fill all five. Two of them failed at 5 s with the
* previous test's slots still held, because a release is fire-and-forget from a
* close handler and the registry client had already been quit inside the test
* body. Per-test users make the leak impossible; the instance owning its own
* module, closed after `sessions.close()`, makes the release possible. */
let user: string;
const connect = (instance: Instance): Recorded => {
const socket = new WebSocket(`${instance.url}?token=any`);
sockets.push(socket);
return record(socket);
};
beforeEach(() => {
user = `u-${randomUUID()}`;
});
afterEach(async () => {
// Sockets before servers. `afterEach` runs in reverse registration order and
// a teardown that closed servers first cost the membership-revocation chapter seven tests and
// eighty-three seconds, every failure naming a hook.
for (const socket of sockets.splice(0)) socket.close();
// A beat for each close handler to run its release before the client goes.
await new Promise((resolve) => setTimeout(resolve, 150));
for (const instance of open.splice(0)) await instance.close();
});
it("accepts five and refuses the sixth with 4004 (FR-001, FR-003, SC-002)", async () => {
const channel = randomUUID();
const instance = await boot({ user, channels: [channel], cap: {} });
open.push(instance);
const five: Recorded[] = [];
for (let i = 0; i < MAX_CONNECTIONS_PER_USER; i += 1) {
const r = connect(instance);
await untilAcked(r);
five.push(r);
}
expect(five).toHaveLength(5);
// THE CLOSE CODE AND THE ERROR CODE, NOT THE FACT OF CLOSING. A socket that
// closes for the wrong reason is identical from outside, which is what
// `contracts/refusal.md` is about — every reuse of the five existing codes
// sends a client to the wrong remedy.
const sixth = connect(instance);
const code = await new Promise<number>((resolve) => {
sixth.socket.on("close", (c) => resolve(c));
});
expect(code).toBe(4004);
const error = sixth.frames.find((f) => f.type === "error");
expect(error?.payload["code"]).toBe("connection_limit_reached");
expect(String(error?.payload["message"])).toContain("close one and reconnect");
// FR-005 and SC-012: the five are undisturbed. A message published now
// reaches all of them — the refusal cost them nothing.
const fanout = createFanout({ url: REDIS, logger: silent });
await fanout.publish({
id: randomUUID(),
channel,
seq: 3,
user,
text: `after the refusal ${randomUUID()}`,
created_at: new Date(0).toISOString(),
});
for (const [i, r] of five.entries()) {
await untilCount(r, "message.created", 1);
expect(count(r, "message.created"), `on ${String(i)}`).toBe(1);
}
await fanout.close();
}, 40_000);
it("frees a slot on close, reusable with NO waiting period (FR-010, SC-003)", async () => {
const channel = randomUUID();
const instance = await boot({ user, channels: [channel], cap: {} });
open.push(instance);
const five: Recorded[] = [];
for (let i = 0; i < MAX_CONNECTIONS_PER_USER; i += 1) {
const r = connect(instance);
await untilAcked(r);
five.push(r);
}
// Close one and take its place. **No clock is involved** — that is the
// observable difference between this refusal and a rate limit, whose remedy
// is to wait.
five[0]?.socket.close();
await new Promise((resolve) => setTimeout(resolve, 300));
const replacement = connect(instance);
await untilAcked(replacement);
expect(count(replacement, "connection.ack")).toBe(1);
}, 40_000);
it("logs the refusal with the user, the environment and the count, and no credential (FR-015, SC-008)", async () => {
const channel = randomUUID();
const lines: Record<string, unknown>[] = [];
const instance = await boot({ user, channels: [channel], cap: {}, lines });
open.push(instance);
for (let i = 0; i < MAX_CONNECTIONS_PER_USER; i += 1) {
await untilAcked(connect(instance));
}
const sixth = connect(instance);
await new Promise<number>((resolve) => {
sixth.socket.on("close", (c) => resolve(c));
});
const rejected = lines.find(
(l) => l["msg"] === "connection.rejected" && l["reason"] === "connection_limit_reached",
);
expect(rejected, "connection.rejected was not logged").toBeTruthy();
expect(rejected?.["user"]).toBe(user);
expect(rejected?.["environment_id"]).toBe(ENVIRONMENT);
expect(rejected?.["held"]).toBe(5);
// NFR-SEC-06. "the key rk_dev_abc… is invalid" is how a live secret reaches a
// support ticket, so the whole line is checked rather than one field.
expect(JSON.stringify(rejected)).not.toContain("token");
expect(JSON.stringify(rejected)).not.toContain("rk_");
}, 40_000);
it("delivers a message to both of one user's connections, each exactly once (FR-014, SC-001)", async () => {
const channel = randomUUID();
const instance = await boot({ user, channels: [channel] });
open.push(instance);
const a = connect(instance);
const b = connect(instance);
await untilAcked(a);
await untilAcked(b);
const fanout = createFanout({ url: REDIS, logger: silent });
const text = `to both tabs ${randomUUID()}`;
await fanout.publish({
id: randomUUID(),
channel,
seq: 1,
user,
text,
created_at: new Date(0).toISOString(),
});
await untilCount(a, "message.created", 1);
await untilCount(b, "message.created", 1);
// Settle, so a duplicate has time to arrive and be counted. Asserting a
// count immediately after the first arrival cannot see a second.
await new Promise((resolve) => setTimeout(resolve, 300));
expect(count(a, "message.created"), "on a").toBe(1);
expect(count(b, "message.created"), "on b").toBe(1);
expect(
(a.frames.find((f) => f.type === "message.created")?.payload as { text: string })
.text,
).toBe(text);
await fanout.close();
}, 30_000);
it("delivers a membership change to both of one user's connections (FR-014)", async () => {
const channel = randomUUID();
const instance = await boot({ user, channels: [channel] });
open.push(instance);
const a = connect(instance);
const b = connect(instance);
await untilAcked(a);
await untilAcked(b);
// The membership fabric, not the message one. `deliverPresence` is
// deliberately UNFILTERED and typing's rule is the opposite, and the two sit
// adjacent in `session.ts` — so a test copied from one to the other asserts
// the wrong thing. This asserts arrival on both, which is FR-014's subject.
//
// PUBLISHED RAW, because the `Membership` module has no `publish`: the api
// publishes a change and the gateway only ever subscribes. `membership.itest.ts`
// reached for a raw client for the same reason, and counting a publish through
// the code that publishes is the shape the fan-out chapter warned about anyway.
const publisher = new Redis(REDIS);
await publisher.publish(
subjectForChannelMembership(channel),
JSON.stringify({
environment: ENVIRONMENT,
channel,
user,
change: "added",
}),
);
await untilCount(a, "membership.changed", 1);
await untilCount(b, "membership.changed", 1);
await publisher.quit();
}, 30_000);
it("keeps delivering to a live connection when an EARLIER one is gone (FR-014)", async () => {
const channel = randomUUID();
const instance = await boot({ user, channels: [channel] });
open.push(instance);
const a = connect(instance);
const b = connect(instance);
await untilAcked(a);
await untilAcked(b);
// THE FIRST CONNECTION IS THE ONE TERMINATED, and the order is the test.
// `subscribersOf` returns `[...byId.values()]` — a `Map`, so insertion order —
// so `a` is delivered to first. Terminating `b` instead would leave a
// delivery loop that dies on its first failure looking correct, because it
// would already have reached `a`. **A test that passes whichever way the
// subject behaves proves nothing**, and the first draft of this test
// terminated `b`.
//
// AND FR-014's SECOND HALF IS NOT WHAT THIS TESTS, because it could not be
// falsified. The clause says one connection's delivery failure must not
// prevent another's. `send` is a bare `socket.send(...)` with no try/catch,
// so the falsification is to make it throw on a socket that is not OPEN —
// and that leaves all three tests green, because `registry.remove` runs from
// the terminated socket's own close handler before any publish arrives.
// **There is no failing send to survive**: a dead connection is gone from
// the registry, not present-and-broken.
//
// So the property is real and unobservable through this fixture, which is
// the membership-revocation chapter's lesson in its own words — a claim about an observable
// difference needs falsifying before the test is written. What this test does
// assert is narrower and still worth having: a surviving connection keeps
// receiving after an earlier one is gone.
a.socket.terminate();
await new Promise((resolve) => setTimeout(resolve, 200));
const fanout = createFanout({ url: REDIS, logger: silent });
await fanout.publish({
id: randomUUID(),
channel,
seq: 2,
user,
text: `after b is gone ${randomUUID()}`,
created_at: new Date(0).toISOString(),
});
await untilCount(b, "message.created", 1);
expect(count(b, "message.created")).toBe(1);
await fanout.close();
}, 30_000);
});
// ---------------------------------------------------------------------------
// PHASE 6 — US2: the count survives the gateway it was counted on.
//
// Two of the three halves `docs/05-sad.md` gets wrong about `conn:{env}:{user}`.
// Line 167 describes it in the present tense as an instance-id lookup that does
// not exist; line 574 calls the same key "Not built". What is actually needed is
// neither: a count that no single gateway owns, so CON-02's "no sticky routing for
// correctness" survives contact with a cap.
// ---------------------------------------------------------------------------
describe("the count survives the gateway it was counted on (US2)", () => {
const open: Instance[] = [];
const sockets: WebSocket[] = [];
let user: string;
const connect = (instance: Instance): Recorded => {
const socket = new WebSocket(`${instance.url}?token=any`);
sockets.push(socket);
return record(socket);
};
/** **BOUNDED, and for the reason `refusedOn` polls.** An unbounded wait on a
* close turns an implementation that keeps the connection into a test that hangs
* until its own timeout — measured at 40,172 ms on the falsification below,
* against a lane with eleven seconds of headroom. */
const untilClosed = async (r: Recorded): Promise<number> => {
const deadline = Date.now() + 5_000;
while (Date.now() < deadline) {
if (r.closed !== undefined) return r.closed;
await new Promise((resolve) => setTimeout(resolve, 25));
}
throw new Error("the socket was still open after 5s");
};
/** Connect and expect the door to be shut, returning both codes. The close code
* AND the error code, never the fact of closing: a socket that closes for the
* wrong reason is identical from outside, which is the whole subject of
* `contracts/refusal.md`.
*
* **THE ACK IS RACED AGAINST THE CLOSE**, and that is a falsification's finding
* rather than a nicety. Waiting only for a close means an implementation that
* ADMITS the connection hangs until the test's own timeout — measured at 40,192
* ms against a per-instance count, one red test costing forty seconds of a lane
* with eleven seconds of headroom. Racing them turns the same defect into a
* one-line diff in about thirty milliseconds. */
const refusedOn = async (
instance: Instance,
): Promise<{ code: number; error: string | undefined }> => {
const r = connect(instance);
// POLLED RATHER THAN `Promise.race`d. `untilAcked` rejects at its own deadline,
// and a rejection nobody is waiting on any more is an unhandled rejection —
// which in this package takes the process down.
const deadline = Date.now() + 5_000;
while (Date.now() < deadline) {
if (r.closed !== undefined) {
const frame = r.frames.find((f) => f.type === "error");
return { code: r.closed, error: frame?.payload["code"] as string | undefined };
}
if (r.frames.some((f) => f.type === "connection.ack")) {
return { code: -1, error: "accepted" };
}
await new Promise((resolve) => setTimeout(resolve, 25));
}
throw new Error("neither refused nor acked within 5s");
};
const fill = async (instance: Instance, howMany: number): Promise<Recorded[]> => {
const held: Recorded[] = [];
for (let i = 0; i < howMany; i += 1) {
const r = connect(instance);
await untilAcked(r);
held.push(r);
}
return held;
};
const slotKey = (slot: number, environment = ENVIRONMENT): string =>
`conn:${environment}:${user}:${String(slot)}`;
beforeEach(() => {
user = `u-${randomUUID()}`;
});
afterEach(async () => {
for (const socket of sockets.splice(0)) socket.close();
await new Promise((resolve) => setTimeout(resolve, 150));
for (const instance of open.splice(0)) await instance.close();
});
it("counts five across two instances and refuses the sixth on either (FR-006, SC-004)", async () => {
// CON-02 AS A TEST. Two gateways, two registry modules, one Redis, and a cap
// that neither instance can compute on its own: three places on A and two on
// B is five, and the sixth has nowhere to go whichever door it knocks on.
// A per-instance count would accept five more on each.
const channel = randomUUID();
const a = await boot({ user, channels: [channel], cap: {} });
const b = await boot({ user, channels: [channel], cap: {} });
open.push(a, b);
await fill(a, 3);
await fill(b, 2);
// ON EITHER, and both halves are asserted. A test that only tries the
// instance holding three would pass against a cap that counts per instance
// and happens to be full there.
expect(await refusedOn(a)).toEqual({
code: 4004,
error: "connection_limit_reached",
});
expect(await refusedOn(b)).toEqual({
code: 4004,
error: "connection_limit_reached",
});
}, 40_000);
it("frees a dead instance's slots after the bound (FR-007, SC-005)", async () => {
// AN INJECTED BOUND, the way `presence.itest.ts` injects `graceMs`. The
// wall-clock version — sixty seconds of waiting — belongs in `quickstart.md`,
// and it is what proves this injected one is telling the truth.
const channel = randomUUID();
const dying = await boot({
user,
channels: [channel],
cap: { boundMs: 1_000, heartbeatMs: 300 },
});
open.push(dying);
await fill(dying, MAX_CONNECTIONS_PER_USER);
await dying.crash();
const survivor = await boot({
user,
channels: [channel],
cap: { boundMs: 1_000, heartbeatMs: 300 },
});
open.push(survivor);
// The last renewal landed at most 300 ms before the crash, so the five keys
// are gone by 1,000 ms after it and no later.
await new Promise((resolve) => setTimeout(resolve, 1_300));
const replacement = connect(survivor);
await untilAcked(replacement);
expect(count(replacement, "connection.ack")).toBe(1);
}, 40_000);
it("still refuses BEFORE the bound elapses (FR-007, SC-005)", async () => {
// **THE HALF USUALLY SKIPPED**, and the two halves were measured to be
// independent rather than assumed to be. Dropping the `PX` from the claim
// turns the test above red and this one green; making the `PX` 1 ms turns this
// one red and the test above green. So this is the assertion that says the
// places are really held, and that one says the holding really ends.
//
// The first draft of this comment said a slot-frees-eventually test "passes
// against an implementation whose keys carry no TTL", which is backwards: with
// no TTL it is the freeing that fails. What this test catches is a cap that
// claims nothing.
const channel = randomUUID();
const dying = await boot({
user,
channels: [channel],
cap: { boundMs: 1_000, heartbeatMs: 300 },
});
open.push(dying);
await fill(dying, MAX_CONNECTIONS_PER_USER);
await dying.crash();
const survivor = await boot({
user,
channels: [channel],
cap: { boundMs: 1_000, heartbeatMs: 300 },
});
open.push(survivor);
// 200 ms in, with expiry no earlier than 700 ms.
await new Promise((resolve) => setTimeout(resolve, 200));
expect(await refusedOn(survivor)).toEqual({
code: 4004,
error: "connection_limit_reached",
});
}, 40_000);
it("frees the slots IMMEDIATELY on a clean shutdown (FR-011a, SC-013)", async () => {
// FOUND BY BUILDING `traceability.md` DURING PLANNING: one task wrote
// `releaseAll()`, another asserted `connections.close()` was *registered* in
// `main.ts`, and nothing asserted anything was released. The crash test above
// covers the opposite path.
//
// THE DEFAULT BOUND IS THE POINT. Sixty seconds, and this test finishes in
// about one — so expiry cannot explain the acceptance, and neither can a
// socket's own close handler, because the sockets are still open. The default
// twenty-second heartbeat matters too: a short one would have a renewal
// re-claim the places straight after `releaseAll()` freed them.
const channel = randomUUID();
const replaced = await boot({ user, channels: [channel], cap: {} });
open.push(replaced);
await fill(replaced, MAX_CONNECTIONS_PER_USER);
const successor = await boot({ user, channels: [channel], cap: {} });
open.push(successor);
// NFR-REL-03 allows a deployment no more than one reconnection cycle, and a
// bound's worth of refusals after every deploy is more than one.
expect(await refusedOn(successor)).toEqual({
code: 4004,
error: "connection_limit_reached",
});
// RECONNECTED WITH NO WAIT AT ALL, which is what found the tombstone defect:
// a 20 ms sleep here made this test pass every time and hid it.
await replaced.shutdown();
const reconnected = connect(successor);
await untilAcked(reconnected);
expect(count(reconnected, "connection.ack")).toBe(1);
}, 40_000);
it("keeps a heartbeating connection's slot across three bounds (FR-008, SC-006)", async () => {
// THE PRESENCE CHAPTER SHIPPED A PRESENCE BUG BY ARMING A CHECK AT EXACTLY ITS OWN
// GRACE PERIOD — two deadlines on one instant, reached by two clocks, and the
// losing side stranded a user online for ever. This is the test that would
// have caught the same mistake here: three renewals per bound, so two
// consecutive misses still do not free a live connection's place.
const channel = randomUUID();
const lines: Record<string, unknown>[] = [];
const instance = await boot({
user,
channels: [channel],
cap: { boundMs: 600, heartbeatMs: 200 },
lines,
});
open.push(instance);
const live = await fill(instance, 1);
const before = await raw.get(slotKey(0));
expect(before, "the connection did not claim slot 0").toBeTruthy();
await new Promise((resolve) => setTimeout(resolve, 2_000));
// THE SAME VALUE IN THE SAME KEY, three bounds later. Non-null alone would be
// satisfied by a slot the connection lost and then re-claimed, which is a
// different outcome — so the log is checked for the re-claim as well.
expect(await raw.get(slotKey(0))).toBe(before);
expect(live[0]?.socket.readyState).toBe(WebSocket.OPEN);
expect(lines.filter((l) => l["msg"] === "connection.reclaimed")).toEqual([]);
expect(lines.filter((l) => l["msg"] === "connection.rejected")).toEqual([]);
}, 40_000);
it("counts each environment separately for one user (FR-012)", async () => {
// CONSTITUTION I. The key carries the environment, so the same person in a
// customer's staging and production environments has two allowances rather
// than one shared between them.
const channel = randomUUID();
const other = `env-${randomUUID()}`;
const here = await boot({ user, channels: [channel], cap: {} });
const there = await boot({
user,
channels: [channel],
environment: other,
cap: {},
});
open.push(here, there);
await fill(here, MAX_CONNECTIONS_PER_USER);
// Accepted in the other environment while this one is full.
const across = connect(there);
await untilAcked(across);
expect(count(across, "connection.ack")).toBe(1);
// AND THE FIRST ENVIRONMENT REALLY WAS FULL, which is the half that makes the
// acceptance above mean something. Without it the test passes against a cap
// that counts nothing.
expect(await refusedOn(here)).toEqual({
code: 4004,
error: "connection_limit_reached",
});
}, 40_000);
it("does not resurrect an expired slot on renewal (FR-011)", async () => {
// THE KEY IS DELETED RATHER THAN WAITED OUT, so the state is exact and no
// sleep is being trusted to be longer than a TTL.
//
// THE LOG LINE IS THE DISCRIMINATOR, and there is no other. Under `IFEQ` the
// renewal is refused and the module re-claims — `connection.reclaimed`. Under a
// plain `SET` the renewal succeeds against a key that does not exist, the slot
// is resurrected, and no line appears. The end state of the two is the same
// key holding the same id, which is why this test reads the logs.
const channel = randomUUID();
const lines: Record<string, unknown>[] = [];
const instance = await boot({
user,
channels: [channel],
cap: { heartbeatMs: 200 },
lines,
});
open.push(instance);
await fill(instance, 1);
expect(await raw.get(slotKey(0))).toBeTruthy();
await raw.del(slotKey(0));
await new Promise((resolve) => setTimeout(resolve, 500));
const reclaimed = lines.filter((l) => l["msg"] === "connection.reclaimed");
expect(reclaimed, "the renewal was not refused").toHaveLength(1);
expect(reclaimed[0]?.["slot"]).toBe(0);
}, 40_000);
it("refuses to renew a slot another connection took, rather than overwriting it (FR-011)", async () => {
// A DIFFERENT STATE FROM THE TEST ABOVE, and the one FR-011's second sentence
// was written for: the slot did not just expire, somebody else has it.
//
// `XX` TESTS EXISTENCE AND NOT OWNERSHIP — measured on Redis 8.10.0, `SET k B
// XX` against a key holding `A` returns OK — so under `XX` this renewal would
// take the rival's place back and six connections would be open against a
// count of five. The test above stays green under `XX`, because `XX` also
// refuses a key that is absent.
//
// T056's task text called this "the only test in the chapter that catches that
// substitution" and the falsification says otherwise: `IFEQ` → `XX` turns this
// test AND the cap-really-full test below red, two of sixteen. The claim was
// inherited from the task list and not re-run — this chapter's most common
// finding, one level up.
//
// PLANTED WITH ONE COMMAND rather than a delete followed by a claim. The two-
// command version has a window: a renewal firing inside it re-claims slot 0
// legitimately, the rival lands on slot 1, and the assertion below fails for a
// reason that is not a defect.
const channel = randomUUID();
const lines: Record<string, unknown>[] = [];
const instance = await boot({
user,
channels: [channel],
cap: { heartbeatMs: 200 },
lines,
});
open.push(instance);
await fill(instance, 1);
const rival = randomUUID();
await raw.set(slotKey(0), rival, "PX", 60_000);
await new Promise((resolve) => setTimeout(resolve, 500));
// Still the rival's. The renewal was refused and the connection went and found
// slot 1 instead.
expect(await raw.get(slotKey(0))).toBe(rival);
const reclaimed = lines.filter((l) => l["msg"] === "connection.reclaimed");
expect(reclaimed, "the renewal was not refused").toHaveLength(1);
expect(reclaimed[0]?.["slot"]).toBe(1);
expect(await raw.get(slotKey(1))).toBeTruthy();
}, 40_000);
it("keeps the connection working on a NEW slot after a re-claim (FR-011b, SC-014)", async () => {
// THE BRANCH A DESIGN THAT CLOSES ON ANY REFUSED RENEWAL GETS WRONG, and the
// one that happens after every brief outage: the slot expired, nothing else
// took it, and the user is under the limit. Closing here would cost somebody
// their connection for the registry's downtime.
//
// THE ASSERTION IS DELIVERY, not the log line the two tests above read. A
// socket can be open and no longer subscribed to anything.
const channel = randomUUID();
const lines: Record<string, unknown>[] = [];
const instance = await boot({
user,
channels: [channel],
cap: { heartbeatMs: 200 },
lines,
});
open.push(instance);
const [live] = await fill(instance, 1);
if (live === undefined) throw new Error("expected a connection");
await raw.del(slotKey(0));
await new Promise((resolve) => setTimeout(resolve, 500));
expect(lines.filter((l) => l["msg"] === "connection.reclaimed")).toHaveLength(1);
const fanout = createFanout({ url: REDIS, logger: silent });
await fanout.publish({
id: randomUUID(),
channel,
seq: 4,
user,
text: `after the re-claim ${randomUUID()}`,
created_at: new Date(0).toISOString(),
});
await untilCount(live, "message.created", 1);
expect(live.socket.readyState).toBe(WebSocket.OPEN);
expect(live.frames.filter((f) => f.type === "error")).toEqual([]);
await fanout.close();
}, 40_000);
it("never admits a sixth under many simultaneous claims (FR-013)", async () => {
// **THE RACE IS OBSERVABLE, AND T058'S FIRST ANSWER WAS THAT IT WAS NOT.**
// Replacing `SET NX` with a `GET` followed by a `SET` — check-then-act, which
// is exclusive when the calls are sequential and racy in the window between the
// two commands — left every one of the sixteen existing tests green. Under this
// test it admits **all twelve**, six runs out of six, because every attempt
// reads slot 0 as free before any of them writes it.
//
// So "nothing went red, therefore the ordering is unobservable" was a statement
// about the suite and not about the system, and T058's own instruction to read
// it that way would have been wrong. The difference is only observable to a
// test that puts several claims in flight at once, and that test did not exist
// until it was written to find out.
//
// TWO INSTANCES, because `Promise.all` of twelve connects against one gateway
// is not a race — they reach Redis through one client on one socket and the
// commands serialise there. Two modules are two clients on two sockets, which
// the server is free to interleave.
const channel = randomUUID();
const a = await boot({ user, channels: [channel], cap: {} });
const b = await boot({ user, channels: [channel], cap: {} });
open.push(a, b);
const ATTEMPTS = 12;
const tried = Array.from({ length: ATTEMPTS }, (_, i) =>
connect(i % 2 === 0 ? a : b),
);
const acked = (r: Recorded): boolean =>
r.frames.some((f) => f.type === "connection.ack");
const settled = (r: Recorded): boolean => r.closed !== undefined || acked(r);
const deadline = Date.now() + 10_000;
while (Date.now() < deadline && !tried.every(settled)) {
await new Promise((resolve) => setTimeout(resolve, 25));
}
expect(
tried.filter((r) => !settled(r)),
"an attempt neither acked nor closed",
).toHaveLength(0);
// FIVE IN, SEVEN OUT, AND THE KEYS AGREE. The count of accepted sockets and
// the count of live keys are two different oracles and both are checked: a
// registry that hands the same slot to two connections satisfies the second
// and fails the first.
expect(tried.filter(acked)).toHaveLength(MAX_CONNECTIONS_PER_USER);
expect(tried.filter((r) => r.closed === 4004)).toHaveLength(
ATTEMPTS - MAX_CONNECTIONS_PER_USER,
);
expect(await raw.keys(`conn:${ENVIRONMENT}:${user}:*`)).toHaveLength(
MAX_CONNECTIONS_PER_USER,
);
}, 40_000);
it("closes the connection when the cap is genuinely full at renewal (FR-011b, SC-014)", async () => {
// THE OTHER BRANCH, and it has to close: the place is gone, all five are held
// by other connections, and leaving this one open is six against a count of
// five. FR-005 is not in tension with this — it governs a REFUSAL, where
// opening a sixth must not cost the five. Here the connection has already lost
// its place to a competitor.
//
// The same code and the same message a refused sixth connection gets, because
// that is what it means.
const channel = randomUUID();
const lines: Record<string, unknown>[] = [];
const instance = await boot({
user,
channels: [channel],
cap: { heartbeatMs: 200 },
lines,
});
open.push(instance);
const [live] = await fill(instance, 1);
if (live === undefined) throw new Error("expected a connection");
// Every place taken by somebody else, this connection's included.
for (let slot = 0; slot < MAX_CONNECTIONS_PER_USER; slot += 1) {
await raw.set(slotKey(slot), randomUUID(), "PX", 60_000);
}
expect(await untilClosed(live)).toBe(4004);
const error = live.frames.find((f) => f.type === "error");
expect(error?.payload["code"]).toBe("connection_limit_reached");
expect(String(error?.payload["message"])).toContain("close one and reconnect");
const rejected = lines.filter(
(l) =>
l["msg"] === "connection.rejected" &&
l["reason"] === "connection_limit_reached",
);
expect(rejected).toHaveLength(1);
expect(rejected[0]?.["held"]).toBe(5);
}, 40_000);
});
// ---------------------------------------------------------------------------
// PHASE 8 — failing open, where the log line is the only evidence.
//
// FR-016 chooses availability over the cap: a registry the gateway cannot reach
// must not stop people connecting. The whole difficulty is that this decision is
// INVISIBLE. An accepted connection is an accepted connection; nothing a client
// sees says whether five was checked. the fan-out chapter found the general case — the
// fan-out's `publish` swallows its errors and resolves, so "the send returned 201
// while Redis was down" is equally true of a publisher that does nothing at all.
// The assertion that carries the requirement is the log line.
// ---------------------------------------------------------------------------
const UNREACHABLE = "redis://127.0.0.1:6399";
describe("the cap fails open, and says so (US4)", () => {
const open: Instance[] = [];
const sockets: WebSocket[] = [];
let user: string;
const connect = (instance: Instance): Recorded => {
const socket = new WebSocket(`${instance.url}?token=any`);
sockets.push(socket);
return record(socket);
};
beforeEach(() => {
user = `u-${randomUUID()}`;
});
afterEach(async () => {
for (const socket of sockets.splice(0)) socket.close();
await new Promise((resolve) => setTimeout(resolve, 150));
for (const instance of open.splice(0)) await instance.close();
});
it("logs that the cap was not enforced when the registry is unreachable (FR-016, SC-011)", async () => {
// THE LOG LINE, NOT THE ACCEPTANCE. `expect(acked).toBe(true)` here would pass
// against a build with no cap at all, against one whose claim always succeeds,
// and against this one. It says nothing.
const channel = randomUUID();
const lines: Record<string, unknown>[] = [];
const instance = await boot({
user,
channels: [channel],
cap: { url: UNREACHABLE },
lines,
});
open.push(instance);
const r = connect(instance);
await untilAcked(r);
const unenforced = lines.filter((l) => l["msg"] === "connection.cap_unenforced");
expect(unenforced, "nothing said the cap went unchecked").toHaveLength(1);
expect(unenforced[0]?.["user"]).toBe(user);
expect(unenforced[0]?.["environment_id"]).toBe(ENVIRONMENT);
// The id the claim was attempted with, so the line joins to the accept line
// below rather than floating free among however many connections are in flight.
const opened = lines.find((l) => l["msg"] === "connection.opened");
expect(unenforced[0]?.["connection_id"]).toBe(opened?.["connection_id"]);
// NFR-SEC-06, the same check the refusal gets: a failure surface is where a
// credential ends up in a support ticket.
expect(JSON.stringify(unenforced[0])).not.toContain("rk_");
}, 40_000);
it("tells 'not enforced' apart from 'enforced and under the limit' (FR-016a, SC-014)", async () => {
// BOTH DIRECTIONS, because one alone is satisfied by a line that always says
// the same thing. A single "accepted" satisfies neither.
const channel = randomUUID();
const blind: Record<string, unknown>[] = [];
const seeing: Record<string, unknown>[] = [];
const withoutRegistry = await boot({
user,
channels: [channel],
cap: { url: UNREACHABLE },
lines: blind,
});
const withRegistry = await boot({
user,
channels: [channel],
cap: {},
lines: seeing,
});
open.push(withoutRegistry, withRegistry);
await untilAcked(connect(withoutRegistry));
await untilAcked(connect(withRegistry));
const openedBlind = blind.find((l) => l["msg"] === "connection.opened");
const openedSeeing = seeing.find((l) => l["msg"] === "connection.opened");
expect(openedBlind?.["cap_enforced"]).toBe(false);
expect(openedSeeing?.["cap_enforced"]).toBe(true);
// And the error-level line exists on one side only, which is what an alert
// would be built on.
expect(blind.some((l) => l["msg"] === "connection.cap_unenforced")).toBe(true);
expect(seeing.some((l) => l["msg"] === "connection.cap_unenforced")).toBe(false);
}, 40_000);
it("does NOT fall back to counting this instance's own connections (FR-016b)", async () => {
// THE SPEC'S Q2, REJECTED EXPLICITLY. Falling back to
// `registry.connectionsFor(user)` looks like defence and is not: five per
// instance across four gateways is an effective cap of twenty wearing the
// label five. A wrong number that looks right is worse than a stated absence,
// and the stated absence is the log line the two tests above assert.
//
// SIX, not five, and the sixth is the assertion. A local fallback refuses it.
const channel = randomUUID();
const instance = await boot({
user,
channels: [channel],
cap: { url: UNREACHABLE },
});
open.push(instance);
for (let i = 0; i < MAX_CONNECTIONS_PER_USER + 1; i += 1) {
const r = connect(instance);
await untilAcked(r);
expect(r.closed, `attempt ${String(i)} was closed`).toBeUndefined();
}
}, 60_000);
});
describe("the slot registry, against a real broker", () => {
let registry: Connections;
let user: string;
const silent = { log: () => {} };
const ENV = `env-${randomUUID()}`;
beforeEach(() => {
registry = createConnections({ url: REDIS, logger: silent });
// A fresh user per test rather than a flush: `FLUSHDB` would delete the keys of
// every other suite running in parallel, and this package's config sets no
// `fileParallelism`.
user = `u-${randomUUID()}`;
});
afterAll(async () => {
await registry.close();
});
// ---- ARM 1 and ARM 2: the walk -----------------------------------------
it("claims the first free slot, and reports how many were held", async () => {
const first = await registry.claim(ENV, user, randomUUID());
expect(first).toEqual({ kind: "claimed", slot: 0, held: 0 });
const second = await registry.claim(ENV, user, randomUUID());
// ARM 1: `SET NX` missed on slot 0 and the walk moved on.
expect(second).toEqual({ kind: "claimed", slot: 1, held: 1 });
});
it("refuses when every slot is held, and says five (FR-001)", async () => {
for (let i = 0; i < MAX_CONNECTIONS_PER_USER; i += 1) {
expect((await registry.claim(ENV, user, randomUUID())).kind).toBe("claimed");
}
// ARM 2: the walk found no free slot.
expect(await registry.claim(ENV, user, randomUUID())).toEqual({
kind: "full",
held: 5,
});
});
it("counts each environment separately for one user identifier (FR-012)", async () => {
const other = `env-${randomUUID()}`;
for (let i = 0; i < MAX_CONNECTIONS_PER_USER; i += 1) {
await registry.claim(ENV, user, randomUUID());
}
expect((await registry.claim(other, user, randomUUID())).kind).toBe("claimed");
});
// ---- ARM 3 and ARM 9: the renewal, and the re-claim --------------------
it("renews a slot it still holds (FR-008)", async () => {
const id = randomUUID();
const claimed = await registry.claim(ENV, user, id);
if (claimed.kind !== "claimed") throw new Error("expected a slot");
expect(await registry.renew(ENV, user, id, claimed.slot)).toEqual({
kind: "renewed",
});
});
it("re-claims when its slot is GONE and nothing else took it (FR-011b)", async () => {
// ARM 3 then ARM 9. A short-lived registry so the bound elapses inside a test
// rather than in a minute: the boundMs option exists for exactly this, the way
// `membership.ts`'s reread interval does — sixty seconds does not fit in a
// package whose whole wall clock is forty-five.
const brief = createConnections({ url: REDIS, logger: silent, boundMs: 60 });
const id = randomUUID();
const claimed = await brief.claim(ENV, user, id);
if (claimed.kind !== "claimed") throw new Error("expected a slot");
await new Promise((resolve) => setTimeout(resolve, 120));
// THE COMMON CASE AFTER ANY BRIEF OUTAGE, and the branch a design that closes
// on every refused renewal gets wrong. The user is under the limit; the slot
// simply expired.
expect(await brief.renew(ENV, user, id, claimed.slot)).toEqual({
kind: "reclaimed",
slot: 0,
});
await brief.close();
});
// ---- ARM 4 and ARM 10: the hijack, and the cap genuinely full ----------
it("refuses to renew a slot ANOTHER connection now holds (FR-011)", async () => {
// ARM 4, and the one test in the chapter that catches `IFEQ` being replaced by
// `XX`. `XX` tests existence and not ownership — measured on 8.10.0,
// `SET k B XX` against a key holding `A` returns OK — so under `XX` this
// renewal would silently take the slot and the count would say five while six
// connections were open.
const brief = createConnections({ url: REDIS, logger: silent, boundMs: 60 });
const mine = randomUUID();
const claimed = await brief.claim(ENV, user, mine);
if (claimed.kind !== "claimed") throw new Error("expected a slot");
await new Promise((resolve) => setTimeout(resolve, 120));
// Somebody else takes the expired slot, and fills the rest so the re-claim has
// nowhere to go — ARM 10.
for (let i = 0; i < MAX_CONNECTIONS_PER_USER; i += 1) {
await brief.claim(ENV, user, randomUUID());
}
expect(await brief.renew(ENV, user, mine, claimed.slot)).toEqual({
kind: "full",
held: 5,
});
await brief.close();
});
// ---- ARM 6, ARM 7 and ARM 8: the release ------------------------------
it("frees a slot it holds, and the slot is reusable at once (FR-010)", async () => {
const id = randomUUID();
const claimed = await registry.claim(ENV, user, id);
if (claimed.kind !== "claimed") throw new Error("expected a slot");
await registry.release(ENV, user, id, claimed.slot);
// NO WAIT, AND THE SLOT IS NOT PINNED — because at the default one-millisecond
// tombstone there are THREE outcomes, not two, and the coverage lane found the
// third by failing here with `slot: 1` where this assertion had demanded 0.
//
// the tombstone is still there `SET NX` fails, `SET IFEQ -` takes it -> 0
// it expired before the walk `SET NX` succeeds -> 0
// it expires BETWEEN the two both fail, the walk moves on -> 1
//
// The third is a millisecond wide and harmless: a slot is skipped, never
// over-admitted, and the connection is accepted. What must not happen is a
// refusal, and that is what this asserts. The determinate version lives in the
// test below, where the window is held open at 500 ms so it cannot race.
//
// This test's FIRST version slept 20 ms and accepted any slot; the sleep is
// what hid the `releaseAll` defect for two phases. Removing the sleep was
// right and pinning the slot with it was not — the two changes arrived
// together and only one of them was justified.
const again = await registry.claim(ENV, user, randomUUID());
expect(again.kind).toBe("claimed");
if (again.kind !== "claimed") throw new Error("unreachable");
expect(again.slot, "a released slot cost more than one place").toBeLessThanOrEqual(1);
});
it("claims a slot whose tombstone has NOT expired (FR-010)", async () => {
// A HALF-SECOND TOMBSTONE, so the window is a window rather than a coin flip.
// With the shipped one-millisecond value this test would pass against the
// broken walk about half the time, which is how the defect survived: two of six
// runs of the clean-shutdown test, reported as `no connection.ack within 5s`.
const slow = createConnections({
url: REDIS,
logger: silent,
tombstoneMs: 500,
});
const id = randomUUID();
const claimed = await slow.claim(ENV, user, id);
if (claimed.kind !== "claimed") throw new Error("expected a slot");
await slow.release(ENV, user, id, claimed.slot);
expect(await slow.claim(ENV, user, randomUUID())).toEqual({
kind: "claimed",
slot: 0,
held: 0,
});
await slow.close();
});
it("accepts a claim immediately after releaseAll frees all five (FR-011a)", async () => {
// THE CASE THAT WAS ACTUALLY BROKEN, and it is a deploy. One slot tombstoned is
// one slot skipped; five tombstoned is a walk that finds nothing free and
// reports `full` — so a client reconnecting to the new instance is refused with
// `connection_limit_reached`, and the remedy that close code names is to close
// one of the connections it already holds. Those went with the old instance.
const slow = createConnections({
url: REDIS,
logger: silent,
tombstoneMs: 500,
});
const held = [];
for (let i = 0; i < MAX_CONNECTIONS_PER_USER; i += 1) {
const id = randomUUID();
const claimed = await slow.claim(ENV, user, id);
if (claimed.kind !== "claimed") throw new Error("expected a slot");
held.push({ environmentId: ENV, user, connectionId: id, slot: claimed.slot });
}
await slow.releaseAll(held);
expect(await slow.claim(ENV, user, randomUUID())).toEqual({
kind: "claimed",
slot: 0,
held: 0,
});
await slow.close();
});
it("does NOT free a slot another connection now holds (FR-010)", async () => {
// ARM 6, and the reason the release is conditional. Under a plain `DEL` this
// would delete the new owner's key and hand out a place that is in use — the
// same ownership hole `IFEQ` closed on the renewal, on the path that fix
// introduced.
const brief = createConnections({ url: REDIS, logger: silent, boundMs: 60 });
const mine = randomUUID();
const claimed = await brief.claim(ENV, user, mine);
if (claimed.kind !== "claimed") throw new Error("expected a slot");
await new Promise((resolve) => setTimeout(resolve, 120));
const theirs = randomUUID();
const retaken = await brief.claim(ENV, user, theirs);
expect(retaken).toEqual({ kind: "claimed", slot: 0, held: 0 });
await brief.release(ENV, user, mine, claimed.slot);
// Still theirs: the release was refused. Renewing proves it.
expect(await brief.renew(ENV, user, theirs, 0)).toEqual({ kind: "renewed" });
await brief.close();
});
it("releases every slot this instance holds (FR-011a)", async () => {
const held = [];
for (let i = 0; i < 3; i += 1) {
const id = randomUUID();
const claimed = await registry.claim(ENV, user, id);
if (claimed.kind !== "claimed") throw new Error("expected a slot");
held.push({ environmentId: ENV, user, connectionId: id, slot: claimed.slot });
}
await registry.releaseAll(held);
await new Promise((resolve) => setTimeout(resolve, 20));
// All three back, so the next three claims all succeed.
for (let i = 0; i < 3; i += 1) {
expect((await registry.claim(ENV, user, randomUUID())).kind).toBe("claimed");
}
});
it("builds without a url, from the environment or from the default", async () => {
// TWO BRANCHES IN ONE LINE, and the ratchet wanted both: the default parameter
// — which every test above steps over by passing `url` — and the `??` inside
// it, whose right-hand side the lane can never reach because it always sets
// `RELAY_REDIS_URL`. `codes.test.ts:128` established the swap-and-restore
// shape for exactly this; the `finally` is what keeps a failure here from
// silently pointing every later suite at a different Redis.
const defaulted = createConnections({ logger: silent });
const outcome = await defaulted.claim(ENV, `u-${randomUUID()}`, randomUUID());
expect(outcome.kind).toBe("claimed");
await defaulted.close();
const before = process.env["RELAY_REDIS_URL"];
try {
delete process.env["RELAY_REDIS_URL"];
// `DEFAULT_REDIS_URL` is localhost:6379, which is where the lane's Redis is,
// so this claims a place rather than failing open — and the assertion is that
// it reached A Redis, not that it reached a particular one.
const fallback = createConnections({ logger: silent });
expect((await fallback.claim(ENV, `u-${randomUUID()}`, randomUUID())).kind).toBe(
"claimed",
);
await fallback.close();
} finally {
if (before === undefined) delete process.env["RELAY_REDIS_URL"];
else process.env["RELAY_REDIS_URL"] = before;
}
});
});@@ -2,12 +2,13 @@ import { CLOSE_CODES, docsUrl, frameSchema } from "@relay/protocol";
import { createLogger, serve, type Logger } from "@relay/service-kit";
import { createApiClient } from "./api-client.js";
import { createFanout } from "./fanout.js";
import { createMembership } from "./membership.js";
import { createPresence } from "./presence.js";
+import { createConnections } from "./connections.js";
import { createTyping } from "./typing.js";
import { attachSessions } from "./session.js";
// The gateway — SAD §4.1: terminates WebSockets and never writes to the
// database (ADR-05). Chapter 1.4 stood up the HTTP half (health, request
// ids, structured logs); chapter 2.5 gives it the job it exists for, and
@@ -51,27 +52,43 @@ export function createServer(logger?: Logger) {
const membership = createMembership({ logger: log });
// THE SIXTH AND SEVENTH REDIS CLIENTS, and this module needs two of its own — a
// publisher and a subscriber — because it is the first fabric this service both
// publishes to and consumes from. `fanout.ts` states why they cannot be one client:
// a subscribed connection cannot issue ordinary commands, and PUBLISH is one.
const typing = createTyping({ logger: log });
+ // The connection registry's own client. Its keys put the environment id FIRST —
+ // `conn:{env}:{user}:{slot}` — so a cross-tenant read would need a caller to hand
+ // this module another environment's id, which the session layer takes from the
+ // api's verified identity and never from a payload.
+ const connections = createConnections({ logger: log });
const sessions = attachSessions({
server,
api: createApiClient(process.env.RELAY_API_URL ?? DEFAULT_API_URL),
logger: log,
fanout,
presence,
membership,
typing,
+ // AND THIS LINE IS WHAT `main.test.ts` GUARDS. Without it the cap does nothing:
+ // `connections?.claim(...)` is an optional chain on `undefined`, so every socket
+ // is admitted and no number moves — `**/main.ts` is excluded from the coverage
+ // ratchet, so no figure could show it. Registering `close()` below is the other
+ // half and neither substitutes for the other: without this line the cap is inert,
+ // without that one every gateway leaks a Redis client.
+ connections,
});
server.on("close", () => {
- sessions.close();
+ // `void`, LIKE ITS SIBLINGS. `sessions.close()` returns a promise as of the
+ // connection cap — it frees the places this instance holds before closing the
+ // socket server — and `server.on("close")` has nowhere to await one.
+ void sessions.close();
void fanout.close();
void presence.close();
void membership.close();
void typing.close();
+ void connections.close();
});
return server;
}
if (import.meta.main) {
const requested = Number(process.env.PORT ?? 4001);@@ -1,12 +1,14 @@
import { readFileSync } from "node:fs";
-import { join } from "node:path";
+import { dirname, join } from "node:path";
+import { fileURLToPath } from "node:url";
import type { AddressInfo } from "node:net";
import type { Server } from "node:http";
import { CLOSE_CODES, frameSchema } from "@relay/protocol";
+
import { describe, expect, it } from "vitest";
import { createLogger } from "@relay/service-kit";
import { createServer } from "./main.js";
@@ -62,12 +64,63 @@ describe("gateway skeleton", () => {
expect(body.code).toBe("not_found");
expect(typeof body.docs_url).toBe("string");
} finally {
server.close();
}
});
+
+ // T042b. THE SHUTDOWN SET, NAMED EXPLICITLY AND FAILING ON AN
+ // UNKNOWN MEMBER.
+ //
+ // Nothing verified the registration before analysis pass 6. `shutdown()` awaits
+ // each module's `close()`, `main.test.ts` called `server.close()` and asserted
+ // nothing about which modules closed, so a missing registration leaks one Redis
+ // client per gateway — silently and for ever.
+ //
+ // THIS IS THE TYPING CHAPTER'S DEFECT INVERTED. There, awaiting `close()` made lint
+ // see a used variable and hid a module that was never passed to
+ // `attachSessions`. Here, NOT awaiting it is what nothing could see. The two
+ // halves need two checks, and this is the second one's.
+ //
+ // Read from the source rather than executed: a shutdown that actually closes
+ // seven Redis clients is not something a unit test can observe without seven
+ // servers. What it CAN observe is that every module the file builds is also
+ // closed, which is the property that breaks when somebody adds an eighth.
+ // THE OTHER HALF OF THE PAIR BELOW, and neither substitutes for the other: a
+ // module that is built and never injected is inert, and one that is built and
+ // never closed leaks a Redis client per gateway. Same derivation, two properties.
+ it("closes every module it builds", () => {
+ const here = dirname(fileURLToPath(import.meta.url));
+ const source = readFileSync(join(here, "main.ts"), "utf8");
+ const built = [...source.matchAll(/^ {2}const (\w+) = create(\w+)\(/gm)].map(
+ (m) => m[1],
+ );
+ // TWO NAMED EXCEPTIONS, and named rather than pattern-excluded. `createLogger`
+ // returns a writer with nothing to release and `createServer` is this file's
+ // own export, not a module it owns. **The first version of this test had
+ // neither and went red on the logger** — which is the check working: an unknown
+ // member fails instead of being quietly skipped, and adding one to this list is
+ // a decision somebody has to write down.
+ const NOT_CLOSEABLE = ["logger", "server"];
+ const closeable = built.filter((name) => !NOT_CLOSEABLE.includes(name ?? ""));
+ // A POSITIVE CONTROL RATHER THAN A COUNT. The published version of this test
+ // asserted `toHaveLength(6)`, which is a number every later chapter that adds a
+ // module has to edit — and a number edited on every change is a number nobody
+ // reads. The loop below is the assertion; this line only says the derivation
+ // found something to loop over.
+ expect(closeable.length, "no `const x = createY(` found in main.ts").toBeGreaterThan(1);
+ // DERIVED, NOT NAMED. The closing site is wherever this file registers one —
+ // `server.on("close", …)` here — and reading the whole source rather than a
+ // named function means a refactor that moves the calls cannot silently pass.
+ for (const name of closeable) {
+ expect(
+ new RegExp(`(void |await )${String(name)}\\.close\\(\\)`).test(source),
+ `${String(name)} is built but never closed`,
+ ).toBe(true);
+ }
+ });
});
describe("every fabric createServer builds is injected", () => {
// A MODULE BUILT, CLOSED, AND NEVER PASSED IN IS INERT AND GREEN.
//
// `signalTyping` calls `typing?.publish(...)`. If `typing` never reaches@@ -764,12 +764,23 @@ describe("presence: the grace period (FR-RTM-06)", () => {
expect(seen.frames.filter((f) => f.payload.state === "offline")).toHaveLength(1);
}, 30_000);
// T045, and FR-RTM-09's five is enforced NOWHERE — `policy.ts:13` mentions it in a
// comment and nothing counts — so the reference count is unbounded and two is the
// easy case. Five connections, closed one at a time: nothing until the last.
+ //
+ // THE CONNECTION-CAP CHAPTER BUILT THE CAP AND THE SENTENCE ABOVE IS STILL TRUE HERE, which is a
+ // decision rather than an oversight. Every gateway module is an optional parameter
+ // and this fixture passes no `connections`, so nothing counts in THIS file after
+ // that chapter either. The cap lives where a fixture asks for it: the new
+ // `connections.itest.ts` and `session.itest.ts`'s "cap at the door" describe.
+ //
+ // The five below is therefore still free, and it is also now exactly the cap. A
+ // future edit adding a sixth connection to this test would be fine here and
+ // refused in either of those two files — worth knowing before somebody copies the
+ // pattern.
it("publishes nothing until the fifth of five connections closes", async () => {
const who = takeSubject();
const watcher = await arrive("linh");
const open = [];
for (let i = 0; i < 5; i += 1) open.push(await arrive(who, i % 2 ? a : b));
const seen = collect(watcher, "presence.changed", who);const DRIVER_EXEMPT_TESTS = [
// …
"services/gateway/src/typing.itest.ts",
// The connection-cap chapter, and NOT for the reason the four above give. This file needs no
// raw client to assert a publish — its subject is delivery, and it asserts on
// frames a socket received. It needs one to CAUSE a state: plant a rival in a
// slot key, delete one, and watch what a live connection does about it.
"services/gateway/src/connections.itest.ts",
];@@ -31,20 +31,43 @@ export default defineConfig({
env: {
RELAY_HARNESS_BAIT: "on",
RELAY_OUTBOX_RELAY: "off",
RELAY_EVENT_CONSUMER: "off",
},
include: ["src/**/*.itest.ts"],
- // ONE FILE AT A TIME, BECAUSE THEY SHARE ONE DATABASE.
- //
- // Every suite here runs migrations before it starts. Vitest runs FILES in parallel
- // by default, so several of them issue `CREATE TYPE` against the same schema at the
- // same moment and Postgres answers `duplicate key value violates unique constraint
- // "pg_type_typname_nsp_index"` — an error about its own catalogue, which reads like
- // a driver fault and is not one.
- //
- // It only bites when a migration is PENDING. With the schema already applied every
- // suite finds nothing to do and the race has no window, which is why serialising the
- // turbo tasks was enough until this chapter added a table.
- fileParallelism: false,
+ // FILES IN PARALLEL AGAIN, AND EIGHT PLACES ARE WHY IT COULD NOT BE.
+ //
+ // This lane ran one file at a time from the outbox chapter, for a real reason: a
+ // PENDING migration lets two suites issue `CREATE TYPE` against one schema, and
+ // Postgres answers `duplicate key value violates unique constraint
+ // "pg_type_typname_nsp_index"` — an error about its own catalogue that reads like a
+ // driver fault. That window closed in the instruments chapter, when `globalSetup`
+ // began migrating once before any file starts; the setting outlived it by eight
+ // chapters. The probe is a fresh database with every file racing, and it passes.
+ //
+ // What actually kept the files apart was EIGHT places scoped wider than their own
+ // subject, spread across eight chapters. The connection-cap chapter has them in a
+ // table with what each one really read. Three were known and carried in; five came
+ // from running the lane without the setting and reading what fell over, one at a
+ // time, over six runs. **Three of the five are not assertions at all.**
+ //
+ // AND THE WORKER COUNT IS MEASURED, NOT INHERITED. Vitest defaults to roughly one
+ // worker per core, which here is nine NestJS applications against one Postgres:
+ //
+ // maxWorkers 1 2 3 5 default(9)
+ // api lane 177s 102s 102s 102s 102s
+ // peak used 4188M 4275M 4467M 4817M 5456M
+ //
+ // **Every second of the saving is in one-to-two.** This lane waits on a shared
+ // database and broker far more than it computes, so a second file fills the first
+ // one's gaps and a third has no gap left to fill — it just pays for another runtime.
+ // The gateway's curve has its knee at FOUR, not two (69s, 46s, 35s, 35s), because its
+ // suites each drive a child process of their own; **the right number is per-lane and
+ // measured, and a default is neither.** Above the knee it stops being free: the
+ // gateway at six workers is no faster than at four and its typing suite starts
+ // missing a presence frame.
+ maxWorkers: 2,
+ //
+ // Each of the eight is fixed where it lives. The migration race is left to the one
+ // run it can happen on.
},
});@@ -12,20 +12,41 @@ export default defineConfig({
// bait, plants it per file. This lane gets exemption
// handling and NO bait: it holds no reader-shape fault, and planting
// would change its workload for no return (FR-022).
globalSetup: ["../../packages/test-harness/src/global-setup.ts"],
setupFiles: ["../../packages/test-harness/src/setup.ts"],
include: ["src/**/*.itest.ts"],
- // ONE FILE AT A TIME, BECAUSE THEY SHARE ONE DATABASE.
+ // FILES IN PARALLEL AGAIN, AND EIGHT PLACES ARE WHY IT COULD NOT BE.
//
- // Every suite here runs migrations before it starts. Vitest runs FILES in parallel
- // by default, so several of them issue `CREATE TYPE` against the same schema at the
- // same moment and Postgres answers `duplicate key value violates unique constraint
- // "pg_type_typname_nsp_index"` — an error about its own catalogue, which reads like
- // a driver fault and is not one.
+ // This lane ran one file at a time from the outbox chapter, for a real reason: a
+ // PENDING migration lets two suites issue `CREATE TYPE` against one schema, and
+ // Postgres answers `duplicate key value violates unique constraint
+ // "pg_type_typname_nsp_index"` — an error about its own catalogue that reads like a
+ // driver fault. That window closed in the instruments chapter, when `globalSetup`
+ // began migrating once before any file starts; the setting outlived it by eight
+ // chapters. The probe is a fresh database with every file racing, and it passes.
//
- // It only bites when a migration is PENDING. With the schema already applied every
- // suite finds nothing to do and the race has no window, which is why serialising the
- // turbo tasks was enough until this chapter added a table.
- fileParallelism: false,
+ // What actually kept the files apart was EIGHT places scoped wider than their own
+ // subject, spread across eight chapters. The connection-cap chapter has them in a
+ // table with what each one really read. Three were known and carried in; five came
+ // from running the lane without the setting and reading what fell over, one at a
+ // time, over six runs. **Three of the five are not assertions at all.**
+ //
+ // AND THE WORKER COUNT IS MEASURED, NOT INHERITED. Vitest defaults to roughly one
+ // worker per core, which here is nine NestJS applications against one Postgres:
+ //
+ // maxWorkers 2 3 4 6
+ // gateway lane 69s 46s 35s 35s
+ // peak used 4313M 4623M 4717M 5064M
+ //
+ // **The knee is at four here and at TWO in the api lane**, whose curve is flat from
+ // two workers on. These suites each drive a child api of their own, so they overlap
+ // further before they start queueing on the same database — **the right number is
+ // per-lane and measured, and a default is neither.** Above the knee it stops being
+ // free: six workers is no faster than four, and `typing.itest.ts` starts missing a
+ // presence frame it waits for.
+ maxWorkers: 4,
+ //
+ // Each of the eight is fixed where it lives. The migration race is left to the one
+ // run it can happen on.
},
});@@ -3,17 +3,23 @@ import { defineConfig } from "vitest/config";
// The guard's own lane. `globalSetup` installs the function and the triggers, the
// same file every other lane uses; there is deliberately NO `setupFiles`, because
// this suite manages its own connections — one carrying the exemption and one
// not — and a setup file that rewrote DATABASE_URL would remove the distinction
// the tests are about.
//
-// AND NO `fileParallelism: false`, unlike the api and gateway lanes. Those need it
-// because several suites call `migrate(pool)` concurrently and race on
-// `pg_type_typname_nsp_index`; here `globalSetup` migrates once and there is one
-// suite. A setting that changes nothing is worth leaving out — it reads as a
-// requirement to whoever copies this file next.
+// AND NO `fileParallelism: false` — WHICH IS NOW TRUE OF EVERY LANE BUT COVERAGE, AND
+// WAS WRITTEN HERE AS THE EXCEPTION. The api and gateway lanes set it because several
+// suites call `migrate(pool)` concurrently and race on `pg_type_typname_nsp_index`. That
+// race is one run wide — the first after a schema change — and serialising every file for
+// ever to cover it cost 175 seconds a battery; both lanes dropped it in this chapter, once
+// the six places that actually needed the files kept apart were fixed where they live.
+//
+// Here it was never needed for either reason: `globalSetup` migrates once and there is one
+// suite. A setting that changes nothing is worth leaving out — it reads as a requirement to
+// whoever copies this file next, which is exactly how it spread to two lanes that then kept
+// it for six chapters after the reason had been fixed.
export default defineConfig({
test: {
globalSetup: ["src/global-setup.ts"],
include: ["src/**/*.itest.ts"],
hookTimeout: 60_000,
},