92 lines
2.4 KiB
TypeScript
92 lines
2.4 KiB
TypeScript
export type Job = () => Promise<void>;
|
|
|
|
type QueueEntry = { name: string; job: Job };
|
|
|
|
type QueueSettings = { maxAttempts: number; baseDelayMs: number };
|
|
|
|
const settings: QueueSettings = { maxAttempts: 3, baseDelayMs: 100 };
|
|
|
|
const pending: QueueEntry[] = [];
|
|
const idleWaiters: Array<() => void> = [];
|
|
let active = false;
|
|
|
|
const sleep = (ms: number): Promise<void> =>
|
|
new Promise((resolve) => {
|
|
setTimeout(resolve, ms);
|
|
});
|
|
|
|
// Delay for attempt n is drawn from [base * 2^(n-1), base * 2^n), so successive
|
|
// waits always grow while jitter keeps retries from synchronizing across jobs.
|
|
const backoffDelay = (attempt: number): number => {
|
|
const window = settings.baseDelayMs * 2 ** (attempt - 1);
|
|
return window + Math.random() * window;
|
|
};
|
|
|
|
const runEntry = async (entry: QueueEntry): Promise<void> => {
|
|
for (let attempt = 1; ; attempt += 1) {
|
|
try {
|
|
await entry.job();
|
|
return;
|
|
} catch (err) {
|
|
if (attempt >= settings.maxAttempts) {
|
|
console.error(
|
|
`[queue] job "${entry.name}" abandoned after ${attempt} attempt(s)`,
|
|
err
|
|
);
|
|
return;
|
|
}
|
|
await sleep(backoffDelay(attempt));
|
|
}
|
|
}
|
|
};
|
|
|
|
const runLoop = async (): Promise<void> => {
|
|
try {
|
|
for (;;) {
|
|
const entry = pending.shift();
|
|
if (!entry) return;
|
|
await runEntry(entry);
|
|
}
|
|
} finally {
|
|
active = false;
|
|
for (const resolve of idleWaiters.splice(0)) resolve();
|
|
}
|
|
};
|
|
|
|
export const enqueue = (name: string, job: Job): void => {
|
|
pending.push({ name, job });
|
|
if (active) return;
|
|
|
|
active = true;
|
|
// Deferred to a microtask so enqueue() returns to its caller — typically a
|
|
// request handler that has already responded — before any job body runs.
|
|
void Promise.resolve()
|
|
.then(runLoop)
|
|
.catch((err: unknown) => {
|
|
active = false;
|
|
console.error('[queue] queue loop stopped unexpectedly', err);
|
|
});
|
|
};
|
|
|
|
export const size = (): number => pending.length;
|
|
|
|
export const drain = (): Promise<void> => {
|
|
if (!active && pending.length === 0) return Promise.resolve();
|
|
|
|
return new Promise<void>((resolve) => {
|
|
idleWaiters.push(resolve);
|
|
});
|
|
};
|
|
|
|
export const configureQueue = (opts: {
|
|
maxAttempts?: number;
|
|
baseDelayMs?: number;
|
|
}): void => {
|
|
if (opts.maxAttempts !== undefined) {
|
|
settings.maxAttempts = Math.max(1, Math.floor(opts.maxAttempts));
|
|
}
|
|
if (opts.baseDelayMs !== undefined) {
|
|
settings.baseDelayMs = Math.max(0, opts.baseDelayMs);
|
|
}
|
|
};
|