fix(webapp): stop concurrent batch subscribers sharing a realtime working set

Batch feeds derived their stream handle from the batchId, so two
subscribers to the same batch on one instance shared a working-set
cache entry and one could permanently suppress the other's deltas,
including terminal status updates. Handles are now minted per
connection with a collision-proof suffix and working-set entries are
keyed by environment, so a client-echoed handle can never read or
overwrite another connection's state.

Also: mid-run metadata updates carry the batchId so batch feeds
fast-wake instead of waiting for the backstop, and the run-engine
metadata handler publishes after the write lands so hydration sees
the new row.

Author: ericallam

apps/webapp/app/routes/api.v1.runs.$runId.metadata.ts

@@ -187,7 +187,7 @@ const { action } = createActionApiRoute(
     if (pgResult) {
       // Reflect metadata.set() on a live feed before the next lifecycle event. Publish the
       // internal id (the router keys single-run feeds by it, not the friendly id from the URL).
-      publishChangeRecord({ runId: pgResult.runId, envId: env.id });
+      publishChangeRecord({ runId: pgResult.runId, envId: env.id, batchId: pgResult.batchId });
       return json({ metadata: pgResult.metadata }, { status: 200 });
     }
 

apps/webapp/app/services/metadata/updateMetadata.server.ts

@@ -308,6 +308,7 @@ export class UpdateMetadataService {
           },
       select: {
         id: true,
+        batchId: true,
         completedAt: true,
         status: true,
         metadata: true,
@@ -355,8 +356,9 @@ export class UpdateMetadataService {
 
     return {
       metadata: newMetadata,
-      // Internal id, so callers can publish realtime records keyed how the router indexes feeds.
+      // Internal id + batchId, so callers can publish realtime records keyed how the router indexes feeds.
       runId: taskRun.id,
+      batchId: taskRun.batchId,
     };
   }
 

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

@@ -364,15 +364,11 @@ export class NotifierRealtimeClient implements RealtimeStreamClient {
     clientVersion?: string,
     signal?: AbortSignal
   ): Promise<Response> {
-    const { offset, isLive, skipColumns } = this.#parseStreamRequest(url, requestOptions);
+    const { offset, handle, isLive, skipColumns } = this.#parseStreamRequest(url, requestOptions);
 
-    // The batch set is fully defined by batchId (the route resolves it from the
-    // friendlyId on every request), so the handle is derived and stable and there's
-    // no createdAt window to pin.
-    const handle = `batch-${batchId}`;
     const filter: RunSetFilter = { batchId };
 
-    if (offset !== INITIAL_OFFSET && isLive) {
+    if (offset !== INITIAL_OFFSET && handle && isLive) {
       return this.#runSetLiveResponse(
         environment,
         filter,
@@ -385,11 +381,13 @@ export class NotifierRealtimeClient implements RealtimeStreamClient {
       );
     }
 
-    // Initial snapshot + non-live catch-up.
+    // Initial snapshot + non-live catch-up. The handle must be per-connection, never
+    // derived from the batchId: working sets are keyed by handle, and a shared handle
+    // lets one subscriber's emit permanently suppress the same row for another.
     return this.#runSetSnapshotResponse(
       environment,
       filter,
-      handle,
+      handle ?? this.#mintBatchHandle(batchId),
       skipColumns,
       apiVersion,
       clientVersion
@@ -519,7 +517,7 @@ export class NotifierRealtimeClient implements RealtimeStreamClient {
       seen.set(row.id, updatedAtMs);
       maxUpdatedAt = Math.max(maxUpdatedAt, updatedAtMs);
     }
-    this.#workingSetCache.set(handle, seen);
+    this.#workingSetCache.set(this.#workingSetKey(environment.id, handle), seen);
 
     return this.#buildResponse(buildRowsBody(changes, skipColumns), apiVersion, clientVersion, {
       offset: encodeOffset(maxUpdatedAt, this.#nextSeq()),
@@ -556,7 +554,8 @@ export class NotifierRealtimeClient implements RealtimeStreamClient {
 
       // Working set we diff against: seeded from the cache (or the offset floor on a
       // miss) and advanced on each refetch within this held request.
-      let prevSeen = this.#workingSetCache.get(handle);
+      const workingSetKey = this.#workingSetKey(environment.id, handle);
+      let prevSeen = this.#workingSetCache.get(workingSetKey);
 
       const emitFromSerialized = (changes: SerializedRowChange[], maxUpdatedAt: number): Response => {
         const seq = this.#nextSeq();
@@ -614,7 +613,7 @@ export class NotifierRealtimeClient implements RealtimeStreamClient {
             // Merge (not replace): the router only surfaced the changed subset, so keep the
             // rest of the working set intact. The backstop full-resolve rebuilds it.
             const merged = this.#mergeWorkingSet(prevSeen, touched);
-            this.#workingSetCache.set(handle, merged);
+            this.#workingSetCache.set(workingSetKey, merged);
             prevSeen = merged;
 
             if (changes.length > 0) {
@@ -636,7 +635,7 @@ export class NotifierRealtimeClient implements RealtimeStreamClient {
             prevSeen,
             offsetFloorMs
           );
-          this.#workingSetCache.set(handle, touched);
+          this.#workingSetCache.set(workingSetKey, touched);
           prevSeen = touched;
 
           if (changes.length > 0) {
@@ -881,7 +880,23 @@ export class NotifierRealtimeClient implements RealtimeStreamClient {
   #mintListHandle(createdAtFilterMs: number): string {
     // Pins the createdAt threshold in the opaque handle so live polls reuse the
     // same lower bound even on a working-set cache miss.
-    return `runs_${Math.trunc(createdAtFilterMs)}_${this.#nextSeq()}`;
+    return `runs_${Math.trunc(createdAtFilterMs)}_${this.#mintUniqueSuffix()}`;
+  }
+
+  #mintBatchHandle(batchId: string): string {
+    return `batch_${batchId}_${this.#mintUniqueSuffix()}`;
+  }
+
+  #mintUniqueSuffix(): string {
+    // The seq alone isn't unique across instances/restarts; behind a non-sticky ALB a
+    // collision would land two connections on one working-set cache entry.
+    return `${this.#nextSeq()}_${randomUUID().slice(0, 8)}`;
+  }
+
+  #workingSetKey(environmentId: string, handle: string): string {
+    // The handle is client-echoed; env-prefix the key so a foreign handle can never
+    // read or overwrite another tenant's working set.
+    return `${environmentId}:${handle}`;
   }
 
   #filterMsFromHandle(handle: string): number | undefined {

apps/webapp/app/v3/runEngineHandlers.server.ts

@@ -571,12 +571,11 @@ export function registerRunEngineEventBusHandlers() {
 
     const { environment, runTags, batchId } = result;
 
-    // Realtime run-changed publish: a full record (env + tags + batchId all from the one
-    // read above), so tag/batch feeds route by index instead of hydrate-to-classify.
-    publishChangeRecord({ runId: run.id, envId: environment.id, tags: runTags, batchId });
-
     try {
       await updateMetadataService.call(run.id, run.metadata, environment);
+      // Realtime run-changed publish, after the write so the router's hydrate sees the new
+      // row. A full record (env + tags + batchId), so feeds route by index.
+      publishChangeRecord({ runId: run.id, envId: environment.id, tags: runTags, batchId });
     } catch (e) {
       if (e instanceof MetadataTooLargeError) {
         logger.warn("[runMetadataUpdated] Failed to update metadata, too large", {

apps/webapp/test/realtime/notifierRunSetCache.test.ts

@@ -127,6 +127,29 @@ describe("NotifierRealtimeClient run-set resolve coalescing + cache", () => {
     await snapshot(client, "batch_1");
     expect(results).toEqual(["miss", "hit"]);
   });
+
+  it("mints a distinct batch handle per connection and echoes a client-provided one", async () => {
+    const { client } = makeClient();
+    // Two subscribers to the SAME batch must never share a handle (the working-set
+    // cache is keyed by it; sharing lets one suppress the other's deltas forever).
+    const res1 = await snapshot(client, "batch_1");
+    const res2 = await snapshot(client, "batch_1");
+    const h1 = res1.headers.get("electric-handle");
+    const h2 = res2.headers.get("electric-handle");
+    expect(h1).toBeTruthy();
+    expect(h1).not.toBe(h2);
+
+    // Catch-up under an existing handle keeps it.
+    const res3 = await client.streamBatch(
+      `http://localhost:3030/realtime/v1/batches/batch_1?offset=123_1&handle=${h1}`,
+      ENV,
+      "batch_1",
+      CURRENT_API_VERSION,
+      undefined,
+      "1.0.0"
+    );
+    expect(res3.headers.get("electric-handle")).toBe(h1);
+  });
 });
 
 describe("NotifierRealtimeClient resolve admission gate (mass-reconnect stampede)", () => {
@@ -304,7 +327,7 @@ describe("NotifierRealtimeClient review fixes", () => {
       livePollTimeoutMs: 50,
     });
     const res = await client.streamBatch(
-      "http://localhost:3030/realtime/v1/batches/batch_1?offset=123_1&live=true",
+      "http://localhost:3030/realtime/v1/batches/batch_1?offset=123_1&live=true&handle=batch_batch_1_7_abc",
       ENV,
       "batch_1",
       CURRENT_API_VERSION,