fix(webapp): realtime subscriptions racing stream creation

Fix intermittent failures when subscribing to newly created realtime
streams. Retry stream-specific not-found responses within the existing
retry limits, preserve resume positions and cancellation, and avoid
treating configuration or authentication errors as empty streams.

Mono-RevId: b373bf50250ab8f79cb447841929c957158c62ef

Author: carderne

.server-changes/fix-stream-subscription-creation-race.md

@@ -0,0 +1,6 @@
+---
+area: webapp
+type: fix
+---
+
+Fix intermittent subscription failures and missed records when reading newly created realtime streams.

apps/webapp/app/services/realtime/s2realtimeStreams.server.ts

@@ -344,7 +344,7 @@ export class S2RealtimeStreams implements StreamResponder, StreamIngestor {
 
   /**
    * Non-blocking read of the records after `afterSeqNum`. Retries transient
-   * failures (network errors, creation conflicts, 5xx, and reads S2 ends with an `error` event)
+   * failures (network errors, stream creation races, 5xx, and reads S2 ends with an `error` event)
    * because the `.wait()` race check depends on this read to find a record
    * sent before the waitpoint registered.
    */
@@ -356,9 +356,18 @@ export class S2RealtimeStreams implements StreamResponder, StreamIngestor {
       try {
         return await this.#readRecordsOnce(s2Stream, afterSeqNum);
       } catch (error) {
-        const status = (error as { status?: number }).status;
+        const { status, streamNotFound } = error as {
+          status?: number;
+          streamNotFound?: boolean;
+        };
+        // A stream that is still absent after retries has no records yet.
+        if (streamNotFound && attempt + 1 >= maxAttempts) return [];
         const isClientError =
-          status !== undefined && status >= 400 && status < 500 && status !== 409;
+          status !== undefined &&
+          status >= 400 &&
+          status < 500 &&
+          status !== 409 &&
+          !streamNotFound;
         if (isClientError || attempt + 1 >= maxAttempts) {
           this.logger.error("S2 readRecords failed", {
             stream: s2Stream,
@@ -401,13 +410,11 @@ export class S2RealtimeStreams implements StreamResponder, StreamIngestor {
     );
 
     if (!res.ok) {
-      // Stream may not exist yet (no data sent)
-      if (res.status === 404) {
-        return [];
-      }
+      const streamNotFound = await isS2StreamNotFound(res);
       await cancelResponseBody(res);
       throw Object.assign(new Error(`S2 readRecords failed: ${res.status} ${res.statusText}`), {
         status: res.status,
+        streamNotFound,
       });
     }
 
@@ -808,7 +815,8 @@ export class S2RealtimeStreams implements StreamResponder, StreamIngestor {
     if (opts.wait != null) qs.set("wait", String(opts.wait));
 
     // Concurrent first readers/writers can race S2's automatic stream creation.
-    // Retry only the conflict, before exposing any SSE bytes to the caller.
+    // S2-lite can also return stream_not_found during this race. Other 404s
+    // (missing basin, token or route) are terminal. Retry before exposing SSE bytes.
     const backoffsMs = [100, 200, 400];
     for (let attempt = 0; ; attempt++) {
       const res = await fetch(
@@ -826,9 +834,11 @@ export class S2RealtimeStreams implements StreamResponder, StreamIngestor {
       );
 
       if (!res.ok) {
+        const streamNotFound = await isS2StreamNotFound(res);
         await cancelResponseBody(res);
+        opts.signal?.throwIfAborted();
         const delay = backoffsMs[attempt];
-        if (res.status === 409 && delay !== undefined) {
+        if ((res.status === 409 || streamNotFound) && delay !== undefined) {
           await sleep(delay, undefined, { signal: opts.signal });
           continue;
         }
@@ -856,6 +866,21 @@ export class S2RealtimeStreams implements StreamResponder, StreamIngestor {
   }
 }
 
+async function isS2StreamNotFound(response: Response): Promise<boolean> {
+  if (response.status !== 404) return false;
+  try {
+    const error: unknown = await response.json();
+    return (
+      typeof error === "object" &&
+      error !== null &&
+      "code" in error &&
+      error.code === "stream_not_found"
+    );
+  } catch {
+    return false;
+  }
+}
+
 async function cancelResponseBody(response: Response): Promise<void> {
   try {
     await response.body?.cancel();

apps/webapp/test/helpers/s2FaultProxy.ts

@@ -18,6 +18,7 @@ type Fault = {
   message: string;
   remaining: number;
   status?: number;
+  code?: string;
 };
 
 /**
@@ -77,8 +78,8 @@ export class S2FaultProxy {
   }
 
   /** Reject before SSE headers, as when concurrent first reads race to create a stream. */
-  rejectNextRead(match: string, status: number, times = 1) {
-    this.faults.push({ match, status, remaining: times, afterBatches: 0, message: "" });
+  rejectNextRead(match: string, status: number, times = 1, code?: string) {
+    this.faults.push({ match, status, code, remaining: times, afterBatches: 0, message: "" });
   }
 
   private takeFault(req: IncomingMessage): Fault | undefined {
@@ -112,7 +113,12 @@ export class S2FaultProxy {
 
     const fault = this.takeFault(req);
     if (fault?.status) {
-      res.writeHead(fault.status).end();
+      if (fault.code) {
+        res.writeHead(fault.status, { "Content-Type": "application/json" });
+        res.end(JSON.stringify({ code: fault.code }));
+      } else {
+        res.writeHead(fault.status).end();
+      }
       return;
     }
     const upstream = await fetch(`${target}${req.url}`, {

apps/webapp/test/s2StreamSubscription.e2e.test.ts

@@ -5,6 +5,20 @@ import { createS2Container, type StartedS2Container } from "@internal/testcontai
 import { S2RealtimeStreams } from "~/services/realtime/s2realtimeStreams.server";
 import { S2FaultProxy } from "./helpers/s2FaultProxy";
 
+const creationErrors = [
+  { status: 409, code: undefined },
+  { status: 404, code: "stream_not_found" },
+];
+
+const terminalErrors = [
+  { status: 401, code: "authn" },
+  { status: 403, code: "permission_denied" },
+  { status: 404, code: "basin_not_found" },
+  { status: 404, code: "access_token_not_found" },
+  { status: 404, code: undefined },
+  { status: 404, code: "unknown" },
+];
+
 describe("S2 subscription handshake", () => {
   let s2: StartedS2Container;
   let streams: S2RealtimeStreams;
@@ -37,50 +51,60 @@ describe("S2 subscription handshake", () => {
     );
   }
 
-  it("retries creation conflicts without losing the requested resume position", async () => {
-    const session = randomUUID();
-    await streams.appendPartToSessionStream("first", "p0", session, "out");
-    await streams.appendPartToSessionStream("second", "p1", session, "out");
-    proxy.rejectNextRead(session, 409, 2);
-
-    const response = await subscribe(session, undefined, "0");
-    const body = await response.text();
-    expect(response.status).toBe(200);
-    expect(response.headers.get("X-Stream-Version")).toBe("v2");
-    expect(body).toContain("second");
-    expect(body).not.toContain("first");
-    expect(proxy.injected.filter((path) => path.includes(session))).toHaveLength(2);
-  });
+  it.each(creationErrors)(
+    "retries HTTP $status ($code) without losing the requested resume position",
+    async ({ status, code }) => {
+      const session = randomUUID();
+      await streams.appendPartToSessionStream("first", "p0", session, "out");
+      await streams.appendPartToSessionStream("second", "p1", session, "out");
+      proxy.rejectNextRead(session, status, 2, code);
 
-  it("fails after four attempts if the conflict persists", async () => {
-    const session = randomUUID();
-    proxy.rejectNextRead(session, 409, 10);
-    await expect(subscribe(session)).rejects.toThrow("S2 stream failed: 409");
-    expect(proxy.injected.filter((path) => path.includes(session))).toHaveLength(4);
-  });
+      const response = await subscribe(session, undefined, "0");
+      const body = await response.text();
+      expect(response.status).toBe(200);
+      expect(response.headers.get("X-Stream-Version")).toBe("v2");
+      expect(body).toContain("second");
+      expect(body).not.toContain("first");
+      expect(proxy.injected.filter((path) => path.includes(session))).toHaveLength(2);
+    }
+  );
 
-  it("retries non-blocking read conflicts without losing the requested resume position", async () => {
-    const session = randomUUID();
-    await streams.appendPartToSessionStream("first", "p0", session, "out");
-    await streams.appendPartToSessionStream("second", "p1", session, "out");
-    proxy.rejectNextRead(session, 409, 2);
+  it.each(creationErrors)(
+    "fails after four attempts if HTTP $status ($code) persists",
+    async ({ status, code }) => {
+      const session = randomUUID();
+      proxy.rejectNextRead(session, status, 10, code);
+      await expect(subscribe(session)).rejects.toThrow(`S2 stream failed: ${status}`);
+      expect(proxy.injected.filter((path) => path.includes(session))).toHaveLength(4);
+    }
+  );
 
-    const records = await streams.readSessionStreamRecords(session, "out", 0);
+  it.each(creationErrors)(
+    "retries non-blocking HTTP $status ($code) without losing the requested resume position",
+    async ({ status, code }) => {
+      const session = randomUUID();
+      const channel = "updates";
+      await streams.appendPartToSessionStream("first", "p0", session, "out", channel);
+      await streams.appendPartToSessionStream("second", "p1", session, "out", channel);
+      proxy.rejectNextRead(session, status, 2, code);
 
-    expect(records.map(({ data, id, seqNum }) => ({ data, id, seqNum }))).toEqual([
-      { data: "second", id: "p1", seqNum: 1 },
-    ]);
-    expect(proxy.injected.filter((path) => path.includes(session))).toHaveLength(2);
-  });
+      const records = await streams.readSessionStreamRecords(session, "out", 0, channel);
+
+      expect(records.map(({ data, id, seqNum }) => ({ data, id, seqNum }))).toEqual([
+        { data: "second", id: "p1", seqNum: 1 },
+      ]);
+      expect(proxy.injected.filter((path) => path.includes(session))).toHaveLength(2);
+    }
+  );
 
   it.each([
-    { status: 409, attempts: 3 },
-    { status: 403, attempts: 1 },
+    { status: 409, attempts: 3, code: undefined },
+    ...terminalErrors.map((error) => ({ ...error, attempts: 1 })),
   ])(
-    "bounds non-blocking reads to $attempts attempts for HTTP $status",
-    async ({ status, attempts }) => {
+    "bounds non-blocking reads to $attempts attempts for HTTP $status ($code)",
+    async ({ status, attempts, code }) => {
       const run = randomUUID();
-      proxy.rejectNextRead(run, status, 10);
+      proxy.rejectNextRead(run, status, 10, code);
 
       await expect(streams.readRecords(run, "input")).rejects.toThrow(
         `S2 readRecords failed: ${status}`
@@ -89,6 +113,14 @@ describe("S2 subscription handshake", () => {
     }
   );
 
+  it("returns no records only after bounded retries for a missing stream", async () => {
+    const session = randomUUID();
+    proxy.rejectNextRead(session, 404, 10, "stream_not_found");
+
+    expect(await streams.readSessionStreamRecords(session, "in")).toEqual([]);
+    expect(proxy.injected.filter((path) => path.includes(session))).toHaveLength(3);
+  });
+
   it("opens concurrent subscriptions to previously unwritten streams", async () => {
     for (let round = 0; round < 8; round++) {
       const session = randomUUID();
@@ -103,24 +135,27 @@ describe("S2 subscription handshake", () => {
     }
   });
 
-  it("does not retry other HTTP errors", async () => {
+  it.each(terminalErrors)("does not retry HTTP $status ($code)", async ({ status, code }) => {
     const session = randomUUID();
-    proxy.rejectNextRead(session, 403, 10);
-    await expect(subscribe(session)).rejects.toThrow("S2 stream failed: 403");
+    proxy.rejectNextRead(session, status, 10, code);
+    await expect(subscribe(session)).rejects.toThrow(`S2 stream failed: ${status}`);
     expect(proxy.injected.filter((path) => path.includes(session))).toHaveLength(1);
   });
 
-  it("stops retrying when the subscriber disconnects during backoff", async () => {
-    const session = randomUUID();
-    const abort = new AbortController();
-    proxy.rejectNextRead(session, 409, 10);
-    const result = subscribe(session, abort.signal).catch((error: unknown) => error);
-    await expect.poll(() => proxy.injected.some((path) => path.includes(session))).toBe(true);
-    abort.abort();
-    expect(await result).toMatchObject({ name: "AbortError" });
-    const attempts = proxy.injected.filter((path) => path.includes(session)).length;
-    // Prove no new request is sent after cancellation, across the first retry delay.
-    await setTimeout(150);
-    expect(proxy.injected.filter((path) => path.includes(session))).toHaveLength(attempts);
-  });
+  it.each(creationErrors)(
+    "stops retrying HTTP $status ($code) when the subscriber disconnects during backoff",
+    async ({ status, code }) => {
+      const session = randomUUID();
+      const abort = new AbortController();
+      proxy.rejectNextRead(session, status, 10, code);
+      const result = subscribe(session, abort.signal).catch((error: unknown) => error);
+      await expect.poll(() => proxy.injected.some((path) => path.includes(session))).toBe(true);
+      abort.abort();
+      expect(await result).toMatchObject({ name: "AbortError" });
+      const attempts = proxy.injected.filter((path) => path.includes(session)).length;
+      // Prove no new request is sent after cancellation, across the first retry delay.
+      await setTimeout(150);
+      expect(proxy.injected.filter((path) => path.includes(session))).toHaveLength(attempts);
+    }
+  );
 });