feat: add pending task drilldown

src/components/backend-core/src/index.ts

@@ -21,6 +21,7 @@ interface RuntimeConfig {
   databaseUrl: string;
   providerToken: string;
   heartbeatTimeoutMs: number;
+  taskPendingTimeoutMs: number;
   logFile: string;
 }
 
@@ -62,6 +63,16 @@ function readNumberEnv(name: string): number {
   return parsed;
 }
 
+function readOptionalNumberEnv(name: string, fallback: number): number {
+  const raw = process.env[name];
+  if (raw === undefined || raw.length === 0) return fallback;
+  const parsed = Number(raw);
+  if (!Number.isFinite(parsed) || parsed <= 0) {
+    throw new Error(`Environment variable ${name} must be a positive number, got ${raw}`);
+  }
+  return parsed;
+}
+
 function readConfig(): RuntimeConfig {
   return {
     port: readNumberEnv("PORT"),
@@ -69,6 +80,7 @@ function readConfig(): RuntimeConfig {
     databaseUrl: requiredEnv("DATABASE_URL"),
     providerToken: requiredEnv("PROVIDER_TOKEN"),
     heartbeatTimeoutMs: readNumberEnv("HEARTBEAT_TIMEOUT_MS"),
+    taskPendingTimeoutMs: readOptionalNumberEnv("TASK_PENDING_TIMEOUT_MS", 10 * 60 * 1000),
     logFile: requiredEnv("LOG_FILE"),
   };
 }
@@ -353,6 +365,47 @@ async function markStaleProvidersOffline(): Promise<void> {
   }
 }
 
+async function markStaleTasksFailed(): Promise<void> {
+  if (!dbReady) return;
+  const timeoutMs = config.taskPendingTimeoutMs;
+  const rows = await sql<Array<{ id: string; provider_id: string; command: string; previous_status: string; updated_at: Date | string }>>`
+    WITH stale AS (
+      SELECT id, provider_id, command, status AS previous_status, updated_at
+      FROM unidesk_tasks
+      WHERE status IN ('queued', 'dispatched', 'running')
+        AND updated_at < now() - (${timeoutMs}::bigint * interval '1 millisecond')
+      FOR UPDATE
+    ),
+    updated AS (
+      UPDATE unidesk_tasks task
+      SET
+        status = 'failed',
+        result = jsonb_build_object(
+          'error', 'task timed out without terminal provider status',
+          'timeoutMs', ${timeoutMs}::bigint,
+          'previousStatus', stale.previous_status,
+          'previousResult', task.result,
+          'timedOutAt', now()
+        ),
+        updated_at = now()
+      FROM stale
+      WHERE task.id = stale.id
+      RETURNING task.id, task.provider_id, task.command, stale.previous_status, stale.updated_at
+    )
+    SELECT * FROM updated
+  `;
+  for (const row of rows) {
+    await recordEvent("task_timeout", row.provider_id, {
+      taskId: row.id,
+      providerId: row.provider_id,
+      command: row.command,
+      previousStatus: row.previous_status,
+      previousUpdatedAt: row.updated_at instanceof Date ? row.updated_at.toISOString() : String(row.updated_at),
+      timeoutMs,
+    });
+  }
+}
+
 function parseMessage(raw: string | Buffer): ProviderToCoreMessage {
   const text = typeof raw === "string" ? raw : raw.toString("utf8");
   const parsed = JSON.parse(text) as unknown;
@@ -532,13 +585,22 @@ async function getEvents(limit: number): Promise<ApiEvent[]> {
   }));
 }
 
-async function getTasks(limit: number): Promise<ApiTask[]> {
-  const rows = await sql<Array<Record<string, unknown>>>`
-    SELECT id, provider_id, command, status, payload, result, created_at, updated_at
-    FROM unidesk_tasks
-    ORDER BY updated_at DESC
-    LIMIT ${limit}
-  `;
+async function getTasks(limit: number, statusFilter = "all"): Promise<ApiTask[]> {
+  await markStaleTasksFailed();
+  const rows = statusFilter === "pending"
+    ? await sql<Array<Record<string, unknown>>>`
+      SELECT id, provider_id, command, status, payload, result, created_at, updated_at
+      FROM unidesk_tasks
+      WHERE status IN ('queued', 'dispatched', 'running')
+      ORDER BY updated_at DESC
+      LIMIT ${limit}
+    `
+    : await sql<Array<Record<string, unknown>>>`
+      SELECT id, provider_id, command, status, payload, result, created_at, updated_at
+      FROM unidesk_tasks
+      ORDER BY updated_at DESC
+      LIMIT ${limit}
+    `;
   return rows.map((row) => ({
     id: String(row.id),
     providerId: String(row.provider_id),
@@ -551,13 +613,22 @@ async function getTasks(limit: number): Promise<ApiTask[]> {
   }));
 }
 
+async function countPendingTasks(): Promise<number> {
+  await markStaleTasksFailed();
+  const rows = await sql<Array<{ count: string | number }>>`
+    SELECT count(*)::int AS count
+    FROM unidesk_tasks
+    WHERE status IN ('queued', 'dispatched', 'running')
+  `;
+  return Number(rows[0]?.count ?? 0);
+}
+
 async function getOverview(): Promise<JsonValue> {
   const nodes = await getNodes();
-  const tasks = await getTasks(50);
+  const pendingTasks = await countPendingTasks();
   const dockerStatuses = await getNodeDockerStatuses();
   const systemStatuses = await getNodeSystemStatuses(1);
   const online = nodes.filter((node) => node.status === "online").length;
-  const pendingTasks = tasks.filter((task) => task.status === "queued" || task.status === "dispatched" || task.status === "running").length;
   return {
     service: "unidesk-core",
     ok: true,
@@ -568,6 +639,7 @@ async function getOverview(): Promise<JsonValue> {
     dockerStatusNodeCount: dockerStatuses.filter((item) => item.dockerStatus !== null).length,
     systemStatusNodeCount: systemStatuses.filter((item) => item.current !== null).length,
     pendingTaskCount: pendingTasks,
+    taskPendingTimeoutMs: config.taskPendingTimeoutMs,
     activeSocketCount: activeProviders.size,
     heartbeatTimeoutMs: config.heartbeatTimeoutMs,
   };
@@ -618,7 +690,7 @@ async function route(req: Request): Promise<Response> {
     if (url.pathname === "/api/nodes/system-status") return jsonResponse({ ok: true, systemStatuses: await getNodeSystemStatuses(readLimit(url, 120)) });
     if (url.pathname === "/api/nodes/docker-status") return jsonResponse({ ok: true, dockerStatuses: await getNodeDockerStatuses() });
     if (url.pathname === "/api/events") return jsonResponse({ ok: true, events: await getEvents(readLimit(url, 100)) });
-    if (url.pathname === "/api/tasks") return jsonResponse({ ok: true, tasks: await getTasks(readLimit(url, 100)) });
+    if (url.pathname === "/api/tasks") return jsonResponse({ ok: true, tasks: await getTasks(readLimit(url, 100), url.searchParams.get("status") ?? "all") });
     if (url.pathname === "/api/dispatch" && req.method === "POST") return dispatchTask(req);
     if (url.pathname === "/logs") return jsonResponse({ ok: true, logs: recentLogs.slice(-readLimit(url, 100)) });
     if (url.pathname === "/favicon.ico") return textResponse("", 204);
@@ -655,6 +727,7 @@ function readLimit(url: URL, defaultLimit: number): number {
 }
 
 await initDatabaseWithRetry();
+markStaleTasksFailed().catch((error) => logger("error", "task_timeout_sweep_failed", { error: errorToJson(error) }));
 
 const apiServer = Bun.serve<WsData>({
   port: config.port,
@@ -690,6 +763,10 @@ setInterval(() => {
   markStaleProvidersOffline().catch((error) => logger("error", "heartbeat_sweep_failed", { error: errorToJson(error) }));
 }, 10_000);
 
+setInterval(() => {
+  markStaleTasksFailed().catch((error) => logger("error", "task_timeout_sweep_failed", { error: errorToJson(error) }));
+}, Math.min(config.taskPendingTimeoutMs, 60_000));
+
 logger("info", "server_listening", {
   apiUrl: `http://0.0.0.0:${apiServer.port}`,
   providerIngressUrl: `ws://0.0.0.0:${providerServer.port}/ws/provider`,