Software/platform
1
1
Fork 0
Code Issues 3 Pull Requests Actions Packages Projects Releases Wiki Activity

Implement custom retry handling of rate limit errors

Browse Source
This commit is contained in:
Dominic Ferrando
2026-07-13 16:32:20 -04:00
parent b408bb5439
commit fd94f22d21
7 changed files with 80 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/api/src/services/event-queue.ts
+18 -6
View File
@@ -5,6 +5,7 @@ import { sleep } from "bun";
import { db } from "../database/db";
import { AssertError, Value } from "@sinclair/typebox/value";
import type { TSchema, Static } from "@sinclair/typebox";
import { RateLimitError } from "@blade-and-brawn/domain";
export type EventSource = "portal" | "printful" | "webflow";
export type EventStatus = "waiting" | "available" | "processing" | "failed" | "fulfilled";
@@ -38,6 +39,7 @@ export function defineEventType<P extends TSchema, S extends TSchema>(opt: Event
export class EventQueue<G extends string, T extends Record<string, EventTypeOptions<any, any>>> {
private static readonly DEQUEUE_RETRY_DELAY_MS = 10000; // 10 seconds
private static readonly DEQUEUE_RETRY_LIMIT = 3;
private static readonly RATE_LIMIT_RETRY_LIMIT = 15;
private _lastCleanDate: Date = new Date(0);
readonly cfg: EventQueueCfg;
@@ -179,7 +181,7 @@ export class EventQueue<G extends string, T extends Record<string, EventTypeOpti
async front(trx?: Transaction<DB>) {
const result = await (trx ?? db).selectFrom("events")
.select(["id", "type", "payload", "available_at", "tries"])
.select(["id", "type", "payload", "available_at", "tries", "rate_limits"])
.where("group", "=", this.group)
.where("started_at", "is", null)
.where("available_at", "<=", new Date())
@@ -220,7 +222,7 @@ export class EventQueue<G extends string, T extends Record<string, EventTypeOpti
.execute();
}
private async processEvent(event: Pick<Selectable<DB["events"]>, "id" | "type" | "tries" | "payload">) {
private async processEvent(event: Pick<Selectable<DB["events"]>, "id" | "type" | "tries" | "payload" | "rate_limits">) {
log.info({ id: event.id, type: event.type, payload: event.payload }, "processing event");
if (!(event.type in this.events)) throw new Error(`event type not registered in this queue (${this.group}): ${event.type}`);
const cfg = this.events[event.type as keyof T]!;
@@ -231,15 +233,24 @@ export class EventQueue<G extends string, T extends Record<string, EventTypeOpti
await this.fulfill(event.id);
}
catch (err) {
if (err instanceof RateLimitError && event.rate_limits < EventQueue.RATE_LIMIT_RETRY_LIMIT) {
log.warn(
{ tries: event.tries, delay: err.retryDelayMs, source: err.source, payload: err.payload },
"encountered a rate limit error"
);
await this.retry(event.id, err.retryDelayMs, { isRateLimit: true });
return;
}
if (err instanceof AssertError) {
log.error({ tries: event.tries }, "could not process event: malformed payload");
await this.fail(event.id);
return;
}
if (event.tries < this.cfg.retry.limit) {
if ((event.tries - event.rate_limits) < this.cfg.retry.limit) {
log.warn({ err, tries: event.tries, delay: this.cfg.retry.delayMs }, "encountered an error during event processing, retrying")
await this.retry(event.id);
await this.retry(event.id, this.cfg.retry.delayMs);
return;
}
@@ -255,14 +266,15 @@ export class EventQueue<G extends string, T extends Record<string, EventTypeOpti
.execute();
}
private async retry(id: string) {
private async retry(id: string, delayMs: number, opt: { isRateLimit?: boolean } = {}) {
await db.updateTable("events")
.set((eb) => ({
tries: eb("tries", "+", 1),
rate_limits: opt.isRateLimit ? eb("rate_limits", "+", 1) : eb.ref("rate_limits"),
started_at: null,
ended_at: null,
has_failed: false,
available_at: new Date(Date.now() + this.cfg.retry.delayMs)
available_at: new Date(Date.now() + delayMs)
}))
.where("id", "=", id)
.execute();