fix(webapp): deliver realtime changes with current content when the read replica lags

The realtime feed hydrates change rows from the read replica, and that
read can race the replica's apply of the very write that triggered it.
Subscribers then receive the previous change's content, and an isolated
final change is only corrected by the backstop poll.

Publishers now stamp change records with the committed row's updatedAt
(taken from writes they already perform, no extra queries), the router
waits out the measured replica lag before hydrating, and a tripwire
retries stale reads, feeding observations back into the lag estimate.
Exhausted retries deliver anyway (liveness over freshness) while echo
re-hydrates emit the fresh row through the normal diff once the replica
catches up.

Also: metadata updates that write nothing no longer publish a change
record, and buffered parent/root metadata operations now publish when
the flusher writes them.

Author: ericallam

.server-changes/realtime-replica-read-consistency.md

@@ -0,0 +1,6 @@
+---
+area: webapp
+type: fix
+---
+
+Realtime feed reads now wait out measured read-replica lag and retry stale reads, so subscribers receive each change's current content instead of trailing one change behind when a read replica races the write.

apps/webapp/app/env.server.ts

@@ -338,6 +338,28 @@ const EnvironmentSchema = z
     // TTL/size of the per-org realtimeBackend flag cache used to pick the serving backend.
     REALTIME_BACKEND_FLAG_CACHE_TTL_MS: z.coerce.number().int().default(30_000),
     REALTIME_BACKEND_FLAG_CACHE_MAX_ENTRIES: z.coerce.number().int().default(50_000),
+    // "1" enables the read-your-writes gate: wake hydrates wait out the measured replica lag
+    // (anchored to the change record's updatedAtMs) and stale reads are retried.
+    REALTIME_BACKEND_NATIVE_REPLICA_LAG_GATE_ENABLED: z.string().default("1"),
+    // Reader-side lag probe cadence while the router is active; probing pauses when idle.
+    REALTIME_BACKEND_NATIVE_REPLICA_LAG_SAMPLE_INTERVAL_MS: z.coerce.number().int().default(250),
+    REALTIME_BACKEND_NATIVE_REPLICA_LAG_IDLE_AFTER_MS: z.coerce.number().int().default(30_000),
+    // The lag estimate is the max sample inside this window (spikes widen it immediately).
+    REALTIME_BACKEND_NATIVE_REPLICA_LAG_WINDOW_MS: z.coerce.number().int().default(5_000),
+    // Estimate before the first sample lands (and the floor when probing is unavailable).
+    REALTIME_BACKEND_NATIVE_REPLICA_LAG_DEFAULT_MS: z.coerce.number().int().default(30),
+    // Safety margin (clock skew + scheduling) added on top of the lag estimate.
+    REALTIME_BACKEND_NATIVE_REPLICA_LAG_MARGIN_MS: z.coerce.number().int().default(10),
+    // Hard cap on any single gate delay — a sick replica degrades freshness, never liveness.
+    REALTIME_BACKEND_NATIVE_REPLICA_LAG_MAX_DELAY_MS: z.coerce.number().int().default(1_000),
+    // Re-hydrate attempts for rows the tripwire still finds stale after the delay.
+    REALTIME_BACKEND_NATIVE_STALE_HYDRATE_RETRIES: z.coerce.number().int().default(3),
+    // How long a tripwire-observed staleness floors the lag estimate (vanilla-PG replicas
+    // can't measure mid-apply lag, so observations carry the estimate between races).
+    REALTIME_BACKEND_NATIVE_REPLICA_LAG_OBSERVED_FLOOR_TTL_MS: z.coerce
+      .number()
+      .int()
+      .default(60_000),
 
     PUBSUB_REDIS_HOST: z
       .string()

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

@@ -100,7 +100,20 @@ export async function routeOperationsToRun(
   const [error, result] = await tryCatch(
     updateMetadataService.call(targetRunId, { operations }, env)
   );
-  if (!error && result !== undefined) return;
+  if (!error && result !== undefined) {
+    // The parent/root run changed too — wake its live feeds (only when something was
+    // actually written here; buffered writes publish from the flusher).
+    if (result.updatedAtMs !== undefined) {
+      publishChangeRecord({
+        runId: result.runId,
+        envId: env.id,
+        tags: result.runTags,
+        batchId: result.batchId,
+        updatedAtMs: result.updatedAtMs,
+      });
+    }
+    return;
+  }
 
   if (error) {
     // PG threw — auxiliary op, stay best-effort and don't surface this
@@ -186,13 +199,18 @@ 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,
-        tags: pgResult.runTags,
-        batchId: pgResult.batchId,
-      });
+      // internal id (the router keys single-run feeds by it, not the friendly id from the
+      // URL) with the committed updatedAt as the read-your-writes watermark. No write
+      // (no-op body, or ops buffered for the flusher) means nothing to announce here.
+      if (pgResult.updatedAtMs !== undefined) {
+        publishChangeRecord({
+          runId: pgResult.runId,
+          envId: env.id,
+          tags: pgResult.runTags,
+          batchId: pgResult.batchId,
+          updatedAtMs: pgResult.updatedAtMs,
+        });
+      }
       return json({ metadata: pgResult.metadata }, { status: 200 });
     }
 

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

@@ -84,20 +84,22 @@ export async function action({ request, params }: ActionFunctionArgs) {
         if (newTags.length === 0) {
           return json({ message: "No new tags to add" }, { status: 200 });
         }
-        await prisma.taskRun.update({
+        const updated = await prisma.taskRun.update({
           where: {
             id: taskRun.id,
             runtimeEnvironmentId: env.id,
           },
           data: { runTags: { push: newTags } },
+          select: { updatedAt: true },
         });
         // Publish a run-changed record with the NEW tag set so tag feeds reindex
-        // (no-op unless enabled).
+        // (no-op unless enabled). updatedAt is the read-your-writes watermark.
         publishChangeRecord({
           runId: taskRun.id,
           envId: env.id,
           tags: existing.concat(newTags),
           batchId: taskRun.batchId,
+          updatedAtMs: updated.updatedAt.getTime(),
         });
         return json({ message: `Successfully set ${newTags.length} new tags.` }, { status: 200 });
       },

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

@@ -30,6 +30,16 @@ export type UpdateMetadataServiceOptions = {
   maximumSize?: number;
   logger?: Logger;
   logLevel?: LogLevel;
+  /** Called after the batched flusher writes a run's buffered operations, with everything
+   * a realtime change record needs — buffered (parent/root) updates otherwise never wake
+   * live feeds. */
+  onRunFlushed?: (run: {
+    runId: string;
+    environmentId: string;
+    tags: string[];
+    batchId: string | null;
+    updatedAtMs: number;
+  }) => void;
   // Testing hooks
   onBeforeUpdate?: (runId: string) => Promise<void>;
   onAfterRead?: (runId: string, metadataVersion: number) => Promise<void>;
@@ -172,12 +182,20 @@ export class UpdateMetadataService {
     operations: BufferedRunMetadataChangeOperation[]
   ) => {
     return Effect.gen(this, function* (_) {
-      // Fetch current run
+      // Fetch current run (+ the realtime membership keys, so a flush can publish)
       const run = yield* _(
         Effect.tryPromise(() =>
           this._prisma.taskRun.findFirst({
             where: { id: runId },
-            select: { id: true, metadata: true, metadataType: true, metadataVersion: true },
+            select: {
+              id: true,
+              metadata: true,
+              metadataType: true,
+              metadataVersion: true,
+              runtimeEnvironmentId: true,
+              runTags: true,
+              batchId: true,
+            },
           })
         )
       );
@@ -237,6 +255,9 @@ export class UpdateMetadataService {
         yield* _(Effect.tryPromise(() => this.options.onBeforeUpdate!(runId)));
       }
 
+      // Stamp updatedAt explicitly so the realtime publish can carry the exact committed
+      // value without a follow-up read (updateMany can't RETURNING).
+      const writeTime = new Date();
       const result = yield* _(
         Effect.tryPromise(() =>
           this._prisma.taskRun.updateMany({
@@ -247,6 +268,7 @@ export class UpdateMetadataService {
             data: {
               metadata: newMetadataPacket.data,
               metadataVersion: { increment: 1 },
+              updatedAt: writeTime,
             },
           })
         )
@@ -262,6 +284,16 @@ export class UpdateMetadataService {
         return yield* _(Effect.fail(new Error("Optimistic lock failed")));
       }
 
+      yield* Effect.sync(() => {
+        this.options.onRunFlushed?.({
+          runId,
+          environmentId: run.runtimeEnvironmentId,
+          tags: run.runTags,
+          batchId: run.batchId,
+          updatedAtMs: writeTime.getTime(),
+        });
+      });
+
       return result;
     });
   };
@@ -346,7 +378,7 @@ export class UpdateMetadataService {
       this.#ingestRunOperations(taskRun.rootTaskRun?.id ?? taskRun.id, body.rootOperations);
     }
 
-    const newMetadata = await this.#updateRunMetadata({
+    const result = await this.#updateRunMetadata({
       runId: taskRun.id,
       body,
       existingMetadata: {
@@ -356,11 +388,14 @@ export class UpdateMetadataService {
     });
 
     return {
-      metadata: newMetadata,
+      metadata: result?.metadata,
       // Internal id + membership keys, so callers can publish full realtime records the router routes by index.
       runId: taskRun.id,
       batchId: taskRun.batchId,
       runTags: taskRun.runTags,
+      // The committed row's updatedAt — the realtime watermark. Undefined when nothing was
+      // written here (no-op, or buffered for the flusher, which publishes itself).
+      updatedAtMs: result?.updatedAtMs,
     };
   }
 
@@ -372,15 +407,18 @@ export class UpdateMetadataService {
     runId: string;
     body: UpdateMetadataRequestBody;
     existingMetadata: IOPacket;
-  }) {
+  }): Promise<{ metadata: Record<string, unknown> | undefined; updatedAtMs?: number }> {
     if (Array.isArray(body.operations)) {
       return this.#updateRunMetadataWithOperations(runId, body.operations);
     } else {
       return this.#updateRunMetadataDirectly(runId, body, existingMetadata);
     }
   }
 
-  async #updateRunMetadataWithOperations(runId: string, operations: RunMetadataChangeOperation[]) {
+  async #updateRunMetadataWithOperations(
+    runId: string,
+    operations: RunMetadataChangeOperation[]
+  ): Promise<{ metadata: Record<string, unknown> | undefined; updatedAtMs?: number }> {
     const MAX_RETRIES = 3;
     let attempts = 0;
 
@@ -408,9 +446,9 @@ export class UpdateMetadataService {
       // Apply operations to the current metadata
       const applyResults = applyMetadataOperations(currentMetadata, operations);
 
-      // If no operations were applied, return the current metadata
+      // If no operations were applied, return the current metadata (nothing written)
       if (applyResults.unappliedOperations.length === operations.length) {
-        return currentMetadata;
+        return { metadata: currentMetadata };
       }
 
       const newMetadataPacket = handleMetadataPacket(
@@ -428,7 +466,9 @@ export class UpdateMetadataService {
         await this.options.onBeforeUpdate(runId);
       }
 
-      // Update with optimistic locking
+      // Update with optimistic locking; updatedAt stamped explicitly so the caller can
+      // publish the exact committed watermark without a follow-up read.
+      const writeTime = new Date();
       const result = await this._prisma.taskRun.updateMany({
         where: {
           id: runId,
@@ -440,6 +480,7 @@ export class UpdateMetadataService {
           metadataVersion: {
             increment: 1,
           },
+          updatedAt: writeTime,
         },
       });
 
@@ -454,9 +495,10 @@ export class UpdateMetadataService {
         }
 
         // If this was our last attempt, buffer the operations and return optimistically
+        // (no watermark — the flusher writes later and publishes itself).
         if (attempts === MAX_RETRIES) {
           this.#ingestRunOperations(runId, operations);
-          return applyResults.newMetadata;
+          return { metadata: applyResults.newMetadata };
         }
 
         // Otherwise sleep and try again
@@ -474,8 +516,10 @@ export class UpdateMetadataService {
       }
 
       // Success! Return the new metadata
-      return applyResults.newMetadata;
+      return { metadata: applyResults.newMetadata, updatedAtMs: writeTime.getTime() };
     }
+
+    return { metadata: undefined };
   }
 
   // Checks to see if a run is updatable
@@ -493,17 +537,19 @@ export class UpdateMetadataService {
     runId: string,
     body: UpdateMetadataRequestBody,
     existingMetadata: IOPacket
-  ) {
+  ): Promise<{ metadata: Record<string, unknown> | undefined; updatedAtMs?: number }> {
     const metadataPacket = handleMetadataPacket(
       body.metadata,
       "application/json",
       this.maximumSize
     );
 
     if (!metadataPacket) {
-      return {};
+      return { metadata: {} };
     }
 
+    let updatedAtMs: number | undefined;
+
     if (
       metadataPacket.data !== "{}" ||
       (existingMetadata.data && metadataPacket.data !== existingMetadata.data)
@@ -515,7 +561,9 @@ export class UpdateMetadataService {
         });
       }
 
-      // Update the metadata without version check
+      // Update the metadata without version check; updatedAt stamped explicitly so the
+      // caller can publish the exact committed watermark.
+      const writeTime = new Date();
       await this._prisma.taskRun.update({
         where: {
           id: runId,
@@ -526,12 +574,14 @@ export class UpdateMetadataService {
           metadataVersion: {
             increment: 1,
           },
+          updatedAt: writeTime,
         },
       });
+      updatedAtMs = writeTime.getTime();
     }
 
     const newMetadata = await parsePacket(metadataPacket);
-    return newMetadata;
+    return { metadata: newMetadata, updatedAtMs };
   }
 
   #ingestRunOperations(runId: string, operations: RunMetadataChangeOperation[]) {

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

@@ -2,6 +2,7 @@ import { singleton } from "~/utils/singleton";
 import { env } from "~/env.server";
 import { UpdateMetadataService } from "./updateMetadata.server";
 import { prisma } from "~/db.server";
+import { publishChangeRecord } from "~/services/realtime/runChangeNotifierInstance.server";
 
 export const updateMetadataService = singleton(
   "update-metadata-service",
@@ -13,5 +14,16 @@ export const updateMetadataService = singleton(
       flushLoggingEnabled: env.BATCH_METADATA_OPERATIONS_FLUSH_LOGGING_ENABLED === "1",
       maximumSize: env.TASK_RUN_METADATA_MAXIMUM_SIZE,
       logLevel: env.BATCH_METADATA_OPERATIONS_FLUSH_LOGGING_ENABLED === "1" ? "debug" : "info",
+      // Buffered (parent/root) operations land via the flusher, not the caller's request —
+      // publish here so those changes wake live feeds too (no-op when the backend is off).
+      onRunFlushed: (run) => {
+        publishChangeRecord({
+          runId: run.runId,
+          envId: run.environmentId,
+          tags: run.tags,
+          batchId: run.batchId,
+          updatedAtMs: run.updatedAtMs,
+        });
+      },
     })
 );