chore: address cubic comment

server/src/external/redis/utils/isTransientRedisError.ts

@@ -1,6 +1,9 @@
 import { RedisUnavailableError } from "./errors.js";
 
-const TRANSIENT_REDIS_ERROR_MESSAGES = new Set(["Command timed out"]);
+const TRANSIENT_REDIS_ERROR_MESSAGES = new Set([
+	"Command timed out",
+	"Connection is closed.",
+]);
 const TRANSIENT_REDIS_ERROR_NAMES = new Set(["MaxRetriesPerRequestError"]);
 
 export const isTransientRedisError = ({

server/src/queue/processMessage.ts

@@ -3,6 +3,7 @@ import * as Sentry from "@sentry/bun";
 import chalk from "chalk";
 import type { Logger } from "pino";
 import { isTransientDbError } from "@/db/dbUtils.js";
+import { isTransientRedisError } from "@/external/redis/utils/isTransientRedisError.js";
 import type { DrizzleCli } from "@/db/initDrizzle.js";
 import { logger } from "@/external/logtail/logtailUtils.js";
 import type { AutumnContext } from "@/honoUtils/HonoEnv.js";
@@ -42,6 +43,27 @@ export interface SqsJob {
 	data: any;
 }
 
+export const shouldRetrySqsJobError = ({
+	jobName,
+	error,
+}: {
+	jobName: string;
+	error: unknown;
+}) => {
+	switch (jobName) {
+		case JobName.SyncBalanceBatchV3:
+		case JobName.SyncBalanceBatchV4:
+		case JobName.RefreshEntityAggregate:
+			return isTransientDbError({ error });
+		case JobName.Track:
+			return (
+				isTransientDbError({ error }) || isTransientRedisError({ error })
+			);
+		default:
+			return false;
+	}
+};
+
 export const processMessage = async ({
 	message,
 	db,
@@ -306,14 +328,9 @@ export const processMessage = async ({
 		// Sync jobs: re-throw infrastructure errors so the message stays in SQS.
 		// Application errors (RecaseError, InternalError) are swallowed — they
 		// won't fix on retry. DB errors (connection, timeout) will.
-		if (
-			(job.name === JobName.SyncBalanceBatchV3 ||
-				job.name === JobName.SyncBalanceBatchV4 ||
-				job.name === JobName.RefreshEntityAggregate) &&
-			isTransientDbError({ error })
-		) {
+		if (shouldRetrySqsJobError({ jobName: job.name, error })) {
 			Sentry.captureException(error);
-			errorLogger.error(`[${job.name}] Retryable DB error, keeping in SQS`, {
+			errorLogger.error(`[${job.name}] Retryable error, keeping in SQS`, {
 				jobName: job.name,
 				error:
 					error instanceof Error

server/tests/unit/balances/track/handle-track-queue-fallback.test.ts

@@ -111,6 +111,24 @@ describe("track queue fallback", () => {
 		});
 	});
 
+	test("queues track when rollout path hits a raw closed-connection Redis error", async () => {
+		mockState.v3Error = new Error("Connection is closed.");
+
+		const response = await runTrackWithRollout({
+			ctx,
+			body,
+			featureDeductions: [],
+		});
+
+		expect(mockState.queueCommands).toHaveLength(1);
+		expect(response).toEqual({
+			customer_id: "cus_123",
+			entity_id: undefined,
+			value: 2,
+			balance: null,
+		});
+	});
+
 	test("throws retryable Redis failure when queue fallback is unavailable", async () => {
 		const error = new RedisUnavailableError({
 			source: "runTrackV3",

server/tests/unit/queue/processMessage.test.ts

@@ -0,0 +1,27 @@
+import { describe, expect, test } from "bun:test";
+import { RedisUnavailableError } from "@/external/redis/utils/errors.js";
+import { JobName } from "@/queue/JobName.js";
+import { shouldRetrySqsJobError } from "@/queue/processMessage.js";
+
+describe("shouldRetrySqsJobError", () => {
+	test("retries track jobs on transient Redis errors", () => {
+		expect(
+			shouldRetrySqsJobError({
+				jobName: JobName.Track,
+				error: new RedisUnavailableError({
+					source: "runTrackV3",
+					reason: "timeout",
+				}),
+			}),
+		).toBe(true);
+	});
+
+	test("does not retry track jobs on non-transient application errors", () => {
+		expect(
+			shouldRetrySqsJobError({
+				jobName: JobName.Track,
+				error: new Error("insufficient balance"),
+			}),
+		).toBe(false);
+	});
+});