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

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

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

@@ -1,19 +1,30 @@
-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
 // the module.exports namespace — which is not constructable. TypeScript
 // says so plainly ("This expression is not constructable"); the fix is to
 // take the named export the package actually provides.
 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 THE FAN-OUT CHAPTER. 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
 // acceptable because it is RECOVERABLE: sequences live in Postgres, cursors
 // live with the client, and 2.7's resume path turns any gap into a
 // backfill. Durability was never this layer's job (constitution IV:
@@ -23,19 +34,12 @@ import { Redis } from "ioredis";
 // connection cannot issue ordinary commands, so publisher and subscriber
 // must be two connections — ioredis models that as two client objects whose
 // lifecycles match the session registry's.
 
 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
    * hand it to. */
   onDelivery(handler: (channelId: string, message: Message) => void): void;
   /** Publish a committed message to its channel's subject. A failure here
@@ -85,13 +89,13 @@ export function createFanout({
     onDelivery(handler) {
       deliver = handler;
     },
     async publish(message) {
       try {
         await publisher.publish(
-          subjectFor(message.channel),
+          subjectForChannel(message.channel),
           JSON.stringify(message),
         );
       } catch (error) {
         // Delivery is allowed to fail; the message is already durable and
         // 2.7's resume will find it. Log and move on.
         logger.log("error", "fanout.publish_failed", {
@@ -100,19 +104,19 @@ 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);
       }
     },
     async close() {
       subscriber.disconnect();

@@ -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 the fan-out chapter — 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,16 +1,61 @@
-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";
 
+/** 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. */
+
+/** THE CONVENTION THIS FOLLOWS: a resource in this api closes through
+ * `OnModuleDestroy`. Two other modules implement it — the outbox relay's and the
+ * consumer's, both of which hold a broker connection — and this is the third, for
+ * the Redis client the publisher holds. 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).
 @Module({
   imports: [AuthModule],
   controllers: [MessagesController],
@@ -38,10 +83,19 @@ import { MessagesService } from "./messages.service";
         // has not proved it may act for. The empty-string fallback is the
         // same as 2.2's: no principal means no scope, and the guard below
         // turns that into a 401 before any handler runs.
         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 {}

@@ -1,21 +1,26 @@
 import {
   BadRequestException,
   Body,
   Controller,
   Get,
+  Inject,
   Param,
   Post,
   Query,
   Req,
   UseGuards,
 } from "@nestjs/common";
 
 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
 // in a decorated signature must be imported as a type or TS1272 refuses
 // to compile it.
 import type { HistoryQuery, SendMessageBody } from "./messages.schema";
@@ -59,12 +64,18 @@ function actingUser(req: RequestWithPrincipal): string | undefined {
 @Accepts("application", "user")
 @UseGuards(CredentialGuard)
 export class MessagesController {
   constructor(
     private readonly messages: MessagesService,
     private readonly repo: Repository,
+    // 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()
   async send(
     @Param("channelId") channelId: string,
     @Body(new ZodValidationPipe(sendMessageBodySchema)) body: SendMessageBody,
@@ -155,12 +166,58 @@ export class MessagesController {
     // wire: the client sees the same body whether this was the original
     // send or the retry that recovered it. Moved down from the service in
     // chapter 2.6, where an internal caller turned out to need the flag.
     // 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 (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,
       seq: message.seq,
       text: message.text,
       created_at: message.created_at,

    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(),
});

    const parsed = messageSchema.safeParse(JSON.parse(raw!));
    expect(parsed.success, JSON.stringify(parsed.error?.issues)).toBe(true);

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", {

      // The weak assertions — both PASS, which is the finding.
      expect((await send(base, text)).status).toBe(201);
      expect(await history(base)).toContain(text);
 
      // The one that carries FR-010 and FR-011 — absent, correctly.
      expect(lines.find((l) => l["msg"] === "fanout.publish_failed")).toBeUndefined();

@@ -16,12 +16,13 @@
     "@nestjs/common": "^11.1.28",
     "@nestjs/core": "^11.1.28",
     "@nestjs/platform-express": "^11.1.28",
     "@relay/protocol": "workspace:*",
     "@relay/service-kit": "workspace:*",
     "drizzle-orm": "^0.45.2",
+    "ioredis": "^6.0.0",
     "jose": "^6.2.7",
     "nats": "^2.29.3",
     "pg": "^8.22.0",
     "reflect-metadata": "^0.2.2",
     "rxjs": "^7.8.2",
     "zod": "^4.4.3"

    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).

@@ -28,12 +28,16 @@ export type PrincipalKind = Principal["kind"];
  * principal is optional at the type level for one honest reason: a request that
  * presented nothing has none, and pre-credential routes (signup) are reached
  * exactly that way. */
 export interface RequestWithPrincipal {
   headers: Record<string, string | string[] | undefined>;
   principal?: Principal;
+  /** 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;
 }
 
 /** How a credential class is named to a human. Used by the wrong-credential
  * error, which must say what was presented and what was expected — and must
  * never quote the credential (NFR-SEC-06). */
 export function describePrincipalKind(kind: PrincipalKind): string {

@@ -14,12 +14,18 @@ import { LOGGER } from "./logger";
 export class RequestContextMiddleware implements NestMiddleware {
   constructor(@Inject(LOGGER) private readonly logger: Logger) {}
 
   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. 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;
     res.on("finish", () => {
       this.logger.log("info", "request", {
         request_id: requestId,
         method: req.method,
         path: req.url,
         status: res.statusCode,

@@ -2,13 +2,13 @@ import { randomUUID } from "node:crypto";
 
 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,
 // same Redis, no knowledge of each other. If a message published by one
 // arrives at the other, the split brain is closed.
 //
@@ -143,10 +143,12 @@ describe("fan-out across instances", () => {
       setTimeout(resolve, 100);
     });
     await expect(nextDelivery(g2, 300)).rejects.toThrow("deadline");
     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 THE FAN-OUT CHAPTER, 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.
 });

@@ -366,12 +366,18 @@ describe("the socket refuses another tenant's identifiers", () => {
   //      `user`, `cursor`, `resume_ok`, `truncated` and no channel list.
   //   2. Sending a message and waiting for `message.created`. **This suite attaches
   //      no fan-out** — `attachSessions({server, api, logger})` passes none, so
   //      `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 this chapter. It 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
   // something.
   it("a removed member's resume cursor is no longer accepted", async () => {
     // A CONTROL AND THE CASE, in that order, on one fixture. Asserting only that a
@@ -617,16 +623,24 @@ describe("the socket refuses another tenant's identifiers", () => {
   // ASSERTED RATHER THAN ASSUMED, because "we did not change the protocol" is the claim a
   // test replaces. A later change that enriched `user` into an object would break every
   // client parsing frames against the published schema.
   //
   // 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.
-  // The isolation harness 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 THIS CHAPTER AND THE FACT DID NOT. This comment used to say
+  // "the api publishes to no fan-out", which was the platform-wide truth the isolation
+  // harness recorded as a finding — a REST-sent message reached no socket, by two
+  // independent mechanisms. The sender chapter removed one and this chapter 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(`${t.apiUrl}/v1/users/${t.victim.userExternalId}`, {
       method: "PATCH",
       headers: {
         "content-type": "application/json",

@@ -226,35 +226,35 @@ describe("a channel, a member and a message, all over the public API", () => {
   //
   // FIXING IT IS A PRODUCT DECISION AND NOT THIS CHAPTER'S. Attributing a public
   // send to an end-user token would change what `user` means on the wire for every
   // 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.
   //
-  // THE SENDER CHAPTER 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. The fan-out chapter 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. The sender chapter 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. This chapter gave
+  // the api a publisher, so the same row now reaches a LIVE socket too. The isolation
+  // harness'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
     // design — it is the one that tests what a customer can reach.
     await post(
       "/v1/users",
       {
         users: [
           {
             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",
           },
         ],
       },
       api.credential,
     );
 
@@ -283,33 +283,56 @@ describe("a channel, a member and a message, all over the public API", () => {
     // WAS `every((m) => m.user === null)`, AND THAT IS THE CHAPTER (
     // T055's class). This assertion existed to prove the rows were senderless, which was
     // why `toFrame` dropped them. Every REST send now names a sender, so the premise it
     // 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 THIS CHAPTER.
+    //
+    // 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.
     //
     // This block asserted `[]`, and the comment said why: "the page came back and every
     // row in it was dropped for having no sender." That was true, and it is the reason
     // the isolation harness'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. The fan-out chapter is that half.
+    // FR-MSG-15 removed the second and this chapter'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));
     const ack = resumed.frames.find((f) => f.type === "connection.ack") as
       | { payload: { cursor: Record<string, number>; resume_ok: boolean } }
       | undefined;

@@ -182,12 +182,50 @@ export default defineConfig({
         "services/api/src/auth/user-token.ts": {
           branches: 96,
           functions: 100,
           lines: 100,
           statements: 96,
         },
+        // CHOSEN BEFORE THE FIRST COVERAGE REPORT, not read off it (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 forces
+        // `close()` and the ioredis `error` listener to be tested rather than assumed.
+        //
+        // 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,
+        },
+        // 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 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,
+        },
       },
     },
   },
   plugins: [
     swc.vite({
       module: { type: "es6" },

@@ -326,6 +326,110 @@ describe("resume across a real fabric", () => {
     await settle(300);
 
     expect(created(frames)).toEqual([41]);
     socket.close();
   });
 });
+
+// ── 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 (the sender chapter'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", () => {
+  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 THE FAN-OUT CHAPTER, and the reason is the same one the broker chapter 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 this chapter
 * 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 the broker chapter'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
// this chapter, 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 (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 are chosen for a request handler and measured against a dead
 *    port, a hung port and a healthy one — each of the three is a different
 *    failure and only one of them is the one people think of.
 *
 * 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.
 *
 * FAILING OPEN IS NOT FREE IF IT FAILS SLOWLY. Bounding the client is not the same
 * as making a dead Redis cheap: with the options alone, every send still pays its
 * timeout before giving up, so a broker that is down costs every request in the
 * outage rather than the first one. Measured on this chapter's own suite — twelve
 * sends at a dead port total 2 ms, and eleven of the twelve never reach the client.
 *
 * The window is the part that buys that, and the first draft of this chapter's
 * contract specified 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, so an outage
    // costs every request in it rather than the first one.
    //
    // 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. A 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);
  });
});