Tin nhắn chưa từng tới

    api -- "publish fan-out" --> redis

    G->>G: publish to Redis chan:{channel_id}

@@ -1,4 +1,8 @@
-import { messageCreatedSchema, type Message } from "@relay/protocol";
+import {
+  messageCreatedSchema,
+  subjectForChannel,
+  type Message,
+} from "@relay/protocol";
 import type { Logger } from "@relay/service-kit";
 // A NAMED import, not a default: ioredis is CommonJS, the gateway is ESM,
 // and without esModuleInterop a default import of a CJS module hands you
@@ -8,9 +12,16 @@ import type { Logger } from "@relay/service-kit";
 import { Redis } from "ioredis";
 
 // The fan-out fabric (chapter 2.6, ADR-07): Redis pub/sub, one subject per
-// channel — `chan:{channel_id}`. The instance that handled a send publishes
-// the committed message AFTER the api's response; every instance hosting a
-// member of that channel is subscribed and delivers to its local sockets.
+// channel — `chan:{channel_id}`. Every instance hosting a member of that
+// channel is subscribed and delivers to its local sockets.
+//
+// WHO PUBLISHES CHANGED IN CHAPTER 3.18. This comment used to say "the
+// instance that handled a send publishes the committed message AFTER the api's
+// response", which was true while a socket was the only way in. There are two
+// publishers now: this one, for a socket send, and the api, for a REST send.
+// The ordering also splits by transport — a socket can ack and then publish
+// because it has two channels, and a request handler cannot, because its
+// response IS the ack.
 //
 // This fabric is AT-MOST-ONCE by design. No acks, no replay, no consumer
 // groups. A frame that misses a subscriber is simply gone — and that is
@@ -26,13 +37,6 @@ import { Redis } from "ioredis";
 
 export const DEFAULT_REDIS_URL = "redis://localhost:6379";
 
-/** One subject per channel: an instance receives only frames it can
- * actually deliver, and a pathological channel saturates its own subject
- * rather than every gateway's inbox. */
-export function subjectFor(channelId: string): string {
-  return `chan:${channelId}`;
-}
-
 export interface Fanout {
   /** Register the delivery callback. Set by the session layer at wiring
    * time — the fabric knows how to receive, the sessions know who to
@@ -88,7 +92,7 @@ export function createFanout({
     async publish(message) {
       try {
         await publisher.publish(
-          subjectFor(message.channel),
+          subjectForChannel(message.channel),
           JSON.stringify(message),
         );
       } catch (error) {
@@ -103,13 +107,13 @@ export function createFanout({
     async subscribe(channelId) {
       const next = (counts.get(channelId) ?? 0) + 1;
       counts.set(channelId, next);
-      if (next === 1) await subscriber.subscribe(subjectFor(channelId));
+      if (next === 1) await subscriber.subscribe(subjectForChannel(channelId));
     },
     async unsubscribe(channelId) {
       const next = (counts.get(channelId) ?? 1) - 1;
       if (next <= 0) {
         counts.delete(channelId);
-        await subscriber.unsubscribe(subjectFor(channelId));
+        await subscriber.unsubscribe(subjectForChannel(channelId));
       } else {
         counts.set(channelId, next);
       }

@@ -1,9 +1,12 @@
 // @relay/protocol — the shared wire contract (ADR-01's payoff, chapter 1.3).
 // One home for frame schemas, their inferred types, the failure
-// vocabulary, and — from chapter 2.5 — the internal service contract the
-// gateway and API service share. Consumed by the gateway and API service
-// from 1.4, and by the SDK in a later part.
+// vocabulary, the internal service contract the gateway and API service share
+// (chapter 2.5), and — from chapter 3.18 — the live fan-out's subject grammar,
+// which needed a shared home the moment a second service published to it.
+// Consumed by the gateway and API service from 1.4, and by the SDK in a later
+// part.
 
 export * from "./frames.js";
 export * from "./codes.js";
 export * from "./internal.js";
+export * from "./fanout.js";

- **Ack after commit, never before** (FR-MSG-05). The Redis fan-out happens after the ack;
  a recipient may see the message milliseconds after the sender's ack, never before durability.

      send(connection.socket, { type: "message.ack", payload: { seq } });
      // ...and only THEN does anyone else hear about it. Durability, then the
      // sender's confirmation, then everybody's copy: no step overtakes the
      // one before it.

    242-byte payload, 2000 samples after 200 warm-up
 
    p50   0.142 ms
    p95   0.226 ms
    p99   0.472 ms
    max   0.965 ms

    const message = await this.messages.send(
      channelId,
      body,
      user.id,
      actingExternalId,
      tokenSubject === undefined,
    );

$ grep -rl "this.messages.send(" services/api/src
services/api/src/internal/internal.controller.ts
services/api/src/messages/messages.controller.ts

  // MESSAGE_PUBLISHER is deliberately absent. See the note above it.
  exports: [Repository, MessagesService],

@@ -1,13 +1,58 @@
-import { Module, Scope } from "@nestjs/common";
+import {
+  Inject,
+  Injectable,
+  Module,
+  Scope,
+  type OnModuleDestroy,
+} from "@nestjs/common";
 import { REQUEST } from "@nestjs/core";
+import type { Logger } from "@relay/service-kit";
 
 import { AuthModule } from "../auth/auth.module";
+import {
+  createMessagePublisher,
+  MESSAGE_PUBLISHER,
+  type MessagePublisher,
+} from "../fanout/publisher";
+import { apiLogger, LOGGER } from "../logger";
 import { createDb, createPool, type Db } from "../db/client";
 import type { RequestWithTenant } from "./request-with-tenant";
 import { Repository } from "../db/repository";
 import { MessagesController } from "./messages.controller";
 import { MessagesService } from "./messages.service";
 
+/** Chapter 3.18. The api publishes to the live fan-out from the send path, so
+ * the module that owns that path owns the client.
+ *
+ * PROVIDED AND NOT EXPORTED, and that is the point. `internal.module.ts` imports
+ * this module and, in its own words, "reuse[s] MessagesModule's providers
+ * wholesale" — so an exported publisher would be injectable from the internal
+ * route, which is the one path that must never publish. The gateway already
+ * publishes for a socket send, and a second publisher there would put the same
+ * message on every member's screen twice (FR-006).
+ *
+ * Withholding it makes that structural rather than a matter of where a call
+ * sits. This module already does the same with `"DB"`.
+ *
+ * The TOKEN lives in `../fanout/publisher` — this module imports the controller,
+ * so a controller importing the token from here would be a cycle. */
+
+/** `limits/limits.module.ts:10` states the convention: "resource in this api
+ * closes through `OnModuleDestroy`". Six modules implement it; this is
+ * `CounterStoreLifecycle` for the analogous Redis client. A `close()` nothing
+ * calls is a leaked handle in a service that boots once per integration
+ * suite. */
+@Injectable()
+export class MessagePublisherLifecycle implements OnModuleDestroy {
+  constructor(
+    @Inject(MESSAGE_PUBLISHER) private readonly publisher: MessagePublisher,
+  ) {}
+
+  async onModuleDestroy(): Promise<void> {
+    await this.publisher.close();
+  }
+}
+
 // The repository stays the plain 2.1 class — the framework's job is only
 // to construct it per request with the authenticated tenant (ADR-15's
 // scope note: guards authenticate, the data layer isolates).
@@ -41,7 +86,16 @@ import { MessagesService } from "./messages.service";
         new Repository(db, req.principal?.environmentId ?? ""),
     },
     MessagesService,
+    { provide: LOGGER, useFactory: apiLogger },
+    {
+      provide: MESSAGE_PUBLISHER,
+      inject: [LOGGER],
+      useFactory: (logger: Logger): MessagePublisher =>
+        createMessagePublisher({ logger }),
+    },
+    MessagePublisherLifecycle,
   ],
+  // MESSAGE_PUBLISHER is deliberately absent. See the note above it.
   exports: [Repository, MessagesService],
 })
 export class MessagesModule {}

@@ -3,6 +3,7 @@ import {
   Body,
   Controller,
   Get,
+  Inject,
   Param,
   Post,
   Query,
@@ -13,6 +14,10 @@ import {
 import { Accepts, CredentialGuard } from "../auth/credential.guard";
 import { Repository } from "../db/repository";
 import { MessagesService } from "./messages.service";
+import {
+  MESSAGE_PUBLISHER,
+  type MessagePublisher,
+} from "../fanout/publisher";
 import { historyQuerySchema, sendMessageBodySchema } from "./messages.schema";
 // `import type` is required, not stylistic: with isolatedModules and
 // emitDecoratorMetadata on (ADR-15's trade-off, chapter 1.4), a type used
@@ -62,6 +67,12 @@ export class MessagesController {
   constructor(
     private readonly messages: MessagesService,
     private readonly repo: Repository,
+    // Chapter 3.18. INJECTED HERE AND NOT INTO THE SERVICE, because two callers
+    // reach `MessagesService.send` — this route and `internal.controller.ts`,
+    // which is the gateway's — and the gateway publishes for its own path
+    // already. A publish in the service would put every socket-sent message on
+    // every member's screen twice (FR-006).
+    @Inject(MESSAGE_PUBLISHER) private readonly fanout: MessagePublisher,
   ) {}
 
   @Post()
@@ -158,6 +169,52 @@ export class MessagesController {
     // The field list is spelled out rather than spread-minus-`duplicate`,
     // so a new column joins the public response only when someone decides
     // it should.
+    // ── the live fan-out (chapter 3.18, FR-004) ────────────────────────────
+    //
+    // AFTER THE COMMIT, BEFORE THE RESPONSE. `docs/05-sad.md` says the fan-out
+    // happens "after the ack", and a socket can do that literally — it writes an
+    // ack frame and then publishes, because it has two channels. A request
+    // handler has one: the response IS the ack, so anything awaited here
+    // precedes it. FR-005 was amended to split by transport rather than pretend
+    // otherwise. What the sentence protects survives either way: the row is
+    // durable before anyone hears about it.
+    //
+    // Not in a `finally`, and not in the service's `try`. A refused send throws
+    // out of `this.messages.send` above and never reaches this line, which is
+    // FR-008 by construction rather than by a flag.
+    //
+    // TWO GUARDS, both mirrored from `session.ts:651`, both load-bearing:
+    //
+    //   !duplicate    A RECOGNISED RETRY WROTE NO ROW. 2.3 made the retry safe
+    //                 for storage; that did not make it safe for delivery, and a
+    //                 client retrying on a flaky link would otherwise put the
+    //                 same message on every member's screen twice.
+    //   text !== null A tombstone recovered by an old idempotency key is not a
+    //                 creation. It has a second, independent reason here:
+    //                 `messageSchema.text` is `z.string()`, not nullable, so a
+    //                 tombstone could not be published anyway — the far end
+    //                 would drop it as an invalid payload while this route
+    //                 answered 201.
+    if (!message.duplicate && message.text !== null) {
+      await this.fanout.publish(
+        {
+          id: message.id,
+          // `channel`, not `channel_id`. The frame's field is `channel`, and
+          // `messageSchema` is a `z.strictObject` — publishing `channel_id`
+          // would deliver NOTHING while this route still answered 201.
+          channel: message.channel_id,
+          seq: message.seq,
+          user: actingExternalId,
+          text: message.text,
+          created_at: message.created_at,
+        },
+        {
+          requestId: req.requestId ?? "unknown",
+          environmentId: req.principal?.environmentId ?? "unknown",
+        },
+      );
+    }
+
     return {
       id: message.id,
       channel_id: message.channel_id,

    if (!message.duplicate && message.text !== null) {
      await this.fanout.publish(

      // A RECOGNISED RETRY IS NOT REPUBLISHED. 2.3 made the retry safe for
      // storage; that did not make it safe for delivery, and a client that
      // retries on a flaky link would otherwise put the same message on
      // every member's screen twice. `text === null` is the same argument:
      // a tombstone recovered by an old key is not a creation.

export const messageSchema = z.strictObject({
  id: z.string().min(1),
  channel: z.string().min(1),
  seq: z.number().int().positive(),
  user: z.string().min(1),
  text: z.string(),
  created_at: z.iso.datetime(),
});

src/index.ts(12,1): error TS2308: Module "./internal.js" has already exported a
member named 'subjectFor'.

events.msg.created.{environment_id}     the spine   — carries the tenant
chan:{channel_id}                       the fan-out — carries a channel

      if (now() < downUntil) return;
      try {
        await redis.publish(
          subjectForChannel(message.channel),
          JSON.stringify(message),
        );
        downUntil = 0;
      } catch (error) {
        downUntil = now() + DOWN_WINDOW_MS;
        logger.log("error", "fanout.publish_failed", {

  // FAILING OPEN IS NOT FREE IF IT FAILS SLOWLY, and the first version of this
  // file was slow. With the store gone, every command waits out its connect
  // timeout before giving up — so each request paid a second or more, twice,

    the shipped publisher, 12 sends at a dead port    total 2 ms
                                                      (2, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0)
    createFanout's options, 12 sends                  HUNG past 420 s

    as first shipped (no commandTimeout)   publish HUNG past a 3,000 ms bound, twice
    with commandTimeout: 100               rejected in 100 ms, then 4 ms, then 100 ms

    registry.add(connection);
    // Subscriptions follow membership: the first local member of a channel
    // makes this instance a subscriber, and the last one to leave releases
    // it (reference-counted in the fabric).

@@ -75,6 +75,10 @@ export const OVER_AUTH_THRESHOLD = Symbol.for("relay:over-auth-threshold");
 export interface RequestWithPrincipal {
   headers: Record<string, string | string[] | undefined>;
   principal?: Principal;
+  /** Chapter 3.18: the id `RequestContextMiddleware` generated for this request.
+   * A handler that logs on its own — the fan-out publish does — needs it, and
+   * NFR-OBS-01 requires it in every structured line. */
+  requestId?: string;
   /** Chapter 3.8: set when this source address has spent its
    * failed-authentication allowance. See `OVER_AUTH_THRESHOLD` above. */
   [OVER_AUTH_THRESHOLD]?: boolean;

@@ -17,6 +17,13 @@ export class RequestContextMiddleware implements NestMiddleware {
   use(req: IncomingMessage, res: ServerResponse, next: () => void): void {
     const requestId = newRequestId();
     res.setHeader("X-Request-Id", requestId);
+    // ...and on the request, so a handler can put it in a line of its own.
+    // Chapter 3.18 needed this: the fan-out publish logs its failure from inside
+    // the send handler, and NFR-OBS-01 wants a request id in every structured
+    // line while NFR-OBS-06 wants five-minute traceability from one. Until now
+    // the id existed only here and on the response header, which a handler
+    // cannot read without taking over the response.
+    (req as { requestId?: string }).requestId = requestId;
     // `originalUrl` first, and this line was WRONG from chapter 2.2 until 3.8.
     // Express rewrites `req.url` relative to the mount point, and this middleware
     // is applied through `forRoutes("{*path}")`, so `req.url` is `/` — every

@@ -5,7 +5,7 @@ import { afterAll, beforeAll, describe, expect, it } from "vitest";
 import { createLogger } from "@relay/service-kit";
 import type { Message } from "@relay/protocol";
 
-import { createFanout, subjectFor, type Fanout } from "./fanout.js";
+import { createFanout, type Fanout } from "./fanout.js";
 
 // Chapter 2.6's real test: the one behaviour a single-process test CANNOT
 // show. Two fabric clients stand in for two gateway instances — same code,
@@ -146,7 +146,9 @@ describe("fan-out across instances", () => {
     await raw.fanout.close();
   });
 
-  it("names subjects per channel, so an instance hears only what it can deliver", () => {
-    expect(subjectFor(CHANNEL)).toBe(`chan:${CHANNEL}`);
-  });
+  // THE SUBJECT GRAMMAR'S TEST MOVED IN CHAPTER 3.18, to
+  // `packages/protocol/src/fanout.test.ts`, along with `subjectFor` itself. It
+  // was a pure string assertion sitting in a suite that needs a running Redis;
+  // it needed neither. What stays here is everything that genuinely needs the
+  // fabric — two clients, a real subject, and a delivery.
 });

@@ -382,6 +382,12 @@ describe("the socket gauntlet", () => {
   //      `fanout?.publish` is a no-op and nothing is ever delivered here. The control
   //      hung for five seconds and timed out.
   //
+  //      STILL TRUE OF THIS SUITE after chapter 3.18. That chapter gave the api a
+  //      publisher and added a THIRD describe to `session.itest.ts` with a fan-out
+  //      attached — deliberately a new block rather than a fourth argument to the
+  //      existing ones, so blocks like this that want no broker keep none. If a
+  //      delivery assertion belongs anywhere, it belongs there.
+  //
   // The ack's `cursor` is what the server ACCEPTED, so it is the one place the
   // session's membership decision is visible from outside. The control below shows a
   // member's cursor being accepted, which is what makes the removal assertion mean
@@ -593,10 +599,18 @@ describe("the socket gauntlet", () => {
   //
   // THE MESSAGE HALF IS CHECKED AGAINST THE SCHEMA AND NOT AGAINST A LIVE FRAME, because
   // no `message.created` ever arrives in this suite: `say()` writes through the
-  // repository, the api publishes to no fan-out, and nothing here drains the outbox.
-  // Chapter 3.12 recorded that as its own finding — a REST-sent message reaches no socket,
-  // ever — and `public-surface.itest.ts` is what pins it. Waiting for a frame here is a
-  // 5-second timeout, which is how this test was written the first time.
+  // repository, THIS SUITE attaches no fan-out, and nothing here drains the outbox.
+  //
+  // THE REASON CHANGED IN CHAPTER 3.18 AND THE FACT DID NOT. This comment used to say
+  // "the api publishes to no fan-out", which was the platform-wide truth chapter 3.12
+  // recorded as a finding — a REST-sent message reached no socket, by two independent
+  // mechanisms. Chapter 3.17 removed one and chapter 3.18 the other, so the api does
+  // publish now; nothing arrives HERE because this suite subscribes to nothing, which is
+  // a property of the fixture rather than of the platform. `public-surface.itest.ts` used
+  // to pin the absence and now pins the arrival.
+  //
+  // Waiting for a frame here is still a 5-second timeout, which is how this test was
+  // written the first time.
   it("keeps the socket's identity a bare external id, whatever the profile holds", async () => {
     // A full profile written through the public route.
     const patched = await fetch(`${api.url}/v1/users/${tenants.victim.userExternalId}`, {

@@ -239,17 +239,17 @@ describe("a channel, a member and a message, all over the public API", () => {
   // existing caller (FR-MSG-13's territory), and a live fan-out from the api is a
   // new coupling between the api and Redis. Both are named in the chapter.
   //
-  // CHAPTER 3.17 DID HALF OF THAT, and this comment is left standing rather than
-  // rewritten because the half it did is not the half that fixes this. FR-MSG-13 was
-  // amended — "on behalf of any user" became "on behalf of a bot user of that tenant" —
-  // so a REST send now names a sender, and the send below names one. What did NOT change
-  // is the fan-out: the api still publishes nothing, so the message still reaches no
-  // socket, live or on resume. Chapter 3.18 is the fan-out.
+  // BOTH HALVES ARE CLOSED NOW, and the two chapters that closed them are worth naming
+  // separately because the gap needed both. Chapter 3.17 amended FR-MSG-13 — "on behalf
+  // of any user" became "on behalf of a bot user of that tenant" — so a REST send names a
+  // sender and `toFrame` stopped dropping the row from a resume. Chapter 3.18 gave the api
+  // a publisher, so the same row now reaches a LIVE socket too. Chapter 3.12's `gaps.md`
+  // G1 listed exactly those two mechanisms; neither remains.
   //
   // THE SENDER IS A BOT, because the caller is a key. A key may not name "tuan" — that
   // is a person and `sender_not_permitted` is the refusal — so the send that this test
   // needs to succeed must name software.
-  it("does NOT deliver a REST-sent message, live or on resume", async () => {
+  it("delivers a REST-sent message, live and on resume", async () => {
     const channelId = await seedOverTheWire("rest", ["tuan"]);
     const token = await mint("tuan");
     // Created over the public route, because this suite has no database handle by
@@ -261,7 +261,7 @@ describe("a channel, a member and a message, all over the public API", () => {
           {
             external_id: "rest-courier",
             kind: "bot",
-            description: "sends over REST so this test can watch nothing arrive",
+            description: "sends over REST so this test can watch it arrive",
           },
         ],
       },
@@ -296,9 +296,32 @@ describe("a channel, a member and a message, all over the public API", () => {
     // rested on is gone.
     expect(history.messages.every((m) => m.user === "rest-courier")).toBe(true);
 
-    // No live delivery.
-    await new Promise((resolve) => setTimeout(resolve, 1_500));
-    expect(live.frames.filter((f) => f.type === "message.created")).toEqual([]);
+    // LIVE DELIVERY, WHICH THIS BLOCK ASSERTED WAS ABSENT UNTIL CHAPTER 3.18.
+    //
+    // It read `toEqual([])`, and the reason was true when it was written: the only
+    // publisher to the fan-out was the gateway's own send handler, so a message a
+    // customer's backend posted committed, returned 201, and reached nobody. The api
+    // publishes now — `messages.controller.ts`, guarded the way `session.ts:651` is —
+    // and the two sends above arrive here in order.
+    //
+    // Waiting for BOTH rather than for the first: one frame arriving would be satisfied
+    // by a publisher that fired once and by one that fired correctly twice.
+    const deadline = Date.now() + 4_000;
+    for (;;) {
+      const created = live.frames.filter((f) => f.type === "message.created");
+      if (created.length >= 2) break;
+      if (Date.now() > deadline) {
+        throw new Error(
+          `expected 2 live frames, saw ${created.length}: ` +
+            live.frames.map((f) => f.type).join(", "),
+        );
+      }
+      await new Promise((resolve) => setTimeout(resolve, 25));
+    }
+    const liveTexts = live.frames
+      .filter((f) => f.type === "message.created")
+      .map((f) => (f as { payload: { text: string } }).payload.text);
+    expect(liveTexts).toEqual([first, second]);
     live.socket.close();
 
     // AND ON RESUME IT NOW ARRIVES — WHICH IS HALF OF THE GAP CLOSING (chapter 3.17).
@@ -308,15 +331,15 @@ describe("a channel, a member and a message, all over the public API", () => {
     // chapter 3.12's `gaps.md` G1 listed TWO independent mechanisms for "a REST-sent
     // message reaches no socket" — nothing publishes, and the public send passes no user.
     //
-    // FR-MSG-15 removes the second. Every REST send now names a sender, `toFrame` has no
-    // reason to drop the row, and the backfill delivers it. So the resume half of G1 is
-    // closed by this chapter and the LIVE half is not: `live.frames` above is still
-    // empty, because only the gateway publishes to the fan-out (`session.ts`) and the api
-    // still publishes nothing. Chapter 3.18 is that half.
+    // FR-MSG-15 removed the second and chapter 3.18's publisher removed the first, so
+    // both legs of this test now assert arrival: live above, and on resume below. The
+    // resume leg is the one that proves the two paths do not double up — a client that
+    // was connected and then reconnects with a cursor gets the backfill, not a replay of
+    // what the fan-out already delivered, because the cursor is what decides.
     //
-    // The test's name is now half wrong and is left alone deliberately: T096a amends the
-    // gap record, and renaming a test is not how a reader learns that a two-mechanism
-    // gap became a one-mechanism gap.
+    // THE TEST'S NAME CHANGED WITH IT. It said "does NOT deliver" and was left half wrong
+    // on purpose while only half the gap was closed; leaving it now would make it wholly
+    // wrong, which is a different thing.
     const resumed = reader(`${wsUrl}/v1/ws?token=${token}&cursor=${channelId}:1`);
     await resumed.opened;
     await new Promise((resolve) => setTimeout(resolve, 1_500));

@@ -1,3 +1,4 @@
+import { randomUUID } from "node:crypto";
 import { beforeAll, describe, expect, it } from "vitest";
 
 // AN INTEGRATION BUILT FROM PUBLISHED DOCUMENTATION ALONE (FR-031, SC-009,
@@ -230,12 +231,23 @@ describe("integrating with Relay from the outside", () => {
     expect(page.messages.map((m) => m.text)).toContain(text);
   });
 
-  it("receives a message on a socket — SENT over the socket", async () => {
-    // THE SEND HAS TO BE ON THE SOCKET, and finding that out is one of the gaps this
-    // exercise recorded. It had TWO causes and chapter 3.17 removed one: the api still
-    // publishes to no fan-out, so nothing arrives LIVE — but the public send now
-    // attributes a sender, so the row is no longer dropped from a resume. Half the gap,
-    // and the half that remains is the fan-out.
+  // WAS `it.fails` FOR THE LENGTH OF THIS CHAPTER'S PHASE 1 AND 2.
+  //
+  // A red lane is not the same as a recorded failure, so the gap was asserted
+  // rather than left broken: 10,114 ms to the deadline having seen only
+  // `connection.ack`, with a 201 in hand. The publish landed in Phase 3 and this
+  // became a plain `it` — the body now succeeds in about 150 ms.
+  it("receives a message on a socket — sent over REST", async () => {
+    // THE SEND NO LONGER HAS TO BE ON THE SOCKET, and that is this chapter.
+    //
+    // The gap this exercise recorded had TWO causes. Chapter 3.17 removed the first:
+    // a public send attributes a sender, so the row is no longer dropped from a
+    // resume. Chapter 3.18 removes the second, which was the whole of what remained
+    // — the api published to no fan-out, so a REST-sent message reached no live
+    // socket. The title of this test used to say "SENT over the socket" in capitals,
+    // because a REST send could not work; it now sends over REST on purpose.
+    //
+    // The send is the one an integrating developer's backend actually makes.
     const socket = new WebSocket(`${ws}/v1/ws?token=${token}`);
     const frames: { type: string; payload?: { text?: string; seq?: number } }[] = [];
     // Listeners attached BEFORE the open await. `connection.ack` arrives the
@@ -268,17 +280,27 @@ describe("integrating with Relay from the outside", () => {
 
     await waitFor((f) => f.type === "connection.ack", "connection.ack");
 
-    const text = `over the socket ${Date.now()}`;
-    socket.send(
-      JSON.stringify({
-        type: "message.send",
-        payload: { idem_key: `outsider-${Date.now()}`, channel: channelId, text },
-      }),
+    const text = `over REST ${Date.now()}`;
+    // NOT `socket.send`. A POST, with the credential a customer's server holds, to
+    // the route their backend calls — and then the socket is watched for the frame.
+    // `user: "outside-bot"` is not optional and not decoration. Chapter 3.17 made an
+    // application credential speak only as a bot user of its tenant, so a POST without
+    // it is a 400 naming `user` — which is how the first run of this inverted test
+    // failed, for a reason that had nothing to do with delivery.
+    const posted = await post(
+      `/v1/channels/${channelId}/messages`,
+      // A UUID, because the REST body demands one: `idempotency_key: z.string().uuid()`
+      // on this route, where the socket frame's `idem_key` is any string up to 255.
+      // Two entrances, two idempotency contracts — the second run of this inverted
+      // test failed on it, with `invalid_request` naming the field.
+      { text, user: "outside-bot", idempotency_key: randomUUID() },
+      credential,
     );
+    expect(posted.status).toBe(201);
 
-    // The sender's own acknowledgement, then the event. Both are documented and
-    // both matter: the ack says it was committed, the event says it was delivered.
-    await waitFor((f) => f.type === "message.ack", "message.ack");
+    // The REST response is the acknowledgement — there is no `message.ack` frame on
+    // this path, because the sender is not holding a socket. What has to arrive is
+    // the delivery, on a socket that was already open before the send.
     await waitFor(
       (f) => f.type === "message.created" && (f as { payload?: { text?: string } }).payload?.text === text,
       "message.created for the text just sent",

@@ -371,6 +371,23 @@ export default defineConfig({
           statements: 97,
         },
 
+        // CHOSEN BEFORE THE FIRST COVERAGE REPORT, not read off it (chapter 3.18,
+        // T011). The requirement is that the failure path be covered: this file's
+        // whole job is to swallow a publish error, log it, and open a window, and
+        // a test that only checks `publish` resolved cannot tell that apart from a
+        // publisher with no body. So every branch, and every function — the last
+        // of which forced `close()` and the ioredis `error` listener to be tested
+        // rather than assumed, which is R10 and the OnModuleDestroy convention.
+        //
+        // Without a pin this file falls to the global floor of 70, which a
+        // ten-line publisher clears with its `catch` untested.
+        "services/api/src/fanout/publisher.ts": {
+          branches: 100,
+          functions: 100,
+          lines: 100,
+          statements: 100,
+        },
+
         // A FLOOR, NOT AN ACHIEVEMENT. `messages.service.ts` measures 70.83 / 61.76 /
         // 100 / 70.83, and the six uncovered statements are all PRE-EXISTING: the quota
         // refusal and its rethrow (chapter 3.10) and the history cursor's decode (chapter
@@ -388,6 +405,30 @@ export default defineConfig({
           statements: 70,
         },
 
+        // 87, AGAINST A MEASURED 87.5 (21/24) — T011 asked for this pin or a recorded
+        // reason, and got neither for eight phases. The publish guard's two branches,
+        // `!message.duplicate && message.text !== null`, are FR-007's entire mechanism
+        // and were sitting under the global floor of 70.
+        //
+        // The FR-007 test moved this file 83.33 -> 87.5 by covering the `duplicate`
+        // side; T058a's traceability map is what noticed the clause had no test at all.
+        //
+        // THE REMAINING UNCOVERED BRANCH IS UNREACHABLE ON THIS ROUTE, and is left
+        // rather than deleted. `message.text !== null` is only ever evaluated for a
+        // NON-duplicate — the `&&` short-circuits otherwise — and a non-duplicate row
+        // was just written from a request whose schema requires `text`. So the false
+        // side cannot be reached from here. The ratchet has removed unreachable code
+        // three times in this repository; this one stays, because `messageSchema` types
+        // `text` as non-nullable and a null would publish a frame the delivery side
+        // drops silently. A guard against a state the type system forbids is cheap; the
+        // alternative is a silent drop.
+        "services/api/src/messages/messages.controller.ts": {
+          branches: 87,
+          functions: 100,
+          lines: 100,
+          statements: 96,
+        },
+
         "services/api/src/webhooks/disable.ts": {
           branches: 100,
           functions: 100,

@@ -356,3 +356,107 @@ describe("resume across a real fabric", () => {
     socket.close();
   });
 });
+
+// ── chapter 3.18: two instances, one fabric (US2) ───────────────────────────
+//
+// `boot()` IS UNTOUCHED. It is called six times above and each call builds its
+// own `createFanout` and its own server, so two calls already give two gateway
+// instances sharing one Redis — which is precisely what SC-002 needs. Changing
+// the fixture to "support" that would have changed six passing tests to prove
+// nothing new (3.17's T040b, the fifth such incident in two features).
+//
+// WHAT THIS PROVES AND WHAT IT DOES NOT. The api here is a stub, as everywhere
+// in this file: the gateway has no database (ADR-05) and these suites are about
+// the fabric. So this is the DELIVERY half of SC-002 — a frame published by
+// somebody else reaches the instance holding a member and only that one. The
+// half where a REAL api publishes lives in `session.itest.ts`, which spawns one.
+// Neither fixture does both, and `chapter-notes.md` says so rather than letting
+// the pair imply it.
+describe("two instances on one fabric (chapter 3.18)", () => {
+  const OTHER_CHANNEL = randomUUID();
+  let member: Harness | undefined;
+  let bystander: Harness | undefined;
+  /** Closed BEFORE the harnesses. `Harness.close()` calls `server.close()`,
+   * which waits for open connections to drain — so a test that leaves a socket
+   * open hangs the teardown, and vitest reports it as "Hook timed out in
+   * 10000ms" pointing at `afterEach`. The first version of these tests looked
+   * like a delivery failure and was a housekeeping one. */
+  const sockets: WebSocket[] = [];
+  const open = async (harness: Harness) => {
+    const socket = new WebSocket(`${harness.url}?token=${await token()}`);
+    sockets.push(socket);
+    return record(socket);
+  };
+
+  const stub = (channels: string[]) => ({
+    session: async () => ({
+      environment_id: "env-1",
+      user: "tuan",
+      banned: false,
+      channel_ids: channels,
+      limits: { connect: 3_000, send: 600 },
+    }),
+    backfill: async () => ({}),
+    sendMessage: async () => {
+      throw new Error("not used");
+    },
+  });
+
+  afterEach(async () => {
+    for (const socket of sockets.splice(0)) socket.close();
+    await member?.close();
+    await bystander?.close();
+    member = undefined;
+    bystander = undefined;
+  });
+
+  it("delivers to the instance holding a member (SC-002)", async () => {
+    member = await boot(stub([CHANNEL]));
+    const frames = await open(member);
+    await settle(200);
+
+    // Published by a THIRD client — neither instance's own — which is what the
+    // api is once it publishes. The instance under test is a subscriber only.
+    await publishFromElsewhere(frame(3_001));
+    await settle(400);
+
+    expect(created(frames)).toEqual([3_001]);
+  });
+
+  it("delivers to NEITHER instance for a channel neither holds (SC-002's negative)", async () => {
+    // The subject is the filter and it is the only one. An instance subscribes
+    // to `chan:{id}` because a session named that channel at connect; a frame on
+    // any other subject is not something it declines to deliver, it is something
+    // it never hears.
+    member = await boot(stub([CHANNEL]));
+    bystander = await boot(stub([OTHER_CHANNEL]));
+    const a = await open(member);
+    const b = await open(bystander);
+    await settle(200);
+
+    // A third channel, which neither session named.
+    await publishFromElsewhere({ ...frame(3_002), channel: randomUUID() });
+    await settle(400);
+
+    expect(created(a)).toEqual([]);
+    expect(created(b)).toEqual([]);
+  });
+
+  it("delivers to the member's instance and not to the bystander's", async () => {
+    // The pair that matters for SC-002: two live instances, one frame, and the
+    // silence on the second is as much of the assertion as the arrival on the
+    // first. Asserted by COUNT on both sides — "the member got it" alone would
+    // be satisfied by a fabric that broadcast to everybody.
+    member = await boot(stub([CHANNEL]));
+    bystander = await boot(stub([OTHER_CHANNEL]));
+    const a = await open(member);
+    const b = await open(bystander);
+    await settle(200);
+
+    await publishFromElsewhere(frame(3_003));
+    await settle(400);
+
+    expect(created(a)).toEqual([3_003]);
+    expect(created(b)).toEqual([]);
+  });
+});

/** The live fan-out's subject grammar (chapter 2.6, ADR-07).
 *
 * MOVED HERE IN CHAPTER 3.18, and the reason is the same one chapter 3.4 gave
 * when it moved the event spine's `subjectFor` into this package: a subject
 * grammar belongs where every party that uses it can agree on it. Until 3.18
 * the gateway was the only publisher, so the grammar could live beside the
 * client that spoke it. The api publishes now too, and the api cannot import
 * from a service.
 *
 * WHAT DID NOT MOVE. `createFanout` stays in the gateway: it holds two ioredis
 * connections and this package has exactly one dependency, `zod`. A client with
 * a socket does not belong in a package of schemas. `DEFAULT_REDIS_URL` did not
 * move either — it is declared in three service files, it is deployment
 * configuration rather than protocol, and consolidating one of three copies
 * into a shared package leaves a shared definition and two locals, which is
 * worse than three locals.
 *
 * The payload never needed moving. `Message` and `messageCreatedSchema` have
 * been in `frames.ts` since 2.2; the fan-out has always carried a wire frame's
 * payload rather than a shape of its own. */
 
/** One subject per channel: an instance receives only frames it can
 * actually deliver, and a pathological channel saturates its own subject
 * rather than every gateway's inbox.
 *
 * NOT `subjectFor`, which this function was called in the gateway. This package
 * already exports a `subjectFor` — `internal.ts`'s, for the event spine's
 * `events.{domain}.{action}.{env}` — and the two cannot share a name here. The
 * compiler said so the moment both were exported:
 *
 *     src/index.ts(12,1): error TS2308: Module "./internal.js" has already
 *     exported a member named 'subjectFor'.
 *
 * The spine's name is chapter 3.4's and is published; this one is new, so this
 * one moves. The collision is the same asymmetry the chapter has to explain
 * anyway: the spine's subject carries the tenant, the fan-out's carries only a
 * channel id, and putting them side by side is what made that visible. */
export function subjectForChannel(channelId: string): string {
  return `chan:${channelId}`;
}

import { describe, expect, it } from "vitest";
 
import { subjectForChannel } from "./fanout.js";
 
// THIS ASSERTION USED TO LIVE IN `services/gateway/src/fanout.itest.ts:150`,
// inside a suite that needs a running Redis. It is a pure string test: it needs
// no broker, no container and no lane. It moved here with the function in
// chapter 3.18, and it exists before the old copy is deleted so the property is
// never untested for the length of a commit.
describe("the fan-out subject grammar", () => {
  it("names one subject per channel", () => {
    expect(subjectForChannel("c1")).toBe("chan:c1");
  });
 
  it("is a prefix and the id, with nothing between them", () => {
    // The gateway subscribes with this and the api publishes with it. A change
    // to the separator, the prefix or the order silently stops delivery while
    // both sides keep working on their own — which is why the shape is pinned
    // rather than left to `chan:${id}` appearing twice in two repositories.
    const id = "954ff4f6-e4da-43ca-9988-6eb92d6e383a";
    expect(subjectForChannel(id)).toBe(`chan:${id}`);
    expect(subjectForChannel(id).startsWith("chan:")).toBe(true);
    expect(subjectForChannel(id).slice("chan:".length)).toBe(id);
  });
 
  it("does not interpret the id", () => {
    // No validation, no escaping, no lowercasing. The channel id is a UUID from
    // the repository by the time anything publishes, and a grammar that quietly
    // rewrote it would be a second source of truth for the subject.
    expect(subjectForChannel("")).toBe("chan:");
    expect(subjectForChannel("Mixed-Case_1")).toBe("chan:Mixed-Case_1");
  });
});

import { subjectForChannel, type Message } from "@relay/protocol";
import type { Logger } from "@relay/service-kit";
// A NAMED import: ioredis is CommonJS and this service is ESM.
import { Redis } from "ioredis";
 
/** The api's half of the live fan-out (chapter 3.18, FR-004).
 *
 * WHY THE API HAS ITS OWN PUBLISHER instead of reusing the gateway's
 * `createFanout`. Three reasons, in order of how much they cost to learn:
 *
 * 1. It cannot import it. `services/api` does not depend on the gateway, and it
 *    should not — the shared thing is the subject grammar, which moved to
 *    `@relay/protocol` for exactly this.
 * 2. It needs half of it. `createFanout` builds two connections because a
 *    subscribed ioredis client cannot issue ordinary commands. The api never
 *    subscribes, so it takes one.
 * 3. IT MUST NOT COPY THE GATEWAY'S CLIENT OPTIONS. `createFanout` uses
 *    `new Redis(url)` with defaults and attaches no `error` listener, which is
 *    survivable for a long-lived gateway and not for a request handler. The
 *    options below come from `limits/store.ts`, which is the api's own Redis
 *    client and learned this the hard way.
 *
 * NO OFF-SWITCH, and that is a decision rather than an omission (T009c).
 * Four api modules carry one — `RELAY_OUTBOX_RELAY`, `RELAY_DELIVERY_RELAY`,
 * `RELAY_NOTIFICATION_RELAY`, `RELAY_EVENT_CONSUMER` — and CI sets them off in
 * the lane because "a background daemon draining the table two suites are
 * asserting on is a race between test files, not a property". Every one of
 * those is a *daemon* that polls shared state. This is a synchronous publish to
 * `chan:{uuid}`, and a suite that did not create that channel cannot observe it.
 * The stronger reason is the one this chapter is about: a switch would let the
 * lane run green with the publish disabled, which is the false-green shape the
 * whole feature exists to remove. */
/** The DI token, declared HERE rather than in `messages.module.ts`.
 *
 * A CIRCULAR IMPORT OTHERWISE, and Nest reports it as a missing dependency
 * rather than as a cycle: "Nest can't resolve dependencies of the
 * MessagesController (MessagesService, Repository, ?)". The module imports the
 * controller, so a controller importing the token from the module closes the
 * loop and the token is `undefined` when the decorator metadata is read. Beside
 * the interface it names, nobody imports anybody twice. */
export const MESSAGE_PUBLISHER = "MESSAGE_PUBLISHER";
 
export interface MessagePublisher {
  /** Publish a committed message to its channel's subject. NEVER REJECTS —
   * delivery is allowed to fail, because the row is already durable and 2.7's
   * resume will find it (ADR-07, constitution IV). */
  publish(message: Message, context: PublishContext): Promise<void>;
  close(): Promise<void>;
}
 
/** What the failure has to be findable by. NFR-OBS-01 wants a request id and a
 * tenant id in every structured log, and NFR-OBS-06 wants five-minute
 * traceability from the former. The gateway's equivalent line carries neither,
 * correctly — it is not inside a request. This one is. */
export interface PublishContext {
  requestId: string;
  environmentId: string;
}
 
export const DEFAULT_FANOUT_REDIS_URL = "redis://localhost:6379";
 
/** How long a known-dead Redis is left alone. Lifted from
 * `limits/store.ts`'s `DOWN_WINDOW_MS`, and the reason is that file's: "FAILING
 * OPEN IS NOT FREE IF IT FAILS SLOWLY… each request paid a second or more,
 * twice." The options alone were the slow version; the window is the fix, and
 * the first draft of chapter 3.18's contract copied the options without it. */
const DOWN_WINDOW_MS = 5_000;
 
export interface PublisherOptions {
  url?: string;
  logger: Logger;
  now?: () => number;
}
 
export function createMessagePublisher({
  url = process.env["RELAY_REDIS_URL"] ?? DEFAULT_FANOUT_REDIS_URL,
  logger,
  now = () => Date.now(),
}: PublisherOptions): MessagePublisher {
  const redis = new Redis(url, {
    // A queued command rejects as soon as the connection attempt fails, rather
    // than waiting out a retry schedule. On a request path that difference is
    // the whole of NFR-PRF-02's 150 ms budget.
    lazyConnect: true,
    maxRetriesPerRequest: 0,
    connectTimeout: 1_000,
    // A CONNECTED SERVER THAT NEVER ANSWERS IS A DIFFERENT FAILURE, and the two
    // options above do nothing for it. Measured against a TCP listener that
    // accepts and never speaks:
    //
    //     without commandTimeout   publish HUNG past a 3,000 ms bound, twice
    //     with commandTimeout 100  rejected at the timeout, then at 3 ms
    //                              (the second is the down-window, already open)
    //
    // `plan.md`'s post-design re-check named this as the residual risk of
    // awaiting the publish and the contract did not close it. 100 ms is ~440x the
    // measured p95 of 0.226 ms and ~100x the worst sample of 0.965 ms, and it
    // sits inside NFR-PRF-02's 150 ms budget for the whole write — a timeout
    // above that budget could not protect it.
    commandTimeout: 100,
  });
  // A dead fan-out is an expected state, not an exception. Without a listener
  // ioredis emits `error` on an EventEmitter with none attached and Node turns
  // that into an unhandled exception — the api would die for the thing it is
  // designed to survive. `createFanout` has no such listener; that is a gap
  // this chapter records rather than inherits.
  redis.on("error", () => {});
 
  let downUntil = 0;
 
  return {
    async publish(message, context) {
      // A known-down store is not retried on the request path. The first
      // failure opens a window; while it is open every call returns
      // immediately, which is the same outcome the caller already handles.
      if (now() < downUntil) return;
      try {
        await redis.publish(
          subjectForChannel(message.channel),
          JSON.stringify(message),
        );
        downUntil = 0;
      } catch (error) {
        downUntil = now() + DOWN_WINDOW_MS;
        logger.log("error", "fanout.publish_failed", {
          channel: message.channel,
          message_id: message.id,
          request_id: context.requestId,
          environment_id: context.environmentId,
          error: String(error),
        });
      }
    },
    async close() {
      redis.disconnect();
    },
  };
}

import { createLogger, type Logger } from "@relay/service-kit";
import { messageSchema, subjectForChannel } from "@relay/protocol";
import { beforeEach, describe, expect, it, vi } from "vitest";
 
import { createMessagePublisher } from "./publisher";
 
// A fake at the ioredis seam. The publisher's contract is "never rejects", so a
// test that only checks it resolved cannot tell a swallowed failure from a
// success — and cannot tell either from a publisher that does nothing. Every
// assertion below therefore names what it would take to fail.
const publishes: Array<[string, string]> = [];
let throwing = false;
let disconnects = 0;
let errorHandler: ((e: Error) => void) | undefined;
vi.mock("ioredis", () => ({
  Redis: class {
    on(event: string, handler: (e: Error) => void): this {
      if (event === "error") errorHandler = handler;
      return this;
    }
    async publish(subject: string, payload: string): Promise<number> {
      if (throwing) throw new Error("ECONNREFUSED");
      publishes.push([subject, payload]);
      return 1;
    }
    disconnect(): void {
      disconnects += 1;
    }
  },
}));
 
const message = {
  id: "m1",
  channel: "c1",
  seq: 1,
  user: "outside-bot",
  text: "hello",
  created_at: "2026-08-27T00:00:00.000Z",
};
const context = { requestId: "req-1", environmentId: "env-1" };
 
function sink(): { lines: Record<string, unknown>[]; logger: Logger } {
  const lines: Record<string, unknown>[] = [];
  const logger = createLogger("publisher-test", (line) =>
    lines.push(JSON.parse(line) as Record<string, unknown>),
  );
  return { lines, logger };
}
 
beforeEach(() => {
  publishes.length = 0;
  throwing = false;
  disconnects = 0;
  errorHandler = undefined;
});
 
describe("the api's fan-out publisher", () => {
  it("publishes to the channel's subject", async () => {
    const { logger } = sink();
    await createMessagePublisher({ logger }).publish(message, context);
    expect(publishes).toHaveLength(1);
    expect(publishes[0]![0]).toBe(subjectForChannel("c1"));
  });
 
  it("publishes a payload the delivery side will accept", async () => {
    // The far end parses with `messageCreatedSchema.shape.payload`, a
    // `z.strictObject` of six fields, and DROPS what does not match — so an
    // extra key delivers nothing while the send still returns 201. Asserting
    // against the schema catches a seventh field, a missing `user`, a
    // non-positive `seq` and a `created_at` that is not RFC 3339, in one line.
    const { logger } = sink();
    await createMessagePublisher({ logger }).publish(message, context);
    const parsed = messageSchema.safeParse(JSON.parse(publishes[0]![1]));
    expect(parsed.success).toBe(true);
    expect(Object.keys(JSON.parse(publishes[0]![1])).sort()).toEqual([
      "channel",
      "created_at",
      "id",
      "seq",
      "text",
      "user",
    ]);
  });
 
  it("resolves when the client throws, and says so in the log", async () => {
    // What would have to be false for this to fail? That the catch exists. The
    // resolution alone proves nothing — a publisher with no body also resolves
    // — so the log line is the assertion that carries FR-010 and FR-011.
    throwing = true;
    const { lines, logger } = sink();
    await expect(
      createMessagePublisher({ logger }).publish(message, context),
    ).resolves.toBeUndefined();
    expect(lines).toHaveLength(1);
    expect(lines[0]!["msg"]).toBe("fanout.publish_failed");
    expect(lines[0]!["level"]).toBe("error");
    // NFR-OBS-01's two fields, and NFR-OBS-06's five-minute traceability.
    expect(lines[0]!["request_id"]).toBe("req-1");
    expect(lines[0]!["environment_id"]).toBe("env-1");
    expect(lines[0]!["channel"]).toBe("c1");
    expect(lines[0]!["message_id"]).toBe("m1");
  });
 
  it("does not touch the client again inside the down-window", async () => {
    // T009b. The window is what makes a dead Redis cheap rather than merely
    // survivable: without it every send pays the connect timeout, which is
    // `limits/store.ts`'s recorded mistake — "each request paid a second or
    // more, twice".
    //
    // The assertion is that the client is NOT CALLED, not that the publish
    // resolved: it resolves either way, window or no window.
    throwing = true;
    const { lines, logger } = sink();
    let clock = 1_000;
    const p = createMessagePublisher({ logger, now: () => clock });
 
    await p.publish(message, context);
    expect(lines).toHaveLength(1); // the first failure opens the window
 
    clock += 4_999;
    await p.publish(message, context);
    expect(lines).toHaveLength(1); // still one: no attempt, so nothing to log
 
    clock += 2; // 5_001 ms after the failure — the window has closed
    await p.publish(message, context);
    expect(lines).toHaveLength(2);
  });
 
  it("survives an ioredis `error` event instead of dying on it", () => {
    // R10, and the reason this listener exists at all. Without one, ioredis
    // emits `error` on an EventEmitter with no listener and Node turns that
    // into an unhandled exception — the api would die for the thing it is built
    // to survive. `createFanout` in the gateway has no such listener.
    const { lines, logger } = sink();
    createMessagePublisher({ logger });
    expect(errorHandler).toBeTypeOf("function");
    expect(() => errorHandler!(new Error("ECONNREFUSED"))).not.toThrow();
    // Deliberately silent: the failure that matters is a failed PUBLISH, which
    // has its own line. A connection-level error on every retry would be noise.
    expect(lines).toHaveLength(0);
  });
 
  it("falls back to the documented default when RELAY_REDIS_URL is unset", async () => {
    // The coverage pin found this one. Every other test either passes `url` or
    // runs with the lane's env set, so the `??` fallback was never taken — 100%
    // of statements, functions and lines, and 5 of 6 branches. The number that
    // caught it is the one chosen from the requirement rather than from a report.
    const saved = process.env["RELAY_REDIS_URL"];
    delete process.env["RELAY_REDIS_URL"];
    try {
      const { logger } = sink();
      await createMessagePublisher({ logger }).publish(message, context);
      // It published, which means it resolved a URL — and the only URL left is
      // `DEFAULT_FANOUT_REDIS_URL`. The mocked client accepts any.
      expect(publishes).toHaveLength(1);
    } finally {
      if (saved === undefined) delete process.env["RELAY_REDIS_URL"];
      else process.env["RELAY_REDIS_URL"] = saved;
    }
  });
 
  it("disconnects on close", async () => {
    // Not a formality. `limits/limits.module.ts:10` states the api's convention —
    // "resource in this api closes through `OnModuleDestroy`" — and a `close()`
    // that nothing calls is a leaked handle in a service that boots once per
    // integration suite. The coverage pin for this file requires 100% of
    // functions precisely so this cannot go untested.
    const { logger } = sink();
    await createMessagePublisher({ logger }).close();
    expect(disconnects).toBe(1);
  });
 
  it("closes the window after a success", async () => {
    const { lines, logger } = sink();
    let clock = 1_000;
    const p = createMessagePublisher({ logger, now: () => clock });
    throwing = true;
    await p.publish(message, context);
    throwing = false;
    clock += 6_000;
    await p.publish(message, context); // succeeds, clears downUntil
    clock += 1;
    throwing = true;
    await p.publish(message, context); // must attempt, and fail, and log
    expect(lines).toHaveLength(2);
    expect(publishes).toHaveLength(1);
  });
});