watchdog.ts · Source

watchdog.tsapps/worker/src/processors/automations/watchdog.tsraw ← automations7,690 B← Back to the app
1/**
2 * apps/worker — automation run watchdog (TASK-0113 §9).
3 *
4 * A repeatable maintenance job (every 5 min) that recovers runs stuck in a
5 * non-terminal state after a worker crash (a SIGKILL mid-run leaves the row
6 * RUNNING; an enqueued-but-never-processed job leaves it PENDING).
7 *
8 * Reclaim rules:
9 *  - RUNNING rows with `startedAt` older than 15 min are STALE (the BullMQ
10 *    lockDuration of 60 s / stalledInterval of 30 s already reprocess
11 *    in-process stalls; this is the cross-restart backstop):
12 *      - retryCount < cap → hop to PENDING (retryCount + 1) and re-enqueue
13 *        the same runId (the run resumes from action 1 — idempotency notes
14 *        in the pipeline apply);
15 *      - retryCount >= cap → FAILED `WORKER_CRASHED` (terminal) + the same
16 *        notification path as dead-letter (admin notification + audit +
17 *        counter) via `deadLetterRun` (which records FAILED→DEAD_LETTER-safe
18 *        bookkeeping).
19 *  - PENDING rows with `createdAt` older than 15 min are STALE (enqueued but
20 *    never picked up — e.g. the enqueue succeeded but the worker was down):
21 *    re-enqueued the same way (no retryCount bump — they never started).
22 *
23 * Idempotent: a run already reclaimed by another watchdog pass (its status
24 * moved on) is skipped by the transition assert. The query uses the
25 * `automation_runs(status)` index — fine at MVP scale (a partial index is a
26 * Phase-15 optimization, per the spec).
27 */
28import type { PrismaClient } from "@prisma/client";
29import type { Redis } from "ioredis";
30import { asJson } from "../../lib/prisma.js";
31import { getLogger } from "../../logger.js";
32import type { WorkerEnv } from "../../config.js";
33import { runLog } from "./log.js";
34import { enqueueRunJob } from "./matcher.js";
35import { autoRunMaxRetries } from "./pipeline.js";
36 
37const log = getLogger("automation-watchdog");
38 
39/** Stale threshold (spec §9: 15 min). */
40export const STALE_MS = 15 * 60_000;
41 
42export interface WatchdogResult {
43  reclaimed: number;
44  deadLettered: number;
45  checked: number;
46}
47 
48/** Run one watchdog pass. Returns the counts for observability. Never throws. */
49export async function runWatchdog(
50  prisma: PrismaClient,
51  redis: Redis,
52  env: WorkerEnv
53): Promise<WatchdogResult> {
54  const now = Date.now();
55  const staleBefore = new Date(now - STALE_MS);
56  const maxRetries = autoRunMaxRetries(env);
57  const result: WatchdogResult = { reclaimed: 0, deadLettered: 0, checked: 0 };
58 
59  // RUNNING + startedAt < now-15m.
60  const runningStale = await prisma.automationRun.findMany({
61    where: { status: "RUNNING", startedAt: { lt: staleBefore } },
62    select: { id: true, automationId: true, workspaceId: true, triggerEvent: true, retryCount: true, createdAt: true, status: true }
63  });
64  // PENDING + createdAt < now-15m (enqueued but never started).
65  const pendingStale = await prisma.automationRun.findMany({
66    where: { status: "PENDING", createdAt: { lt: staleBefore } },
67    select: { id: true, automationId: true, workspaceId: true, triggerEvent: true, retryCount: true, createdAt: true, status: true }
68  });
69 
70  for (const run of [...runningStale, ...pendingStale]) {
71    result.checked++;
72    try {
73      await reclaimRun(prisma, redis, env, run, maxRetries, run.status === "RUNNING");
74      if (run.status === "RUNNING" && run.retryCount >= maxRetries) result.deadLettered++;
75      else result.reclaimed++;
76    } catch (err) {
77      log.warn({ err: (err as Error).message, runId: run.id, status: run.status }, "watchdog reclaim failed");
78    }
79  }
80 
81  if (result.checked > 0) {
82    log.info({ ...result }, "watchdog pass complete");
83  }
84  return result;
85}
86 
87interface StaleRun {
88  id: string;
89  automationId: string;
90  workspaceId: string;
91  triggerEvent: unknown;
92  retryCount: number;
93  createdAt: Date;
94  status: string;
95}
96 
97async function reclaimRun(
98  prisma: PrismaClient,
99  redis: Redis,
100  env: WorkerEnv,
101  run: StaleRun,
102  maxRetries: number,
103  wasRunning: boolean
104): Promise<void> {
105  const te = (run.triggerEvent as Record<string, unknown> | null) ?? {};
106  const triggerType = typeof te.type === "string" ? te.type : "unknown";
107  const depth = (te.automationContext as { depth?: number } | undefined)?.depth ?? 0;
108 
109  const automation = await prisma.automation.findUnique({
110    where: { id: run.automationId },
111    select: { id: true, name: true, workspaceId: true }
112  });
113  const automationName = automation?.name ?? "unknown automation";
114 
115  const reenqueue = async (): Promise<void> => {
116    await enqueueRunJob(redis, {
117      runId: run.id,
118      automationId: run.automationId,
119      workspaceId: run.workspaceId,
120      triggerEvent: te,
121      eventId: (te.eventId as string) ?? run.id
122    });
123  };
124 
125  if (wasRunning) {
126    if (run.retryCount < maxRetries) {
127      await prisma.automationRun.update({
128        where: { id: run.id },
129        data: { status: "PENDING", retryCount: run.retryCount + 1 }
130      });
131      await runLog(prisma, run.id, "warn", "watchdog reclaimed stale RUNNING run", {
132        retryCount: run.retryCount + 1,
133        startedAgoMs: STALE_MS
134      });
135      await reenqueue();
136      return;
137    }
138    // At the cap → FAILED WORKER_CRASHED (spec §9: "marks FAILED
139    // WORKER_CRASHED at the cap (log + same notification path as
140    // dead-letter)" — FAILED, not DEAD_LETTER; the retry endpoint can
141    // still revive it).
142    try {
143      await prisma.automationRun.update({
144        where: { id: run.id },
145        data: {
146          status: "FAILED",
147          finishedAt: new Date(),
148          error: asJson({ code: "WORKER_CRASHED", message: "run stuck RUNNING past the watchdog threshold at the retry cap", attempts: run.retryCount })
149        }
150      });
151      await runLog(prisma, run.id, "error", "run FAILED WORKER_CRASHED (watchdog, retry cap reached)", {
152        attempts: run.retryCount,
153        triggerType
154      });
155      const { emitAutomationExecuted } = await import("./dead-letter.js");
156      await emitAutomationExecuted({
157        redis,
158        workspaceId: run.workspaceId,
159        automationId: run.automationId,
160        runId: run.id,
161        status: "FAILED",
162        triggerType,
163        durationMs: STALE_MS,
164        error: { code: "WORKER_CRASHED" },
165        depth
166      });
167      const { notifyWorkspaceAdmins } = await import("../../lib/notify.js");
168      await notifyWorkspaceAdmins(
169        prisma,
170        redis,
171        env.SMTP_HOST,
172        run.workspaceId,
173        `Automation "${automationName}" failed (worker crash)`,
174        "The run was stuck RUNNING past the watchdog threshold and exceeded the retry cap (WORKER_CRASHED).",
175        { type: "automation", id: run.automationId },
176        "AUTOMATION"
177      );
178      await prisma.auditLog.create({
179        data: {
180          actorType: "SYSTEM",
181          actorId: `automation:${run.automationId}`,
182          workspaceId: run.workspaceId,
183          resourceType: "automation_run",
184          resourceId: run.id,
185          action: "automation.worker_crashed",
186          metadata: asJson({ automationId: run.automationId, runId: run.id, code: "WORKER_CRASHED", attempts: run.retryCount })
187        }
188      });
189      await redis.incr(`automations:dead_letter:${run.workspaceId}`).catch(() => undefined);
190      log.error({ runId: run.id, automationId: run.automationId }, "stale RUNNING run → FAILED WORKER_CRASHED (cap)");
191    } catch (err) {
192      log.error({ err: (err as Error).message, runId: run.id }, "watchdog crash-cap bookkeeping failed");
193    }
194    return;
195  }
196 
197  // PENDING (never started) → re-enqueue, no retryCount bump.
198  await reenqueue();
199  await runLog(prisma, run.id, "warn", "watchdog re-enqueued stale PENDING run", {
200    pendingForMs: STALE_MS
201  });
202}
203