root/sase.tr
1
0
Fork 0
Code Issues 5 Pull Requests 3 Actions 7 Packages Projects Releases Wiki Activity

Merge pull request 'fix(api): defer rate-limited prefetch jobs instead of failing them' (#63) from dev into main

Browse Source 
Reviewed-on: #63
This commit was merged in pull request #63.
This commit is contained in:
root
2026-05-27 15:05:59 +00:00
parent 4196638a8c c7e59b90e0
commit 6e440f55de
1 changed files with 43 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

73
apps/api/src/jobs/prefetch-worker.service.ts
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}`);
}
/**