From ef998a518ba1a39ebce9f206d58d8e3f736e3bad Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Mon, 15 Jun 2026 11:55:49 +0100 Subject: [PATCH] fix(webapp): make native realtime change publishing fail-safe (#3946) Two defensive fixes to the native realtime backend's run-change publishing (behind a feature flag, off by default), so turning it on can never destabilize the run lifecycle. **Never throws at the caller.** Publish sites run synchronously on the run-engine event bus and the metadata flush loop. The internal publish was already wrapped in try/catch, but lazy construction (singleton + metrics) and record encoding ran before that guard, so a throw could propagate into a run lifecycle operation. The public `publishChangeRecord` / `publishManyChangeRecords` helpers now wrap the whole call and log-and-drop on failure. **Bounds outage buffering.** The publisher connection caps `maxRetriesPerRequest` at 1 (vs ioredis's default of 20), so during a pub/sub Redis outage a publish rejects after ~1 reconnect cycle instead of holding commands in memory for ~20s. A dropped publish is latency-only, since the consumer has a periodic backstop full-resolve. The offline queue stays on, so the first publish after a process boots still flushes once the connection is ready. --- .../harden-realtime-change-publishing.md | 6 ++++++ apps/webapp/app/redis.server.ts | 8 ++++++++ .../realtime/runChangeNotifier.server.ts | 8 +++++++- .../runChangeNotifierInstance.server.ts | 17 +++++++++++++++-- 4 files changed, 36 insertions(+), 3 deletions(-) create mode 100644 .server-changes/harden-realtime-change-publishing.md diff --git a/.server-changes/harden-realtime-change-publishing.md b/.server-changes/harden-realtime-change-publishing.md new file mode 100644 index 000000000..0dff1e1ea --- /dev/null +++ b/.server-changes/harden-realtime-change-publishing.md @@ -0,0 +1,6 @@ +--- +area: webapp +type: improvement +--- + +Harden the native realtime backend's run-change publishing so a publish can never throw into a run lifecycle operation and never buffers commands in memory during a pub/sub Redis outage. diff --git a/apps/webapp/app/redis.server.ts b/apps/webapp/app/redis.server.ts index 01efa0d3e..211c97829 100644 --- a/apps/webapp/app/redis.server.ts +++ b/apps/webapp/app/redis.server.ts @@ -11,6 +11,8 @@ export type RedisWithClusterOptions = { clusterMode?: boolean; clusterOptions?: Omit; keyPrefix?: string; + /** Cap retries for a command before it rejects; `null` means unlimited (default: ioredis's default of 20). */ + maxRetriesPerRequest?: number | null; }; export type RedisClient = Redis | Cluster; @@ -44,6 +46,9 @@ export function createRedisClient( password: options.password, enableAutoPipelining: true, reconnectOnError: defaultReconnectOnError, + ...(options.maxRetriesPerRequest !== undefined + ? { maxRetriesPerRequest: options.maxRetriesPerRequest } + : {}), ...(options.tlsDisabled ? { checkServerIdentity: () => { @@ -72,6 +77,9 @@ export function createRedisClient( enableAutoPipelining: true, keyPrefix: options.keyPrefix, reconnectOnError: defaultReconnectOnError, + ...(options.maxRetriesPerRequest !== undefined + ? { maxRetriesPerRequest: options.maxRetriesPerRequest } + : {}), ...(options.tlsDisabled ? {} : { tls: {} }), }); } diff --git a/apps/webapp/app/services/realtime/runChangeNotifier.server.ts b/apps/webapp/app/services/realtime/runChangeNotifier.server.ts index f295c02d3..d413dcb93 100644 --- a/apps/webapp/app/services/realtime/runChangeNotifier.server.ts +++ b/apps/webapp/app/services/realtime/runChangeNotifier.server.ts @@ -225,7 +225,13 @@ export class RunChangeNotifier { #ensurePublisher(): RedisClient { if (!this.#publisher) { - this.#publisher = createRedisClient(`${this.#connectionName}:pub`, this.options.redis); + // Publishes are fire-and-forget with a consumer-side backstop, so a dropped publish is + // latency-only. Cap retries (vs ioredis's default 20) so a pub/sub outage rejects publishes + // after ~1 reconnect cycle instead of buffering them in memory across the fleet. + this.#publisher = createRedisClient(`${this.#connectionName}:pub`, { + ...this.options.redis, + maxRetriesPerRequest: 1, + }); } return this.#publisher; } diff --git a/apps/webapp/app/services/realtime/runChangeNotifierInstance.server.ts b/apps/webapp/app/services/realtime/runChangeNotifierInstance.server.ts index b656052c3..7300d081b 100644 --- a/apps/webapp/app/services/realtime/runChangeNotifierInstance.server.ts +++ b/apps/webapp/app/services/realtime/runChangeNotifierInstance.server.ts @@ -1,6 +1,7 @@ import { getMeter } from "@internal/tracing"; import { env } from "~/env.server"; import { singleton } from "~/utils/singleton"; +import { logger } from "../logger.server"; import { RunChangeNotifier, type ChangeRecordInput } from "./runChangeNotifier.server"; /** @@ -74,12 +75,24 @@ export function publishChangeRecord(input: ChangeRecordInput): void { if (!nativeBackendEnabled) { return; } - getRunChangeNotifier().publish(input); + // Publish runs on the run-engine event bus / metadata flush loop; lazy init + encoding happen + // before the notifier's own try/catch, so guard the whole call — it must never throw at its caller. + try { + getRunChangeNotifier().publish(input); + } catch (error) { + logger.error("[runChangeNotifier] publishChangeRecord threw; dropping notification", { error }); + } } export function publishManyChangeRecords(inputs: ChangeRecordInput[]): void { if (!nativeBackendEnabled) { return; } - getRunChangeNotifier().publishMany(inputs); + try { + getRunChangeNotifier().publishMany(inputs); + } catch (error) { + logger.error("[runChangeNotifier] publishManyChangeRecords threw; dropping notifications", { + error, + }); + } }