Files
triggerdotdev--trigger.dev/apps/webapp/app/services/autoIncrementCounter.server.ts
Eric Allam 567e2a2c32 feat(webapp,redis): handle READONLY / LOADING during ElastiCache failover (#3548)
## 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
2026-05-11 07:17:07 +01:00

85 lines
2.5 KiB
TypeScript

import Redis, { RedisOptions } from "ioredis";
import { defaultReconnectOnError } from "@internal/redis";
import { Prisma, PrismaClientOrTransaction, PrismaTransactionOptions, prisma } from "~/db.server";
import { env } from "~/env.server";
import { singleton } from "~/utils/singleton";
export type AutoIncrementCounterOptions = {
redis: RedisOptions;
};
export class AutoIncrementCounter {
private _redis: Redis;
constructor(private options: AutoIncrementCounterOptions) {
this._redis = new Redis({ reconnectOnError: defaultReconnectOnError, ...options.redis });
}
async incrementInTransaction<T>(
key: string,
callback: (num: number, tx: PrismaClientOrTransaction) => Promise<T>,
backfiller?: (key: string, db: PrismaClientOrTransaction) => Promise<number | undefined>,
client: PrismaClientOrTransaction = prisma,
transactionOptions?: PrismaTransactionOptions
): Promise<T | undefined> {
let performedIncrement = false;
let performedBackfill = false;
try {
let newNumber = await this.#increment(key);
performedIncrement = true;
if (newNumber === 1 && backfiller) {
const backfilledNumber = await backfiller(key, client);
if (backfilledNumber && backfilledNumber > 1) {
newNumber = backfilledNumber + 1;
await this._redis.set(key, newNumber);
performedBackfill = true;
}
}
return await callback(newNumber, client);
} catch (e) {
if (
e instanceof Prisma.PrismaClientKnownRequestError ||
e instanceof Prisma.PrismaClientUnknownRequestError ||
e instanceof Prisma.PrismaClientValidationError
) {
if (performedIncrement && !performedBackfill) {
await this._redis.decr(key);
}
}
throw e;
}
}
async #increment(key: string): Promise<number> {
return await this._redis.incr(key);
}
}
export const autoIncrementCounter = singleton("auto-increment-counter", getAutoIncrementCounter);
function getAutoIncrementCounter() {
if (!env.REDIS_HOST || !env.REDIS_PORT) {
throw new Error(
"Could not initialize auto-increment counter because process.env.REDIS_HOST and process.env.REDIS_PORT are required to be set. "
);
}
return new AutoIncrementCounter({
redis: {
keyPrefix: "auto-counter:",
port: env.REDIS_PORT,
host: env.REDIS_HOST,
username: env.REDIS_USERNAME,
password: env.REDIS_PASSWORD,
enableAutoPipelining: true,
...(env.REDIS_TLS_DISABLED === "true" ? {} : { tls: {} }),
},
});
}