diff --git a/standalone/src/db.js b/standalone/src/db.js index 62c1bd41..434f858c 100644 --- a/standalone/src/db.js +++ b/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]); diff --git a/standalone/test/db-reconnect.test.js b/standalone/test/db-reconnect.test.js new file mode 100644 index 00000000..4727d082 --- /dev/null +++ b/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); + }); +}