Eric Allam
567e2a2c32
## Summary
During an ElastiCache role swap (failover) or node-type change (vertical
scale), the ioredis TCP/TLS connection stays open but the server starts
answering with `READONLY` (the client is talking to a node that became a
replica) or `LOADING` (node still loading data from disk). Without an
explicit hook, those errors surface to caller code as `ReplyError`
instances — every write op on the affected connection fails until the
cluster fully cuts over.
This PR adds `reconnectOnError` to every prod ioredis client so the
disconnect + reconnect + retry cycle absorbs these errors and caller
code never sees them.
## Fix
```ts
export function defaultReconnectOnError(err: Error): boolean | 1 | 2 {
const msg = err.message ?? "";
if (msg.startsWith("READONLY") || msg.startsWith("LOADING")) return 2;
return false;
}
```
Returning `2` tells ioredis to disconnect, reconnect, and re-issue the
failed command. After reconnect, DNS / SG state routes the new socket to
a writable node.
The helper lives in `@internal/redis` and is wired into both the shared
`createRedisClient` (which covers RunQueue, schedule-engine,
redis-worker, and every other internal-package consumer) and the direct
`new Redis(...)` call sites in the webapp.
V1-only marqs files are intentionally not migrated.
## Test plan
- [x] `pnpm run typecheck --filter webapp`
- [x] `pnpm run typecheck --filter @internal/run-engine`
- [x] Verified end-to-end against a live ElastiCache vertical-scale
event — caller-surfaced errors went from tens of thousands during the
cutover window down to a handful per ioredis client
- [ ] Confirm steady-state behavior unchanged after deploy
85 lines
2.1 KiB
TypeScript
85 lines
2.1 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;
|
|
};
|
|
|
|
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.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.tlsDisabled ? {} : { tls: {} }),
|
|
});
|
|
}
|
|
|
|
redis.on("error", (error) => {
|
|
logger.error("Redis client error", { connectionName, error });
|
|
});
|
|
|
|
return redis;
|
|
}
|