diff --git a/server/src/external/redis/customerRedisRouting.ts b/server/src/external/redis/customerRedisRouting.ts index e06ddc95d..fcd8f4668 100644 --- a/server/src/external/redis/customerRedisRouting.ts +++ b/server/src/external/redis/customerRedisRouting.ts @@ -5,6 +5,7 @@ import { isCacheV2RampActive, } from "@/internal/misc/cacheV2Ramp/index.js"; import { getActiveRedisV2Instance } from "@/internal/misc/redisV2Cache/redisV2CacheStore.js"; +import { redisV2 as redisV2Primary } from "./initRedisV2.js"; import { getOrgRedis, type OrgWithRedisConfig } from "./orgRedisPool.js"; import { resolveRedisV2 } from "./resolveRedisV2.js"; @@ -51,7 +52,7 @@ export const resolveCustomerRedisRouting = ({ return { ...routingInfo, - redis: resolveRedisV2({ orgId: org.id, customerId }), + redis: resolveRedisV2({ customerId }), }; }; @@ -119,7 +120,11 @@ export const getRedisTargetsForCustomer = ({ if (getActiveRedisV2Instance() === "dragonfly" && isCacheV2RampActive()) { const destination = getRampDestinationRedis(); if (destination) { - redisTargets.push(destination, resolveRedisV2()); + // Use redisV2Primary directly: at 100% ramp, resolveRedisV2() with no + // args still goes through isCacheV2RampEnabled (which returns true + // when migrationPercent >= 100 regardless of customerId), so it + // would return the destination — leaving primary out of the fan-out. + redisTargets.push(destination, redisV2Primary); } } return [...new Set(redisTargets)]; diff --git a/server/src/external/redis/orgRedisUtils/orgRedisMigrationUtils.ts b/server/src/external/redis/orgRedisUtils/orgRedisMigrationUtils.ts index 1c3854e81..c3c8fd6f5 100644 --- a/server/src/external/redis/orgRedisUtils/orgRedisMigrationUtils.ts +++ b/server/src/external/redis/orgRedisUtils/orgRedisMigrationUtils.ts @@ -5,6 +5,7 @@ import { isCacheV2RampActive, } from "@/internal/misc/cacheV2Ramp/index.js"; import { getActiveRedisV2Instance } from "@/internal/misc/redisV2Cache/redisV2CacheStore.js"; +import { redisV2 as redisV2Primary } from "../initRedisV2.js"; import { getOrgRedis } from "../orgRedisPool.js"; import { resolveRedisV2 } from "../resolveRedisV2.js"; @@ -27,7 +28,10 @@ const withRampClustersIfActive = ({ if (!isCacheV2RampActive()) return candidates; const destination = getRampDestinationRedis(); if (!destination) return candidates; - return [...candidates, destination, resolveRedisV2()]; + // Use redisV2Primary directly: at 100% ramp, resolveRedisV2() with no + // args returns the destination (isCacheV2RampEnabled short-circuits to + // true when migrationPercent >= 100 regardless of customerId). + return [...candidates, destination, redisV2Primary]; }; export const getRedisV2LockReceiptCandidates = ({ diff --git a/server/src/external/redis/resolveRedisV2.ts b/server/src/external/redis/resolveRedisV2.ts index 9546a5d79..e5c0c4268 100644 --- a/server/src/external/redis/resolveRedisV2.ts +++ b/server/src/external/redis/resolveRedisV2.ts @@ -26,10 +26,7 @@ let publicRouteWarned = false; * * Called by every ctx-building middleware/worker — request-path code reads * ctx.redisV2 rather than calling this. */ -export const resolveRedisV2 = (opts?: { - orgId?: string; - customerId?: string; -}): Redis => { +export const resolveRedisV2 = (opts?: { customerId?: string }): Redis => { const activeInstance = getActiveRedisV2Instance(); if (activeInstance !== lastLoggedInstance) { diff --git a/server/src/internal/admin/handleAdminCacheV2Ramp.ts b/server/src/internal/admin/handleAdminCacheV2Ramp.ts index d589f2460..eff9cda55 100644 --- a/server/src/internal/admin/handleAdminCacheV2Ramp.ts +++ b/server/src/internal/admin/handleAdminCacheV2Ramp.ts @@ -78,22 +78,17 @@ export const handleUpsertAdminCacheV2Ramp = createRoute({ }); } - const current = getCacheV2RampConfig(); - if (current && current.migrationPercent > 0) { - throw new RecaseError({ - message: `Cannot update destination while migrationPercent is ${current.migrationPercent}%. Set it to 0 first.`, - code: ErrCode.InvalidRequest, - statusCode: 400, - }); - } - + // Store enforces "refuse while migrationPercent > 0" atomically against + // fresh S3 state — no separate handler-level guard against the polled + // snapshot (which can lag in multi-instance deployments). + const wasConfigured = !!getCacheV2RampConfig(); await upsertCacheV2RampConnection({ connectionString: encryptData(connectionString), url: redisUrl.host, }); logger.info( - `[admin/handleUpsertAdminCacheV2Ramp] ${current ? "updated" : "created"}, url=${redisUrl.host}, actor=${actorString(ctx)}`, + `[admin/handleUpsertAdminCacheV2Ramp] ${wasConfigured ? "updated" : "created"}, url=${redisUrl.host}, actor=${actorString(ctx)}`, ); return c.json({ success: true }); @@ -112,26 +107,22 @@ export const handleUpdateAdminCacheV2RampMigration = createRoute({ const { logger } = ctx; const { migrationPercent } = c.req.valid("json"); - const current = getCacheV2RampConfig(); - if (!current) { - throw new RecaseError({ - message: - "No cache V2 ramp config set. Configure destination first via PATCH /admin/cache-v2-ramp.", - code: ErrCode.InvalidRequest, - statusCode: 400, - }); - } - + // Snapshot the previous percent for logging only; the store enforces + // "config must exist" atomically against fresh S3 state. + const previousSnapshot = getCacheV2RampConfig(); await updateCacheV2RampMigrationPercent({ migrationPercent }); // Warm the destination client on first ramp-up so the first ramped // requests don't pay the connect-handshake latency. - if (migrationPercent > 0 && current.migrationPercent === 0) { + if ( + migrationPercent > 0 && + (previousSnapshot?.migrationPercent ?? 0) === 0 + ) { getRampDestinationRedis(); } logger.info( - `[admin/handleUpdateAdminCacheV2RampMigration] ${current.migrationPercent}% -> ${migrationPercent}%, actor=${actorString(ctx)}`, + `[admin/handleUpdateAdminCacheV2RampMigration] ${previousSnapshot?.migrationPercent ?? 0}% -> ${migrationPercent}%, actor=${actorString(ctx)}`, ); return c.json({ success: true }); @@ -147,16 +138,8 @@ export const handleDeleteAdminCacheV2Ramp = createRoute({ handler: async (c) => { const ctx = c.get("ctx"); const { logger } = ctx; - const current = getCacheV2RampConfig(); - - if (current && current.migrationPercent > 0) { - throw new RecaseError({ - message: `Cannot remove cache V2 ramp while migrationPercent is ${current.migrationPercent}%. Set it to 0 first.`, - code: ErrCode.InvalidRequest, - statusCode: 400, - }); - } - + // Store enforces "refuse while migrationPercent > 0" atomically against + // fresh S3 state. await removeCacheV2RampConfig(); closeRampDestinationClient(); diff --git a/server/src/internal/customers/cusProducts/actions/expireOneOffCustomerProductResults.ts b/server/src/internal/customers/cusProducts/actions/expireOneOffCustomerProductResults.ts index aa0c4ab9a..a16a149a3 100644 --- a/server/src/internal/customers/cusProducts/actions/expireOneOffCustomerProductResults.ts +++ b/server/src/internal/customers/cusProducts/actions/expireOneOffCustomerProductResults.ts @@ -49,7 +49,7 @@ export const expireOneOffCustomerProductResults = async ({ org: group.org, env: group.env, logger: ctx.logger, - redisV2: resolveRedisV2({ orgId: group.org.id }), + redisV2: resolveRedisV2(), }; await batchUpdateCustomerProducts({ diff --git a/server/src/internal/misc/cacheV2Ramp/cacheV2RampStore.ts b/server/src/internal/misc/cacheV2Ramp/cacheV2RampStore.ts index b8e42d0a1..104049d01 100644 --- a/server/src/internal/misc/cacheV2Ramp/cacheV2RampStore.ts +++ b/server/src/internal/misc/cacheV2Ramp/cacheV2RampStore.ts @@ -1,4 +1,4 @@ -import { ms } from "@autumn/shared"; +import { ErrCode, ms, RecaseError } from "@autumn/shared"; import { ADMIN_CACHE_V2_RAMP_CONFIG_KEY } from "@/external/aws/s3/adminS3Config.js"; import { registerEdgeConfig } from "@/internal/misc/edgeConfig/edgeConfigRegistry.js"; import { createEdgeConfigStore } from "@/internal/misc/edgeConfig/edgeConfigStore.js"; @@ -21,7 +21,9 @@ export const getCacheV2RampConfig = (): CacheV2RampConfig => store.get(); export const getCacheV2RampStatus = () => store.getStatus(); /** Create-or-update the connection details. Preserves migration state if a - * config already exists; initializes with migrationPercent=0 otherwise. */ + * config already exists; initializes with migrationPercent=0 otherwise. + * Invariant check (refuse while migrationPercent > 0) runs AFTER readFromSource + * to avoid the multi-instance race where a handler's polled snapshot lags S3. */ export const upsertCacheV2RampConnection = async ({ connectionString, url, @@ -30,6 +32,13 @@ export const upsertCacheV2RampConnection = async ({ url: string; }) => { const current = await store.readFromSource(); + if (current && current.migrationPercent > 0) { + throw new RecaseError({ + message: `Cannot update destination while migrationPercent is ${current.migrationPercent}%. Set it to 0 first.`, + code: ErrCode.InvalidRequest, + statusCode: 400, + }); + } const now = Date.now(); const next: CacheV2RampConfig = current ? { ...current, connectionString, url } @@ -50,9 +59,12 @@ export const updateCacheV2RampMigrationPercent = async ({ }) => { const current = await store.readFromSource(); if (!current) { - throw new Error( - "No cache V2 ramp config set. Configure destination first.", - ); + throw new RecaseError({ + message: + "No cache V2 ramp config set. Configure destination first via PATCH /admin/cache-v2-ramp.", + code: ErrCode.InvalidRequest, + statusCode: 400, + }); } await store.writeToSource({ config: { @@ -65,6 +77,14 @@ export const updateCacheV2RampMigrationPercent = async ({ }; export const removeCacheV2RampConfig = async () => { + const current = await store.readFromSource(); + if (current && current.migrationPercent > 0) { + throw new RecaseError({ + message: `Cannot remove cache V2 ramp while migrationPercent is ${current.migrationPercent}%. Set it to 0 first.`, + code: ErrCode.InvalidRequest, + statusCode: 400, + }); + } await store.writeToSource({ config: null }); }; diff --git a/server/src/queue/createWorkerContext.ts b/server/src/queue/createWorkerContext.ts index 4055ca554..e21de3a40 100644 --- a/server/src/queue/createWorkerContext.ts +++ b/server/src/queue/createWorkerContext.ts @@ -74,7 +74,7 @@ export const createWorkerContext = async ({ db, dbGeneral: db, logger: workerLogger, - redisV2: resolveRedisV2({ orgId: org.id, customerId }), + redisV2: resolveRedisV2({ customerId }), id: requestId || generateId("job"), timestamp: Date.now(), diff --git a/server/src/utils/workerUtils/createAutumnContext.ts b/server/src/utils/workerUtils/createAutumnContext.ts index e0d022710..22c0e76e7 100644 --- a/server/src/utils/workerUtils/createAutumnContext.ts +++ b/server/src/utils/workerUtils/createAutumnContext.ts @@ -59,7 +59,7 @@ export const createWorkerAutumnContext = async ({ db, dbGeneral: db, logger, - redisV2: resolveRedisV2({ orgId: org.id }), + redisV2: resolveRedisV2(), expand: [], id: workerId,