262 lines
8.0 KiB
TypeScript
262 lines
8.0 KiB
TypeScript
import { createExecutionContext, type ExecutionContext } from "@wrnexus/core";
|
|
import type { AddOptions, Job, JobHandler } from "./index.ts";
|
|
|
|
/**
|
|
* Persistence contract for the durable queue.
|
|
*
|
|
* Distributed drivers should implement `claim()` atomically and exclude leased
|
|
* jobs from `due()` until their lease expires. Calling `put()` must replace the
|
|
* stored record and release any previous lease for that job.
|
|
*/
|
|
export interface QueueStore {
|
|
put(job: Job): Promise<void>;
|
|
get(id: string): Promise<Job | null>;
|
|
remove(id: string): Promise<void>;
|
|
due(now: number, limit: number): Promise<Job[]>;
|
|
list(name?: string): Promise<Job[]>;
|
|
claim?(id: string, worker: string, leaseUntil: number): Promise<boolean>;
|
|
}
|
|
|
|
export function memoryQueueStore(): QueueStore {
|
|
const jobs = new Map<string, Job>();
|
|
|
|
return {
|
|
async put(job) {
|
|
jobs.set(job.id, structuredClone(job));
|
|
},
|
|
async get(id) {
|
|
const job = jobs.get(id);
|
|
return job ? structuredClone(job) : null;
|
|
},
|
|
async remove(id) {
|
|
jobs.delete(id);
|
|
},
|
|
async due(now, limit) {
|
|
return [...jobs.values()]
|
|
.filter((job) => job.runAt <= now)
|
|
.sort(
|
|
(left, right) =>
|
|
right.priority - left.priority ||
|
|
left.runAt - right.runAt ||
|
|
left.createdAt - right.createdAt,
|
|
)
|
|
.slice(0, limit)
|
|
.map((job) => structuredClone(job));
|
|
},
|
|
async list(name) {
|
|
return [...jobs.values()]
|
|
.filter((job) => !name || job.name === name)
|
|
.map((job) => structuredClone(job));
|
|
},
|
|
};
|
|
}
|
|
|
|
export interface DurableQueueOptions {
|
|
store?: QueueStore;
|
|
workerId?: string;
|
|
maxAttempts?: number;
|
|
concurrency?: number;
|
|
capacity?: number;
|
|
leaseMs?: number;
|
|
backoff?: (attempt: number) => number;
|
|
now?: () => number;
|
|
onDeadLetter?: (job: Job, error: unknown) => void | Promise<void>;
|
|
context?: (job: Job, signal: AbortSignal) => ExecutionContext;
|
|
}
|
|
|
|
export interface DurableQueue {
|
|
add<T>(name: string, data: T, options?: AddOptions): Promise<Job<T>>;
|
|
process<T>(name: string, handler: JobHandler<T>): void;
|
|
drain(): Promise<number>;
|
|
cancel(id: string): Promise<boolean>;
|
|
shutdown(options?: { force?: boolean }): Promise<void>;
|
|
list(name?: string): Promise<Job[]>;
|
|
failed(): Job[];
|
|
retry(id: string): Promise<boolean>;
|
|
}
|
|
|
|
function positiveInteger(value: number, label: string): number {
|
|
if (!Number.isInteger(value) || value < 1) {
|
|
throw new RangeError(`${label} must be a positive integer`);
|
|
}
|
|
return value;
|
|
}
|
|
|
|
function nonNegativeNumber(value: number, label: string): number {
|
|
if (!Number.isFinite(value) || value < 0) {
|
|
throw new RangeError(`${label} must be a non-negative number`);
|
|
}
|
|
return value;
|
|
}
|
|
|
|
export function createDurableQueue(options: DurableQueueOptions = {}): DurableQueue {
|
|
const store = options.store ?? memoryQueueStore();
|
|
const handlers = new Map<string, JobHandler>();
|
|
const deadLetters = new Map<string, Job>();
|
|
const now = options.now ?? Date.now;
|
|
const workerId = options.workerId?.trim() || `worker-${crypto.randomUUID()}`;
|
|
const defaultMaxAttempts = positiveInteger(options.maxAttempts ?? 3, "queue maxAttempts");
|
|
const concurrency = positiveInteger(options.concurrency ?? 10, "queue concurrency");
|
|
const capacity = positiveInteger(options.capacity ?? 10_000, "queue capacity");
|
|
const leaseMs = positiveInteger(options.leaseMs ?? 30_000, "queue leaseMs");
|
|
let sequence = 0;
|
|
let draining = false;
|
|
let accepting = true;
|
|
const active = new Map<string, { controller: AbortController; promise: Promise<boolean> }>();
|
|
|
|
function beginJob(job: Job): Promise<boolean> {
|
|
const controller = new AbortController();
|
|
const promise = runJob(job, controller).finally(() => active.delete(job.id));
|
|
active.set(job.id, { controller, promise });
|
|
return promise;
|
|
}
|
|
|
|
async function runJob(job: Job, controller: AbortController): Promise<boolean> {
|
|
const handler = handlers.get(job.name);
|
|
if (!handler) return false;
|
|
|
|
if (store.claim && !(await store.claim(job.id, workerId, now() + leaseMs))) {
|
|
return false;
|
|
}
|
|
|
|
job.attempts += 1;
|
|
|
|
try {
|
|
await handler(job, {
|
|
signal: controller.signal,
|
|
execution:
|
|
options.context?.(job, controller.signal) ??
|
|
createExecutionContext({
|
|
kind: job.repeat ? "cron" : "queue",
|
|
signal: controller.signal,
|
|
metadata: { jobId: job.id, jobName: job.name, attempt: job.attempts },
|
|
}),
|
|
});
|
|
|
|
if (job.repeat && job.repeat > 0) {
|
|
job.attempts = 0;
|
|
job.runAt = now() + job.repeat;
|
|
await store.put(job);
|
|
} else {
|
|
await store.remove(job.id);
|
|
}
|
|
} catch (error) {
|
|
if (controller.signal.aborted) {
|
|
await store.remove(job.id);
|
|
return true;
|
|
}
|
|
if (job.attempts < job.maxAttempts) {
|
|
const delay = nonNegativeNumber(
|
|
options.backoff?.(job.attempts) ?? Math.min(60_000, 1_000 * 2 ** (job.attempts - 1)),
|
|
"queue retry backoff",
|
|
);
|
|
job.runAt = now() + delay;
|
|
await store.put(job);
|
|
} else {
|
|
await store.remove(job.id);
|
|
deadLetters.set(job.id, structuredClone(job));
|
|
await options.onDeadLetter?.(structuredClone(job), error);
|
|
}
|
|
}
|
|
|
|
return true;
|
|
}
|
|
|
|
return {
|
|
async add<T>(name: string, data: T, add: AddOptions = {}) {
|
|
if (!accepting) throw new Error("WRN-QUEUE-CLOSED: queue is shutting down");
|
|
if (!name.trim()) throw new TypeError("queue job name cannot be empty");
|
|
|
|
const maxAttempts = positiveInteger(add.maxAttempts ?? defaultMaxAttempts, "job maxAttempts");
|
|
const delayMs = nonNegativeNumber(add.delayMs ?? 0, "job delayMs");
|
|
const priority = add.priority ?? 0;
|
|
if (!Number.isFinite(priority)) {
|
|
throw new RangeError("job priority must be a finite number");
|
|
}
|
|
if (add.repeat !== undefined) {
|
|
positiveInteger(add.repeat, "job repeat");
|
|
}
|
|
|
|
if (add.idempotencyKey) {
|
|
const existing = (await store.list(name)).find(
|
|
(job) => job.idempotencyKey === add.idempotencyKey,
|
|
);
|
|
if (existing) return existing as Job<T>;
|
|
}
|
|
if ((await store.list()).length >= capacity) {
|
|
throw new Error(`WRN-QUEUE-CAPACITY: queue capacity of ${capacity} reached`);
|
|
}
|
|
|
|
const createdAt = now();
|
|
const job: Job<T> = {
|
|
id: `job-${createdAt.toString(36)}-${(++sequence).toString(36)}`,
|
|
name,
|
|
data,
|
|
attempts: 0,
|
|
maxAttempts,
|
|
runAt: createdAt + delayMs,
|
|
repeat: add.repeat,
|
|
priority,
|
|
idempotencyKey: add.idempotencyKey,
|
|
createdAt,
|
|
};
|
|
await store.put(job);
|
|
return structuredClone(job);
|
|
},
|
|
|
|
process(name, handler) {
|
|
if (!name.trim()) throw new TypeError("queue worker name cannot be empty");
|
|
handlers.set(name, handler as JobHandler);
|
|
},
|
|
|
|
async drain() {
|
|
if (draining) return 0;
|
|
draining = true;
|
|
try {
|
|
const due = await store.due(now(), concurrency);
|
|
const results = await Promise.all(due.map(beginJob));
|
|
return results.filter(Boolean).length;
|
|
} finally {
|
|
draining = false;
|
|
}
|
|
},
|
|
|
|
failed() {
|
|
return [...deadLetters.values()].map((job) => structuredClone(job));
|
|
},
|
|
|
|
list(name) {
|
|
return store.list(name);
|
|
},
|
|
|
|
async cancel(id) {
|
|
const running = active.get(id);
|
|
if (running) {
|
|
running.controller.abort();
|
|
return true;
|
|
}
|
|
if (!(await store.get(id))) return false;
|
|
await store.remove(id);
|
|
return true;
|
|
},
|
|
|
|
async shutdown(shutdownOptions = {}) {
|
|
accepting = false;
|
|
if (shutdownOptions.force) {
|
|
for (const { controller } of active.values()) controller.abort();
|
|
}
|
|
await Promise.allSettled([...active.values()].map(({ promise }) => promise));
|
|
},
|
|
|
|
async retry(id) {
|
|
const job = deadLetters.get(id);
|
|
if (!job) return false;
|
|
deadLetters.delete(id);
|
|
job.attempts = 0;
|
|
job.runAt = now();
|
|
await store.put(job);
|
|
return true;
|
|
},
|
|
};
|
|
}
|