107 lines
3.3 KiB
TypeScript
107 lines
3.3 KiB
TypeScript
import { test, expect } from "bun:test";
|
|
import { createQueue } from "../src/index.ts";
|
|
|
|
test("processes a job", async () => {
|
|
const queue = createQueue();
|
|
const done: string[] = [];
|
|
queue.process<{ to: string }>("email", (job) => {
|
|
done.push(job.data.to);
|
|
});
|
|
await queue.add("email", { to: "a@b.com" });
|
|
expect(queue.size()).toBe(1);
|
|
await queue.drain();
|
|
expect(done).toEqual(["a@b.com"]);
|
|
expect(queue.size()).toBe(0);
|
|
});
|
|
|
|
test("delayed jobs only run once due", async () => {
|
|
let clock = 1000;
|
|
const queue = createQueue({ now: () => clock });
|
|
const ran: number[] = [];
|
|
queue.process("x", () => {
|
|
ran.push(clock);
|
|
});
|
|
await queue.add("x", 1, { delayMs: 500 });
|
|
await queue.drain(1200); // not yet due (runAt = 1500)
|
|
expect(ran.length).toBe(0);
|
|
clock = 1600;
|
|
await queue.drain();
|
|
expect(ran.length).toBe(1);
|
|
});
|
|
|
|
test("retries with backoff, then dead-letters", async () => {
|
|
let clock = 0;
|
|
const failed: unknown[] = [];
|
|
const queue = createQueue({
|
|
maxAttempts: 3,
|
|
backoffMs: 100,
|
|
now: () => clock,
|
|
onFailed: (job) => failed.push(job.id),
|
|
});
|
|
let calls = 0;
|
|
queue.process("flaky", () => {
|
|
calls++;
|
|
throw new Error("boom");
|
|
});
|
|
await queue.add("flaky", {});
|
|
await queue.drain(); // attempt 1 → reschedule at 100
|
|
expect(calls).toBe(1);
|
|
clock = 100;
|
|
await queue.drain(); // attempt 2 → reschedule at 0+200
|
|
clock = 300;
|
|
await queue.drain(); // attempt 3 → dead-letter
|
|
expect(calls).toBe(3);
|
|
expect(failed.length).toBe(1);
|
|
expect(queue.size()).toBe(0);
|
|
});
|
|
|
|
test("repeat re-enqueues a recurring job after each success", async () => {
|
|
let clock = 0;
|
|
const queue = createQueue({ now: () => clock });
|
|
let runs = 0;
|
|
queue.process("tick", () => {
|
|
runs++;
|
|
});
|
|
await queue.add("tick", {}, { repeat: 1000 });
|
|
await queue.drain(); // run 1 → re-enqueued at 1000
|
|
expect(runs).toBe(1);
|
|
await queue.drain(); // not due yet
|
|
expect(runs).toBe(1);
|
|
clock = 1000;
|
|
await queue.drain(); // run 2 → re-enqueued at 2000
|
|
expect(runs).toBe(2);
|
|
expect(queue.size()).toBe(1); // always one pending
|
|
});
|
|
|
|
test("jobs without a worker stay queued", async () => {
|
|
const queue = createQueue();
|
|
await queue.add("later", {});
|
|
await queue.drain();
|
|
expect(queue.size()).toBe(1); // no handler yet
|
|
queue.process("later", () => {});
|
|
await queue.drain();
|
|
expect(queue.size()).toBe(0);
|
|
});
|
|
|
|
test("rejects invalid queue and job timing options", async () => {
|
|
expect(() => createQueue({ pollMs: 0 })).toThrow("pollMs");
|
|
expect(() => createQueue({ maxAttempts: 0 })).toThrow("maxAttempts");
|
|
const queue = createQueue();
|
|
await expect(queue.add("", {})).rejects.toThrow("name");
|
|
await expect(queue.add("job", {}, { delayMs: -1 })).rejects.toThrow("delayMs");
|
|
await expect(queue.add("job", {}, { repeat: 0 })).rejects.toThrow("repeat");
|
|
});
|
|
|
|
test("drains independent due jobs concurrently", async () => {
|
|
const queue = createQueue();
|
|
const releases: Array<() => void> = [];
|
|
queue.process("work", () => new Promise<void>((resolve) => releases.push(resolve)));
|
|
await queue.add("work", 1);
|
|
await queue.add("work", 2);
|
|
const draining = queue.drain();
|
|
await Promise.resolve();
|
|
expect(releases).toHaveLength(2);
|
|
releases.forEach((release) => release());
|
|
expect(await draining).toBe(2);
|
|
});
|