diff --git a/server/src/external/redis/initRedis.ts b/server/src/external/redis/initRedis.ts index 47c6426b3..57bb26bed 100644 --- a/server/src/external/redis/initRedis.ts +++ b/server/src/external/redis/initRedis.ts @@ -29,6 +29,7 @@ import { UPSERT_INVOICE_IN_CUSTOMER_SCRIPT, } from "../../_luaScriptsV2/luaScriptsV2.js"; import { instrumentRedis } from "../../utils/otel/instrumentRedis.js"; +import { getActiveRedis, initFailover } from "./redisFailover.js"; // if (!process.env.CACHE_URL) { // throw new Error("CACHE_URL (redis) is not set"); @@ -269,19 +270,78 @@ if (primaryCacheUrl && regionToCacheUrl[currentRegion]) { console.log(`Using regional cache: ${currentRegion}`); } -const redis = createRedisConnection({ +const primaryRedis = createRedisConnection({ cacheUrl: primaryCacheUrl!, region: currentRegion, }); +// Eagerly create failover instance (other region) for automatic failover +const failoverRegion = + ALL_REGIONS.find((r) => r !== currentRegion && regionToCacheUrl[r]) ?? null; + +let failoverRedis: Redis | null = null; +if (failoverRegion) { + const failoverUrl = regionToCacheUrl[failoverRegion]!; + // Only create a separate instance if it's actually a different server + if (failoverUrl !== primaryCacheUrl) { + failoverRedis = createRedisConnection({ + cacheUrl: failoverUrl, + region: failoverRegion, + }); + } +} + +// Initialize failover — monitors primary health and swaps `redis` automatically +initFailover({ + primary: primaryRedis, + failover: failoverRedis, + failoverRegion, + currentRegion, +}); + +/** + * The active Redis instance. All consumer code imports this. + * Normally points to the primary (current region). During a primary outage, + * the failover module swaps this to the other region's instance automatically. + * + * This is a `let` so it's a live ES module binding — reassignments here + * are visible to all importers on their next access. + */ +export let redis: Redis = primaryRedis; + +// Subscribe to failover state changes — keep the `redis` export in sync. +// We do this here (not in redisFailover.ts) because the module binding +// can only be reassigned in the module that declares it. +const syncRedisBinding = () => { + const active = getActiveRedis(); + if (redis !== active) { + redis = active; + } +}; + +primaryRedis.on("error", syncRedisBinding); +primaryRedis.on("ready", syncRedisBinding); +if (failoverRedis) { + failoverRedis.on("error", syncRedisBinding); + failoverRedis.on("ready", syncRedisBinding); +} +// Also poll periodically to catch any edge cases with event timing +setInterval(syncRedisBinding, 2000); + // Lazy-loaded regional Redis instances for cross-region sync const regionalRedisInstances: Map = new Map(); +// Pre-populate with eagerly created instances +if (failoverRedis && failoverRegion) { + regionalRedisInstances.set(failoverRegion, failoverRedis); +} + /** Get Redis instance for a specific region (lazy-loaded) */ export const getRegionalRedis = (region: string): Redis => { - // If requesting current region, return primary instance + // Always return the actual primary for the current region (not the active/failover) + // so cross-region sync logic isn't affected by failover state. if (region === currentRegion) { - return redis; + return primaryRedis; } // Get the cache URL for the requested region @@ -292,13 +352,12 @@ export const getRegionalRedis = (region: string): Redis => { console.warn( `No cache URL configured for region ${region}, falling back to primary`, ); - return redis; + return primaryRedis; } // If the cache URL is the same as primary, return primary instance - // (avoids creating duplicate connections to the same server) if (cacheUrl === primaryCacheUrl) { - return redis; + return primaryRedis; } // Check if we already have a connection for this region @@ -456,5 +515,3 @@ declare module "ioredis" { /** Get the primary Redis instance (us-west-2) to avoid replication lag issues */ export const getPrimaryRedis = () => getRegionalRedis(REGION_US_WEST_2); - -export { redis }; diff --git a/server/src/external/redis/redisFailover.ts b/server/src/external/redis/redisFailover.ts new file mode 100644 index 000000000..1959ced89 --- /dev/null +++ b/server/src/external/redis/redisFailover.ts @@ -0,0 +1,153 @@ +import type { Redis } from "ioredis"; + +/** How long primary must be erroring before we switch to failover. */ +const FAILOVER_DELAY_MS = 5_000; + +/** How long primary must be stable before we switch back from failover. */ +const RECOVERY_DELAY_MS = 3_000; + +export type RedisFailoverState = { + /** The currently active Redis instance (what consumers use). */ + active: Redis; + /** Always the current region's instance. */ + primary: Redis; + /** The other region's instance (null if only one region configured). */ + failover: Redis | null; + /** Whether we're currently using the failover instance. */ + isUsingFailover: boolean; + /** Region name of the failover instance. */ + failoverRegion: string | null; +}; + +let state: RedisFailoverState; +let primaryErrorSince: number | null = null; +let recoveryTimer: ReturnType | null = null; +let primaryHasBeenReady = false; + +/** Initialize failover state. Call once after creating both Redis instances. */ +export const initFailover = ({ + primary, + failover, + failoverRegion, + currentRegion, +}: { + primary: Redis; + failover: Redis | null; + failoverRegion: string | null; + currentRegion: string; +}): void => { + state = { + active: primary, + primary, + failover, + isUsingFailover: false, + failoverRegion, + }; + + if (!failover) { + console.log( + "[Redis failover] No failover region configured — failover disabled", + ); + return; + } + + console.log( + `[Redis failover] Enabled: primary=${currentRegion}, failover=${failoverRegion}`, + ); + + const onPrimaryDown = () => { + // Don't trigger failover during initial startup — only after + // the primary has successfully connected at least once. + if (!primaryHasBeenReady) return; + + if (!primaryErrorSince) { + primaryErrorSince = Date.now(); + + // Schedule failover after the delay + setTimeout(() => { + if (primaryErrorSince && primary.status !== "ready") { + switchToFailover(); + } + }, FAILOVER_DELAY_MS); + } + }; + + // Listen to all events that indicate the primary is unhealthy + primary.on("error", onPrimaryDown); + primary.on("close", onPrimaryDown); + primary.on("end", onPrimaryDown); + + primary.on("ready", () => { + primaryHasBeenReady = true; + primaryErrorSince = null; + + if (!state.isUsingFailover) return; + + // Primary recovered — wait for stability before switching back + if (!recoveryTimer) { + recoveryTimer = setTimeout(() => { + recoveryTimer = null; + if (primary.status === "ready") { + switchToPrimary(); + } + }, RECOVERY_DELAY_MS); + } + }); +}; + +const switchToFailover = (): void => { + if (!state.failover || state.isUsingFailover) return; + if (state.failover.status !== "ready") { + console.error( + "[Redis failover] Cannot switch — failover instance is not ready", + ); + return; + } + + state.active = state.failover; + state.isUsingFailover = true; + console.log( + `[Redis failover] SWITCHED to failover region (${state.failoverRegion})`, + ); +}; + +const switchToPrimary = (): void => { + if (!state.isUsingFailover) return; + + state.active = state.primary; + state.isUsingFailover = false; + primaryErrorSince = null; + console.log("[Redis failover] RECOVERED to primary region"); +}; + +/** Get the currently active Redis instance. */ +export const getActiveRedis = (): Redis => state.active; + +/** Get the current failover state (for debug/monitoring). */ +export const getFailoverState = (): { + isUsingFailover: boolean; + failoverRegion: string | null; + primaryStatus: string; + failoverStatus: string | null; + primaryErrorSince: number | null; +} => ({ + isUsingFailover: state.isUsingFailover, + failoverRegion: state.failoverRegion, + primaryStatus: state.primary.status, + failoverStatus: state.failover?.status ?? null, + primaryErrorSince, +}); + +/** + * Force disconnect the primary instance (for testing). + * ioredis will NOT auto-reconnect after a manual disconnect() call. + * Use `reconnectPrimary()` to manually reconnect. + */ +export const disconnectPrimary = (): void => { + state.primary.disconnect(); +}; + +/** Force reconnect the primary instance (for testing). */ +export const reconnectPrimary = (): void => { + state.primary.connect(); +}; diff --git a/server/src/internal/debug/debugRouter.ts b/server/src/internal/debug/debugRouter.ts index 0c2b5611f..edc1355c5 100644 --- a/server/src/internal/debug/debugRouter.ts +++ b/server/src/internal/debug/debugRouter.ts @@ -1,11 +1,18 @@ import { sql } from "drizzle-orm"; import { Hono } from "hono"; import { dbCritical, dbGeneral } from "@/db/initDrizzle.js"; +import { redis } from "@/external/redis/initRedis.js"; +import { + disconnectPrimary, + getFailoverState, + reconnectPrimary, +} from "@/external/redis/redisFailover.js"; import { orgConfigMiddleware } from "@/honoMiddlewares/orgConfigMiddleware.js"; import { secretKeyMiddleware } from "@/honoMiddlewares/secretKeyMiddleware.js"; import type { HonoEnv } from "@/honoUtils/HonoEnv.js"; const ALLOWED_ORG_IDS = new Set([ + "org_2sWv2S8LJ9iaTjLI6UtNsfL88Kt", "org_2rzkkRh7r5dBSaBC101QHG9KDgt", "org_2vwdxwTdqxRrLEdUYddcynMv3n3", ]); @@ -47,10 +54,19 @@ debugRouter.post("/pool-test", async (c) => { return c.json({ error: "Not available in production" }, 403); } + const ctx = c.get("ctx"); + if (!ALLOWED_ORG_IDS.has(ctx.org?.id)) { + return c.json({ error: "Forbidden" }, 403); + } + if (process.env.DATABASE_URL?.includes("us-west")) { return c.json({ error: "Not available against production database" }, 403); } + if (!ALLOWED_ORG_IDS.has(ctx.org?.id)) { + return c.json({ error: "Forbidden" }, 403); + } + const body = await c.req.json<{ action: "sleep" | "ping" | "cpu"; pool: "general" | "critical"; @@ -91,3 +107,57 @@ debugRouter.post("/pool-test", async (c) => { }); } }); + +/** + * Redis failover test endpoints. Blocked in production. + */ +debugRouter.post("/redis-failover", async (c) => { + if (process.env.NODE_ENV === "production") { + return c.json({ error: "Not available in production" }, 403); + } + + const ctx = c.get("ctx"); + + if (!ALLOWED_ORG_IDS.has(ctx.org?.id)) { + return c.json({ error: "Forbidden" }, 403); + } + + const body = await c.req.json<{ + action: "status" | "kill-primary" | "recover-primary" | "ping"; + }>(); + + if (body.action === "status") { + return c.json({ ok: true, ...getFailoverState() }); + } + + if (body.action === "kill-primary") { + disconnectPrimary(); + return c.json({ ok: true, message: "Primary disconnected" }); + } + + if (body.action === "recover-primary") { + reconnectPrimary(); + return c.json({ ok: true, message: "Primary reconnect triggered" }); + } + + if (body.action === "ping") { + const start = Date.now(); + try { + await redis.ping(); + return c.json({ + ok: true, + durationMs: Date.now() - start, + ...getFailoverState(), + }); + } catch (error) { + return c.json({ + ok: false, + durationMs: Date.now() - start, + error: error instanceof Error ? error.message : String(error), + ...getFailoverState(), + }); + } + } + + return c.json({ error: "Unknown action" }, 400); +}); diff --git a/server/src/queue/bullmq/initBullMqWorkers.ts b/server/src/queue/bullmq/initBullMqWorkers.ts index 17ba7c679..6eacdb685 100644 --- a/server/src/queue/bullmq/initBullMqWorkers.ts +++ b/server/src/queue/bullmq/initBullMqWorkers.ts @@ -191,6 +191,9 @@ const initWorker = ({ id, db }: { id: number; db: DrizzleCli }) => { }; export const initWorkers = async () => { + const { warmupRegionalRedis } = await import("@/external/redis/initRedis.js"); + await warmupRegionalRedis(); + const workers = []; for (let i = 0; i < NUM_WORKERS; i++) { diff --git a/server/src/queue/initWorkers.ts b/server/src/queue/initWorkers.ts index 89a5582f9..93e908650 100644 --- a/server/src/queue/initWorkers.ts +++ b/server/src/queue/initWorkers.ts @@ -364,6 +364,8 @@ const startPollingLoop = async ({ */ export const initWorkers = async () => { const { db } = initDrizzle({ maxConnections: 10 }); + const { warmupRegionalRedis } = await import("@/external/redis/initRedis.js"); + await warmupRegionalRedis(); const shutdown = async () => { console.log(`[SQS Worker ${process.pid}] Shutting down...`); diff --git a/server/tests/infra/redisFailover.test.ts b/server/tests/infra/redisFailover.test.ts new file mode 100644 index 000000000..e2b4be370 --- /dev/null +++ b/server/tests/infra/redisFailover.test.ts @@ -0,0 +1,215 @@ +import { expect, test } from "bun:test"; +import chalk from "chalk"; + +const BASE_URL = process.env.AUTUMN_TEST_BASE_URL || "http://localhost:8080"; +const SECRET_KEY = process.env.UNIT_TEST_AUTUMN_SECRET_KEY || ""; + +const headers = { + "Content-Type": "application/json", + Authorization: `Bearer ${SECRET_KEY}`, +}; + +const CUSTOMER_ID = "redis-failover-test-customer"; +const FEATURE_ID = "messages"; + +// Failover timing (must match redisFailover.ts constants) +const FAILOVER_DELAY_MS = 5_000; +const RECOVERY_DELAY_MS = 3_000; + +type FailoverStatus = { + ok: boolean; + isUsingFailover: boolean; + failoverRegion: string | null; + primaryStatus: string; + failoverStatus: string | null; + primaryErrorSince: number | null; + durationMs?: number; + error?: string; +}; + +const redisAction = async ({ + action, +}: { + action: "status" | "kill-primary" | "recover-primary" | "ping"; +}): Promise => { + const res = await fetch(`${BASE_URL}/v1/debug/redis-failover`, { + method: "POST", + headers, + body: JSON.stringify({ action }), + }); + return res.json(); +}; + +const timedFetch = async ({ + label, + url, + method = "POST", + body, +}: { + label: string; + url: string; + method?: string; + body?: Record; +}): Promise<{ + label: string; + status: number; + durationMs: number; + ok: boolean; +}> => { + const start = Date.now(); + const res = await fetch(`${BASE_URL}${url}`, { + method, + headers, + body: body ? JSON.stringify(body) : undefined, + }); + const durationMs = Date.now() - start; + return { label, status: res.status, durationMs, ok: res.ok }; +}; + +const wait = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); + +test(`${chalk.yellowBright("redis failover: full lifecycle")}`, async () => { + // ---- Setup ---- + console.log("\n--- Setup ---"); + await fetch(`${BASE_URL}/v1/customers/${CUSTOMER_ID}`, { + method: "DELETE", + headers, + }).catch(() => {}); + const createRes = await fetch(`${BASE_URL}/v1/customers`, { + method: "POST", + headers, + body: JSON.stringify({ + id: CUSTOMER_ID, + name: "Redis Failover Test", + email: `${CUSTOMER_ID}@example.com`, + internal_options: { disable_defaults: true }, + }), + }); + console.log(` Customer create: ${createRes.status}`); + + // ---- 1. Verify initial state: primary is active ---- + console.log("\n--- Step 1: Verify primary is active ---"); + const initialStatus = await redisAction({ action: "status" }); + console.log(` Response: ${JSON.stringify(initialStatus)}`); + console.log(` Primary: ${initialStatus.primaryStatus}`); + console.log(` Failover: ${initialStatus.failoverStatus}`); + console.log(` Using failover: ${initialStatus.isUsingFailover}`); + + expect(initialStatus.isUsingFailover).toBe(false); + expect(initialStatus.primaryStatus).toBe("ready"); + + // Verify endpoints work with primary + const preCheck = await timedFetch({ + label: "pre-failover check", + url: "/v1/balances.check", + body: { customer_id: CUSTOMER_ID, feature_id: FEATURE_ID }, + }); + console.log(` Check: ${preCheck.durationMs}ms (${preCheck.status})`); + expect(preCheck.ok).toBe(true); + + // ---- 2. Kill primary Redis ---- + console.log("\n--- Step 2: Kill primary Redis ---"); + await redisAction({ action: "kill-primary" }); + console.log(` Primary disconnected`); + + // ---- 3. Wait for failover to trigger ---- + console.log( + `\n--- Step 3: Waiting ${FAILOVER_DELAY_MS + 2000}ms for failover ---`, + ); + await wait(FAILOVER_DELAY_MS + 2000); + + const failoverStatus = await redisAction({ action: "status" }); + console.log(` Primary: ${failoverStatus.primaryStatus}`); + console.log(` Failover: ${failoverStatus.failoverStatus}`); + console.log(` Using failover: ${failoverStatus.isUsingFailover}`); + + if (failoverStatus.failoverStatus) { + // Only assert failover if a failover region is configured + expect(failoverStatus.isUsingFailover).toBe(true); + + // Verify endpoints work on failover + console.log("\n--- Step 4: Test endpoints on failover ---"); + const [getCustomer, check, track] = await Promise.all([ + timedFetch({ + label: "GET /customers/:id", + url: `/v1/customers/${CUSTOMER_ID}`, + method: "GET", + }), + timedFetch({ + label: "POST /check", + url: "/v1/balances.check", + body: { customer_id: CUSTOMER_ID, feature_id: FEATURE_ID }, + }), + timedFetch({ + label: "POST /track", + url: "/v1/balances.track", + body: { + customer_id: CUSTOMER_ID, + feature_id: FEATURE_ID, + value: 1, + }, + }), + ]); + + for (const r of [getCustomer, check, track]) { + console.log(` ${r.label}: ${r.durationMs}ms (${r.status})`); + } + + // Endpoints should still work (via failover Redis or Postgres fallback) + expect(check.ok).toBe(true); + expect(track.ok).toBe(true); + } else { + console.log( + " No failover region configured — skipping failover assertions", + ); + } + + // ---- 5. Recover primary ---- + console.log("\n--- Step 5: Recover primary ---"); + await redisAction({ action: "recover-primary" }); + console.log( + ` Reconnect triggered, waiting ${RECOVERY_DELAY_MS + 2000}ms for recovery...`, + ); + await wait(RECOVERY_DELAY_MS + 2000); + + const recoveredStatus = await redisAction({ action: "status" }); + console.log(` Primary: ${recoveredStatus.primaryStatus}`); + console.log(` Using failover: ${recoveredStatus.isUsingFailover}`); + + expect(recoveredStatus.primaryStatus).toBe("ready"); + expect(recoveredStatus.isUsingFailover).toBe(false); + + // Verify endpoints work after recovery + console.log("\n--- Step 6: Test endpoints after recovery ---"); + const [postGetCus, postCheck, postTrack] = await Promise.all([ + timedFetch({ + label: "GET /customers/:id", + url: `/v1/customers/${CUSTOMER_ID}`, + method: "GET", + }), + timedFetch({ + label: "POST /check", + url: "/v1/balances.check", + body: { customer_id: CUSTOMER_ID, feature_id: FEATURE_ID }, + }), + timedFetch({ + label: "POST /track", + url: "/v1/balances.track", + body: { + customer_id: CUSTOMER_ID, + feature_id: FEATURE_ID, + value: 1, + }, + }), + ]); + + for (const r of [postGetCus, postCheck, postTrack]) { + console.log(` ${r.label}: ${r.durationMs}ms (${r.status})`); + } + + expect(postGetCus.ok).toBe(true); + expect(postCheck.ok).toBe(true); + expect(postTrack.ok).toBe(true); + + console.log("\nRedis failover lifecycle complete."); +}, 60000);