fix: recover redis clients and moderation workers after outages

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<ModuleShared> {
+  private workerRecovery = new AbortController()
   /**
    * Worker that executes the actual ban/unban commands
    *
@@ -211,14 +213,19 @@ export class BanAllQueue extends Module<ModuleShared> {
 
     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(),

src/redis/index.ts

@@ -3,22 +3,18 @@ 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,
     port: env.REDIS_PORT,
     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<boolean> {
 
   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.")

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<Worker, "run" | "isRunning" | "isPaused" | "closing">
+
+/** Restart an unexpectedly failed worker loop; respect deliberate shutdown and pause. */
+export async function runWorkerWithRecovery(
+  worker: RecoverableWorker,
+  signal: AbortSignal,
+  onError: (error: unknown) => void
+): Promise<void> {
+  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
+    }
+  }
+}

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<typeof import("redis")>()
+  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<Socket>
+let client: typeof import("@/redis")["redis"] | undefined
+
+async function listen(port = 0) {
+  await new Promise<void>((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<void>((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)
+  })
+})

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<void>>(),
+    isRunning: vi.fn(() => false),
+    isPaused: vi.fn(() => false),
+    closing: undefined as Promise<void> | 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()
+  })
+})