fix(standalone): reconnect to redis after connection loss

standalone/src/db.js

@@ -4,12 +4,12 @@ const redisUrl =
   process.env.REDIS_URL || process.env.VALKEY_URL || "redis://localhost:6379";
 const prefix = process.env.REDIS_PREFIX || "";
 
-const client = new RedisClient(redisUrl);
+let client = new RedisClient(redisUrl);
+let pendingReconnect = null;
 
 await client.send("PING", []);
 
-const addPrefix = (k) =>
-  typeof k === "string" && k.length ? prefix + k : k;
+const addPrefix = (k) => (typeof k === "string" && k.length ? prefix + k : k);
 const stripPrefix = (k) =>
   typeof k === "string" && k.startsWith(prefix) ? k.slice(prefix.length) : k;
 
@@ -42,39 +42,119 @@ const KEY_ALL = new Set(["del", "unlink", "exists", "mget"]);
 
 const MULTI_KEY_CMDS = new Set(["DEL", "UNLINK", "MGET", "EXISTS"]);
 
+const RECONNECT_BASE_DELAY_MS = 1000;
+const RECONNECT_MAX_DELAY_MS = 60000;
+const RECONNECT_ATTEMPT_TIMEOUT_MS = 3000;
+
+function attemptConnect() {
+  const next = new RedisClient(redisUrl);
+  const ping = next.send("PING", []);
+  ping.catch(() => {});
+  let timer;
+  const timeout = new Promise((_, reject) => {
+    timer = setTimeout(
+      () => reject(new Error("redis reconnect attempt timed out")),
+      RECONNECT_ATTEMPT_TIMEOUT_MS,
+    );
+  });
+  return Promise.race([ping, timeout])
+    .then(
+      () => next,
+      (error) => {
+        next.close();
+        throw error;
+      },
+    )
+    .finally(() => clearTimeout(timer));
+}
+
+async function runReconnectLoop() {
+  let delay = RECONNECT_BASE_DELAY_MS;
+  for (;;) {
+    try {
+      const next = await attemptConnect();
+      client.close();
+      client = next;
+      console.error("[cap] redis reconnected");
+      return;
+    } catch {
+      await Bun.sleep(delay);
+      delay = Math.min(delay * 2, RECONNECT_MAX_DELAY_MS);
+    }
+  }
+}
+
+function ensureReconnect() {
+  if (!pendingReconnect) {
+    console.error("[cap] redis connection lost, reconnecting");
+    pendingReconnect = runReconnectLoop().finally(() => {
+      pendingReconnect = null;
+    });
+  }
+  return pendingReconnect;
+}
+
+function isConnectionError(error) {
+  return (
+    error?.code === "ERR_REDIS_CONNECTION_CLOSED" ||
+    error?.code === "ERR_REDIS_CONNECTION_TIMEOUT"
+  );
+}
+
+async function withReconnect(run, retried = false) {
+  try {
+    return await run();
+  } catch (error) {
+    if (retried || !isConnectionError(error)) throw error;
+    const recovery = ensureReconnect();
+    if (error.code === "ERR_REDIS_CONNECTION_TIMEOUT") throw error;
+    await Promise.race([
+      recovery,
+      Bun.sleep(RECONNECT_ATTEMPT_TIMEOUT_MS + 500),
+    ]);
+    return withReconnect(run, true);
+  }
+}
+
+function rawSend(cmd, args = []) {
+  return withReconnect(() => client.send(cmd, args));
+}
+
 function prefixedSend(cmd, args = []) {
   const upper = cmd.toUpperCase();
-  if (upper === "PING" || !args.length) return client.send(cmd, args);
+  if (upper === "PING" || !args.length) return rawSend(cmd, args);
   if (upper === "KEYS") {
-    return client
-      .send(cmd, [addPrefix(args[0]), ...args.slice(1)])
-      .then((res) => (Array.isArray(res) ? res.map(stripPrefix) : res));
+    return rawSend(cmd, [addPrefix(args[0]), ...args.slice(1)]).then((res) =>
+      Array.isArray(res) ? res.map(stripPrefix) : res,
+    );
   }
   if (MULTI_KEY_CMDS.has(upper)) {
-    return client.send(cmd, args.map(addPrefix));
+    return rawSend(cmd, args.map(addPrefix));
   }
-  return client.send(cmd, [addPrefix(args[0]), ...args.slice(1)]);
+  return rawSend(cmd, [addPrefix(args[0]), ...args.slice(1)]);
 }
 
-const db = prefix
-  ? new Proxy(client, {
-      get(target, prop, receiver) {
-        if (prop === "send") return prefixedSend;
-        const value = Reflect.get(target, prop, receiver);
-        if (typeof value !== "function") return value;
-        if (KEY_FIRST.has(prop)) {
-          return (...args) => {
-            if (args.length) args[0] = addPrefix(args[0]);
-            return value.apply(target, args);
-          };
-        }
-        if (KEY_ALL.has(prop)) {
-          return (...args) => value.apply(target, args.map(addPrefix));
-        }
-        return value.bind(target);
-      },
-    })
-  : client;
+const db = new Proxy(
+  {},
+  {
+    get(_target, prop) {
+      if (prop === "send") return prefix ? prefixedSend : rawSend;
+      const value = client[prop];
+      if (typeof value !== "function") return value;
+      if (prefix && KEY_FIRST.has(prop)) {
+        return (...args) => {
+          if (args.length) args[0] = addPrefix(args[0]);
+          return withReconnect(() => client[prop](...args));
+        };
+      }
+      if (prefix && KEY_ALL.has(prop)) {
+        return (...args) =>
+          withReconnect(() => client[prop](...args.map(addPrefix)));
+      }
+      return (...args) => withReconnect(() => client[prop](...args));
+    },
+  },
+);
 
 export async function hgetall(key) {
   const data = await db.send("HGETALL", [key]);

standalone/test/db-reconnect.test.js

@@ -0,0 +1,106 @@
+import { afterAll, describe, expect, test } from "bun:test";
+import { RedisClient } from "bun";
+
+const REDIS_URL =
+  process.env.REDIS_URL || process.env.VALKEY_URL || "redis://127.0.0.1:6379";
+
+let redisAvailable = false;
+let probeClient;
+try {
+  probeClient = new RedisClient(REDIS_URL);
+  await probeClient.send("PING", []);
+  redisAvailable = true;
+} catch (e) {
+  console.warn(
+    "[db-reconnect-test] redis not available, skipping reconnect tests:",
+    e.message,
+  );
+}
+
+if (!redisAvailable) {
+  test.skip(`db reconnect skipped (no redis at ${REDIS_URL})`, () => {});
+} else {
+  const backend = new URL(REDIS_URL);
+  const backendHost = backend.hostname;
+  const backendPort = Number(backend.port || 6379);
+
+  let listener = null;
+
+  function startProxy(port) {
+    listener = Bun.listen({
+      hostname: "127.0.0.1",
+      port,
+      socket: {
+        open(socket) {
+          socket.data = { upstream: null, buffered: [] };
+          Bun.connect({
+            hostname: backendHost,
+            port: backendPort,
+            socket: {
+              open(upstream) {
+                upstream.data = socket;
+                socket.data.upstream = upstream;
+                for (const chunk of socket.data.buffered) upstream.write(chunk);
+                socket.data.buffered = [];
+              },
+              data(upstream, chunk) {
+                upstream.data.write(chunk);
+              },
+              close(upstream) {
+                upstream.data.end();
+              },
+            },
+          }).catch(() => socket.end());
+        },
+        data(socket, chunk) {
+          if (socket.data.upstream) {
+            socket.data.upstream.write(chunk);
+          } else {
+            socket.data.buffered.push(chunk);
+          }
+        },
+        close(socket) {
+          socket.data.upstream?.end();
+        },
+      },
+    });
+    return listener.port;
+  }
+
+  function stopProxy() {
+    listener.stop(true);
+    listener = null;
+  }
+
+  const proxyPort = startProxy(0);
+  const originalRedisUrl = process.env.REDIS_URL;
+  process.env.REDIS_URL = `redis://127.0.0.1:${proxyPort}`;
+  const { db } = await import("../src/db.js?reconnect");
+  process.env.REDIS_URL = originalRedisUrl;
+
+  afterAll(async () => {
+    try {
+      await db.del("test:reconnect");
+    } catch {}
+    listener?.stop(true);
+    probeClient?.close();
+  });
+
+  describe("db reconnect", () => {
+    test("recovers after the redis connection is lost", async () => {
+      await db.set("test:reconnect", "before");
+      expect(await db.get("test:reconnect")).toBe("before");
+
+      stopProxy();
+      await Bun.sleep(50);
+      await expect(db.get("test:reconnect")).rejects.toThrow();
+
+      startProxy(proxyPort);
+      await Bun.sleep(50);
+
+      expect(await db.get("test:reconnect")).toBe("before");
+      await db.set("test:reconnect", "after");
+      expect(await db.get("test:reconnect")).toBe("after");
+    }, 60000);
+  });
+}