From 82e6315b61d7b9ddde205ef29f3e1cd9f55544c2 Mon Sep 17 00:00:00 2001 From: Jefferson Youashi <119521983+clintjeff2@users.noreply.github.com> Date: Tue, 30 Jun 2026 16:29:18 +0100 Subject: [PATCH] Persist task dead-letter queue --- __tests__/api/tasks.test.ts | 77 ++++++++++++- app/api/tasks/dead-letter/[id]/retry/route.ts | 16 +++ app/api/tasks/dead-letter/[id]/route.ts | 16 +++ app/api/tasks/dead-letter/route.ts | 21 ++++ lib/agent-runtime/task-queue.ts | 102 ++++++++++++++++++ 5 files changed, 230 insertions(+), 2 deletions(-) create mode 100644 app/api/tasks/dead-letter/[id]/retry/route.ts create mode 100644 app/api/tasks/dead-letter/[id]/route.ts create mode 100644 app/api/tasks/dead-letter/route.ts diff --git a/__tests__/api/tasks.test.ts b/__tests__/api/tasks.test.ts index 0d0fc872..7309b54d 100644 --- a/__tests__/api/tasks.test.ts +++ b/__tests__/api/tasks.test.ts @@ -1,15 +1,25 @@ import { beforeEach, describe, expect, it } from "vitest" +import { mkdtempSync, readFileSync } from "node:fs" +import { tmpdir } from "node:os" +import { join } from "node:path" +import { DELETE as discardDeadLetterTask } from "@/app/api/tasks/dead-letter/[id]/route" +import { POST as retryDeadLetterTask } from "@/app/api/tasks/dead-letter/[id]/retry/route" +import { GET as listDeadLetterTasks } from "@/app/api/tasks/dead-letter/route" import { GET as getTask } from "@/app/api/tasks/[id]/route" import { POST as cancelTask } from "@/app/api/tasks/[id]/cancel/route" import { POST as failTask } from "@/app/api/tasks/[id]/fail/route" import { POST as retryTask } from "@/app/api/tasks/[id]/retry/route" import { GET as listTasks, POST as enqueueTask } from "@/app/api/tasks/route" -import { resetTaskQueueForTests } from "@/lib/agent-runtime/task-queue" +import { loadDeadLetterQueueForTests, resetTaskQueueForTests } from "@/lib/agent-runtime/task-queue" const context = (id: string) => ({ params: Promise.resolve({ id }) }) describe("/api/tasks", () => { - beforeEach(() => resetTaskQueueForTests()) + beforeEach(() => { + const tempDir = mkdtempSync(join(tmpdir(), "open-stellar-task-dlq-")) + process.env.TASK_DLQ_FILE = join(tempDir, "task-dlq.json") + resetTaskQueueForTests() + }) it("enqueues and retrieves a prioritized task", async () => { const createRes = await enqueueTask(new Request("http://localhost/api/tasks", { @@ -83,4 +93,67 @@ describe("/api/tasks", () => { expect(retry.task.status).toBe("pending") expect(retry.task.retryCount).toBe(0) }) + + it("persists dead-letter entries and reloads them from file", async () => { + const createRes = await enqueueTask(new Request("http://localhost/api/tasks", { + method: "POST", + body: JSON.stringify({ id: "persisted-task", type: "agent.persist", maxRetries: 0, payload: { sector: "beta" } }), + headers: { "Content-Type": "application/json" }, + })) + expect(createRes.status).toBe(201) + + const failRes = await failTask(new Request("http://localhost/api/tasks/persisted-task/fail", { + method: "POST", + body: JSON.stringify({ error: "permanent" }), + }), context("persisted-task")) + expect(failRes.status).toBe(200) + + const persisted = JSON.parse(readFileSync(process.env.TASK_DLQ_FILE!, "utf8")) + expect(persisted).toHaveLength(1) + expect(persisted[0]).toMatchObject({ + task: { id: "persisted-task", status: "dead-letter" }, + errorMessage: "permanent", + retryCount: 0, + }) + + resetTaskQueueForTests() + loadDeadLetterQueueForTests() + const listRes = await listDeadLetterTasks(new Request("http://localhost/api/tasks/dead-letter")) + const list = await listRes.json() + expect(list.tasks[0].task.id).toBe("persisted-task") + }) + + it("lists, retries, and discards dead-letter tasks via dedicated endpoints", async () => { + for (const id of ["dlq-1", "dlq-2"]) { + await enqueueTask(new Request("http://localhost/api/tasks", { + method: "POST", + body: JSON.stringify({ id, type: "agent.dead", maxRetries: 0 }), + headers: { "Content-Type": "application/json" }, + })) + await failTask(new Request(`http://localhost/api/tasks/${id}/fail`, { + method: "POST", + body: JSON.stringify({ error: `${id} failed` }), + }), context(id)) + } + + const listRes = await listDeadLetterTasks(new Request("http://localhost/api/tasks/dead-letter?limit=1&offset=1")) + const list = await listRes.json() + expect(listRes.status).toBe(200) + expect(list.tasks).toHaveLength(1) + expect(list.pagination.total).toBe(2) + expect(list.tasks[0]).toMatchObject({ errorMessage: "dlq-2 failed", retryCount: 0 }) + + const retryRes = await retryDeadLetterTask(new Request("http://localhost/api/tasks/dead-letter/dlq-1/retry", { method: "POST" }), context("dlq-1")) + const retried = await retryRes.json() + expect(retryRes.status).toBe(200) + expect(retried.task.status).toBe("pending") + + const discardRes = await discardDeadLetterTask(new Request("http://localhost/api/tasks/dead-letter/dlq-2", { method: "DELETE" }), context("dlq-2")) + const discarded = await discardRes.json() + expect(discardRes.status).toBe(200) + expect(discarded.task.status).toBe("dead-letter") + + const notFoundRes = await discardDeadLetterTask(new Request("http://localhost/api/tasks/dead-letter/missing", { method: "DELETE" }), context("missing")) + expect(notFoundRes.status).toBe(404) + }) }) diff --git a/app/api/tasks/dead-letter/[id]/retry/route.ts b/app/api/tasks/dead-letter/[id]/retry/route.ts new file mode 100644 index 00000000..64303151 --- /dev/null +++ b/app/api/tasks/dead-letter/[id]/retry/route.ts @@ -0,0 +1,16 @@ +import { NextResponse } from "next/server" +import { retryDeadLetterTask } from "@/lib/agent-runtime/task-queue" + +type TaskRouteContext = { + params: Promise<{ id: string }> +} + +export async function POST(_req: Request, context: TaskRouteContext) { + const { id } = await context.params + try { + return NextResponse.json({ ok: true, task: retryDeadLetterTask(id) }) + } catch (error) { + const message = error instanceof Error ? error.message : "Failed to retry dead-letter task" + return NextResponse.json({ ok: false, error: message }, { status: message === "Task not found" ? 404 : 400 }) + } +} diff --git a/app/api/tasks/dead-letter/[id]/route.ts b/app/api/tasks/dead-letter/[id]/route.ts new file mode 100644 index 00000000..ef1f9559 --- /dev/null +++ b/app/api/tasks/dead-letter/[id]/route.ts @@ -0,0 +1,16 @@ +import { NextResponse } from "next/server" +import { discardDeadLetterTask } from "@/lib/agent-runtime/task-queue" + +type TaskRouteContext = { + params: Promise<{ id: string }> +} + +export async function DELETE(_req: Request, context: TaskRouteContext) { + const { id } = await context.params + try { + return NextResponse.json({ ok: true, task: discardDeadLetterTask(id) }) + } catch (error) { + const message = error instanceof Error ? error.message : "Failed to discard dead-letter task" + return NextResponse.json({ ok: false, error: message }, { status: message === "Task not found" ? 404 : 400 }) + } +} diff --git a/app/api/tasks/dead-letter/route.ts b/app/api/tasks/dead-letter/route.ts new file mode 100644 index 00000000..fda506f3 --- /dev/null +++ b/app/api/tasks/dead-letter/route.ts @@ -0,0 +1,21 @@ +import { NextResponse } from "next/server" +import { listDeadLetterEntries } from "@/lib/agent-runtime/task-queue" + +export const dynamic = "force-dynamic" + +export async function GET(req: Request) { + const url = new URL(req.url) + const offset = Number(url.searchParams.get("offset") ?? 0) + const limit = Number(url.searchParams.get("limit") ?? 50) + const { entries, total, offset: normalizedOffset, limit: normalizedLimit } = listDeadLetterEntries({ offset, limit }) + + return NextResponse.json({ + ok: true, + tasks: entries, + pagination: { + total, + offset: normalizedOffset, + limit: normalizedLimit, + }, + }) +} diff --git a/lib/agent-runtime/task-queue.ts b/lib/agent-runtime/task-queue.ts index 7b718145..3116ec70 100644 --- a/lib/agent-runtime/task-queue.ts +++ b/lib/agent-runtime/task-queue.ts @@ -1,3 +1,5 @@ +import { existsSync, mkdirSync, readFileSync, renameSync, writeFileSync } from "node:fs" +import { dirname } from "node:path" import { DISTRICTS } from "@/lib/data" import type { DistrictId } from "@/lib/types" @@ -23,6 +25,13 @@ export interface QueuedTask { deadLetteredAt?: string } +export interface DeadLetterTaskEntry { + task: QueuedTask + failedAt: string + errorMessage: string + retryCount: number +} + export interface EnqueueTaskInput { id?: string type: string @@ -61,6 +70,9 @@ const PRIORITY_WEIGHT: Record = { } const RETRY_BACKOFF_SECONDS = [5, 15, 45, 135] as const +function getDeadLetterQueuePath(): string { + return process.env.TASK_DLQ_FILE ?? "/.data/task-dlq.json" +} function nextTaskId(): string { queueState.sequence += 1 @@ -72,6 +84,10 @@ export function resetTaskQueueForTests(): void { queueState.sequence = 0 } +export function loadDeadLetterQueueForTests(): void { + loadDeadLetterQueue() +} + export function isQueuedTaskPriority(value: unknown): value is QueuedTaskPriority { return value === "critical" || value === "high" || value === "normal" || value === "low" } @@ -91,6 +107,69 @@ export function isValidTaskDistrict(value: unknown): value is DistrictId { return typeof value === "string" && DISTRICTS.some((district) => district.id === value) } +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value) +} + +function toDeadLetterEntry(task: QueuedTask): DeadLetterTaskEntry { + return { + task, + failedAt: task.deadLetteredAt ?? task.updatedAt, + errorMessage: task.error ?? "Task failed", + retryCount: task.retryCount, + } +} + +function persistDeadLetterQueue(): void { + const dlqFilePath = getDeadLetterQueuePath() + const entries = [...queueState.tasks.values()] + .filter((task) => task.status === "dead-letter") + .map(toDeadLetterEntry) + mkdirSync(dirname(dlqFilePath), { recursive: true }) + const tempPath = `${dlqFilePath}.${process.pid}.${Date.now()}.tmp` + writeFileSync(tempPath, `${JSON.stringify(entries, null, 2)}\n`, "utf8") + renameSync(tempPath, dlqFilePath) +} + +function hydrateTask(value: unknown): QueuedTask | null { + if (!isRecord(value)) return null + if (typeof value.id !== "string" || typeof value.type !== "string") return null + if (!isQueuedTaskPriority(value.priority) || value.status !== "dead-letter") return null + const now = new Date().toISOString() + return { + id: value.id, + type: value.type, + payload: isRecord(value.payload) ? value.payload : {}, + priority: value.priority, + targetAgentId: typeof value.targetAgentId === "string" ? value.targetAgentId : undefined, + targetDistrict: isValidTaskDistrict(value.targetDistrict) ? value.targetDistrict : undefined, + targetCapability: typeof value.targetCapability === "string" ? value.targetCapability : undefined, + retryCount: typeof value.retryCount === "number" ? value.retryCount : 0, + maxRetries: typeof value.maxRetries === "number" ? value.maxRetries : 0, + createdAt: typeof value.createdAt === "string" ? value.createdAt : now, + scheduledFor: typeof value.scheduledFor === "string" ? value.scheduledFor : undefined, + status: "dead-letter", + result: value.result, + error: typeof value.error === "string" ? value.error : undefined, + updatedAt: typeof value.updatedAt === "string" ? value.updatedAt : now, + deadLetteredAt: typeof value.deadLetteredAt === "string" ? value.deadLetteredAt : undefined, + } +} + +function loadDeadLetterQueue(): void { + const dlqFilePath = getDeadLetterQueuePath() + if (!existsSync(dlqFilePath)) return + const parsed: unknown = JSON.parse(readFileSync(dlqFilePath, "utf8")) + if (!Array.isArray(parsed)) return + for (const entry of parsed) { + const taskValue = isRecord(entry) && "task" in entry ? entry.task : entry + const task = hydrateTask(taskValue) + if (task) queueState.tasks.set(task.id, task) + } +} + +loadDeadLetterQueue() + function assertNonEmpty(value: string, field: string): string { const trimmed = value.trim() if (!trimmed) throw new Error(`${field} is required`) @@ -165,6 +244,18 @@ export function listDeadLetterTasks(): QueuedTask[] { return listTasks({ status: "dead-letter", includeDeadLetter: true }) } +export function listDeadLetterEntries(options: { offset?: number; limit?: number } = {}): { + entries: DeadLetterTaskEntry[] + total: number + offset: number + limit: number +} { + const offset = Math.max(0, Math.floor(options.offset ?? 0)) + const limit = Math.min(100, Math.max(1, Math.floor(options.limit ?? 50))) + const entries = listDeadLetterTasks().map(toDeadLetterEntry) + return { entries: entries.slice(offset, offset + limit), total: entries.length, offset, limit } +} + export function cancelTask(id: string): QueuedTask { const task = queueState.tasks.get(id) if (!task) throw new Error("Task not found") @@ -181,6 +272,7 @@ export function failTask(id: string, error: string): QueuedTask { if (task.retryCount >= task.maxRetries) { const dead = { ...task, status: "dead-letter" as const, error, updatedAt: now.toISOString(), deadLetteredAt: now.toISOString() } queueState.tasks.set(id, dead) + persistDeadLetterQueue() return dead } const backoffSeconds = RETRY_BACKOFF_SECONDS[Math.min(task.retryCount, RETRY_BACKOFF_SECONDS.length - 1)] @@ -209,5 +301,15 @@ export function retryDeadLetterTask(id: string): QueuedTask { updatedAt: new Date().toISOString(), } queueState.tasks.set(id, retry) + persistDeadLetterQueue() return retry } + +export function discardDeadLetterTask(id: string): QueuedTask { + const task = queueState.tasks.get(id) + if (!task) throw new Error("Task not found") + if (task.status !== "dead-letter") throw new Error("Only dead-letter tasks can be discarded") + queueState.tasks.delete(id) + persistDeadLetterQueue() + return task +}