/** * ExecutionQueue.ts — FIFO task queue with retry, cancel, and timeout * * Manages all code execution tasks — prevents race conditions and * ensures stable ordering for Judge0 and worker-based runs. */ export type TaskStatus = "pending" | "running" | "success" | "error" | "timeout" | "cancelled"; export interface ExecutionTask { id: string; status: TaskStatus; result?: T; error?: string; startedAt?: number; finishedAt?: number; attempts: number; } export interface QueueOptions { maxRetries?: number; retryDelay?: number; defaultTimeout?: number; concurrency?: number; } type TaskRunner = (signal: AbortSignal) => Promise; interface InternalTask { id: string; runner: TaskRunner; resolve: (val: T) => void; reject: (err: unknown) => void; timeout: number; retries: number; retryDelay: number; attempts: number; abortCtrl?: AbortController; } export class ExecutionQueue { private queue: InternalTask[] = []; private running = 0; private readonly concurrency: number; private readonly defaultTimeout: number; private readonly defaultRetries: number; private readonly defaultRetryDelay: number; private readonly tasks = new Map(); constructor(opts: QueueOptions = {}) { this.concurrency = opts.concurrency ?? 2; this.defaultTimeout = opts.defaultTimeout ?? 30_000; this.defaultRetries = opts.maxRetries ?? 2; this.defaultRetryDelay = opts.retryDelay ?? 1000; } enqueue( id: string, runner: TaskRunner, opts: { timeout?: number; retries?: number; retryDelay?: number } = {}, ): Promise { return new Promise((resolve, reject) => { const task: InternalTask = { id, runner, resolve: resolve as (v: unknown) => void, reject, timeout: opts.timeout ?? this.defaultTimeout, retries: opts.retries ?? this.defaultRetries, retryDelay: opts.retryDelay ?? this.defaultRetryDelay, attempts: 0, }; this.tasks.set(id, { id, status: "pending", attempts: 0 }); this.queue.push(task as InternalTask); this.drain(); }); } cancel(id: string): boolean { const inQueue = this.queue.findIndex(t => t.id === id); if (inQueue !== -1) { const [task] = this.queue.splice(inQueue, 1); const record = this.tasks.get(id); if (record) record.status = "cancelled"; task.reject(new Error("Esecuzione annullata")); return true; } const task = this.queue.find(t => t.id === id); if (task?.abortCtrl) { task.abortCtrl.abort(); return true; } return false; } getStatus(id: string): ExecutionTask | undefined { return this.tasks.get(id); } private drain() { while (this.running < this.concurrency && this.queue.length > 0) { const task = this.queue.shift()!; this.runTask(task); } } private async runTask(task: InternalTask) { this.running++; const record = this.tasks.get(task.id)!; record.status = "running"; record.startedAt = Date.now(); while (task.attempts <= task.retries) { task.attempts++; record.attempts = task.attempts; const ctrl = new AbortController(); task.abortCtrl = ctrl; const timer = setTimeout(() => ctrl.abort(), task.timeout); try { const result = await task.runner(ctrl.signal); clearTimeout(timer); record.status = "success"; record.result = result; record.finishedAt = Date.now(); task.resolve(result); break; } catch (err) { clearTimeout(timer); if (ctrl.signal.aborted && ctrl.signal.reason === undefined) { record.status = "timeout"; record.error = "Timeout esecuzione"; record.finishedAt = Date.now(); task.reject(new Error("Timeout esecuzione")); break; } const isAborted = err instanceof Error && err.name === "AbortError"; if (isAborted) { record.status = "cancelled"; record.finishedAt = Date.now(); task.reject(err); break; } if (task.attempts > task.retries) { record.status = "error"; record.error = err instanceof Error ? err.message : String(err); record.finishedAt = Date.now(); task.reject(err); break; } await new Promise(r => setTimeout(r, task.retryDelay * task.attempts)); } } this.running--; this.drain(); } } export const executionQueue = new ExecutionQueue({ concurrency: 2, maxRetries: 2 });