ai review fixes

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)];

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 = ({

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

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();
 

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({

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

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(),

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,