diff --git a/server/src/external/redis/utils/isTransientRedisError.ts b/server/src/external/redis/utils/isTransientRedisError.ts index f8c99ddee..2c781cbe0 100644 --- a/server/src/external/redis/utils/isTransientRedisError.ts +++ b/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 = ({ diff --git a/server/src/queue/processMessage.ts b/server/src/queue/processMessage.ts index dce2b279c..896fb9bc3 100644 --- a/server/src/queue/processMessage.ts +++ b/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 diff --git a/server/tests/unit/balances/track/handle-track-queue-fallback.test.ts b/server/tests/unit/balances/track/handle-track-queue-fallback.test.ts index 360fa676c..b87f60935 100644 --- a/server/tests/unit/balances/track/handle-track-queue-fallback.test.ts +++ b/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", diff --git a/server/tests/unit/queue/processMessage.test.ts b/server/tests/unit/queue/processMessage.test.ts new file mode 100644 index 000000000..0048bebd7 --- /dev/null +++ b/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); + }); +});