From 6834ba35bda85c4c895ca4c699b26daa8b4c2d79 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Fri, 31 Jul 2026 18:19:16 -0700 Subject: [PATCH] fix(engine): hung-pass watchdogs on scheduler, merge queue, and continuation drain Co-Authored-By: Claude Fable 5 --- .changeset/pump-watchdogs.md | 7 +++++++ packages/engine/src/project-engine.ts | 9 ++++++++- packages/engine/src/runtimes/in-process-runtime.ts | 12 ++++++++++-- packages/engine/src/scheduler.ts | 9 ++++++++- 4 files changed, 33 insertions(+), 4 deletions(-) create mode 100644 .changeset/pump-watchdogs.md diff --git a/.changeset/pump-watchdogs.md b/.changeset/pump-watchdogs.md new file mode 100644 index 0000000000..9e40e290ef --- /dev/null +++ b/.changeset/pump-watchdogs.md @@ -0,0 +1,7 @@ +--- +"@runfusion/fusion": patch +--- + +summary: Scheduler, merge queue, and continuation drain recover loudly from a hung pass instead of dying silently. +category: fix +dev: Same shape as the triage-poll death — a stuck re-entrance guard dropped every later tick without a log. Each pump now records its pass start and force-opens the guard past a duration sized to its legitimate work (schedule 10min, merge drain 30min, continuation drain 5min), logging a WARN with the stuck duration. diff --git a/packages/engine/src/project-engine.ts b/packages/engine/src/project-engine.ts index c821a0bd07..17924952c7 100644 --- a/packages/engine/src/project-engine.ts +++ b/packages/engine/src/project-engine.ts @@ -458,6 +458,7 @@ export class ProjectEngine { private unregisterMergeAdmissionProvider?: () => void; private pausedReviewTaskIds = new Set(); private mergeRunning = false; + private mergeRunningSince = 0; private activeMergeSession: { dispose: () => void } | null = null; private activeMergeTaskId: string | null = null; /** Wall-clock when `activeMergeTaskId` was claimed; self-healing uses this when agent logs are silent. */ @@ -3297,8 +3298,14 @@ export class ProjectEngine { } private async drainMergeQueue(): Promise { - if (this.mergeRunning) return; + if (this.mergeRunning) { + /* FNXC:PumpWatchdog 2026-08-01-02:00: one hung pass leaves the guard closed forever and every later tick/wake drops SILENTLY (the triage-poll death, 00769fad7c/e51ebff381). Past the threshold, warn with the stuck duration and force the guard open; the hung pass's own finally re-clearing it later is harmless. A legitimate merge runs many minutes (rebase, verification, land), so the threshold is 30min — past it the merge is wedged (stale-merge sweeps own the TASK, this owns the PUMP flag). */ + const stuckMs = this.mergeRunningSince > 0 ? Date.now() - this.mergeRunningSince : 0; + if (stuckMs < 1_800_000) return; + runtimeLog.warn(`merge-queue watchdog: previous drain still marked in-flight after ${Math.round(stuckMs / 1000)}s — forcing the guard open so merging resumes`); + } this.mergeRunning = true; + this.mergeRunningSince = Date.now(); try { this.reconcileStaleMergeActive(); diff --git a/packages/engine/src/runtimes/in-process-runtime.ts b/packages/engine/src/runtimes/in-process-runtime.ts index 059d057a51..69d9995fcb 100644 --- a/packages/engine/src/runtimes/in-process-runtime.ts +++ b/packages/engine/src/runtimes/in-process-runtime.ts @@ -620,6 +620,7 @@ export class InProcessRuntime private triageProcessor?: TriageProcessor; private workflowContinuationTimer?: ReturnType; private workflowContinuationDrainActive = false; + private workflowContinuationDrainSince = 0; private messageStore?: MessageStore; /** FNXC:TaskDeleteNotice 2026-07-26-16:10: identity-guarded teardown for the delete-notice mailbox seam. */ private unregisterTaskDeleteNoticeMailbox?: () => void; @@ -2307,7 +2308,15 @@ export class InProcessRuntime * adapters that bind the pass to this runtime's store and executor. */ private async drainWorkflowContinuations(): Promise { - if (this.workflowContinuationDrainActive || this.status !== "active") return; + if (this.status !== "active") return; + if (this.workflowContinuationDrainActive) { + /* FNXC:PumpWatchdog 2026-08-01-02:00: one hung pass leaves the guard closed forever and every later tick/wake drops SILENTLY (the triage-poll death, 00769fad7c/e51ebff381). Past the threshold, warn with the stuck duration and force the guard open; the hung pass's own finally re-clearing it later is harmless. */ + const stuckMs = this.workflowContinuationDrainSince > 0 ? Date.now() - this.workflowContinuationDrainSince : 0; + if (stuckMs < 300_000) return; + runtimeLog.warn(`continuation-drain watchdog: previous drain still marked in-flight after ${Math.round(stuckMs / 1000)}s — forcing the guard open`); + } + this.workflowContinuationDrainActive = true; + this.workflowContinuationDrainSince = Date.now(); /* FNXC:EnginePause 2026-08-01-00:20: A pause-suspended run persists a runnable continuation (same mechanism as capacity). Without @@ -2322,7 +2331,6 @@ export class InProcessRuntime } catch { /* unreadable settings: proceed as before rather than wedging the pump */ } - this.workflowContinuationDrainActive = true; try { await drainDuePlanningContinuations({ listDue: () => this.taskStore.listDueWorkflowWorkItems({ diff --git a/packages/engine/src/scheduler.ts b/packages/engine/src/scheduler.ts index 3d2c26f442..1abad98cad 100644 --- a/packages/engine/src/scheduler.ts +++ b/packages/engine/src/scheduler.ts @@ -893,6 +893,7 @@ export interface SchedulerOptions { export class Scheduler { private running = false; private scheduling = false; + private schedulingSince = 0; private wasWorktreeLimited = false; private wasGlobalPaused = false; private wasEnginePaused = false; @@ -1969,8 +1970,14 @@ export class Scheduler { */ async schedule(): Promise { if (!this.running) return; - if (this.scheduling) return; + if (this.scheduling) { + /* FNXC:PumpWatchdog 2026-08-01-02:00: one hung pass leaves the guard closed forever and every later tick/wake drops SILENTLY (the triage-poll death, 00769fad7c/e51ebff381). Past the threshold, warn with the stuck duration and force the guard open; the hung pass's own finally re-clearing it later is harmless. Dispatch prep runs real git work per candidate, so the threshold is generous. */ + const stuckMs = this.schedulingSince > 0 ? Date.now() - this.schedulingSince : 0; + if (stuckMs < 600_000) return; + schedulerLog.warn(`schedule watchdog: previous pass still marked in-flight after ${Math.round(stuckMs / 1000)}s — forcing the guard open so dispatch resumes`); + } this.scheduling = true; + this.schedulingSince = Date.now(); try { let tasks = await this.store.listTasks({ slim: true, includeArchived: false, startupMemo: false });