From d7e4af787ade4e6b5e8248303ac06b95b75f5ea6 Mon Sep 17 00:00:00 2001 From: Charlie Lamb Date: Fri, 24 Apr 2026 16:40:36 +0100 Subject: [PATCH] chore: init fail open track --- server/src/external/aws/s3/adminS3Config.ts | 6 + server/src/init.ts | 1 + server/src/internal/admin/adminRouter.ts | 4 + .../admin/handleGetAdminJobQueueConfig.ts | 22 ++ .../admin/handleUpsertAdminJobQueueConfig.ts | 14 + .../internal/balances/track/runQueuedTrack.ts | 40 +++ .../balances/track/runTrackWithRollout.ts | 31 +- .../balances/track/utils/queueTrack.ts | 7 +- .../track/v3/handleRedisTrackErrorV3.ts | 3 - .../misc/jobQueues/jobQueueSchemas.ts | 14 + .../internal/misc/jobQueues/jobQueueStore.ts | 55 ++++ server/src/queue/createWorkerContext.ts | 5 +- server/src/queue/initWorkers.ts | 66 ++++- server/src/queue/processMessage.ts | 15 + server/src/queue/queueUtils.ts | 3 + server/src/workers.ts | 1 + .../admin/job-queue-config.test.ts | 246 ++++++++++++++++ .../queue/fixtures/livePollingWorker.ts | 35 +++ .../queue/live-worker-process.test.ts | 271 ++++++++++++++++++ .../queue/start-polling-loop.test.ts | 89 ++++++ .../track-v3/handleRedisTrackErrorV3.test.ts | 50 ++-- .../track/handle-track-queue-fallback.test.ts | 4 + .../unit/balances/track/queueTrack.test.ts | 75 +++++ .../balances/track/runQueuedTrack.test.ts | 88 ++++++ server/tests/unit/queue/queue-utils.test.ts | 26 +- .../views/admin/components/EdgeConfigTab.tsx | 24 ++ .../admin/components/JobQueuesDialog.tsx | 235 +++++++++++++++ 27 files changed, 1348 insertions(+), 82 deletions(-) create mode 100644 server/src/internal/admin/handleGetAdminJobQueueConfig.ts create mode 100644 server/src/internal/admin/handleUpsertAdminJobQueueConfig.ts create mode 100644 server/src/internal/balances/track/runQueuedTrack.ts create mode 100644 server/src/internal/misc/jobQueues/jobQueueSchemas.ts create mode 100644 server/src/internal/misc/jobQueues/jobQueueStore.ts create mode 100644 server/tests/integration/admin/job-queue-config.test.ts create mode 100644 server/tests/integration/queue/fixtures/livePollingWorker.ts create mode 100644 server/tests/integration/queue/live-worker-process.test.ts create mode 100644 server/tests/integration/queue/start-polling-loop.test.ts create mode 100644 server/tests/unit/balances/track/queueTrack.test.ts create mode 100644 server/tests/unit/balances/track/runQueuedTrack.test.ts create mode 100644 vite/src/views/admin/components/JobQueuesDialog.tsx diff --git a/server/src/external/aws/s3/adminS3Config.ts b/server/src/external/aws/s3/adminS3Config.ts index 512071c04..72b1f3102 100644 --- a/server/src/external/aws/s3/adminS3Config.ts +++ b/server/src/external/aws/s3/adminS3Config.ts @@ -6,6 +6,7 @@ export const ADMIN_CUSTOMER_BLOCK_CONFIG_KEY = export const ADMIN_ORG_LIMITS_CONFIG_KEY = "admin/org-limits-config.json"; export const ADMIN_REDIS_V2_CACHE_CONFIG_KEY = "admin/redis-v2-cache-config.json"; +export const ADMIN_JOB_QUEUE_CONFIG_KEY = "admin/job-queue-config.json"; const bucket = process.env.S3_BUCKET || "autumn-prod-server"; const region = process.env.S3_REGION || "us-east-2"; @@ -42,6 +43,11 @@ export const getAdminEdgeConfigSources = () => ({ label: "V2 Redis Instance", key: ADMIN_REDIS_V2_CACHE_CONFIG_KEY, }, + { + id: "job-queues", + label: "Job Queues", + key: ADMIN_JOB_QUEUE_CONFIG_KEY, + }, { id: "stripe-sync", label: "Stripe Sync", diff --git a/server/src/init.ts b/server/src/init.ts index f67964c2f..2c4443038 100644 --- a/server/src/init.ts +++ b/server/src/init.ts @@ -25,6 +25,7 @@ import "./internal/misc/customerBlocks/customerBlockStore.js"; import "./internal/misc/edgeConfig/orgLimitsStore.js"; import "./internal/misc/stripeSync/stripeSyncStore.js"; import "./internal/misc/redisV2Cache/redisV2CacheStore.js"; +import "./internal/misc/jobQueues/jobQueueStore.js"; import { closeStripeSyncEngine } from "@autumn/stripe-sync"; import { startRedisMonitor, diff --git a/server/src/internal/admin/adminRouter.ts b/server/src/internal/admin/adminRouter.ts index 5041967b0..750b7dcb8 100644 --- a/server/src/internal/admin/adminRouter.ts +++ b/server/src/internal/admin/adminRouter.ts @@ -3,6 +3,7 @@ import type { HonoEnv } from "../../honoUtils/HonoEnv"; import { handleGetAdminCustomerBlockConfig } from "./handleGetAdminCustomerBlockConfig"; import { handleGetAdminEdgeConfigSources } from "./handleGetAdminEdgeConfigSources"; import { handleGetAdminFeatureFlagsConfig } from "./handleGetAdminFeatureFlagsConfig"; +import { handleGetAdminJobQueueConfig } from "./handleGetAdminJobQueueConfig"; import { handleGetAdminOrgLimitsConfig } from "./handleGetAdminOrgLimitsConfig"; import { handleGetAdminOrgRequestBlock } from "./handleGetAdminOrgRequestBlock"; import { handleGetAdminRequestBlockConfig } from "./handleGetAdminRequestBlockConfig"; @@ -17,6 +18,7 @@ import { handleListAdminUsers } from "./handleListAdminUsers"; import { handleListOAuthClients } from "./handleListOAuthClients"; import { handleUpsertAdminCustomerBlockConfig } from "./handleUpsertAdminCustomerBlockConfig"; import { handleUpsertAdminFeatureFlagsConfig } from "./handleUpsertAdminFeatureFlagsConfig"; +import { handleUpsertAdminJobQueueConfig } from "./handleUpsertAdminJobQueueConfig"; import { handleUpsertAdminOrgLimitsConfig } from "./handleUpsertAdminOrgLimitsConfig"; import { handleUpsertAdminOrgRequestBlock } from "./handleUpsertAdminOrgRequestBlock"; import { handleUpsertAdminRequestBlockConfig } from "./handleUpsertAdminRequestBlockConfig"; @@ -67,6 +69,8 @@ honoAdminRouter.put( ); honoAdminRouter.get("/org-limits-config", ...handleGetAdminOrgLimitsConfig); honoAdminRouter.put("/org-limits-config", ...handleUpsertAdminOrgLimitsConfig); +honoAdminRouter.get("/job-queue-config", ...handleGetAdminJobQueueConfig); +honoAdminRouter.put("/job-queue-config", ...handleUpsertAdminJobQueueConfig); honoAdminRouter.get("/stripe-sync-config", ...handleGetAdminStripeSyncConfig); honoAdminRouter.put( "/stripe-sync-config", diff --git a/server/src/internal/admin/handleGetAdminJobQueueConfig.ts b/server/src/internal/admin/handleGetAdminJobQueueConfig.ts new file mode 100644 index 000000000..328183d79 --- /dev/null +++ b/server/src/internal/admin/handleGetAdminJobQueueConfig.ts @@ -0,0 +1,22 @@ +import { createRoute } from "@/honoMiddlewares/routeHandler.js"; +import { + getJobQueueConfigFromSource, + getJobQueueConfigStatus, + KNOWN_JOB_QUEUES, +} from "@/internal/misc/jobQueues/jobQueueStore.js"; + +export const handleGetAdminJobQueueConfig = createRoute({ + handler: async (c) => { + const status = getJobQueueConfigStatus(); + const config = await getJobQueueConfigFromSource(); + + return c.json({ + ...config, + knownQueues: KNOWN_JOB_QUEUES, + configHealthy: status.healthy, + configConfigured: status.configured, + lastSuccessAt: status.lastSuccessAt ?? null, + error: status.error ?? null, + }); + }, +}); diff --git a/server/src/internal/admin/handleUpsertAdminJobQueueConfig.ts b/server/src/internal/admin/handleUpsertAdminJobQueueConfig.ts new file mode 100644 index 000000000..705e7e1fb --- /dev/null +++ b/server/src/internal/admin/handleUpsertAdminJobQueueConfig.ts @@ -0,0 +1,14 @@ +import { createRoute } from "@/honoMiddlewares/routeHandler.js"; +import { JobQueueConfigSchema } from "@/internal/misc/jobQueues/jobQueueSchemas.js"; +import { updateFullJobQueueConfig } from "@/internal/misc/jobQueues/jobQueueStore.js"; + +export const handleUpsertAdminJobQueueConfig = createRoute({ + body: JobQueueConfigSchema, + handler: async (c) => { + const body = c.req.valid("json"); + + await updateFullJobQueueConfig({ config: body }); + + return c.json({ success: true }); + }, +}); diff --git a/server/src/internal/balances/track/runQueuedTrack.ts b/server/src/internal/balances/track/runQueuedTrack.ts new file mode 100644 index 000000000..9d2effc5d --- /dev/null +++ b/server/src/internal/balances/track/runQueuedTrack.ts @@ -0,0 +1,40 @@ +import { ErrCode, RecaseError, type ApiVersion, type TrackParams } from "@autumn/shared"; +import type { AutumnContext } from "@/honoUtils/HonoEnv.js"; +import { getTrackFeatureDeductionsForBody } from "./utils/getFeatureDeductions.js"; +import { runTrackV3 } from "./v3/runTrackV3.js"; + +export const runQueuedTrack = async ({ + ctx, + body, + apiVersion, +}: { + ctx: AutumnContext; + body: TrackParams; + apiVersion?: ApiVersion; +}) => { + const featureDeductions = getTrackFeatureDeductionsForBody({ ctx, body }); + + try { + await runTrackV3({ + ctx, + body, + featureDeductions, + apiVersion, + }); + } catch (error) { + if ( + !(error instanceof RecaseError) || + error.code !== ErrCode.DuplicateIdempotencyKey + ) { + throw error; + } + + ctx.logger.info("[track] queued replay already applied", { + type: "track_queue_replay_duplicate", + customer_id: body.customer_id, + entity_id: body.entity_id, + feature_id: body.feature_id, + event_name: body.event_name, + }); + } +}; diff --git a/server/src/internal/balances/track/runTrackWithRollout.ts b/server/src/internal/balances/track/runTrackWithRollout.ts index 4966ad803..23e88a116 100644 --- a/server/src/internal/balances/track/runTrackWithRollout.ts +++ b/server/src/internal/balances/track/runTrackWithRollout.ts @@ -1,10 +1,7 @@ import type { ApiVersion, TrackParams, TrackResponseV3 } from "@autumn/shared"; -import { Result } from "better-result"; +import { withRedisFailOpen } from "@/external/redis/utils/withRedisFailOpen.js"; import type { AutumnContext } from "@/honoUtils/HonoEnv.js"; -import { - isFullSubjectRolloutEnabled, - isRetryableFullSubjectRolloutError, -} from "@/internal/misc/rollouts/fullSubjectRolloutUtils.js"; +import { isFullSubjectRolloutEnabled } from "@/internal/misc/rollouts/fullSubjectRolloutUtils.js"; import type { FeatureDeduction } from "../utils/types/featureDeduction.js"; import { runTrackV2 } from "./runTrackV2.js"; import { queueTrack } from "./utils/queueTrack.js"; @@ -27,29 +24,21 @@ export const runTrackWithRollout = async ({ apiVersion?: ApiVersion; }): Promise => { if (shouldUseTrackV3({ ctx })) { - const result = await Result.tryPromise({ - try: () => + return withRedisFailOpen({ + source: "runTrackWithRollout", + run: () => runTrackV3({ ctx, body, featureDeductions, apiVersion, }), - catch: (error) => error, + fallback: async (error) => { + const queuedResponse = await queueTrack({ ctx, body }); + if (queuedResponse) return queuedResponse; + throw error; + }, }); - - if (Result.isOk(result)) return result.value; - - const error = result.error; - if (!isRetryableFullSubjectRolloutError({ error })) { - throw error; - } - - const queuedResponse = await queueTrack({ ctx, body }); - - if (queuedResponse) return queuedResponse; - - throw error; } return runTrackV2({ diff --git a/server/src/internal/balances/track/utils/queueTrack.ts b/server/src/internal/balances/track/utils/queueTrack.ts index 42382a826..8d25a588f 100644 --- a/server/src/internal/balances/track/utils/queueTrack.ts +++ b/server/src/internal/balances/track/utils/queueTrack.ts @@ -23,11 +23,14 @@ export const queueTrack = async ({ await addTaskToQueue({ jobName: JobName.Track, queueUrl, - messageGroupId: `${ctx.org.id}:${ctx.env}:${body.customer_id}`, - messageDeduplicationId: body.idempotency_key, + messageGroupId: `${ctx.org.id}:${ctx.env}:${body.customer_id}:${body.entity_id ?? "none"}`, + messageDeduplicationId: body.idempotency_key || ctx.id, payload: { orgId: ctx.org.id, env: ctx.env, + customerId: body.customer_id, + entityId: body.entity_id, + requestId: ctx.id, apiVersion: ctx.apiVersion.value, body, }, diff --git a/server/src/internal/balances/track/v3/handleRedisTrackErrorV3.ts b/server/src/internal/balances/track/v3/handleRedisTrackErrorV3.ts index 03d385cdc..c91bac399 100644 --- a/server/src/internal/balances/track/v3/handleRedisTrackErrorV3.ts +++ b/server/src/internal/balances/track/v3/handleRedisTrackErrorV3.ts @@ -12,7 +12,6 @@ import { RedisDeductionError, RedisDeductionErrorCode, } from "../../utils/types/redisDeductionError.js"; -import { queueTrack } from "../utils/queueTrack.js"; import { runPostgresTrackV3 } from "./runPostgresTrackV3.js"; /** Handles errors from V2 Redis deduction. Falls back to Postgres V3 path. */ @@ -58,8 +57,6 @@ export const handleRedisTrackErrorV3 = async ({ } if (error.isRedisUnavailable()) { - const queuedResponse = await queueTrack({ ctx, body }); - if (queuedResponse) return queuedResponse; throw error; } diff --git a/server/src/internal/misc/jobQueues/jobQueueSchemas.ts b/server/src/internal/misc/jobQueues/jobQueueSchemas.ts new file mode 100644 index 000000000..954a7b45d --- /dev/null +++ b/server/src/internal/misc/jobQueues/jobQueueSchemas.ts @@ -0,0 +1,14 @@ +import { z } from "zod/v4"; + +export const JobQueueConfigSchema = z.object({ + queues: z + .record( + z.string(), + z.object({ + enabled: z.boolean().default(false), + }), + ) + .default({}), +}); + +export type JobQueueConfig = z.infer; diff --git a/server/src/internal/misc/jobQueues/jobQueueStore.ts b/server/src/internal/misc/jobQueues/jobQueueStore.ts new file mode 100644 index 000000000..c822abec0 --- /dev/null +++ b/server/src/internal/misc/jobQueues/jobQueueStore.ts @@ -0,0 +1,55 @@ +import { ADMIN_JOB_QUEUE_CONFIG_KEY } from "@/external/aws/s3/adminS3Config.js"; +import { registerEdgeConfig } from "@/internal/misc/edgeConfig/edgeConfigRegistry.js"; +import { createEdgeConfigStore } from "@/internal/misc/edgeConfig/edgeConfigStore.js"; +import { + type JobQueueConfig, + JobQueueConfigSchema, +} from "./jobQueueSchemas.js"; + +export const JOB_QUEUE_IDS = { + primary: "primary", + track: "track", +} as const; + +export const KNOWN_JOB_QUEUES = [ + { + id: JOB_QUEUE_IDS.primary, + label: "Primary Queue", + description: "Shared SQS queue for standard background jobs.", + defaultEnabled: true, + }, + { + id: JOB_QUEUE_IDS.track, + label: "Track Replay Queue", + description: "Dedicated async track replay queue used during fail-open recovery.", + defaultEnabled: true, + }, +] as const; + +const store = createEdgeConfigStore({ + s3Key: ADMIN_JOB_QUEUE_CONFIG_KEY, + schema: JobQueueConfigSchema, + defaultValue: () => ({ queues: {} }), +}); + +registerEdgeConfig({ store }); + +export const isJobQueueEnabled = ({ + queue, + defaultEnabled = true, +}: { + queue: string; + defaultEnabled?: boolean; +}) => store.get().queues[queue]?.enabled ?? defaultEnabled; + +export const getJobQueueConfigStatus = () => store.getStatus(); + +export const getJobQueueConfigFromSource = async () => store.readFromSource(); + +export const updateFullJobQueueConfig = async ({ + config, +}: { + config: JobQueueConfig; +}) => { + await store.writeToSource({ config }); +}; diff --git a/server/src/queue/createWorkerContext.ts b/server/src/queue/createWorkerContext.ts index f21b30b38..5bc2e61d1 100644 --- a/server/src/queue/createWorkerContext.ts +++ b/server/src/queue/createWorkerContext.ts @@ -18,10 +18,11 @@ export const createWorkerContext = async ({ orgId?: string; env?: AppEnv; customerId?: string; + requestId?: string; }; logger: Logger; }) => { - const { orgId, env, customerId } = payload; + const { orgId, env, customerId, requestId } = payload; if (!orgId || !env) return; // Fetch org with features once for all items @@ -74,7 +75,7 @@ export const createWorkerContext = async ({ logger: workerLogger, redisV2: resolveRedisV2(), - id: generateId("job"), + id: requestId || generateId("job"), timestamp: Date.now(), isPublic: false, authType: AuthType.Worker, diff --git a/server/src/queue/initWorkers.ts b/server/src/queue/initWorkers.ts index 2f4326b4c..0d0390615 100644 --- a/server/src/queue/initWorkers.ts +++ b/server/src/queue/initWorkers.ts @@ -12,6 +12,10 @@ import * as Sentry from "@sentry/bun"; import { type DrizzleCli, initDrizzle } from "@/db/initDrizzle.js"; import { logger } from "@/external/logtail/logtailUtils.js"; import { verifyCacheConsistency } from "@/internal/billing/v2/workflows/verifyCacheConsistency/verifyCacheConsistency.js"; +import { + isJobQueueEnabled, + JOB_QUEUE_IDS, +} from "@/internal/misc/jobQueues/jobQueueStore.js"; import { generateId } from "@/utils/genUtils.js"; import { withTimeout } from "@/utils/withTimeout.js"; import { hatchet } from "../external/hatchet/initHatchet.js"; @@ -21,7 +25,7 @@ import { processMessage, type SqsJob } from "./processMessage.js"; // ============ Shared State ============ let isRunning = true; -let abortController: AbortController; +const abortControllers = new Set(); // Process recycling — exit after processing this many messages to prevent memory leaks const MAX_MESSAGES_BEFORE_RECYCLE = 50_000; @@ -47,18 +51,20 @@ const logPrefix = ({ queueUrl }: { queueUrl: string }) => // ============ Polling Loop (per-queue, per-loop state) ============ -const startPollingLoop = async ({ +export const startPollingLoop = async ({ db, queueUrl, isFifo, getSqsClientFn, recreateSqsClientFn, + shouldPoll = () => true, }: { db: DrizzleCli; queueUrl: string; isFifo: boolean; getSqsClientFn: () => SQSClient; recreateSqsClientFn: () => SQSClient; + shouldPoll?: () => boolean; }) => { // Per-loop state let messagesProcessed = 0; @@ -70,6 +76,8 @@ const startPollingLoop = async ({ let consecutiveZeroMessageIntervals = 0; const prefix = logPrefix({ queueUrl }); + let abortController = new AbortController(); + abortControllers.add(abortController); const alertZeroMessages = () => { const minutes = consecutiveZeroMessageIntervals; @@ -105,6 +113,13 @@ const startPollingLoop = async ({ }; const logStatsAndCheckZeroMessages = () => { + if (!shouldPoll()) { + consecutiveZeroMessageIntervals = 0; + messagesProcessed = 0; + lastStatsTime = Date.now(); + return; + } + const elapsedSeconds = ((Date.now() - lastStatsTime) / 1000).toFixed(0); const mem = process.memoryUsage(); console.log( @@ -245,6 +260,7 @@ const startPollingLoop = async ({ ); consecutiveEmptyPolls = 0; abortController = new AbortController(); + abortControllers.add(abortController); return recreateSqsClientFn(); } @@ -270,6 +286,7 @@ const startPollingLoop = async ({ console.warn(`${prefix} Repeated errors - recreating SQS client`); consecutiveEmptyPolls = 0; abortController = new AbortController(); + abortControllers.add(abortController); await new Promise((resolve) => setTimeout(resolve, 5000)); return recreateSqsClientFn(); } @@ -284,6 +301,12 @@ const startPollingLoop = async ({ while (isRunning) { try { + if (!shouldPoll()) { + consecutiveEmptyPolls = 0; + await new Promise((resolve) => setTimeout(resolve, 5000)); + continue; + } + const response = await sqs.send(createReceiveCommand(), { abortSignal: abortController.signal, }); @@ -348,6 +371,7 @@ const startPollingLoop = async ({ } } + abortControllers.delete(abortController); clearInterval(statsInterval); console.log(`${prefix} Stopped`); }; @@ -370,7 +394,9 @@ export const initWorkers = async ({ const shutdown = async () => { console.log(`[SQS Worker ${process.pid}] Shutting down...`); isRunning = false; - if (abortController) abortController.abort(); + for (const controller of abortControllers) { + controller.abort(); + } const isProd = process.env.NODE_ENV === "production"; if (isProd) { @@ -386,20 +412,36 @@ export const initWorkers = async ({ process.on("SIGTERM", shutdown); process.on("SIGINT", shutdown); - abortController = new AbortController(); - const startupDurationMs = Date.now() - startupStartedAt; console.log( `[Worker ${process.pid}] ${queueImplementation} worker ready in ${startupDurationMs}ms`, ); + const pollingLoops = [ + startPollingLoop({ + db, + queueUrl: QUEUE_URL, + isFifo: QUEUE_URL.endsWith(".fifo"), + getSqsClientFn: getSqsClient, + recreateSqsClientFn: recreateSqsClient, + shouldPoll: () => isJobQueueEnabled({ queue: JOB_QUEUE_IDS.primary }), + }), + ]; - await startPollingLoop({ - db, - queueUrl: QUEUE_URL, - isFifo: QUEUE_URL.endsWith(".fifo"), - getSqsClientFn: getSqsClient, - recreateSqsClientFn: recreateSqsClient, - }); + const trackQueueUrl = process.env.TRACK_SQS_QUEUE_URL; + if (trackQueueUrl) { + pollingLoops.push( + startPollingLoop({ + db, + queueUrl: trackQueueUrl, + isFifo: trackQueueUrl.endsWith(".fifo"), + getSqsClientFn: getSqsClient, + recreateSqsClientFn: recreateSqsClient, + shouldPoll: () => isJobQueueEnabled({ queue: JOB_QUEUE_IDS.track }), + }), + ); + } + + await Promise.all(pollingLoops); }; export const initHatchetWorker = async () => { diff --git a/server/src/queue/processMessage.ts b/server/src/queue/processMessage.ts index 02f2cac57..dce2b279c 100644 --- a/server/src/queue/processMessage.ts +++ b/server/src/queue/processMessage.ts @@ -10,6 +10,7 @@ import { runActionHandlerTask } from "@/internal/analytics/runActionHandlerTask. import { autoTopup } from "@/internal/balances/autoTopUp/autoTopup.js"; import { runInsertEventBatch } from "@/internal/balances/events/runInsertEventBatch.js"; import { expireLock } from "@/internal/balances/finalizeLock/expireLock.js"; +import { runQueuedTrack } from "@/internal/balances/track/runQueuedTrack.js"; import { refreshEntityAggregateCache } from "@/internal/balances/utils/refreshEntityAggregate/index.js"; import { syncItemV3 } from "@/internal/balances/utils/sync/syncItemV3.js"; import { syncItemV4 } from "@/internal/balances/utils/sync/syncItemV4.js"; @@ -175,6 +176,20 @@ export const processMessage = async ({ return; } + if (job.name === JobName.Track) { + if (!ctx) { + workerLogger.error("No context found for track job"); + return; + } + + await runQueuedTrack({ + ctx, + body: job.data.body, + apiVersion: job.data.apiVersion, + }); + return; + } + if (job.name === JobName.RefreshEntityAggregate) { if (!ctx) { workerLogger.error("No context found for refresh entity aggregate job"); diff --git a/server/src/queue/queueUtils.ts b/server/src/queue/queueUtils.ts index 34bf6c92d..6401b94ce 100644 --- a/server/src/queue/queueUtils.ts +++ b/server/src/queue/queueUtils.ts @@ -69,6 +69,9 @@ export interface Payloads { [JobName.Track]: { orgId: string; env: AppEnv; + customerId: string; + entityId?: string; + requestId: string; apiVersion: ApiVersion; body: TrackParams; }; diff --git a/server/src/workers.ts b/server/src/workers.ts index bac4ac87f..5c4995972 100644 --- a/server/src/workers.ts +++ b/server/src/workers.ts @@ -10,6 +10,7 @@ import { import "./internal/misc/requestBlocks/requestBlockStore.js"; import "./internal/misc/rollouts/rolloutConfigStore.js"; import "./internal/misc/redisV2Cache/redisV2CacheStore.js"; +import "./internal/misc/jobQueues/jobQueueStore.js"; // Number of worker processes (defaults to CPU cores) const NUM_PROCESSES = process.env.NODE_ENV === "development" ? 3 : 4; diff --git a/server/tests/integration/admin/job-queue-config.test.ts b/server/tests/integration/admin/job-queue-config.test.ts new file mode 100644 index 000000000..23a92fcd6 --- /dev/null +++ b/server/tests/integration/admin/job-queue-config.test.ts @@ -0,0 +1,246 @@ +import { beforeEach, describe, expect, mock, test } from "bun:test"; +import { AppEnv, ErrCode } from "@autumn/shared"; +import { Hono } from "hono"; +import { errorMiddleware } from "@/honoMiddlewares/errorMiddleware.js"; +import type { HonoEnv } from "@/honoUtils/HonoEnv.js"; + +type MockJobQueueConfig = { + queues: Record; +}; + +type MockStatus = { + healthy: boolean; + configured: boolean; + lastSuccessAt: string | null; + error: string | null; +}; + +const mockState = { + config: { + queues: { + primary: { enabled: true }, + track: { enabled: false }, + }, + } as MockJobQueueConfig, + status: { + healthy: true, + configured: true, + lastSuccessAt: "2026-04-24T10:00:00.000Z", + error: null, + } as MockStatus, + updateCalls: [] as unknown[], +}; + +mock.module("@/internal/misc/jobQueues/jobQueueStore.js", () => ({ + KNOWN_JOB_QUEUES: [ + { + id: "primary", + label: "Primary Queue", + description: "Shared SQS queue for standard background jobs.", + defaultEnabled: true, + }, + { + id: "track", + label: "Track Replay Queue", + description: + "Dedicated async track replay queue used during fail-open recovery.", + defaultEnabled: true, + }, + ], + getJobQueueConfigFromSource: async () => mockState.config, + getJobQueueConfigStatus: () => mockState.status, + updateFullJobQueueConfig: async ({ config }: { config: unknown }) => { + mockState.updateCalls.push(config); + }, +})); + +import { handleGetAdminJobQueueConfig } from "@/internal/admin/handleGetAdminJobQueueConfig.js"; +import { handleUpsertAdminJobQueueConfig } from "@/internal/admin/handleUpsertAdminJobQueueConfig.js"; + +const buildApp = () => { + const app = new Hono(); + + app.use("*", async (c, next) => { + c.set("ctx", { + env: AppEnv.Sandbox, + org: { slug: "tests-org" }, + logger: { + warn: () => undefined, + error: () => undefined, + }, + } as any); + await next(); + }); + + app.get("/admin/job-queue-config", ...handleGetAdminJobQueueConfig); + app.put("/admin/job-queue-config", ...handleUpsertAdminJobQueueConfig); + app.onError(errorMiddleware); + + return app; +}; + +describe("admin job queue config", () => { + beforeEach(() => { + mockState.config = { + queues: { + primary: { enabled: true }, + track: { enabled: false }, + }, + }; + mockState.status = { + healthy: true, + configured: true, + lastSuccessAt: "2026-04-24T10:00:00.000Z", + error: null, + }; + mockState.updateCalls = []; + }); + + test("GET returns the stored config, status, and known queues", async () => { + const app = buildApp(); + + const response = await app.request("http://localhost/admin/job-queue-config"); + const body = await response.json(); + + expect(response.status).toBe(200); + expect(body).toMatchObject({ + queues: { + primary: { enabled: true }, + track: { enabled: false }, + }, + configHealthy: true, + configConfigured: true, + lastSuccessAt: "2026-04-24T10:00:00.000Z", + error: null, + }); + expect(body.knownQueues).toEqual([ + { + id: "primary", + label: "Primary Queue", + description: "Shared SQS queue for standard background jobs.", + defaultEnabled: true, + }, + { + id: "track", + label: "Track Replay Queue", + description: + "Dedicated async track replay queue used during fail-open recovery.", + defaultEnabled: true, + }, + ]); + }); + + test("GET handles an unconfigured empty config", async () => { + mockState.config = { queues: {} }; + mockState.status = { + healthy: false, + configured: false, + lastSuccessAt: null, + error: "missing s3 object", + }; + + const app = buildApp(); + const response = await app.request("http://localhost/admin/job-queue-config"); + const body = await response.json(); + + expect(response.status).toBe(200); + expect(body).toMatchObject({ + queues: {}, + configHealthy: false, + configConfigured: false, + lastSuccessAt: null, + error: "missing s3 object", + }); + expect(body.knownQueues).toHaveLength(2); + }); + + test("PUT saves a validated config payload", async () => { + const app = buildApp(); + + const response = await app.request("http://localhost/admin/job-queue-config", { + method: "PUT", + headers: { + "Content-Type": "application/json", + }, + body: JSON.stringify({ + queues: { + primary: { enabled: false }, + track: { enabled: true }, + }, + }), + }); + + expect(response.status).toBe(200); + expect(await response.json()).toEqual({ success: true }); + expect(mockState.updateCalls).toEqual([ + { + queues: { + primary: { enabled: false }, + track: { enabled: true }, + }, + }, + ]); + }); + + test("PUT preserves unknown queues for future config expansion", async () => { + const app = buildApp(); + + const response = await app.request("http://localhost/admin/job-queue-config", { + method: "PUT", + headers: { + "Content-Type": "application/json", + }, + body: JSON.stringify({ + queues: { + reports: { enabled: false }, + }, + }), + }); + + expect(response.status).toBe(200); + expect(mockState.updateCalls).toEqual([ + { + queues: { + reports: { enabled: false }, + }, + }, + ]); + }); + + test("PUT accepts an empty payload and writes the schema default", async () => { + const app = buildApp(); + + const response = await app.request("http://localhost/admin/job-queue-config", { + method: "PUT", + headers: { + "Content-Type": "application/json", + }, + body: JSON.stringify({}), + }); + + expect(response.status).toBe(200); + expect(mockState.updateCalls).toEqual([{ queues: {} }]); + }); + + test("PUT rejects invalid queue payloads", async () => { + const app = buildApp(); + + const response = await app.request("http://localhost/admin/job-queue-config", { + method: "PUT", + headers: { + "Content-Type": "application/json", + }, + body: JSON.stringify({ + queues: { + track: { enabled: "yes" }, + }, + }), + }); + + const body = await response.json(); + + expect(response.status).toBe(400); + expect(body.code).toBe(ErrCode.InvalidInputs); + expect(mockState.updateCalls).toHaveLength(0); + }); +}); diff --git a/server/tests/integration/queue/fixtures/livePollingWorker.ts b/server/tests/integration/queue/fixtures/livePollingWorker.ts new file mode 100644 index 000000000..4dfab199f --- /dev/null +++ b/server/tests/integration/queue/fixtures/livePollingWorker.ts @@ -0,0 +1,35 @@ +import { SQSClient } from "@aws-sdk/client-sqs"; +import { startPollingLoop } from "@/queue/initWorkers.js"; + +const queueUrl = process.env.TEST_QUEUE_URL; +const endpoint = process.env.TEST_SQS_ENDPOINT; +const shouldPoll = process.env.TEST_SHOULD_POLL === "true"; + +if (!queueUrl || !endpoint) { + throw new Error("TEST_QUEUE_URL and TEST_SQS_ENDPOINT are required"); +} + +const createClient = () => + new SQSClient({ + region: "us-east-1", + endpoint, + credentials: { + accessKeyId: "x", + secretAccessKey: "x", + }, + }); + +let client = createClient(); + +await startPollingLoop({ + db: {} as never, + queueUrl, + isFifo: queueUrl.endsWith(".fifo"), + getSqsClientFn: () => client, + recreateSqsClientFn: () => { + client.destroy(); + client = createClient(); + return client; + }, + shouldPoll: () => shouldPoll, +}); diff --git a/server/tests/integration/queue/live-worker-process.test.ts b/server/tests/integration/queue/live-worker-process.test.ts new file mode 100644 index 000000000..c07707ca3 --- /dev/null +++ b/server/tests/integration/queue/live-worker-process.test.ts @@ -0,0 +1,271 @@ +import { mkdtemp, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { randomUUID } from "node:crypto"; +import { spawn, type ChildProcess } from "node:child_process"; +import { afterAll, afterEach, beforeAll, describe, expect, test } from "bun:test"; +import { + CreateQueueCommand, + DeleteQueueCommand, + GetQueueAttributesCommand, + SendMessageCommand, + SQSClient, +} from "@aws-sdk/client-sqs"; + +const ELASTICMQ_JAR = `${process.env.HOME}/.autumn-agent/elasticmq/elasticmq.jar`; +const WORKER_FIXTURE_PATH = new URL( + "./fixtures/livePollingWorker.ts", + import.meta.url, +).pathname; + +const children: ChildProcess[] = []; +const queueUrls: string[] = []; +let elasticmq: ChildProcess | null = null; +let tempDir: string | null = null; +let sqsEndpoint = ""; +let sqs: SQSClient; + +const waitForExit = async ({ + child, + forceKillAfterMs = 1_000, +}: { + child: ChildProcess; + forceKillAfterMs?: number; +}) => { + if (child.exitCode !== null || child.signalCode !== null) { + return; + } + + await new Promise((resolve) => { + const timeout = setTimeout(() => { + child.kill("SIGKILL"); + }, forceKillAfterMs); + + child.once("exit", () => { + clearTimeout(timeout); + resolve(); + }); + }); +}; + +const waitFor = async ({ + check, + timeoutMs = 8_000, + intervalMs = 200, +}: { + check: () => Promise; + timeoutMs?: number; + intervalMs?: number; +}) => { + const deadline = Date.now() + timeoutMs; + + while (Date.now() < deadline) { + if (await check()) return; + await new Promise((resolve) => setTimeout(resolve, intervalMs)); + } + + throw new Error(`Condition not met within ${timeoutMs}ms`); +}; + +beforeAll(async () => { + const port = 19_000 + Math.floor(Math.random() * 1_000); + const statsPort = port + 1; + tempDir = await mkdtemp(join(tmpdir(), "elasticmq-")); + const configPath = join(tempDir, "elasticmq.conf"); + sqsEndpoint = `http://127.0.0.1:${port}`; + + await writeFile( + configPath, + `include classpath("application.conf") +node-address { + protocol = http + host = "127.0.0.1" + port = ${port} + context-path = "" +} +rest-sqs { + enabled = true + bind-port = ${port} + bind-hostname = "127.0.0.1" + sqs-limits = strict +} +generate-node-address = false +rest-stats { + enabled = true + bind-port = ${statsPort} + bind-hostname = "127.0.0.1" +} +queues {} +`, + ); + + elasticmq = spawn( + "java", + [`-Dconfig.file=${configPath}`, "-jar", ELASTICMQ_JAR], + { + stdio: ["ignore", "pipe", "pipe"], + }, + ); + + sqs = new SQSClient({ + region: "us-east-1", + endpoint: sqsEndpoint, + credentials: { + accessKeyId: "x", + secretAccessKey: "x", + }, + }); + + await waitFor({ + timeoutMs: 15_000, + check: async () => { + try { + const response = await fetch( + `${sqsEndpoint}/?Action=ListQueues&Version=2012-11-05`, + ); + return response.ok; + } catch { + return false; + } + }, + }); +}); + +afterAll(async () => { + if (elasticmq) { + elasticmq.kill("SIGTERM"); + await waitForExit({ child: elasticmq }); + } + + if (tempDir) { + await rm(tempDir, { recursive: true, force: true }); + } +}); + +const createTestQueue = async () => { + const name = `autumn-live-${randomUUID()}.fifo`; + const response = await sqs.send( + new CreateQueueCommand({ + QueueName: name, + Attributes: { + FifoQueue: "true", + ContentBasedDeduplication: "true", + VisibilityTimeout: "5", + }, + }), + ); + + if (!response.QueueUrl) { + throw new Error("Failed to create test queue"); + } + + queueUrls.push(response.QueueUrl); + return response.QueueUrl; +}; + +const getQueueCounts = async ({ queueUrl }: { queueUrl: string }) => { + const response = await sqs.send( + new GetQueueAttributesCommand({ + QueueUrl: queueUrl, + AttributeNames: [ + "ApproximateNumberOfMessages", + "ApproximateNumberOfMessagesNotVisible", + ], + }), + ); + + return { + visible: Number.parseInt( + response.Attributes?.ApproximateNumberOfMessages ?? "0", + 10, + ), + notVisible: Number.parseInt( + response.Attributes?.ApproximateNumberOfMessagesNotVisible ?? "0", + 10, + ), + }; +}; + +const startWorker = ({ + queueUrl, + shouldPoll, +}: { + queueUrl: string; + shouldPoll: boolean; +}) => { + const child = spawn( + "bun", + [WORKER_FIXTURE_PATH], + { + cwd: process.cwd(), + env: { + ...process.env, + TEST_QUEUE_URL: queueUrl, + TEST_SQS_ENDPOINT: sqsEndpoint, + TEST_SHOULD_POLL: shouldPoll ? "true" : "false", + }, + stdio: ["ignore", "pipe", "pipe"], + }, + ); + + children.push(child); + return child; +}; + +const sendTestMessage = async ({ queueUrl }: { queueUrl: string }) => { + await sqs.send( + new SendMessageCommand({ + QueueUrl: queueUrl, + MessageBody: JSON.stringify({ + name: "integration-test-job", + data: {}, + }), + MessageGroupId: "test", + MessageDeduplicationId: randomUUID(), + }), + ); +}; + +afterEach(async () => { + for (const child of children.splice(0)) { + child.kill("SIGTERM"); + await waitForExit({ child }); + } + + for (const queueUrl of queueUrls.splice(0)) { + await sqs.send( + new DeleteQueueCommand({ + QueueUrl: queueUrl, + }), + ); + } +}); + +describe("live worker process queue polling", () => { + test("enabled worker consumes a live SQS message", async () => { + const queueUrl = await createTestQueue(); + + startWorker({ queueUrl, shouldPoll: true }); + await sendTestMessage({ queueUrl }); + + await waitFor({ + check: async () => { + const counts = await getQueueCounts({ queueUrl }); + return counts.visible === 0 && counts.notVisible === 0; + }, + }); + }, 15_000); + + test("disabled worker leaves the live SQS message untouched", async () => { + const queueUrl = await createTestQueue(); + + startWorker({ queueUrl, shouldPoll: false }); + await sendTestMessage({ queueUrl }); + + await new Promise((resolve) => setTimeout(resolve, 1_500)); + + const counts = await getQueueCounts({ queueUrl }); + expect(counts.visible).toBe(1); + expect(counts.notVisible).toBe(0); + }, 15_000); +}); diff --git a/server/tests/integration/queue/start-polling-loop.test.ts b/server/tests/integration/queue/start-polling-loop.test.ts new file mode 100644 index 000000000..5ea4dbb9f --- /dev/null +++ b/server/tests/integration/queue/start-polling-loop.test.ts @@ -0,0 +1,89 @@ +import { afterEach, beforeEach, describe, expect, test } from "bun:test"; +import { startPollingLoop } from "@/queue/initWorkers.js"; + +const originalSetTimeout = globalThis.setTimeout; + +const makeAbortError = () => { + const error = new Error("aborted") as Error & { name: string }; + error.name = "AbortError"; + return error; +}; + +describe("startPollingLoop", () => { + beforeEach(() => { + globalThis.setTimeout = (((callback: TimerHandler) => { + if (typeof callback === "function") { + callback(); + } + return 0 as unknown as ReturnType; + }) as unknown) as typeof setTimeout; + }); + + afterEach(() => { + globalThis.setTimeout = originalSetTimeout; + }); + + test("does not poll while the queue is disabled", async () => { + let shouldPollCalls = 0; + let sendCalls = 0; + + await startPollingLoop({ + db: {} as never, + queueUrl: "https://sqs.eu-west-1.amazonaws.com/123/track.fifo", + isFifo: true, + getSqsClientFn: () => + ({ + send: async () => { + sendCalls++; + throw makeAbortError(); + }, + }) as never, + recreateSqsClientFn: () => + ({ + send: async () => { + sendCalls++; + throw makeAbortError(); + }, + }) as never, + shouldPoll: () => { + shouldPollCalls++; + return shouldPollCalls > 1; + }, + }); + + expect(shouldPollCalls).toBeGreaterThan(1); + expect(sendCalls).toBe(1); + }); + + test("polls immediately when the queue is enabled", async () => { + let shouldPollCalls = 0; + let sendCalls = 0; + + await startPollingLoop({ + db: {} as never, + queueUrl: "https://sqs.eu-west-1.amazonaws.com/123/primary.fifo", + isFifo: true, + getSqsClientFn: () => + ({ + send: async () => { + sendCalls++; + throw makeAbortError(); + }, + }) as never, + recreateSqsClientFn: () => + ({ + send: async () => { + sendCalls++; + throw makeAbortError(); + }, + }) as never, + shouldPoll: () => { + shouldPollCalls++; + return true; + }, + }); + + expect(shouldPollCalls).toBe(1); + expect(sendCalls).toBe(1); + }); +}); diff --git a/server/tests/unit/balances/track-v3/handleRedisTrackErrorV3.test.ts b/server/tests/unit/balances/track-v3/handleRedisTrackErrorV3.test.ts index d850206b0..0602f1bf1 100644 --- a/server/tests/unit/balances/track-v3/handleRedisTrackErrorV3.test.ts +++ b/server/tests/unit/balances/track-v3/handleRedisTrackErrorV3.test.ts @@ -7,21 +7,9 @@ import { } from "@/internal/balances/utils/types/redisDeductionError.js"; const mockState = { - queueCalls: [] as Record[], postgresCalls: [] as Record[], }; -mock.module("@/internal/balances/track/utils/queueTrack.js", () => ({ - queueTrack: async (args: Record) => { - mockState.queueCalls.push(args); - return { - customer_id: "cus_123", - feature_id: "messages", - balance: null, - }; - }, -})); - mock.module( "@/internal/balances/track/v3/runPostgresTrackV3.js", () => ({ @@ -45,31 +33,29 @@ const ctx = { describe("handleRedisTrackErrorV3", () => { beforeEach(() => { - mockState.queueCalls = []; mockState.postgresCalls = []; }); - test("queues track instead of falling back to Postgres when Redis is unavailable", async () => { - const response = await handleRedisTrackErrorV3({ - ctx, - error: new RedisDeductionError({ - message: "Redis not ready for deduction", - code: RedisDeductionErrorCode.RedisUnavailable, - }), - body: { - customer_id: "cus_123", - feature_id: "messages", - value: 1, - }, - fullSubject: {} as never, - featureDeductions: [], + test("rethrows when Redis is unavailable", async () => { + const error = new RedisDeductionError({ + message: "Redis not ready for deduction", + code: RedisDeductionErrorCode.RedisUnavailable, }); - expect(mockState.queueCalls).toHaveLength(1); + await expect( + handleRedisTrackErrorV3({ + ctx, + error, + body: { + customer_id: "cus_123", + feature_id: "messages", + value: 1, + }, + fullSubject: {} as never, + featureDeductions: [], + }), + ).rejects.toBe(error); + expect(mockState.postgresCalls).toHaveLength(0); - expect(response).toMatchObject({ - customer_id: "cus_123", - balance: null, - }); }); }); 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 7692f018f..16621a77d 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 @@ -42,6 +42,10 @@ mock.module("@/internal/balances/track/utils/queueTrack.js", () => ({ }, })); +mock.module("@/external/redis/initUtils/redisV2Availability.js", () => ({ + shouldUseRedisV2: () => true, +})); + import { runTrackWithRollout } from "@/internal/balances/track/runTrackWithRollout.js"; const ctx = { diff --git a/server/tests/unit/balances/track/queueTrack.test.ts b/server/tests/unit/balances/track/queueTrack.test.ts new file mode 100644 index 000000000..ea604fd47 --- /dev/null +++ b/server/tests/unit/balances/track/queueTrack.test.ts @@ -0,0 +1,75 @@ +import { afterEach, beforeEach, describe, expect, mock, test } from "bun:test"; +import { ApiVersion, ApiVersionClass, AppEnv } from "@autumn/shared"; +import type { AutumnContext } from "@/honoUtils/HonoEnv.js"; + +const mockState = { + queueCalls: [] as Record[], +}; + +mock.module("@/queue/queueUtils.js", () => ({ + addTaskToQueue: async (args: Record) => { + mockState.queueCalls.push(args); + }, +})); + +mock.module("@/internal/balances/track/utils/getQueuedTrackResponse.js", () => ({ + getQueuedTrackResponse: () => ({ + customer_id: "cus_123", + value: 2, + balance: null, + }), +})); + +import { queueTrack } from "@/internal/balances/track/utils/queueTrack.js"; + +describe("queueTrack", () => { + const originalTrackQueueUrl = process.env.TRACK_SQS_QUEUE_URL; + + beforeEach(() => { + mockState.queueCalls = []; + process.env.TRACK_SQS_QUEUE_URL = + "https://sqs.eu-west-1.amazonaws.com/123456789012/track-dev.fifo"; + }); + + test("queues track with request identity and entity-scoped grouping", async () => { + const ctx = { + id: "req_123", + org: { id: "org_123" }, + env: AppEnv.Sandbox, + apiVersion: new ApiVersionClass(ApiVersion.V2_1), + logger: { + warn: mock(() => {}), + }, + } as unknown as AutumnContext; + + await queueTrack({ + ctx, + body: { + customer_id: "cus_123", + entity_id: "ent_123", + feature_id: "messages", + value: 2, + }, + }); + + expect(mockState.queueCalls).toHaveLength(1); + expect(mockState.queueCalls[0]).toMatchObject({ + queueUrl: + "https://sqs.eu-west-1.amazonaws.com/123456789012/track-dev.fifo", + messageGroupId: "org_123:sandbox:cus_123:ent_123", + messageDeduplicationId: "req_123", + payload: { + orgId: "org_123", + env: AppEnv.Sandbox, + customerId: "cus_123", + entityId: "ent_123", + requestId: "req_123", + apiVersion: ApiVersion.V2_1, + }, + }); + }); + + afterEach(() => { + process.env.TRACK_SQS_QUEUE_URL = originalTrackQueueUrl; + }); +}); diff --git a/server/tests/unit/balances/track/runQueuedTrack.test.ts b/server/tests/unit/balances/track/runQueuedTrack.test.ts new file mode 100644 index 000000000..ccfe87571 --- /dev/null +++ b/server/tests/unit/balances/track/runQueuedTrack.test.ts @@ -0,0 +1,88 @@ +import { beforeEach, describe, expect, mock, test } from "bun:test"; +import { ApiVersion, ApiVersionClass, AppEnv, ErrCode, RecaseError } from "@autumn/shared"; +import type { AutumnContext } from "@/honoUtils/HonoEnv.js"; + +const mockState = { + runTrackV3Calls: [] as Record[], + getFeatureDeductionCalls: [] as Record[], + runTrackV3Error: null as unknown, +}; + +mock.module("@/internal/balances/track/utils/getFeatureDeductions.js", () => ({ + getTrackFeatureDeductionsForBody: (args: Record) => { + mockState.getFeatureDeductionCalls.push(args); + return []; + }, +})); + +mock.module("@/internal/balances/track/v3/runTrackV3.js", () => ({ + runTrackV3: async (args: Record) => { + mockState.runTrackV3Calls.push(args); + if (mockState.runTrackV3Error) throw mockState.runTrackV3Error; + return { customer_id: "cus_123", balance: null }; + }, +})); + +import { runQueuedTrack } from "@/internal/balances/track/runQueuedTrack.js"; + +const ctx = { + id: "req_123", + env: AppEnv.Sandbox, + org: { id: "org_123" }, + apiVersion: new ApiVersionClass(ApiVersion.V2_1), + logger: { + info: mock(() => {}), + }, +} as unknown as AutumnContext; + +describe("runQueuedTrack", () => { + beforeEach(() => { + mockState.runTrackV3Calls = []; + mockState.getFeatureDeductionCalls = []; + mockState.runTrackV3Error = null; + }); + + test("replays queued track through runTrackV3", async () => { + await runQueuedTrack({ + ctx, + body: { + customer_id: "cus_123", + feature_id: "messages", + value: 1, + }, + apiVersion: ApiVersion.V2_1, + }); + + expect(mockState.getFeatureDeductionCalls).toHaveLength(1); + expect(mockState.runTrackV3Calls).toHaveLength(1); + expect(mockState.runTrackV3Calls[0]).toMatchObject({ + ctx, + body: { + customer_id: "cus_123", + feature_id: "messages", + }, + featureDeductions: [], + apiVersion: ApiVersion.V2_1, + }); + }); + + test("treats duplicate idempotency as already applied", async () => { + mockState.runTrackV3Error = new RecaseError({ + message: "duplicate", + code: ErrCode.DuplicateIdempotencyKey, + statusCode: 409, + }); + + await expect( + runQueuedTrack({ + ctx, + body: { + customer_id: "cus_123", + feature_id: "messages", + value: 1, + }, + apiVersion: ApiVersion.V2_1, + }), + ).resolves.toBeUndefined(); + }); +}); diff --git a/server/tests/unit/queue/queue-utils.test.ts b/server/tests/unit/queue/queue-utils.test.ts index a27f4aa29..e7d4f7fa4 100644 --- a/server/tests/unit/queue/queue-utils.test.ts +++ b/server/tests/unit/queue/queue-utils.test.ts @@ -1,20 +1,14 @@ -import { afterEach, beforeEach, describe, expect, mock, test } from "bun:test"; +import { afterEach, beforeEach, describe, expect, test } from "bun:test"; import { ApiVersion, AppEnv } from "@autumn/shared"; +import type { SQSClient } from "@aws-sdk/client-sqs"; const mockState = { commands: [] as Record[], + originalSend: null as null | SQSClient["send"], }; -mock.module("@/queue/initSqs.js", () => ({ - QUEUE_URL: "https://sqs.eu-west-1.amazonaws.com/123456789012/primary.fifo", - getSqsClient: () => ({ - send: async (command: { input: Record }) => { - mockState.commands.push(command.input); - }, - }), -})); - import { JobName } from "@/queue/JobName.js"; +import { getSqsClient } from "@/queue/initSqs.js"; import { addTaskToQueue } from "@/queue/queueUtils.js"; describe("addTaskToQueue queue override", () => { @@ -23,11 +17,21 @@ describe("addTaskToQueue queue override", () => { beforeEach(() => { mockState.commands = []; + const sqsClient = getSqsClient(); + mockState.originalSend = sqsClient.send.bind(sqsClient); + sqsClient.send = (async (command: { input: Record }) => { + mockState.commands.push(command.input); + return {}; + }) as typeof sqsClient.send; delete process.env.SQS_QUEUE_URL; delete process.env.QUEUE_URL; }); afterEach(() => { + if (mockState.originalSend) { + const sqsClient = getSqsClient(); + sqsClient.send = mockState.originalSend as typeof sqsClient.send; + } process.env.SQS_QUEUE_URL = originalSqsQueueUrl; process.env.QUEUE_URL = originalQueueUrl; }); @@ -42,6 +46,8 @@ describe("addTaskToQueue queue override", () => { payload: { orgId: "org_123", env: AppEnv.Sandbox, + customerId: "cus_123", + requestId: "req_123", apiVersion: ApiVersion.V2_1, body: { customer_id: "cus_123", diff --git a/vite/src/views/admin/components/EdgeConfigTab.tsx b/vite/src/views/admin/components/EdgeConfigTab.tsx index 0c867a94d..659f4d896 100644 --- a/vite/src/views/admin/components/EdgeConfigTab.tsx +++ b/vite/src/views/admin/components/EdgeConfigTab.tsx @@ -5,6 +5,7 @@ import { useAxiosInstance } from "@/services/useAxiosInstance"; import { CustomerBlockDialog } from "./CustomerBlockDialog"; import { EdgeConfigDialog } from "./EdgeConfigDialog"; import { FeatureFlagsDialog } from "./FeatureFlagsDialog"; +import { JobQueuesDialog } from "./JobQueuesDialog"; import { OrgLimitsDialog } from "./OrgLimitsDialog"; import { RawEdgeConfigDialog } from "./RawEdgeConfigDialog"; import { RedisV2CacheDialog } from "./RedisV2CacheDialog"; @@ -27,6 +28,7 @@ export function EdgeConfigTab() { const [featureFlagsOpen, setFeatureFlagsOpen] = useState(false); const [customerBlockOpen, setCustomerBlockOpen] = useState(false); const [orgLimitsOpen, setOrgLimitsOpen] = useState(false); + const [jobQueuesOpen, setJobQueuesOpen] = useState(false); const [stripeSyncOpen, setStripeSyncOpen] = useState(false); const [redisV2CacheOpen, setRedisV2CacheOpen] = useState(false); @@ -156,6 +158,23 @@ export function EdgeConfigTab() { +
+
+
Job Queues
+
+ Pause or resume worker consumption for shared and dedicated SQS + queues. +
+
+ +
+
Stripe Sync
@@ -215,6 +234,11 @@ export function EdgeConfigTab() { + + ; + knownQueues: KnownQueue[]; + configHealthy: boolean; + configConfigured: boolean; + lastSuccessAt: string | null; + error: string | null; +}; + +const DEFAULT_CONFIG: JobQueueConfig = { + queues: {}, + knownQueues: [], + configHealthy: false, + configConfigured: false, + lastSuccessAt: null, + error: null, +}; + +export function JobQueuesDialog({ + open, + onOpenChange, +}: { + open: boolean; + onOpenChange: (open: boolean) => void; +}) { + const axiosInstance = useAxiosInstance(); + const [loading, setLoading] = useState(false); + const [saving, setSaving] = useState(false); + const [config, setConfig] = useState(DEFAULT_CONFIG); + const [initialEnabledByQueue, setInitialEnabledByQueue] = useState< + Record + >({}); + + useEffect(() => { + if (!open) return; + + let cancelled = false; + setLoading(true); + + void axiosInstance + .get("/admin/job-queue-config") + .then(({ data }) => { + if (cancelled) return; + const nextConfig = { ...DEFAULT_CONFIG, ...data }; + setConfig(nextConfig); + setInitialEnabledByQueue( + Object.fromEntries( + nextConfig.knownQueues.map((queue) => [ + queue.id, + nextConfig.queues[queue.id]?.enabled ?? queue.defaultEnabled, + ]), + ), + ); + }) + .catch((error) => { + if (!cancelled) { + toast.error(getBackendErr(error, "Failed to load job queue config")); + } + }) + .finally(() => { + if (!cancelled) setLoading(false); + }); + + return () => { + cancelled = true; + }; + }, [axiosInstance, open]); + + const effectiveQueues = useMemo( + () => + config.knownQueues.map((queue) => ({ + ...queue, + enabled: + config.queues[queue.id]?.enabled ?? queue.defaultEnabled, + })), + [config], + ); + + const toggleQueue = ({ + queueId, + enabled, + }: { + queueId: string; + enabled: boolean; + }) => { + setConfig((current) => ({ + ...current, + queues: { + ...current.queues, + [queueId]: { enabled }, + }, + })); + }; + + const dirty = effectiveQueues.some( + (queue) => queue.enabled !== initialEnabledByQueue[queue.id], + ); + + const handleSave = async () => { + setSaving(true); + try { + await axiosInstance.put("/admin/job-queue-config", { + queues: Object.fromEntries( + effectiveQueues.map((queue) => [ + queue.id, + { enabled: queue.enabled }, + ]), + ), + }); + toast.success("Job queue config saved"); + onOpenChange(false); + } catch (error) { + toast.error(getBackendErr(error, "Failed to save job queue config")); + } finally { + setSaving(false); + } + }; + + return ( + + + + Job Queues + + Control which SQS queues workers actively poll. Disabling a queue + pauses consumption without changing producers. + + + + {loading ? ( +
Loading...
+ ) : ( +
+
+ {effectiveQueues.map((queue) => ( +
+
+
+ {queue.label} +
+
+ {queue.description} +
+
+ Default: {queue.defaultEnabled ? "enabled" : "disabled"} +
+
+ + toggleQueue({ queueId: queue.id, enabled }) + } + /> +
+ ))} +
+ +
+
+ + {config.configHealthy + ? "Config healthy" + : "Config unavailable"} + + {config.lastSuccessAt && ( + + Last refresh:{" "} + {new Date(config.lastSuccessAt).toLocaleString()} + + )} +
+
+ {config.configConfigured === false + ? "S3 job queue config is not configured." + : config.error || + "Queue polling changes propagate to workers within ~60 seconds."} +
+
+
+ )} + + + + + + + + ); +}