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.
This commit is contained in:
@@ -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.
|
||||
@@ -11,6 +11,8 @@ export type RedisWithClusterOptions = {
|
||||
clusterMode?: boolean;
|
||||
clusterOptions?: Omit<ClusterOptions, "redisOptions">;
|
||||
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: {} }),
|
||||
});
|
||||
}
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
@@ -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,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user