fix(webapp): keep useRealtimeRun stream open across the buffered window

Customers subscribing to a freshly-triggered run via useRealtimeRun
silently hung when the gate diverted the run into the mollifier buffer.
The route's findResource looked up the PG TaskRun by friendlyId, found
nothing, and returned 404. Electric SQL's ShapeStream treats the
initial 404 as terminal — no retry, no error surfaced to the hook,
and crucially no recovery after the drainer eventually INSERTed the
PG row. The customer's component shows the empty state indefinitely
even though the run is alive and progressing.

When the PG lookup misses but the buffer has the run, return a
synthetic resource whose `id` is derived from the friendlyId — the
same value engine.trigger will write when the drainer materialises
this run. The route then opens the Electric subscription against
`WHERE id='<id>'`, Electric streams an empty initial snapshot, and
the SDK long-polls until the drainer's INSERT propagates through.
Empirically validated end-to-end: trigger a buffered run, open the
subscription, simulate the drainer's PG INSERT + UPDATE, and the
SDK iterator yields the QUEUED and EXECUTING events in real time.

Adds a `mollifier.realtime_subscriptions.buffered` counter and a
structured log line. The observability gate fires once per cold
subscription (Electric's `handle` query param is the dedup signal),
not on every ~20s long-poll reconnect; that gate is unit-tested.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: d-cs

.server-changes/mollifier-realtime-buffered-subscription.md

@@ -0,0 +1,6 @@
+---
+area: webapp
+type: fix
+---
+
+`useRealtimeRun` / `subscribeToRun` previously hung silently when the run was still in the mollifier buffer: the realtime route returned 404, Electric's `ShapeStream` stopped on the first response, and the hook never recovered even after the drainer materialised the run. Open the Electric shape stream against a synthetic resource derived from the buffer entry instead — the stream returns an empty initial snapshot and streams the `INSERT` to the client when the drainer creates the PG row. Adds a `mollifier.realtime_subscriptions.buffered` counter and a structured log line on the initial connect for visibility into how often customers subscribe inside the buffered window.

apps/webapp/app/routes/realtime.v1.runs.$runId.ts

@@ -1,4 +1,3 @@
-import { json } from "@remix-run/server-runtime";
 import { z } from "zod";
 import { $replica } from "~/db.server";
 import { getRequestAbortSignal } from "~/services/httpAsyncStorage.server";
@@ -7,6 +6,12 @@ import {
   anyResource,
   createLoaderApiRoute,
 } from "~/services/routeBuilders/apiBuilder.server";
+import { logger } from "~/services/logger.server";
+import { findRunByIdWithMollifierFallback } from "~/v3/mollifier/readFallback.server";
+import {
+  isInitialBufferedSubscriptionRequest,
+  recordRealtimeBufferedSubscription,
+} from "~/v3/mollifier/mollifierTelemetry.server";
 
 const ParamsSchema = z.object({
   runId: z.string(),
@@ -18,7 +23,7 @@ export const loader = createLoaderApiRoute(
     allowJWT: true,
     corsStrategy: "all",
     findResource: async (params, authentication) => {
-      return $replica.taskRun.findFirst({
+      const pgRun = await $replica.taskRun.findFirst({
         where: {
           friendlyId: params.runId,
           runtimeEnvironmentId: authentication.environment.id,
@@ -31,6 +36,37 @@ export const loader = createLoaderApiRoute(
           },
         },
       });
+      if (pgRun) return pgRun;
+
+      // Buffered fallback. If the run is sitting in the mollifier buffer
+      // (no PG row yet), open the Electric subscription anyway: the
+      // shape stream returns an empty initial snapshot, and when the
+      // drainer INSERTs the PG row Electric streams it to the client.
+      // Without this branch the route 404s, ShapeStream stops on the
+      // first response, and the hook silently hangs even after the run
+      // materialises (no auto-recovery).
+      const synthetic = await findRunByIdWithMollifierFallback({
+        runId: params.runId,
+        environmentId: authentication.environment.id,
+        organizationId: authentication.environment.organizationId,
+      });
+      if (!synthetic) return null;
+
+      // Shape findResource expects: friendlyId, taskIdentifier, runTags,
+      // batch (for authorization), and id (for streamRun's WHERE clause).
+      // The synthetic.id is derived from friendlyId via RunId — the same
+      // value engine.trigger will write when the drainer materialises
+      // this run, so the Electric subscription matches on INSERT.
+      // `__bufferedDwellMs` flags this resource as buffer-sourced for
+      // the loader body's observability hook below.
+      return {
+        id: synthetic.id,
+        friendlyId: synthetic.friendlyId,
+        taskIdentifier: synthetic.taskIdentifier ?? "",
+        runTags: synthetic.runTags,
+        batch: null as { friendlyId: string } | null,
+        __bufferedDwellMs: Date.now() - synthetic.createdAt.getTime(),
+      };
     },
     authorization: {
       action: "read",
@@ -48,6 +84,22 @@ export const loader = createLoaderApiRoute(
     },
   },
   async ({ authentication, request, resource: run, apiVersion }) => {
+    // Observability for buffered-window subscriptions. The gate keeps
+    // the counter at one tick per subscription instead of one tick per
+    // ~20s live-poll iteration (see `isInitialBufferedSubscriptionRequest`).
+    const bufferedDwellMs = (run as { __bufferedDwellMs?: number }).__bufferedDwellMs;
+    if (
+      typeof bufferedDwellMs === "number" &&
+      isInitialBufferedSubscriptionRequest(request.url)
+    ) {
+      recordRealtimeBufferedSubscription(authentication.environment.id);
+      logger.info("mollifier.realtime.buffered_subscription", {
+        runId: run.friendlyId,
+        envId: authentication.environment.id,
+        bufferDwellMs: bufferedDwellMs,
+      });
+    }
+
     return realtimeClient.streamRun(
       request.url,
       authentication.environment,

apps/webapp/app/v3/mollifier/mollifierTelemetry.server.ts

@@ -15,3 +15,30 @@ export function recordDecision(outcome: DecisionOutcome, reason?: DecisionReason
     ...(reason ? { reason } : {}),
   });
 }
+
+// Counts subscriptions hitting `/realtime/v1/runs/<id>` for a run that
+// lives only in the mollifier buffer (no PG row yet). The route opens
+// the Electric stream anyway so the eventual drainer-INSERT propagates
+// to the client; this counter is the signal of how often customers
+// subscribe inside the buffered window.
+export const realtimeBufferedSubscriptionsCounter = meter.createCounter(
+  "mollifier.realtime_subscriptions.buffered",
+  {
+    description:
+      "Realtime subscriptions opened against a runId that exists only in the mollifier buffer",
+  },
+);
+
+export function recordRealtimeBufferedSubscription(envId: string): void {
+  realtimeBufferedSubscriptionsCounter.add(1, { envId });
+}
+
+// Electric SQL's shape-stream protocol adds a `handle=` query param on
+// every reconnect after the initial GET. Gating the realtime-buffered
+// log/counter on its absence keeps the signal at one tick per
+// subscription instead of one tick per ~20s live-poll iteration —
+// without it the counter would over-count by the long-poll factor.
+export function isInitialBufferedSubscriptionRequest(url: string | URL): boolean {
+  const u = typeof url === "string" ? new URL(url) : url;
+  return !u.searchParams.has("handle");
+}

apps/webapp/test/mollifierRealtimeSubscription.test.ts

@@ -0,0 +1,46 @@
+import { describe, expect, it, vi } from "vitest";
+
+vi.mock("~/db.server", () => ({
+  prisma: {},
+  $replica: {},
+}));
+
+import { isInitialBufferedSubscriptionRequest } from "~/v3/mollifier/mollifierTelemetry.server";
+
+describe("isInitialBufferedSubscriptionRequest", () => {
+  // Electric's shape-stream protocol returns a `handle=<shape-id>` in
+  // the first response. The SDK echoes that handle on every reconnect /
+  // live-poll iteration thereafter. The realtime route logs +
+  // increments the mollifier.realtime_subscriptions.buffered counter
+  // only on the initial connect (handle absent) so each subscription
+  // produces a single observability event instead of one per
+  // long-poll round-trip (~20s).
+  it("returns true for the SDK's initial GET (no handle param)", () => {
+    expect(
+      isInitialBufferedSubscriptionRequest(
+        "http://localhost:3030/realtime/v1/runs/run_x?log=full&offset=-1",
+      ),
+    ).toBe(true);
+  });
+
+  it("returns false for Electric's reconnects (handle present)", () => {
+    expect(
+      isInitialBufferedSubscriptionRequest(
+        "http://localhost:3030/realtime/v1/runs/run_x?handle=100344308-1779&log=full&offset=0_0",
+      ),
+    ).toBe(false);
+  });
+
+  it("returns false for Electric live-poll reconnects (handle + cursor)", () => {
+    expect(
+      isInitialBufferedSubscriptionRequest(
+        "http://localhost:3030/realtime/v1/runs/run_x?cursor=51020980&handle=100344308&live=true&log=full&offset=0_inf",
+      ),
+    ).toBe(false);
+  });
+
+  it("accepts a URL instance as well as a string", () => {
+    const url = new URL("http://localhost:3030/realtime/v1/runs/run_x?log=full");
+    expect(isInitialBufferedSubscriptionRequest(url)).toBe(true);
+  });
+});

docs/realtime/react-hooks/subscribe.mdx

@@ -21,6 +21,13 @@ Trigger a task and immediately subscribe to its run. Details in the [triggering]
 
 The `useRealtimeRun` hook allows you to subscribe to a run by its ID.
 
+<Note>
+  During sustained traffic bursts the platform may briefly buffer new triggers before
+  materialising them. `useRealtimeRun` keeps the subscription open across this window and
+  begins streaming as soon as the run is materialised — typically sub-second.
+</Note>
+
+
 ```tsx
 "use client"; // This is needed for Next.js App Router or other RSC frameworks