fix(api): defer rate-limited prefetch jobs instead of failing them
Some checks failed
QA Gate (P0/P1) / Test affected app (pull_request) Has been cancelled

The worker threw a custom RateLimitError for active-user cooldown and off-hours
windows, but BullMQ treated that as a normal failure: it retried with the
queue's exponential backoff (ignoring retryAfterMs) and dropped the job to
"failed" after attempts:3. So during busy hours sub-jobs were discarded en
masse and the requested defer time (e.g. "wait until 09:00") never applied.

Convert RateLimitError into job.moveToDelayed(now + retryAfterMs, token) +
DelayedError — BullMQ's contract for "not done, not failed, retry later". This
does not consume an attempt and honours the exact delay, so cooldown jobs wait
~60-120s and off-hours jobs wait until the window reopens, then resume. Genuine
errors still fail/retry normally.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-05-27 18:02:36 +03:00
parent dae76085ef
commit c7e59b90e0

View File

@@ -5,7 +5,7 @@ import {
type OnModuleDestroy,
type OnModuleInit,
} from "@nestjs/common";
import { type Job, type Queue, Worker } from "bullmq";
import { DelayedError, type Job, type Queue, Worker } from "bullmq";
import { and, asc, eq, gt, inArray, isNull, notExists, sql } from "drizzle-orm";
import { CategoriesService } from "../categories/categories.service";
import { DATABASE, type Database } from "../database/database.provider";
@@ -52,20 +52,20 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
) {}
onModuleInit() {
this.worker = new Worker(QUEUE_NAMES.CATALOG_PREFETCH, (job) => this.process(job), {
connection: getBullConnection(),
concurrency: 1,
limiter: { max: 5, duration: 60_000 },
});
this.worker = new Worker(
QUEUE_NAMES.CATALOG_PREFETCH,
(job, token) => this.process(job, token),
{
connection: getBullConnection(),
concurrency: 1,
limiter: { max: 5, duration: 60_000 },
},
);
this.worker.on("failed", (job, err) => {
if (err instanceof RateLimitError) {
this.logger.debug(
`[prefetch] Job ${job?.name} rate-limited, will retry in ${err.retryAfterMs}ms`,
);
} else {
this.logger.warn(`[prefetch] Job ${job?.name} failed: ${err.message}`);
}
// RateLimitError is handled in process() via moveToDelayed (a DelayedError,
// not a failure), so anything reaching here is a genuine failure.
this.logger.warn(`[prefetch] Job ${job?.name} failed: ${err.message}`);
});
this.worker.on("error", (err) => {
@@ -82,25 +82,38 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
}
}
private async process(job: Job): Promise<void> {
const data = job.data as { source?: string };
if (data.source === "parts-catalogs") {
await new Promise((r) => setTimeout(r, 15_000));
}
private async process(job: Job, token?: string): Promise<void> {
try {
const data = job.data as { source?: string };
if (data.source === "parts-catalogs") {
await new Promise((r) => setTimeout(r, 15_000));
}
if (job.name === "backfill-scan") {
return this.processBackfillScan();
if (job.name === "backfill-scan") {
return await this.processBackfillScan();
}
if (job.name === "prefetch-init") {
return await this.processInit(job as Job<PrefetchInitJobData>);
}
if (job.name === "prefetch-children") {
return await this.processChildren(job as Job<PrefetchCategoryJobData>);
}
if (job.name === "prefetch-parts") {
return await this.processParts(job as Job<PrefetchCategoryJobData>);
}
this.logger.warn(`[prefetch] Unknown job name: ${job.name}`);
} catch (err) {
// Rate-limited (active user cooldown or off-hours window): defer the job to
// its requested time WITHOUT consuming a retry attempt. moveToDelayed +
// DelayedError is BullMQ's contract for "not done, not failed — try later".
if (err instanceof RateLimitError) {
const delayMs = Math.max(1000, err.retryAfterMs);
await job.moveToDelayed(Date.now() + delayMs, token);
this.logger.debug(`[prefetch] Job ${job.name} deferred ${delayMs}ms (rate-limited)`);
throw new DelayedError();
}
throw err;
}
if (job.name === "prefetch-init") {
return this.processInit(job as Job<PrefetchInitJobData>);
}
if (job.name === "prefetch-children") {
return this.processChildren(job as Job<PrefetchCategoryJobData>);
}
if (job.name === "prefetch-parts") {
return this.processParts(job as Job<PrefetchCategoryJobData>);
}
this.logger.warn(`[prefetch] Unknown job name: ${job.name}`);
}
/**