From 511589c3e91e2d2cf0faaba3d8c850801cba7f70 Mon Sep 17 00:00:00 2001 From: Lorenzo Corallo Date: Thu, 17 Sep 2026 22:35:17 +0200 Subject: [PATCH] fix: recover redis clients and moderation workers after outages --- src/modules/moderation/ban-all.ts | 11 +++- src/redis/index.ts | 15 ++--- src/utils/worker-recovery.ts | 28 ++++++++ tests/redis-recovery.test.ts | 106 ++++++++++++++++++++++++++++++ tests/worker-recovery.test.ts | 66 +++++++++++++++++++ 5 files changed, 214 insertions(+), 12 deletions(-) create mode 100644 src/utils/worker-recovery.ts create mode 100644 tests/redis-recovery.test.ts create mode 100644 tests/worker-recovery.test.ts diff --git a/src/modules/moderation/ban-all.ts b/src/modules/moderation/ban-all.ts index 8b80e4b..a5bda90 100644 --- a/src/modules/moderation/ban-all.ts +++ b/src/modules/moderation/ban-all.ts @@ -6,6 +6,7 @@ import { logger } from "@/logger" import { serialize } from "@/utils/serialize" import { throttleAsyncByKey } from "@/utils/throttle" import type { ModuleShared } from "@/utils/types" +import { runWorkerWithRecovery } from "@/utils/worker-recovery" import { modules } from ".." import { type BanAll, type BanAllState, isBanAllState } from "../tg-logger/ban-all" import { Moderation } from "." @@ -59,6 +60,7 @@ const connection: ConnectionOptions = { * when all bans are executed */ export class BanAllQueue extends Module { + private workerRecovery = new AbortController() /** * Worker that executes the actual ban/unban commands * @@ -211,14 +213,19 @@ export class BanAllQueue extends Module { this.orchestrateQueue.on("progress", handleProgress) this.orchestrator.on("progress", handleProgress) - void this.executor.run().catch((error) => logger.error({ error }, "[BanAllQueue] Executor stopped")) - void this.orchestrator.run().catch((error) => logger.error({ error }, "[BanAllQueue] Orchestrator stopped")) + void runWorkerWithRecovery(this.executor, this.workerRecovery.signal, (error) => + logger.error({ error }, "[BanAllQueue] Executor stopped; restarting after delay") + ) + void runWorkerWithRecovery(this.orchestrator, this.workerRecovery.signal, (error) => + logger.error({ error }, "[BanAllQueue] Orchestrator stopped; restarting after delay") + ) } /** * Gracefully close all the queues and workers */ override async stop() { + this.workerRecovery.abort() await Promise.all([ this.executor.close(), this.orchestrator.close(), diff --git a/src/redis/index.ts b/src/redis/index.ts index 11a1899..5fd4fac 100644 --- a/src/redis/index.ts +++ b/src/redis/index.ts @@ -3,7 +3,6 @@ import { createClient, SocketClosedUnexpectedlyError } from "redis" import { env } from "@/env" import { logger } from "@/logger" -let openSuccess: boolean = false const client = createClient({ socket: { host: env.REDIS_HOST, @@ -11,14 +10,11 @@ const client = createClient({ reconnectStrategy: (retries) => { const n = retries + 1 logger.debug(`[REDIS] reconnect retry #${n}`) - if (openSuccess && n < 5) { - const jitter = Math.floor(Math.random() * 200) - const delay = Math.min(2 ** retries * 50, 2000) - return delay + jitter - } - - if (n < 3) return 1000 - return false + // A storage outage must not permanently close the shared client. Keep + // retrying so the bot and its in-memory adapters recover when Redis does. + const jitter = Math.floor(Math.random() * 200) + const delay = Math.min(2 ** retries * 50, 2000) + return delay + jitter }, }, username: env.REDIS_USERNAME, @@ -43,7 +39,6 @@ async function ready(): Promise { try { await client.connect() - openSuccess = true return true } catch (_) { logger.error("[REDIS] connection failed. Some functions may not work correctly. This should be addressed ASAP.") diff --git a/src/utils/worker-recovery.ts b/src/utils/worker-recovery.ts new file mode 100644 index 0000000..13bd7c6 --- /dev/null +++ b/src/utils/worker-recovery.ts @@ -0,0 +1,28 @@ +import { setTimeout as delay } from "node:timers/promises" +import type { Worker } from "bullmq" + +type RecoverableWorker = Pick + +/** Restart an unexpectedly failed worker loop; respect deliberate shutdown and pause. */ +export async function runWorkerWithRecovery( + worker: RecoverableWorker, + signal: AbortSignal, + onError: (error: unknown) => void +): Promise { + while (!signal.aborted && !worker.closing && !worker.isPaused() && !worker.isRunning()) { + try { + await worker.run() + return + } catch (error) { + onError(error) + } + + if (signal.aborted || worker.closing || worker.isPaused()) return + try { + await delay(1000, undefined, { signal }) + } catch (error) { + if (signal.aborted) return + throw error + } + } +} diff --git a/tests/redis-recovery.test.ts b/tests/redis-recovery.test.ts new file mode 100644 index 0000000..f1c4864 --- /dev/null +++ b/tests/redis-recovery.test.ts @@ -0,0 +1,106 @@ +import { createServer, type Server, type Socket } from "node:net" +import type { RedisClientOptions } from "redis" +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest" + +const fixtures = vi.hoisted(() => ({ + env: { REDIS_HOST: "127.0.0.1", REDIS_PORT: 0 }, + logger: { debug: vi.fn(), info: vi.fn(), error: vi.fn() }, +})) + +vi.mock("@/env", () => ({ env: fixtures.env })) +vi.mock("@/logger", () => ({ logger: fixtures.logger })) +vi.mock("redis", async (importOriginal) => { + const actual = await importOriginal() + return { + ...actual, + createClient: (options: RedisClientOptions) => { + const strategy = options.socket?.reconnectStrategy + return actual.createClient({ + ...options, + // The test server implements only RPUSH. Retry decisions and the client + // are real; only the delay is shortened to keep outage tests fast. + disableClientInfo: true, + socket: { + ...options.socket, + reconnectStrategy: (retries, cause) => { + const delay = typeof strategy === "function" ? strategy(retries, cause) : strategy + return typeof delay === "number" ? 5 : (delay ?? 5) + }, + }, + }) + }, + } +}) + +let server: Server +let sockets: Set +let client: typeof import("@/redis")["redis"] | undefined + +async function listen(port = 0) { + await new Promise((resolve) => server.listen(port, "127.0.0.1", resolve)) + const address = server.address() + if (!address || typeof address === "string") throw new Error("Expected TCP address") + fixtures.env.REDIS_PORT = address.port +} + +async function stopServer() { + for (const socket of sockets) socket.destroy() + if (server.listening) await new Promise((resolve) => server.close(() => resolve())) +} + +async function waitForRetry(attempt: number) { + await vi.waitFor(() => { + expect(fixtures.logger.debug).toHaveBeenCalledWith(`[REDIS] reconnect retry #${attempt}`) + }) +} + +beforeEach(() => { + vi.resetModules() + vi.clearAllMocks() + sockets = new Set() + server = createServer((socket) => { + sockets.add(socket) + socket.on("close", () => sockets.delete(socket)) + let request = "" + socket.on("data", (chunk) => { + request += chunk.toString() + if (request.endsWith("$5\r\nentry\r\n")) { + socket.write(":1\r\n") + request = "" + } + }) + }) +}) + +afterEach(async () => { + if (client?.isOpen) await client.disconnect() + client = undefined + await stopServer() +}) + +describe("Redis outage recovery", () => { + it("runs RPUSH after Redis starts later than the initial retry budget", async () => { + await listen() + const port = fixtures.env.REDIS_PORT + await stopServer() + client = (await import("@/redis")).redis + await waitForRetry(3) + + await listen(port) + await expect(client.rPush("moderation:test", "entry")).resolves.toBe(1) + expect(client.isReady).toBe(true) + }) + + it("runs RPUSH after a connected server outlasts the reconnect budget", async () => { + await listen() + const port = fixtures.env.REDIS_PORT + client = (await import("@/redis")).redis + await vi.waitFor(() => expect(client?.isReady).toBe(true)) + await stopServer() + await waitForRetry(5) + + await listen(port) + await expect(client.rPush("moderation:test", "entry")).resolves.toBe(1) + expect(client.isReady).toBe(true) + }) +}) diff --git a/tests/worker-recovery.test.ts b/tests/worker-recovery.test.ts new file mode 100644 index 0000000..a0bcf9b --- /dev/null +++ b/tests/worker-recovery.test.ts @@ -0,0 +1,66 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest" +import { runWorkerWithRecovery } from "@/utils/worker-recovery" + +beforeEach(() => vi.useFakeTimers()) +afterEach(() => vi.useRealTimers()) + +function fixture() { + return { + run: vi.fn<() => Promise>(), + isRunning: vi.fn(() => false), + isPaused: vi.fn(() => false), + closing: undefined as Promise | undefined, + } +} + +describe("worker recovery", () => { + it("restarts a loop that stopped on a connection error and then processes work", async () => { + const worker = fixture() + const onError = vi.fn() + const processed = vi.fn() + const failure = new Error("Connection is closed") + worker.run.mockRejectedValueOnce(failure).mockImplementationOnce(async () => processed()) + + const task = runWorkerWithRecovery(worker, new AbortController().signal, onError) + await vi.advanceTimersByTimeAsync(999) + expect(processed).not.toHaveBeenCalled() + await vi.advanceTimersByTimeAsync(1) + await task + + expect(onError).toHaveBeenCalledWith(failure) + expect(processed).toHaveBeenCalledOnce() + expect(worker.run).toHaveBeenCalledTimes(2) + }) + + it("cancels a pending restart when the module stops", async () => { + const worker = fixture() + const controller = new AbortController() + worker.run.mockRejectedValueOnce(new Error("Connection is closed")) + const task = runWorkerWithRecovery(worker, controller.signal, vi.fn()) + await vi.advanceTimersByTimeAsync(100) + controller.abort() + await task + await vi.advanceTimersByTimeAsync(2000) + expect(worker.run).toHaveBeenCalledOnce() + }) + + it.each(["paused", "closing", "running"])("does not restart a %s worker", async (state) => { + const worker = fixture() + if (state === "paused") worker.isPaused.mockReturnValue(true) + if (state === "running") worker.isRunning.mockReturnValue(true) + if (state === "closing") worker.closing = Promise.resolve() + await runWorkerWithRecovery(worker, new AbortController().signal, vi.fn()) + expect(worker.run).not.toHaveBeenCalled() + }) + + it("does not restart after a deliberate pause following an error", async () => { + const worker = fixture() + worker.run.mockImplementationOnce(async () => { + worker.isPaused.mockReturnValue(true) + throw new Error("Connection is closed") + }) + await runWorkerWithRecovery(worker, new AbortController().signal, vi.fn()) + await vi.advanceTimersByTimeAsync(2000) + expect(worker.run).toHaveBeenCalledOnce() + }) +})