test(webapp): pin realtime buffered-resource resolution

d-cs · claude · d-cs · commit 69e8535106ed · 2026-05-22T13:55:13.000+01:00
The previous commit's regression coverage was thin: only the log-dedup
gate was unit-tested. The load-bearing logic — synthesise a resource
when PG misses but the buffer has the run, with an `id` matching what
the drainer will eventually write — had no regression test, so a
future change that removed the buffered fallback would put the silent-
hang back into prod without anything failing in CI.

Extract the resource-resolution rules from the route's findResource
into `resolveRealtimeRunResource`, a pure function. Cover the
branching with unit tests (PG hit, PG hit during drain race, PG miss
+ buffer hit, missing taskIdentifier default, both miss) and pin the
full chain with a container-backed test that uses a real
MollifierBuffer + the real readFallback helper and asserts the
synthesised `id` matches `RunId.fromFriendlyId(friendlyId)`. That
identity is what Electric's `WHERE id='&lt;id&gt;'` clause depends on when
the drainer eventually INSERTs the row.

12 tests total across the three Phase-5.2 suites; one empirical probe
run after the refactor confirmed end-to-end behaviour unchanged.

Co-Authored-By: Claude Opus 4.7 (1M context) &lt;noreply@anthropic.com&gt;
diff --git a/apps/webapp/app/routes/realtime.v1.runs.$runId.ts b/apps/webapp/app/routes/realtime.v1.runs.$runId.ts
@@ -12,6 +12,7 @@ import {
   isInitialBufferedSubscriptionRequest,
   recordRealtimeBufferedSubscription,
 } from "~/v3/mollifier/mollifierTelemetry.server";
+import { resolveRealtimeRunResource } from "~/v3/mollifier/realtimeRunResource.server";
 
 const ParamsSchema = z.object({
   runId: z.string(),
@@ -36,7 +37,6 @@ 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
@@ -45,28 +45,15 @@ export const loader = createLoaderApiRoute(
       // 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;
+      const bufferedSynthetic = pgRun
+        ? null
+        : await findRunByIdWithMollifierFallback({
+            runId: params.runId,
+            environmentId: authentication.environment.id,
+            organizationId: authentication.environment.organizationId,
+          });
 
-      // 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(),
-      };
+      return resolveRealtimeRunResource({ pgRun, bufferedSynthetic });
     },
     authorization: {
       action: "read",
diff --git a/apps/webapp/app/v3/mollifier/realtimeRunResource.server.ts b/apps/webapp/app/v3/mollifier/realtimeRunResource.server.ts
@@ -0,0 +1,57 @@
+import type { SyntheticRun } from "./readFallback.server";
+
+// Shape `realtime.v1.runs.$runId.ts`'s findResource hands to the route's
+// authorization callback + loader body. The PG-resident case is the
+// canonical shape (a TaskRun row with the batch join); the buffered
+// case below mirrors it from the synthetic run.
+export type RealtimeRunResource = {
+  id: string;
+  friendlyId: string;
+  taskIdentifier: string;
+  runTags: string[];
+  batch: { friendlyId: string } | null;
+  // Present only when this resource was resolved from the mollifier
+  // buffer (no PG row yet). Stamped at resolve time so the loader body
+  // can emit observability for buffered-window subscriptions. The flag
+  // doubles as the discriminant — PG-sourced resources never carry it.
+  __bufferedDwellMs?: number;
+};
+
+export type RealtimeRunResourcePgRun = {
+  id: string;
+  friendlyId: string;
+  taskIdentifier: string;
+  runTags: string[];
+  batch: { friendlyId: string } | null;
+};
+
+// Given the results of the PG and buffer lookups, produce the resource
+// shape the realtime route returns from findResource. PG-first: if the
+// run is PG-resident, return it unchanged (the buffered fallback only
+// fires when no PG row exists yet). When only the buffer has the run,
+// synthesise a matching shape whose `id` is the deterministic value
+// engine.trigger will write when the drainer materialises this run —
+// this is what lets the Electric subscription's `WHERE id=<id>` match
+// the eventual INSERT.
+export function resolveRealtimeRunResource(input: {
+  pgRun: RealtimeRunResourcePgRun | null;
+  bufferedSynthetic: Pick<
+    SyntheticRun,
+    "id" | "friendlyId" | "taskIdentifier" | "runTags" | "createdAt"
+  > | null;
+  now?: () => number;
+}): RealtimeRunResource | null {
+  if (input.pgRun) return input.pgRun;
+  if (input.bufferedSynthetic) {
+    const now = (input.now ?? Date.now)();
+    return {
+      id: input.bufferedSynthetic.id,
+      friendlyId: input.bufferedSynthetic.friendlyId,
+      taskIdentifier: input.bufferedSynthetic.taskIdentifier ?? "",
+      runTags: input.bufferedSynthetic.runTags,
+      batch: null,
+      __bufferedDwellMs: now - input.bufferedSynthetic.createdAt.getTime(),
+    };
+  }
+  return null;
+}
diff --git a/apps/webapp/test/mollifierRealtimeRunResource.test.ts b/apps/webapp/test/mollifierRealtimeRunResource.test.ts
@@ -0,0 +1,90 @@
+import { describe, expect, it, vi } from "vitest";
+
+vi.mock("~/db.server", () => ({ prisma: {}, $replica: {} }));
+
+import { resolveRealtimeRunResource } from "~/v3/mollifier/realtimeRunResource.server";
+
+const pgRun = {
+  id: "pg_internal_id",
+  friendlyId: "run_pg_friendly",
+  taskIdentifier: "hello-world",
+  runTags: ["a", "b"],
+  batch: { friendlyId: "batch_1" },
+};
+
+const bufferedSynthetic = {
+  id: "buffered_id",
+  friendlyId: "run_buffered_id",
+  taskIdentifier: "hello-world",
+  runTags: ["c"],
+  // Six seconds ago against the fixed `now` below.
+  createdAt: new Date("2026-05-22T12:00:00.000Z"),
+};
+
+const fixedNow = () => new Date("2026-05-22T12:00:06.000Z").getTime();
+
+describe("resolveRealtimeRunResource", () => {
+  it("returns the PG run unchanged when one exists", () => {
+    // PG wins even if the buffer also has the entry — the drainer may
+    // be racing the route call and the PG row is the canonical source.
+    expect(
+      resolveRealtimeRunResource({ pgRun, bufferedSynthetic: null }),
+    ).toEqual(pgRun);
+    expect(
+      resolveRealtimeRunResource({ pgRun, bufferedSynthetic }),
+    ).toEqual(pgRun);
+  });
+
+  it("never stamps __bufferedDwellMs on a PG-sourced resource", () => {
+    // The loader body uses __bufferedDwellMs as a discriminant for
+    // emitting buffered-subscription observability. A PG-resident run
+    // must never carry it or every PG subscription would over-count.
+    const result = resolveRealtimeRunResource({ pgRun, bufferedSynthetic });
+    expect(result).not.toHaveProperty("__bufferedDwellMs");
+  });
+
+  it("synthesises a resource from the buffered entry when PG misses", () => {
+    // Load-bearing assertion: `id` must equal `bufferedSynthetic.id`.
+    // The realtime route hands this `id` to streamRun, which builds
+    // Electric's `WHERE id='<id>'` clause. When the drainer materialises
+    // the run, engine.trigger writes the row with that same id (derived
+    // deterministically from friendlyId), and Electric streams the
+    // INSERT to the client. If the synthesised `id` ever drifts from
+    // what the drainer writes, the customer subscribes to a shape that
+    // never matches and the hook silently hangs even after materialise.
+    const result = resolveRealtimeRunResource({
+      pgRun: null,
+      bufferedSynthetic,
+      now: fixedNow,
+    });
+    expect(result).toEqual({
+      id: "buffered_id",
+      friendlyId: "run_buffered_id",
+      taskIdentifier: "hello-world",
+      runTags: ["c"],
+      batch: null,
+      __bufferedDwellMs: 6000,
+    });
+  });
+
+  it("defaults a missing taskIdentifier to empty string", () => {
+    const result = resolveRealtimeRunResource({
+      pgRun: null,
+      bufferedSynthetic: { ...bufferedSynthetic, taskIdentifier: undefined },
+      now: fixedNow,
+    });
+    expect(result?.taskIdentifier).toBe("");
+  });
+
+  it("returns null when neither PG nor buffer have the run", () => {
+    // This is the genuine not-found case — typo'd runId, deleted run,
+    // etc. The api-builder maps null to 404. Critically, the buffered-
+    // fallback must NOT promote a missing run to a synthetic resource —
+    // that would cause Electric to open a shape for a runId that may
+    // never exist, which is also a silent-hang situation but for a
+    // different reason.
+    expect(
+      resolveRealtimeRunResource({ pgRun: null, bufferedSynthetic: null }),
+    ).toBeNull();
+  });
+});
diff --git a/apps/webapp/test/mollifierRealtimeRunResourceBuffer.test.ts b/apps/webapp/test/mollifierRealtimeRunResourceBuffer.test.ts
@@ -0,0 +1,152 @@
+import { describe, expect, vi } from "vitest";
+import { redisTest } from "@internal/testcontainers";
+import { MollifierBuffer } from "@trigger.dev/redis-worker";
+import { RunId } from "@trigger.dev/core/v3/isomorphic";
+
+vi.mock("~/db.server", () => ({ prisma: {}, $replica: {} }));
+
+import { findRunByIdWithMollifierFallback } from "~/v3/mollifier/readFallback.server";
+import { resolveRealtimeRunResource } from "~/v3/mollifier/realtimeRunResource.server";
+
+const SNAPSHOT_BASE = {
+  friendlyId: "run_phase52e2e",
+  taskIdentifier: "hello-world",
+  payload: '{"x":1}',
+  payloadType: "application/json",
+  traceContext: { traceparent: "00-0123456789abcdef0123456789abcdef-fedcba9876543210-01" },
+  traceId: "0123456789abcdef0123456789abcdef",
+  spanId: "fedcba9876543210",
+  queue: "task/hello-world",
+  tags: ["realtime-e2e"],
+  depth: 0,
+  isTest: false,
+  taskEventStore: "taskEvent",
+};
+
+// End-to-end: a real MollifierBuffer has an entry, the real
+// readFallback helper deserialises it, and the resolveRealtimeRunResource
+// helper produces the resource shape the realtime route returns from
+// findResource. Regression intent: if any link in the chain breaks —
+// buffer interface rename, snapshot field rename, id-derivation drift,
+// synthetic-shape change — this test fails. The route file itself is
+// then a thin glue layer over tested pieces.
+describe("realtime buffered-subscription resource resolution (testcontainers)", () => {
+  redisTest(
+    "synthesises a resource whose `id` matches RunId.fromFriendlyId",
+    async ({ redisOptions }) => {
+      const buffer = new MollifierBuffer({ redisOptions, entryTtlSeconds: 60 });
+      try {
+        await buffer.accept({
+          runId: SNAPSHOT_BASE.friendlyId,
+          envId: "env_a",
+          orgId: "org_1",
+          payload: JSON.stringify(SNAPSHOT_BASE),
+        });
+
+        const bufferedSynthetic = await findRunByIdWithMollifierFallback(
+          {
+            runId: SNAPSHOT_BASE.friendlyId,
+            environmentId: "env_a",
+            organizationId: "org_1",
+          },
+          { getBuffer: () => buffer },
+        );
+        expect(bufferedSynthetic).not.toBeNull();
+
+        const resource = resolveRealtimeRunResource({
+          pgRun: null,
+          bufferedSynthetic,
+        });
+
+        // The load-bearing contract: the resolved `id` MUST equal what
+        // engine.trigger will write to PG.TaskRun.id when the drainer
+        // materialises this run. Electric's `WHERE id='<id>'` clause
+        // depends on this match — drift means a silent-hang regression.
+        expect(resource?.id).toBe(RunId.fromFriendlyId(SNAPSHOT_BASE.friendlyId));
+        expect(resource?.friendlyId).toBe(SNAPSHOT_BASE.friendlyId);
+        expect(resource?.taskIdentifier).toBe("hello-world");
+        expect(resource?.runTags).toEqual(["realtime-e2e"]);
+        expect(resource?.batch).toBeNull();
+        expect(resource?.__bufferedDwellMs).toBeTypeOf("number");
+        expect(resource?.__bufferedDwellMs).toBeGreaterThanOrEqual(0);
+      } finally {
+        await buffer.close();
+      }
+    },
+  );
+
+  redisTest(
+    "returns null when neither PG nor the buffer have the entry",
+    async ({ redisOptions }) => {
+      const buffer = new MollifierBuffer({ redisOptions, entryTtlSeconds: 60 });
+      try {
+        const bufferedSynthetic = await findRunByIdWithMollifierFallback(
+          {
+            runId: "run_does_not_exist",
+            environmentId: "env_a",
+            organizationId: "org_1",
+          },
+          { getBuffer: () => buffer },
+        );
+        expect(bufferedSynthetic).toBeNull();
+
+        const resource = resolveRealtimeRunResource({
+          pgRun: null,
+          bufferedSynthetic,
+        });
+        // The api builder relies on this null to emit a real 404 for
+        // genuinely missing runs. If we ever promote unknown runIds to
+        // synthetic resources here, the route opens an Electric shape
+        // for a run that may never exist — a different silent-hang
+        // failure mode for typos, deleted runs, etc.
+        expect(resource).toBeNull();
+      } finally {
+        await buffer.close();
+      }
+    },
+  );
+
+  redisTest(
+    "does not fall back to buffer when PG has the row",
+    async ({ redisOptions }) => {
+      const buffer = new MollifierBuffer({ redisOptions, entryTtlSeconds: 60 });
+      try {
+        await buffer.accept({
+          runId: SNAPSHOT_BASE.friendlyId,
+          envId: "env_a",
+          orgId: "org_1",
+          payload: JSON.stringify(SNAPSHOT_BASE),
+        });
+
+        // Simulate the drainer having materialised the run: PG has the
+        // canonical row, the buffer still has its entry (would be
+        // ack'd & removed in real ops). The resolver must return the
+        // PG row and NOT carry the __bufferedDwellMs flag — otherwise
+        // the loader body would emit a buffered-subscription log for a
+        // run that's actually PG-resident, over-counting the signal.
+        const pgRun = {
+          id: RunId.fromFriendlyId(SNAPSHOT_BASE.friendlyId),
+          friendlyId: SNAPSHOT_BASE.friendlyId,
+          taskIdentifier: "hello-world",
+          runTags: ["realtime-e2e"],
+          batch: null,
+        };
+
+        const bufferedSynthetic = await findRunByIdWithMollifierFallback(
+          {
+            runId: SNAPSHOT_BASE.friendlyId,
+            environmentId: "env_a",
+            organizationId: "org_1",
+          },
+          { getBuffer: () => buffer },
+        );
+
+        const resource = resolveRealtimeRunResource({ pgRun, bufferedSynthetic });
+        expect(resource).toEqual(pgRun);
+        expect(resource).not.toHaveProperty("__bufferedDwellMs");
+      } finally {
+        await buffer.close();
+      }
+    },
+  );
+});
