The sixth connection, and where the count lives

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 passed

import { 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;
    }
  });
});