added redis failover for improved resiliency

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<string, Redis> = 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 };

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<typeof setTimeout> | 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();
+};

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);
+});

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++) {

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...`);

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<FailoverStatus> => {
+	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<string, unknown>;
+}): 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);