Merge pull request #40 from onkernel/mason/redis-failure-handling

fix(redis): add reconnect logic for transient errors

Author: masnwilliams

src/lib/redis.ts

@@ -1,22 +1,49 @@
 import { createClient } from "redis";
 import { createHmac } from "crypto";
 
+// Connect on first use
+let isConnected = false;
+let connectPromise: Promise<void> | null = null;
+
 const client = createClient({
   url: process.env.REDIS_URL,
+  socket: {
+    // Modest backoff to smooth over first-hit cold connections
+    reconnectStrategy: (retries) => Math.min(500 + retries * 100, 2000),
+  },
 });
 
 client.on("error", (err) => {
+  // Reset connection state so the next command will re-connect
+  isConnected = false;
   console.error("Redis Client Error", err);
 });
-
-// Connect on first use
-let isConnected = false;
+client.on("end", () => {
+  isConnected = false;
+});
+client.on("ready", () => {
+  isConnected = true;
+});
 
 async function ensureConnected(): Promise<void> {
-  if (!isConnected) {
-    await client.connect();
-    isConnected = true;
-  }
+  // Prefer the client's readiness state when available
+  // @ts-ignore node-redis exposes isReady at runtime
+  if ((client as any).isReady) return;
+  if (client.isOpen && isConnected) return;
+  if (connectPromise) return await connectPromise;
+  connectPromise = client
+    .connect()
+    .then(() => {
+      // 'ready' event will flip isConnected when the client can process commands
+    })
+    .catch((err) => {
+      isConnected = false;
+      throw err;
+    })
+    .finally(() => {
+      connectPromise = null;
+    });
+  return await connectPromise;
 }
 
 // Hash JWT using HMAC-SHA256 with CLERK_SECRET_KEY for secure Redis storage
@@ -50,7 +77,7 @@ export async function setOrgIdForClientId({
 }): Promise<void> {
   await ensureConnected();
   const key = `client:${clientId}`;
-  await client.setEx(key, ttlSeconds, orgId);
+  await withReconnect(() => client.setEx(key, ttlSeconds, orgId));
 }
 
 export async function getOrgIdForClientId({
@@ -60,7 +87,7 @@ export async function getOrgIdForClientId({
 }): Promise<string | null> {
   await ensureConnected();
   const key = `client:${clientId}`;
-  return await client.get(key);
+  return await withReconnect(() => client.get(key));
 }
 
 export async function setOrgIdForJwt({
@@ -75,7 +102,7 @@ export async function setOrgIdForJwt({
   await ensureConnected();
   const hashedJwt = hashJwt(jwt);
   const key = `jwt:${hashedJwt}`;
-  await client.setEx(key, ttlSeconds, orgId);
+  await withReconnect(() => client.setEx(key, ttlSeconds, orgId));
 }
 
 export { client as redisClient };
@@ -92,7 +119,7 @@ export async function setOrgIdForRefreshToken({
   await ensureConnected();
   const hashed = hashOpaqueToken(refreshToken);
   const key = `refresh:${hashed}`;
-  await client.setEx(key, ttlSeconds, orgId);
+  await withReconnect(() => client.setEx(key, ttlSeconds, orgId));
 }
 
 export async function getOrgIdForRefreshTokenSliding({
@@ -105,10 +132,10 @@ export async function getOrgIdForRefreshTokenSliding({
   await ensureConnected();
   const hashed = hashOpaqueToken(refreshToken);
   const key = `refresh:${hashed}`;
-  const orgId = await client.get(key);
+  const orgId = await withReconnect(() => client.get(key));
   if (orgId) {
     // Refresh TTL to implement sliding expiration on active tokens
-    await client.expire(key, ttlSeconds);
+    await withReconnect(() => client.expire(key, ttlSeconds));
   }
   return orgId;
 }
@@ -121,5 +148,28 @@ export async function deleteOrgIdForRefreshToken({
   await ensureConnected();
   const hashed = hashOpaqueToken(refreshToken);
   const key = `refresh:${hashed}`;
-  await client.del(key);
+  await withReconnect(() => client.del(key));
+}
+
+function isTransientSocketError(error: unknown): boolean {
+  const message = String((error as any)?.message ?? error ?? "");
+  return (
+    message.includes("Socket closed") ||
+    message.includes("ECONNRESET") ||
+    message.includes("EPIPE") ||
+    message.includes("ENETUNREACH")
+  );
+}
+
+async function withReconnect<T>(operation: () => Promise<T>): Promise<T> {
+  try {
+    return await operation();
+  } catch (err) {
+    if (isTransientSocketError(err)) {
+      isConnected = false;
+      await ensureConnected();
+      return await operation();
+    }
+    throw err;
+  }
 }