Eric Allam
ef998a518b
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.
93 lines
2.5 KiB
TypeScript
93 lines
2.5 KiB
TypeScript
import { Cluster, Redis, type ClusterNode, type ClusterOptions } from "ioredis";
|
|
import { defaultReconnectOnError } from "@internal/redis";
|
|
import { logger } from "./services/logger.server";
|
|
|
|
export type RedisWithClusterOptions = {
|
|
host?: string;
|
|
port?: number;
|
|
username?: string;
|
|
password?: string;
|
|
tlsDisabled?: boolean;
|
|
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;
|
|
|
|
export function createRedisClient(
|
|
connectionName: string,
|
|
options: RedisWithClusterOptions
|
|
): Redis | Cluster {
|
|
let redis: Redis | Cluster;
|
|
|
|
if (options.clusterMode) {
|
|
const nodes: ClusterNode[] = [
|
|
{
|
|
host: options.host,
|
|
port: options.port,
|
|
},
|
|
];
|
|
|
|
logger.debug("Creating a redis cluster client", {
|
|
connectionName,
|
|
host: options.host,
|
|
port: options.port,
|
|
});
|
|
|
|
redis = new Redis.Cluster(nodes, {
|
|
...options.clusterOptions,
|
|
redisOptions: {
|
|
connectionName,
|
|
keyPrefix: options.keyPrefix,
|
|
username: options.username,
|
|
password: options.password,
|
|
enableAutoPipelining: true,
|
|
reconnectOnError: defaultReconnectOnError,
|
|
...(options.maxRetriesPerRequest !== undefined
|
|
? { maxRetriesPerRequest: options.maxRetriesPerRequest }
|
|
: {}),
|
|
...(options.tlsDisabled
|
|
? {
|
|
checkServerIdentity: () => {
|
|
// disable TLS verification
|
|
return undefined;
|
|
},
|
|
}
|
|
: { tls: {} }),
|
|
},
|
|
dnsLookup: (address, callback) => callback(null, address),
|
|
slotsRefreshTimeout: 10000,
|
|
});
|
|
} else {
|
|
logger.debug("Creating a redis client", {
|
|
connectionName,
|
|
host: options.host,
|
|
port: options.port,
|
|
});
|
|
|
|
redis = new Redis({
|
|
connectionName,
|
|
host: options.host,
|
|
port: options.port,
|
|
username: options.username,
|
|
password: options.password,
|
|
enableAutoPipelining: true,
|
|
keyPrefix: options.keyPrefix,
|
|
reconnectOnError: defaultReconnectOnError,
|
|
...(options.maxRetriesPerRequest !== undefined
|
|
? { maxRetriesPerRequest: options.maxRetriesPerRequest }
|
|
: {}),
|
|
...(options.tlsDisabled ? {} : { tls: {} }),
|
|
});
|
|
}
|
|
|
|
redis.on("error", (error) => {
|
|
logger.error("Redis client error", { connectionName, error });
|
|
});
|
|
|
|
return redis;
|
|
}
|