Commit và publish là hai khoảnh khắc

const committed = await api.sendMessage(connection.identity, { … });
…                                    // the ack, then nineteen lines of comment
…
if (!committed.duplicate && committed.text !== null) {
  await fanout?.publish({ id: committed.id, channel: committed.channel_id, seq: committed.seq, … });
}

@@ -192,4 +192,120 @@ describe("resume across a real fabric", () => {
     expect(created(frames)).toEqual([42, 44]);
     socket.close();
   });
+
+  it("suppresses a frame the backfill already delivered, published after the resume", async () => {
+    // THE FOURTH QUADRANT, and the defect this chapter exists to close.
+    //
+    // The three tests above cover: published while buffering and in the backfill
+    // (deduplicated by `flushable`); published while buffering and NOT in the
+    // backfill (delivered by the flush); published after going live with a
+    // sequence ABOVE the mark (delivered). The cell they leave empty is this one —
+    // published after going live, with a sequence the backfill already sent.
+    //
+    // It is not a contrived case. A message is durable at one instant and
+    // announced at another: the gateway commits through the api and only then
+    // publishes to Redis. A backfill query landing between those two instants
+    // returns a message the fabric has not yet delivered, and the delivery arrives
+    // after the resume has finished — when chapter 2.7's dedup window has already
+    // closed, because `marks` was a local variable that `resume()` discarded.
+    //
+    // One number different from the test above it. That is the whole bug.
+    harness = await boot({
+      session: async () => ({
+        environment_id: "env-1",
+        user: "tuan",
+        channel_ids: [CHANNEL],
+      }),
+      backfill: async () => ({
+        [CHANNEL]: { messages: [frame(42)], truncated: false },
+      }),
+      sendMessage: async () => {
+        throw new Error("not used");
+      },
+    });
+    const socket = new WebSocket(
+      `${harness.url}?token=${await token()}&cursor=${CHANNEL}:41`,
+    );
+    const frames = record(socket);
+    await settle(400);
+    // The resume has completed. NOW the fabric catches up with a message the
+    // backfill already delivered — the publish that was still in flight while the
+    // backfill query ran.
+    await publishFromElsewhere(frame(42));
+    await settle(300);
+
+    expect(created(frames)).toEqual([42]);
+    socket.close();
+  });
+
+  it("still suppresses when two instances publish out of order", async () => {
+    // Sequences COMMIT in order under a channel row lock; they are PUBLISHED by
+    // whichever gateway instance handled each send, and those do not coordinate.
+    // A prompt publish of 43 can precede a stalled publish of 42.
+    //
+    // This is the case that made the spec's first design unsafe. It proposed
+    // retiring the mark once a higher sequence arrived — which would see the 43,
+    // drop the mark, and then deliver the 42 (research R3).
+    harness = await boot({
+      session: async () => ({
+        environment_id: "env-1",
+        user: "tuan",
+        channel_ids: [CHANNEL],
+      }),
+      backfill: async () => ({
+        [CHANNEL]: { messages: [frame(42)], truncated: false },
+      }),
+      sendMessage: async () => {
+        throw new Error("not used");
+      },
+    });
+    const socket = new WebSocket(
+      `${harness.url}?token=${await token()}&cursor=${CHANNEL}:41`,
+    );
+    const frames = record(socket);
+    await settle(400);
+    // 43 is ABOVE the mark and must be delivered. A rule that retired the mark on
+    // seeing it would then have nothing left to compare the delayed 42 against.
+    await publishFromElsewhere(frame(43));
+    await settle(150);
+    // 42 is the mark itself, arriving late from an instance that stalled between
+    // its api call and its publish. It must still be suppressed.
+    await publishFromElsewhere(frame(42));
+    await settle(300);
+
+    expect(created(frames)).toEqual([42, 43]);
+    socket.close();
+  });
+
+  it("suppresses nothing when the resume degraded", async () => {
+    // A degraded resume tells the client to page history for every channel, so the
+    // backfill it received is a fragment or nothing at all. A mark taken from it
+    // would suppress messages the client never got — turning this chapter's
+    // duplicate into a gap, which constitution II ranks worse.
+    harness = await boot({
+      session: async () => ({
+        environment_id: "env-1",
+        user: "tuan",
+        channel_ids: [CHANNEL],
+      }),
+      backfill: async () => {
+        throw new Error("backfill unavailable");
+      },
+      sendMessage: async () => {
+        throw new Error("not used");
+      },
+    });
+    const socket = new WebSocket(
+      `${harness.url}?token=${await token()}&cursor=${CHANNEL}:41`,
+    );
+    const frames = record(socket);
+    await settle(400);
+    // A sequence at or below the presented cursor. With no mark retained it must
+    // still arrive: the client was told to page history, not to expect silence.
+    await publishFromElsewhere(frame(41));
+    await settle(300);
+
+    expect(created(frames)).toEqual([41]);
+    socket.close();
+  });
 });

$ pnpm --filter @relay/gateway test:integration src/resume.itest.ts
 × suppresses a frame the backfill already delivered, published after the resume
 × still suppresses when two instances publish out of order
 Tests  2 failed | 4 passed (6)
 
AssertionError: expected [ 42, 42 ] to deeply equal [ 42 ]

@@ -107,6 +107,51 @@ export function flushable(
   return buffer.filter((frame) => frame.seq > (marks[frame.channel] ?? 0));
 }
 
+/** Step 5's half, and the reason this chapter exists (FR-RTM-03).
+ *
+ * `flushable` answers this question for frames sitting in the buffer. This answers
+ * it for frames arriving AFTER the connection has gone live, which is the case
+ * chapter 2.7 did not have — its marks were a local variable that `resume()`
+ * discarded the moment it flushed.
+ *
+ * A message is durable at one instant and announced at another: the gateway
+ * commits through the api and only then publishes to the fabric. A backfill query
+ * landing between those two instants returns a message the fabric has not yet
+ * delivered, so the delivery arrives once the resume is over — and before this
+ * function there was nothing left to compare it against.
+ *
+ * `<=`, not `<`. The mark IS a sequence the backfill delivered rather than the one
+ * after it, which is the same off-by-one `flushable` documents one screen up. */
+export function suppressed(
+  marks: Record<string, number> | null,
+  frame: Message,
+): boolean {
+  if (marks === null) return false;
+  const mark = marks[frame.channel];
+  return mark !== undefined && frame.seq <= mark;
+}
+
+/** The marks a connection keeps, bounded to the channels it actually presented
+ * cursors for.
+ *
+ * `highWaterMarks` seeds from the cursors and then adds a key for every channel
+ * the backfill answered with, so on its own it bounds nothing. The api derives its
+ * response from the cursors it was given, so the two agree today — but a bound
+ * this service claims should not live in another service's response shape. This is
+ * `scopeCursors` applied one step later, and it is here rather than in `session.ts`
+ * so that a unit test can reach it.
+ *
+ * At most `MAX_RESUME_CHANNELS` entries, because that is what the resume contract
+ * already caps the cursor map at. */
+export function scopeMarks(
+  marks: Record<string, number>,
+  cursors: Record<string, number>,
+): Record<string, number> {
+  return Object.fromEntries(
+    Object.entries(marks).filter(([channelId]) => channelId in cursors),
+  );
+}
+
 /** A promise that resolves false instead of hanging forever. */
 export async function withDeadline(
   work: Promise<unknown>,

@@ -29,6 +29,24 @@ export interface Connection {
   /** Set when the buffer hit its ceiling. The frames are gone, so the
    * client must be told to page history instead of trusting the stream. */
   overflowed: boolean;
+  /** Chapter 3.7. Per channel, the highest sequence this connection's backfill
+   * delivered — kept for the connection's life so that a frame the fabric
+   * announces AFTER the resume has finished can still be recognised as one the
+   * client already holds.
+   *
+   * Three states, and the middle one is not the same as the first:
+   *
+   *   null            a fresh connect, or a resume that degraded — suppress nothing
+   *   {}              a resume whose cursor set was empty after scoping
+   *   { chan: 42 }    suppress at or below 42 on that channel
+   *
+   * NEVER RETIRED while the connection lives. Retiring on a higher sequence looks
+   * like the natural way to bound this and hands the duplicate straight back:
+   * sequences commit in order under a channel row lock but are published by
+   * whichever gateway instance handled each send, so a prompt 43 can beat a stalled
+   * 42. Bounded instead by `MAX_RESUME_CHANNELS`, which already caps the cursors
+   * these are scoped to. */
+  marks: Record<string, number> | null;
 }
 
 export class Registry {

@@ -19,6 +19,8 @@ import {
   SUBSCRIBE_DEADLINE_MS,
   flushable,
   highWaterMarks,
+  scopeMarks,
+  suppressed,
   parseCursors,
   scopeCursors,
   withDeadline,
@@ -94,6 +96,13 @@ export function attachSessions({
         connection.buffer.push(message);
         continue;
       }
+      // Chapter 3.7. A live connection is not necessarily a connection with
+      // nothing to remember: a frame at or below what its backfill already
+      // delivered is one it has, however long ago the resume finished. Before
+      // this, delivery consulted `phase` and nothing else, and the marks were
+      // discarded the moment the connection went live — which is precisely when
+      // the fabric could still be catching up.
+      if (suppressed(connection.marks, message)) continue;
       send(connection.socket, { type: "message.created", payload: message });
     }
   }
@@ -161,6 +170,9 @@ export function attachSessions({
       phase: presented === undefined ? "live" : "buffering",
       buffer: [],
       overflowed: false,
+      // A fresh connect suppresses nothing; a resume fills this in when it
+      // succeeds, and leaves it null when it degrades.
+      marks: null,
     };
 
     registry.add(connection);
@@ -263,6 +275,11 @@ export function attachSessions({
      * fragment of a stream the client is about to refetch in full. */
     const degrade = (reason: string): void => {
       connection.buffer = [];
+      // AND THE MARKS. A degraded resume tells the client to page history for
+      // every channel, so the backfill behind these marks is a fragment or
+      // nothing at all — suppressing on them would turn this chapter's duplicate
+      // into a gap, which constitution II ranks worse.
+      connection.marks = null;
       connection.phase = "live";
       ack(connection, {
         cursor: cursors,
@@ -337,6 +354,10 @@ export function attachSessions({
       send(connection.socket, { type: "message.created", payload: message });
     }
     connection.buffer = [];
+    // KEPT, where chapter 2.7 discarded them. Scoped to the cursors this
+    // connection actually presented, so the bound is this service's rather than
+    // one inherited from the shape of the api's response.
+    connection.marks = scopeMarks(marks, cursors);
     // Step 5.
     connection.phase = "live";
     logger.log("info", "resume.completed", {

resume completes, mark = 4
seq 5 arrives  → above the mark → deliver, and RETIRE
seq 4 arrives  → no mark left   → deliver   ← the duplicate, again

@@ -372,7 +372,15 @@ export const consumedEvents = pgTable(
 // the platform's own bookkeeping AND no tenant-visible content. An endpoint is
 // customer configuration; a dead letter holds a payload that was being sent to a
 // customer. Both fail the test on both halves, so both are scoped and both join
-// chapter 3.7's cross-tenant gauntlet as targets.
+// the cross-tenant gauntlet as targets.
+//
+// NAMED, NOT NUMBERED. This line used to say "chapter 3.7's cross-tenant
+// gauntlet". The gauntlet was 3.7 when that was written, became 3.8 when a chapter
+// was inserted ahead of it, and is now 3.9 after a second insertion — and the
+// comment was carried neither time. A chapter number in a source comment is a
+// reference that ages every time the plan changes, and this file is fenced
+// byte-exact into a published chapter, so correcting it costs a fence amendment.
+// The subject does not move; the ordinal does.
 // ---------------------------------------------------------------------------
 
 // DECISION (chapter 3.5): no source document defines this table. FR-WHK-01 and
@@ -593,8 +601,9 @@ export const webhookDeadLetters = pgTable(
 // `deliveredAt` is the honest column, and it exists in this chapter solely in
 // order to be null. FR-WHK-07 asks for the endpoint to be disabled "and the
 // organisation notified by email", and this platform has no email transport of
-// any kind. Chapter 3.7 needs the same transport for quotas, so building one here
-// would mean building it for its second consumer first.
+// any kind. A later chapter needs the same transport for quotas, so building one
+// here would mean building it for its second consumer first. (Named rather than
+// numbered: see the note on the dead-letter table above.)
 //
 // A schema that recorded only the disablement would let a future reader believe
 // the requirement was finished. This one says, in a column, which half is

@@ -450,7 +450,7 @@ if (WATCH_DISABLE) {
     console.log("  `delivered_at` is null and stays null. FR-WHK-07 asks for the");
     console.log("  organisation to be notified BY EMAIL, and this platform has no email");
     console.log("  transport of any kind. The row is the obligation; the null is the");
-    console.log("  admission. Chapter 3.7 needs the same transport for quotas.\n");
+    console.log("  admission. A later chapter needs the same transport for quotas.\n");
 
     // Running it again must change nothing. At most once per run, enforced by the
     // `enabled = true` predicate in the update rather than by a check.

@@ -7,6 +7,8 @@ import {
   highWaterMarks,
   parseCursors,
   scopeCursors,
+  scopeMarks,
+  suppressed,
   withDeadline,
 } from "./resume.js";
 
@@ -134,3 +136,85 @@ describe("withDeadline", () => {
     );
   });
 });
+
+// Chapter 3.7. `flushable` decides what a resuming connection may hand over from
+// its buffer; `suppressed` decides what a LIVE connection must still refuse. The
+// two comparisons are the same and the second one did not exist, which is the
+// whole of the defect.
+describe("suppressed", () => {
+  const at = (channel: string, seq: number): Message => ({
+    id: `id-${seq}`,
+    channel,
+    seq,
+    user: "dispatcher",
+    text: `m${seq}`,
+    created_at: "2026-08-19T00:00:00.000Z",
+  });
+
+  it("suppresses a frame at the mark", () => {
+    // `<=`, not `<`: the mark IS a sequence the backfill delivered.
+    expect(suppressed({ a: 42 }, at("a", 42))).toBe(true);
+  });
+
+  it("suppresses a frame below the mark", () => {
+    expect(suppressed({ a: 42 }, at("a", 7))).toBe(true);
+  });
+
+  it("delivers a frame above the mark", () => {
+    // The half that matters most: suppression must never become a gap.
+    expect(suppressed({ a: 42 }, at("a", 43))).toBe(false);
+  });
+
+  it("delivers on a channel with no mark", () => {
+    expect(suppressed({ a: 42 }, at("b", 1))).toBe(false);
+  });
+
+  it("suppresses nothing when the connection never resumed", () => {
+    expect(suppressed(null, at("a", 1))).toBe(false);
+  });
+
+  it("suppresses nothing when the mark set is empty", () => {
+    expect(suppressed({}, at("a", 1))).toBe(false);
+  });
+
+  it("keeps channels apart", () => {
+    // A mark on one channel must never silence another. Two connections' worth of
+    // confusion would look exactly like message loss.
+    const marks = { a: 42, b: 1 };
+    expect(suppressed(marks, at("a", 40))).toBe(true);
+    expect(suppressed(marks, at("b", 40))).toBe(false);
+  });
+});
+
+describe("scopeMarks", () => {
+  it("drops a channel the cursors never named", () => {
+    // `highWaterMarks` adds a key for every channel the BACKFILL answered with.
+    // The api derives its response from the cursors it was given, so this cannot
+    // happen today — and a bound this service claims should not depend on another
+    // service's response shape.
+    expect(scopeMarks({ a: 42, surprise: 9 }, { a: 41 })).toEqual({ a: 42 });
+  });
+
+  it("keeps every channel the cursors did name", () => {
+    expect(scopeMarks({ a: 42, b: 7 }, { a: 41, b: 1 })).toEqual({ a: 42, b: 7 });
+  });
+
+  it("never exceeds the cursor set, which the resume contract already caps", () => {
+    const cursors = Object.fromEntries(
+      Array.from({ length: 200 }, (_, i) => [`c${i}`, i]),
+    );
+    const marks = { ...cursors, extra: 1 };
+    expect(Object.keys(scopeMarks(marks, cursors))).toHaveLength(200);
+  });
+
+  it("keeps a cursor far above anything the channel holds", () => {
+    // A client-supplied cursor is checked only for being a non-negative integer.
+    // A nonsense value seeds a mark rather than being clamped, and that is the
+    // documented consequence rather than a defect: the client asserted it holds
+    // those sequences, and the backfill has always taken that at face value.
+    // The blast radius is one connection and reconnecting recovers it.
+    expect(scopeMarks({ a: 999_999_999 }, { a: 999_999_999 })).toEqual({
+      a: 999_999_999,
+    });
+  });
+});

@@ -467,6 +467,73 @@ describe("the socket (chapter 2.5)", () => {
     socket.close();
   });
 
+  it("chapter 3.7: a frame at the mark, arriving after the resume, is not delivered", async () => {
+    // The same property `resume.itest.ts` proves against a real broker, held here
+    // against a stubbed one so it fails fast and without Redis. This is the wiring
+    // rather than the predicate: that `deliver()` consults the marks at all.
+    const fanout = stubFanout();
+    harness = await boot(
+      stubApi({
+        backfill: async () => ({
+          [CHANNEL]: { messages: [frame(42)], truncated: false },
+        }),
+      }),
+      undefined,
+      fanout,
+    );
+    const socket = new WebSocket(
+      `${harness.url}?token=${await token()}&cursor=${CHANNEL}:41`,
+    );
+    const frames = record(socket);
+    await nextFrame(socket, "connection.ack");
+    await settle();
+    // The resume is over. The fabric catches up with what the backfill sent.
+    fanout.emit(frame(42));
+    await settle();
+    expect(created(frames)).toEqual([42]);
+    socket.close();
+  });
+
+  it("chapter 3.7: a frame above the mark, arriving after the resume, IS delivered", async () => {
+    // The half that stops a duplicate fix becoming a gap (FR-RTM-03 is one
+    // property, not two: no gap AND no double).
+    const fanout = stubFanout();
+    harness = await boot(
+      stubApi({
+        backfill: async () => ({
+          [CHANNEL]: { messages: [frame(42)], truncated: false },
+        }),
+      }),
+      undefined,
+      fanout,
+    );
+    const socket = new WebSocket(
+      `${harness.url}?token=${await token()}&cursor=${CHANNEL}:41`,
+    );
+    const frames = record(socket);
+    await nextFrame(socket, "connection.ack");
+    await settle();
+    fanout.emit(frame(43));
+    await settle();
+    expect(created(frames)).toEqual([42, 43]);
+    socket.close();
+  });
+
+  it("chapter 3.7: a connection that never resumed suppresses nothing", async () => {
+    // A fresh connect presents no cursor, so it holds no marks and behaves
+    // exactly as chapter 2.6 left it.
+    const fanout = stubFanout();
+    harness = await boot(stubApi({}), undefined, fanout);
+    const socket = new WebSocket(`${harness.url}?token=${await token()}`);
+    const frames = record(socket);
+    await nextFrame(socket, "connection.ack");
+    await settle();
+    fanout.emit(frame(1));
+    await settle();
+    expect(created(frames)).toEqual([1]);
+    socket.close();
+  });
+
   it("stages the other interleaving: published during backfill, absent from it", async () => {
     // Same window, the other order: seq 43 committed AFTER the backfill
     // query's snapshot, so it exists ONLY in the buffer. This is the

@@ -116,8 +116,18 @@ export default defineConfig({
           lines: 100,
           statements: 100,
         },
+        // CHAPTER 3.7 RAISED THIS, 93 -> 95. The chapter added two pure functions
+        // to this file — the live-path suppression predicate and the scoping that
+        // bounds the marks — and both are fully covered.
+        //
+        // Not 100, and the missing branch is named rather than chased: it is
+        // `if (timer)` in chapter 2.7's `withDeadline`, whose falsy arm cannot be
+        // reached because a Promise executor runs synchronously and always assigns
+        // the timer before the `finally` can see it. Pinning 100 here would pin a
+        // number the file cannot reach without deleting a defensive check that
+        // costs nothing.
         "services/gateway/src/resume.ts": {
-          branches: 93,
+          branches: 95,
           functions: 100,
           lines: 100,
           statements: 100,

=============== mutation 1: never keep the marks after a successful resume
  RESULT: caught  × suppresses a frame the backfill already delivered…
=============== mutation 2: use < instead of <= in the predicate
  RESULT: caught  × suppresses a frame the backfill already delivered…
=============== mutation 3: suppress on any channel rather than the frame's own
  RESULT: caught  × delivers on a channel with no mark
                  × keeps channels apart
=============== mutation 4: ignore the cursors when scoping the marks
  RESULT: caught  × drops a channel the cursors never named
=============== mutation 5: retire a mark once a higher sequence is delivered
  RESULT: caught  × still suppresses when two instances publish out of order
=============== files restored byte-identical?
  YES