feat(webapp): share realtime replay cursors across instances

A load balancer hop previously made a connection's inter-poll gap
unprovable, forcing a cold resolve and a full-window replay on the new
instance. Per-connection replay cursors (one timestamp each) now live in
Redis behind REALTIME_BACKEND_NATIVE_SHARED_REPLAY_CURSORS (default on),
so any instance can read the true gap. Store reads have a bounded
deadline and degrade to the old cold-probe behavior on any Redis trouble.

Author: ericallam

apps/webapp/app/env.server.ts

@@ -318,6 +318,8 @@ const EnvironmentSchema = z
     REALTIME_BACKEND_NATIVE_RUNSET_CREATED_AT_BUCKET_MS: z.coerce.number().int().default(60_000),
     // Leading-edge throttle (ms) on per-env wake delivery; 0 wakes on every change.
     REALTIME_BACKEND_NATIVE_ENV_WAKE_COALESCE_WINDOW_MS: z.coerce.number().int().default(250),
+    // "1" shares per-connection replay cursors fleet-wide via Redis, so a load-balancer hop reads the connection's true inter-poll gap instead of cold-resolving.
+    REALTIME_BACKEND_NATIVE_SHARED_REPLAY_CURSORS: z.string().default("1"),
     // "1" holds a multi-run live poll open on a non-matching wake instead of replying up-to-date.
     REALTIME_BACKEND_NATIVE_HOLD_ON_EMPTY: z.string().default("1"),
     // Max concurrent fresh ClickHouse resolves per instance (reconnect-stampede gate); 0 disables.

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

@@ -33,6 +33,7 @@ import {
 } from "./envChangeRouter.server";
 import { type RunHydrator, type RunListResolver } from "./runReader.server";
 import { type RealtimeConcurrencyLimiter } from "./realtimeConcurrencyLimiter.server";
+import { InMemoryReplayCursorStore, type ReplayCursorStore } from "./replayCursorStore.server";
 
 /** Widened with projectId so the tag-list feed can resolve ids via ClickHouse (needs org + project + env). */
 export type RealtimeListEnvironment = RealtimeEnvironment & { projectId: string };
@@ -106,6 +107,10 @@ export type NativeRealtimeClientOptions = {
   holdOnEmpty?: boolean;
   /** Max concurrent fresh ClickHouse resolves (cache misses) per instance, bounding a distinct-filter stampede. Defaults to 16; 0 disables. */
   resolveAdmissionLimit?: number;
+  /** Per-connection replay-cursor store. Inject a fleet-shared (Redis) store so an instance
+   * hop reads the connection's true inter-poll gap instead of cold-probing; defaults to a
+   * per-instance in-memory cache. */
+  replayCursorStore?: ReplayCursorStore;
   /** Observability hook: why a live request woke (notify vs timeout vs abort). */
   onWakeup?: (reason: WakeupReason) => void;
   /** Observability hook: how a live poll resolved (fast path vs full resolve). */
@@ -206,18 +211,21 @@ export class NativeRealtimeClient implements RealtimeStreamClient {
   /** Bounds concurrent fresh CH resolves (undefined => unbounded). */
   readonly #admissionGate?: ResolveAdmissionGate;
   /** Per-connection: when this connection's last response was sent, so the router's
-   * replay covers exactly the inter-poll gap instead of rewinding a full window. */
-  readonly #replayCursorCache: BoundedTtlCache<number>;
+   * replay covers exactly the inter-poll gap instead of rewinding a full window.
+   * Fleet-shared when a store is injected (hops stop looking like unknown gaps). */
+  readonly #replayCursors: ReplayCursorStore;
 
   constructor(private readonly options: NativeRealtimeClientOptions) {
     this.#workingSetCache = new BoundedTtlCache(
       options.workingSetCacheTtlMs ?? LIST_CACHE_TTL_MS,
       options.listCacheMaxEntries ?? LIST_CACHE_MAX_ENTRIES
     );
-    this.#replayCursorCache = new BoundedTtlCache(
-      options.workingSetCacheTtlMs ?? LIST_CACHE_TTL_MS,
-      options.listCacheMaxEntries ?? LIST_CACHE_MAX_ENTRIES
-    );
+    this.#replayCursors =
+      options.replayCursorStore ??
+      new InMemoryReplayCursorStore(
+        options.workingSetCacheTtlMs ?? LIST_CACHE_TTL_MS,
+        options.listCacheMaxEntries ?? LIST_CACHE_MAX_ENTRIES
+      );
     this.#runSetCache = new BoundedTtlCache(
       options.runSetResolveCacheTtlMs ?? DEFAULT_RUNSET_CACHE_TTL_MS,
       options.runSetResolveCacheMaxEntries ?? DEFAULT_RUNSET_CACHE_MAX_ENTRIES
@@ -528,7 +536,7 @@ export class NativeRealtimeClient implements RealtimeStreamClient {
       maxUpdatedAt = Math.max(maxUpdatedAt, updatedAtMs);
     }
     this.#workingSetCache.set(this.#workingSetKey(environment.id, handle), seen);
-    this.#replayCursorCache.set(this.#workingSetKey(environment.id, handle), Date.now());
+    this.#replayCursors.set(this.#workingSetKey(environment.id, handle), Date.now());
 
     return this.#buildResponse(buildRowsBody(changes, skipColumns), apiVersion, clientVersion, {
       offset: encodeOffset(maxUpdatedAt, this.#nextSeq()),
@@ -559,7 +567,7 @@ export class NativeRealtimeClient implements RealtimeStreamClient {
       const workingSetKey = this.#workingSetKey(environment.id, handle);
       let prevSeen = this.#workingSetCache.get(workingSetKey);
 
-      const markPollEnd = () => this.#replayCursorCache.set(workingSetKey, Date.now());
+      const markPollEnd = () => this.#replayCursors.set(workingSetKey, Date.now());
       const emitFromSerialized = (changes: SerializedRowChange[], maxUpdatedAt: number): Response => {
         const seq = this.#nextSeq();
         markPollEnd();
@@ -588,12 +596,14 @@ export class NativeRealtimeClient implements RealtimeStreamClient {
         });
       };
 
+      // When this connection last received data, so replay covers exactly its gap. A store
+      // error degrades to undefined (cold probe), never a failed poll.
+      const replaySinceMs = await this.#replayCursors.get(workingSetKey);
       const registration = this.options.router.register(
         environment.id,
         this.#feedFilter(filter),
         skipColumns,
-        // When this connection last received data, so replay covers exactly its gap.
-        { replaySinceMs: this.#replayCursorCache.get(workingSetKey) }
+        { replaySinceMs }
       );
 
       // Cold start (fresh env subscription, e.g. an instance hop): resolve once up front

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

@@ -9,6 +9,7 @@ import { EnvChangeRouter } from "./envChangeRouter.server";
 import { NativeRealtimeClient } from "./nativeRealtimeClient.server";
 import { RealtimeConcurrencyLimiter } from "./realtimeConcurrencyLimiter.server";
 import { getRunChangeNotifier } from "./runChangeNotifierInstance.server";
+import { RedisReplayCursorStore } from "./replayCursorStore.server";
 import { RunHydrator } from "./runReader.server";
 
 // Process-singleton wiring for the native realtime client; only constructed when a
@@ -77,6 +78,11 @@ function initializeNativeRealtimeClient(): NativeRealtimeClient {
     description: "Polls rejected (429) by the per-env concurrency limiter.",
   });
 
+  const replayCursorOps = meter.createCounter("realtime_native.replay_cursor_ops", {
+    description:
+      "Shared replay-cursor store operations by outcome. Errors degrade hops to cold resolves (watch live_polls{path='cold-resolve'} rise with them), never failed polls.",
+  });
+
   const limiter = new RealtimeConcurrencyLimiter({
     keyPrefix: "tr:realtime:native:concurrency",
     redis: {
@@ -89,6 +95,24 @@ function initializeNativeRealtimeClient(): NativeRealtimeClient {
     },
   });
 
+  // Fleet-shared replay cursors (one timestamp per connection) on the same Redis as the
+  // change channel, so a load-balancer hop reads the connection's true inter-poll gap.
+  const replayCursorStore =
+    env.REALTIME_BACKEND_NATIVE_SHARED_REPLAY_CURSORS === "1"
+      ? new RedisReplayCursorStore({
+          redis: {
+            host: env.REALTIME_BACKEND_NATIVE_PUBSUB_REDIS_HOST,
+            port: env.REALTIME_BACKEND_NATIVE_PUBSUB_REDIS_PORT,
+            username: env.REALTIME_BACKEND_NATIVE_PUBSUB_REDIS_USERNAME,
+            password: env.REALTIME_BACKEND_NATIVE_PUBSUB_REDIS_PASSWORD,
+            tlsDisabled: env.REALTIME_BACKEND_NATIVE_PUBSUB_REDIS_TLS_DISABLED === "true",
+            clusterMode: env.REALTIME_BACKEND_NATIVE_PUBSUB_REDIS_CLUSTER_MODE_ENABLED === "1",
+          },
+          ttlMs: env.REALTIME_BACKEND_NATIVE_WORKING_SET_TTL_MS,
+          onResult: (op, ok) => replayCursorOps.add(1, { op, result: ok ? "ok" : "error" }),
+        })
+      : undefined;
+
   // One RunHydrator shared by the router and the client, so its single-flight + short-TTL cache covers both.
   const runReader = new RunHydrator({
     replica: $replica,
@@ -138,6 +162,7 @@ function initializeNativeRealtimeClient(): NativeRealtimeClient {
     runSetCreatedAtBucketMs: env.REALTIME_BACKEND_NATIVE_RUNSET_CREATED_AT_BUCKET_MS,
     holdOnEmpty: env.REALTIME_BACKEND_NATIVE_HOLD_ON_EMPTY === "1",
     resolveAdmissionLimit: env.REALTIME_BACKEND_NATIVE_RESOLVE_ADMISSION_LIMIT,
+    replayCursorStore,
     onWakeup: (reason) => wakeups.add(1, { reason }),
     onLivePollPath: (path) => livePollPaths.add(1, { path }),
     onRunSetResolve: (result) => runSetResolves.add(1, { result }),

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

@@ -0,0 +1,145 @@
+import { createRedisClient, type RedisClient, type RedisWithClusterOptions } from "~/redis.server";
+import { logger } from "../logger.server";
+import { BoundedTtlCache } from "./boundedTtlCache";
+
+/**
+ * Per-connection replay cursors ("when did this connection last receive data"), keyed by the
+ * env-prefixed working-set key. Sharing them fleet-wide makes an instance hop look like a normal
+ * inter-poll gap instead of an unknown one, so hops stop triggering cold resolves and full-window
+ * replays. Values are single timestamps, so the shared store stays cheap.
+ */
+export interface ReplayCursorStore {
+  /** The connection's last-response timestamp; undefined on miss OR error (the caller
+   * degrades to a cold probe / full-window replay, never blocks the poll). */
+  get(key: string): Promise<number | undefined>;
+  /** Fire-and-forget stamp; must never throw. */
+  set(key: string, ms: number): void;
+}
+
+/** Per-instance fallback with the same shape (used when the shared store is disabled, and in tests). */
+export class InMemoryReplayCursorStore implements ReplayCursorStore {
+  readonly #cache: BoundedTtlCache<number>;
+
+  constructor(ttlMs: number, maxEntries: number) {
+    this.#cache = new BoundedTtlCache<number>(ttlMs, maxEntries);
+  }
+
+  async get(key: string): Promise<number | undefined> {
+    return this.#cache.get(key);
+  }
+
+  set(key: string, ms: number): void {
+    this.#cache.set(key, ms);
+  }
+}
+
+export type RedisReplayCursorStoreOptions = {
+  redis: RedisWithClusterOptions;
+  /** Entry TTL (ms); matches the working-set TTL so both views of a connection age out together. */
+  ttlMs: number;
+  /** Read deadline (ms): a slow or down Redis degrades the poll to a cold probe instead of stalling it. */
+  getTimeoutMs?: number;
+  keyPrefix?: string;
+  connectionName?: string;
+  /** Observability hook: a store op settled (errors are the degradation signal, not failures). */
+  onResult?: (op: "get" | "set", ok: boolean) => void;
+};
+
+const DEFAULT_KEY_PREFIX = "realtime:replay-cursor:";
+const DEFAULT_GET_TIMEOUT_MS = 250;
+const TIMED_OUT = Symbol("replay-cursor-get-timeout");
+
+export class RedisReplayCursorStore implements ReplayCursorStore {
+  #client: RedisClient | undefined;
+
+  constructor(private readonly options: RedisReplayCursorStoreOptions) {}
+
+  async get(key: string): Promise<number | undefined> {
+    try {
+      const raw = await this.#getWithDeadline(this.#key(key));
+      if (raw === TIMED_OUT) {
+        this.options.onResult?.("get", false);
+        logger.warn("[replayCursorStore] replay-cursor read timed out", { key });
+        return undefined;
+      }
+      this.options.onResult?.("get", true);
+      if (raw === null) {
+        return undefined;
+      }
+      const ms = Number(raw);
+      return Number.isFinite(ms) && ms > 0 ? ms : undefined;
+    } catch (error) {
+      this.options.onResult?.("get", false);
+      logger.error("[replayCursorStore] failed to read a replay cursor", { error, key });
+      return undefined;
+    }
+  }
+
+  /** GET raced against the read deadline (ioredis queues commands while disconnected, which
+   * would otherwise stall every poll start through an outage). */
+  #getWithDeadline(key: string): Promise<string | null | typeof TIMED_OUT> {
+    return new Promise((resolve, reject) => {
+      const timer = setTimeout(
+        () => resolve(TIMED_OUT),
+        this.options.getTimeoutMs ?? DEFAULT_GET_TIMEOUT_MS
+      );
+      timer.unref?.();
+      this.#ensureClient()
+        .get(key)
+        .then(
+          (value) => {
+            clearTimeout(timer);
+            resolve(value);
+          },
+          (error) => {
+            clearTimeout(timer);
+            reject(error);
+          }
+        );
+    });
+  }
+
+  set(key: string, ms: number): void {
+    try {
+      this.#ensureClient()
+        .set(this.#key(key), String(ms), "PX", this.options.ttlMs)
+        .then(
+          () => this.options.onResult?.("set", true),
+          (error) => {
+            this.options.onResult?.("set", false);
+            logger.error("[replayCursorStore] failed to write a replay cursor", { error, key });
+          }
+        );
+    } catch (error) {
+      this.options.onResult?.("set", false);
+      logger.error("[replayCursorStore] failed to write a replay cursor", { error, key });
+    }
+  }
+
+  async quit(): Promise<void> {
+    const client = this.#client;
+    this.#client = undefined;
+    if (!client) return;
+    try {
+      // Bounded graceful QUIT; cursor writes are best-effort, so force-close beyond it.
+      await Promise.race([client.quit(), new Promise((resolve) => setTimeout(resolve, 500))]);
+    } catch {
+      // force-close below
+    }
+    client.disconnect();
+  }
+
+  #key(key: string): string {
+    return `${this.options.keyPrefix ?? DEFAULT_KEY_PREFIX}${key}`;
+  }
+
+  #ensureClient(): RedisClient {
+    if (!this.#client) {
+      this.#client = createRedisClient(
+        this.options.connectionName ?? "trigger:realtime:replay-cursors",
+        this.options.redis
+      );
+    }
+    return this.#client;
+  }
+}