comi-logic/cfw-autumn
8
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

redis failover mechanism reworked

Browse Source
This commit is contained in:
John Yeo
2026-03-20 11:01:44 +00:00
parent 7d68e6a2f3
commit 99c71c9df3
5 changed files with 288 additions and 140 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
server/src/db/initDrizzle.ts
View File

@@ -40,18 +40,18 @@ export const initDrizzle = ({
// -- Critical pool: used by check, track, getOrCreateCustomer --
export const { db: dbCritical, client: clientCritical } = initDrizzle({
connectTimeout: 2,
// connectTimeout: 10,
});
// -- General pool: used by all other endpoints --
export const { db: dbGeneral, client: clientGeneral } = initDrizzle({
connectTimeout: 5,
// connectTimeout: 5,
});
// -- Replica pool: used as fallback when primary is degraded --
// Only created if DATABASE_REPLICA_URL is configured.
const replicaResult = process.env.DATABASE_REPLICA_URL
? initDrizzle({ replica: true, maxConnections: 5, connectTimeout: 2 })
? initDrizzle({ replica: true, maxConnections: 5, connectTimeout: undefined })
: null;
export const dbReplica = replicaResult?.db ?? null;
export const clientReplica = replicaResult?.client ?? null;

26
server/src/external/redis/initRedis.ts vendored
View File

@@ -29,7 +29,11 @@ import {
UPSERT_INVOICE_IN_CUSTOMER_SCRIPT,
} from "../../_luaScriptsV2/luaScriptsV2.js";
import { instrumentRedis } from "../../utils/otel/instrumentRedis.js";
import { getActiveRedis, initFailover } from "./redisFailover.js";
import {
getActiveRedis,
initFailover,
onActiveChange,
} from "./redisFailover.js";
// if (!process.env.CACHE_URL) {
// throw new Error("CACHE_URL (redis) is not set");
@@ -309,24 +313,12 @@ initFailover({
*/
export let redis: Redis = primaryRedis;
// Subscribe to failover state changes — keep the `redis` export in sync.
// Keep the `redis` export in sync with the failover state machine.
// 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);
onActiveChange(() => {
redis = getActiveRedis();
});
// Lazy-loaded regional Redis instances for cross-region sync
const regionalRedisInstances: Map<string, Redis> = new Map();

315
server/src/external/redis/redisFailover.ts vendored
View File

@@ -1,31 +1,190 @@
import type { Redis } from "ioredis";
import { logger } from "@/external/logtail/logtailUtils.js";
/** How long primary must be erroring before we switch to failover. */
const FAILOVER_DELAY_MS = 5_000;
// ── Config ──────────────────────────────────────────────────────────
/** How long primary must stay down before we switch to failover. */
const FAILOVER_THRESHOLD_MS = 15_000;
/** How long primary must be stable before we switch back from failover. */
const RECOVERY_DELAY_MS = 3_000;
/** How long primary must stay healthy before we switch back. */
const RECOVERY_THRESHOLD_MS = 5_000;
export type RedisFailoverState = {
/** The currently active Redis instance (what consumers use). */
/** Health-check polling interval. */
const POLL_INTERVAL_MS = 2_000;
/** Log a warning if blip count exceeds this in the trailing window. */
const BLIP_WARN_THRESHOLD = 10;
/** Trailing window for blip counting. */
const BLIP_WINDOW_MS = 60 * 60 * 1_000; // 1 hour
// ── State machine ───────────────────────────────────────────────────
type FailoverPhase = "NORMAL" | "DEGRADED" | "FAILOVER" | "RECOVERING";
type FailoverState = {
phase: FailoverPhase;
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;
/** Timestamp when the current phase was entered. */
phaseEnteredAt: number;
};
let state: RedisFailoverState;
let primaryErrorSince: number | null = null;
let recoveryTimer: ReturnType<typeof setTimeout> | null = null;
let state: FailoverState;
let primaryHasBeenReady = false;
let pollTimer: ReturnType<typeof setInterval> | null = null;
/** Initialize failover state. Call once after creating both Redis instances. */
/** Tracks timestamps of recent transient blips (DEGRADED → NORMAL). */
const blipTimestamps: number[] = [];
// ── Callbacks ───────────────────────────────────────────────────────
type StateChangeCallback = () => void;
const onChangeCallbacks: StateChangeCallback[] = [];
/** Register a callback invoked whenever `active` changes. */
export const onActiveChange = (cb: StateChangeCallback): void => {
onChangeCallbacks.push(cb);
};
const notifyChange = (): void => {
for (const cb of onChangeCallbacks) {
try {
cb();
} catch {}
}
};
// ── Helpers ─────────────────────────────────────────────────────────
const isPrimaryReady = (): boolean => state.primary.status === "ready";
const isFailoverReady = (): boolean => state.failover?.status === "ready";
const setPhase = (phase: FailoverPhase): void => {
state.phase = phase;
state.phaseEnteredAt = Date.now();
};
const msInPhase = (): number => Date.now() - state.phaseEnteredAt;
const pruneBlips = (): void => {
const cutoff = Date.now() - BLIP_WINDOW_MS;
while (blipTimestamps.length > 0 && blipTimestamps[0] < cutoff) {
blipTimestamps.shift();
}
};
const recordBlip = ({ durationMs }: { durationMs: number }): void => {
blipTimestamps.push(Date.now());
pruneBlips();
logger.warn(
`[Redis failover] Primary blip #${blipTimestamps.length} (recovered in ${durationMs}ms)`,
{
type: "redis_failover_blip",
blipCount: blipTimestamps.length,
durationMs,
},
);
if (blipTimestamps.length >= BLIP_WARN_THRESHOLD) {
logger.error(
`[Redis failover] ${blipTimestamps.length} blips in the last hour — check Redis health`,
{
type: "redis_failover_blip_alert",
blipCount: blipTimestamps.length,
},
);
}
};
// ── Core poll tick ──────────────────────────────────────────────────
const tick = (): void => {
const ready = isPrimaryReady();
switch (state.phase) {
case "NORMAL": {
if (!ready && primaryHasBeenReady) {
setPhase("DEGRADED");
logger.warn("[Redis failover] Primary unhealthy — entering DEGRADED", {
type: "redis_failover_degraded",
primaryStatus: state.primary.status,
});
}
break;
}
case "DEGRADED": {
if (ready) {
// Blip — primary recovered before we had to failover
recordBlip({ durationMs: msInPhase() });
setPhase("NORMAL");
break;
}
if (msInPhase() >= FAILOVER_THRESHOLD_MS) {
if (!state.failover || !isFailoverReady()) {
logger.error(
"[Redis failover] Threshold reached but failover instance not ready",
{
type: "redis_failover_switch",
failoverStatus: state.failover?.status ?? "none",
},
);
break;
}
state.active = state.failover;
setPhase("FAILOVER");
notifyChange();
logger.error(
`[Redis failover] SWITCHED to failover region (${state.failoverRegion})`,
{
type: "redis_failover_switch",
failoverRegion: state.failoverRegion,
},
);
}
break;
}
case "FAILOVER": {
if (ready) {
setPhase("RECOVERING");
logger.info("[Redis failover] Primary back — entering RECOVERING", {
type: "redis_failover_recovering",
});
}
break;
}
case "RECOVERING": {
if (!ready) {
// Primary dropped again — go back to failover
setPhase("FAILOVER");
logger.warn(
"[Redis failover] Primary dropped during recovery — back to FAILOVER",
{ type: "redis_failover_recovery_failed" },
);
break;
}
if (msInPhase() >= RECOVERY_THRESHOLD_MS) {
state.active = state.primary;
setPhase("NORMAL");
notifyChange();
logger.info("[Redis failover] RECOVERED to primary region", {
type: "redis_failover_recovered",
});
}
break;
}
}
};
// ── Public API ──────────────────────────────────────────────────────
/** Initialize failover. Call once after creating both Redis instances. */
export const initFailover = ({
primary,
failover,
@@ -38,135 +197,75 @@ export const initFailover = ({
currentRegion: string;
}): void => {
state = {
phase: "NORMAL",
active: primary,
primary,
failover,
isUsingFailover: false,
failoverRegion,
phaseEnteredAt: Date.now(),
};
if (!failover) {
logger.info(
"[Redis failover] No failover region configured — failover disabled",
{
type: "redis_failover_init",
},
{ type: "redis_failover_init" },
);
return;
}
logger.info(
`[Redis failover] Enabled: primary=${currentRegion}, failover=${failoverRegion}`,
{
type: "redis_failover_init",
currentRegion,
failoverRegion,
},
{ type: "redis_failover_init", currentRegion, 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);
// Track when primary first connects so we don't failover during startup
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") {
logger.error(
"[Redis failover] Cannot switch — failover instance is not ready",
{
type: "redis_failover_switch",
failoverStatus: state.failover.status,
},
);
return;
}
state.active = state.failover;
state.isUsingFailover = true;
logger.error(
`[Redis failover] SWITCHED to failover region (${state.failoverRegion})`,
{
type: "redis_failover_switch",
failoverRegion: state.failoverRegion,
},
);
};
const switchToPrimary = (): void => {
if (!state.isUsingFailover) return;
state.active = state.primary;
state.isUsingFailover = false;
primaryErrorSince = null;
logger.info("[Redis failover] RECOVERED to primary region", {
type: "redis_failover_recovered",
});
// Start the single polling loop
pollTimer = setInterval(tick, POLL_INTERVAL_MS);
};
/** Get the currently active Redis instance. */
export const getActiveRedis = (): Redis => state.active;
/** Get the current failover state (for debug/monitoring). */
/** Get current failover state (for debug/monitoring). */
export const getFailoverState = (): {
phase: FailoverPhase;
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,
});
msInPhase: number;
blipsLastHour: number;
} => {
pruneBlips();
return {
phase: state.phase,
isUsingFailover: state.phase === "FAILOVER" || state.phase === "RECOVERING",
failoverRegion: state.failoverRegion,
primaryStatus: state.primary.status,
failoverStatus: state.failover?.status ?? null,
msInPhase: msInPhase(),
blipsLastHour: blipTimestamps.length,
};
};
/**
* Force disconnect the primary instance (for testing).
* ioredis will NOT auto-reconnect after a manual disconnect() call.
* Use `reconnectPrimary()` to manually reconnect.
*/
/** Force disconnect the primary (for testing). */
export const disconnectPrimary = (): void => {
state.primary.disconnect();
};
/** Force reconnect the primary instance (for testing). */
/** Force reconnect the primary (for testing). */
export const reconnectPrimary = (): void => {
state.primary.connect();
};
/** Stop the polling loop (for testing/cleanup). */
export const stopFailoverPolling = (): void => {
if (pollTimer) {
clearInterval(pollTimer);
pollTimer = null;
}
};

1
server/src/initHono.ts
View File

@@ -16,7 +16,6 @@ import { errorMiddleware } from "./honoMiddlewares/errorMiddleware.js";
import { traceEnrichMiddleware } from "./honoMiddlewares/traceMiddleware.js";
import type { HonoEnv } from "./honoUtils/HonoEnv.js";
import { handleHealthCheck } from "./honoUtils/handleHealthCheck.js";
import { debugRouter } from "./internal/debug/debugRouter.js";
import { cliRouter } from "./internal/dev/cli/cliRouter.js";
import { handleOAuthCallback } from "./internal/orgs/handlers/stripeHandlers/handleOAuthCallback.js";
import { apiRouter } from "./routers/apiRouter.js";

80
server/tests/infra/redisFailover.test.ts
View File

@@ -13,16 +13,19 @@ 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;
const FAILOVER_THRESHOLD_MS = 15_000;
const RECOVERY_THRESHOLD_MS = 5_000;
const POLL_INTERVAL_MS = 2_000;
type FailoverStatus = {
ok: boolean;
phase: string;
isUsingFailover: boolean;
failoverRegion: string | null;
primaryStatus: string;
failoverStatus: string | null;
primaryErrorSince: number | null;
msInPhase: number;
blipsLastHour: number;
durationMs?: number;
error?: string;
};
@@ -90,7 +93,7 @@ test(`${chalk.yellowBright("redis failover: full lifecycle")}`, async () => {
// ---- 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(` Phase: ${initialStatus.phase}`);
console.log(` Primary: ${initialStatus.primaryStatus}`);
console.log(` Failover: ${initialStatus.failoverStatus}`);
console.log(` Using failover: ${initialStatus.isUsingFailover}`);
@@ -113,12 +116,13 @@ test(`${chalk.yellowBright("redis failover: full lifecycle")}`, async () => {
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);
// Must wait: threshold + up to one poll interval + buffer
const failoverWait = FAILOVER_THRESHOLD_MS + POLL_INTERVAL_MS + 2000;
console.log(`\n--- Step 3: Waiting ${failoverWait}ms for failover ---`);
await wait(failoverWait);
const failoverStatus = await redisAction({ action: "status" });
console.log(` Phase: ${failoverStatus.phase}`);
console.log(` Primary: ${failoverStatus.primaryStatus}`);
console.log(` Failover: ${failoverStatus.failoverStatus}`);
console.log(` Using failover: ${failoverStatus.isUsingFailover}`);
@@ -167,12 +171,14 @@ test(`${chalk.yellowBright("redis failover: full lifecycle")}`, async () => {
// ---- 5. Recover primary ----
console.log("\n--- Step 5: Recover primary ---");
await redisAction({ action: "recover-primary" });
const recoveryWait = RECOVERY_THRESHOLD_MS + POLL_INTERVAL_MS + 2000;
console.log(
` Reconnect triggered, waiting ${RECOVERY_DELAY_MS + 2000}ms for recovery...`,
` Reconnect triggered, waiting ${recoveryWait}ms for recovery...`,
);
await wait(RECOVERY_DELAY_MS + 2000);
await wait(recoveryWait);
const recoveredStatus = await redisAction({ action: "status" });
console.log(` Phase: ${recoveredStatus.phase}`);
console.log(` Primary: ${recoveredStatus.primaryStatus}`);
console.log(` Using failover: ${recoveredStatus.isUsingFailover}`);
@@ -212,4 +218,56 @@ test(`${chalk.yellowBright("redis failover: full lifecycle")}`, async () => {
expect(postTrack.ok).toBe(true);
console.log("\nRedis failover lifecycle complete.");
}, 60000);
}, 90_000);
test(`${chalk.yellowBright("redis failover: blip does NOT trigger failover")}`, async () => {
// ---- 1. Verify we start in NORMAL ----
console.log("\n--- Step 1: Verify NORMAL state ---");
const initial = await redisAction({ action: "status" });
console.log(` Phase: ${initial.phase}, blips: ${initial.blipsLastHour}`);
expect(initial.phase).toBe("NORMAL");
expect(initial.isUsingFailover).toBe(false);
const blipsBefore = initial.blipsLastHour;
// ---- 2. Kill primary (simulate BGSAVE blip) ----
console.log("\n--- Step 2: Kill primary (simulating ~8s BGSAVE blip) ---");
await redisAction({ action: "kill-primary" });
// Wait 4s — should be in DEGRADED but NOT yet FAILOVER (threshold is 15s)
await wait(4_000);
const degraded = await redisAction({ action: "status" });
console.log(
` Phase after 4s: ${degraded.phase} (msInPhase: ${degraded.msInPhase})`,
);
expect(degraded.phase).toBe("DEGRADED");
expect(degraded.isUsingFailover).toBe(false);
// ---- 3. Recover primary before threshold ----
console.log("\n--- Step 3: Recover primary (before 15s threshold) ---");
await redisAction({ action: "recover-primary" });
// Wait for primary to reconnect + next poll tick
await wait(POLL_INTERVAL_MS + 2_000);
const afterBlip = await redisAction({ action: "status" });
console.log(` Phase: ${afterBlip.phase}, blips: ${afterBlip.blipsLastHour}`);
// Should be back in NORMAL, never hit FAILOVER
expect(afterBlip.phase).toBe("NORMAL");
expect(afterBlip.isUsingFailover).toBe(false);
// Blip counter should have incremented
expect(afterBlip.blipsLastHour).toBe(blipsBefore + 1);
// ---- 4. Verify endpoints still work on primary ----
console.log("\n--- Step 4: Verify endpoints work ---");
const check = await timedFetch({
label: "POST /check",
url: "/v1/balances.check",
body: { customer_id: CUSTOMER_ID, feature_id: FEATURE_ID },
});
console.log(` ${check.label}: ${check.durationMs}ms (${check.status})`);
expect(check.ok).toBe(true);
console.log("\nBlip test complete — failover was NOT triggered.");
}, 30_000);