From 192d2cbc06a0295dbfdc35f05d192ae7ed5273e6 Mon Sep 17 00:00:00 2001 From: Luca <102960244+Luca-Timo@users.noreply.github.com> Date: Tue, 23 Jun 2026 13:39:17 +0200 Subject: [PATCH] =?UTF-8?q?feat(workflows):=20crash=20recovery=20=E2=80=94?= =?UTF-8?q?=20resume=20runs=20orphaned=20mid-flow?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes the crash-safety gap: a run left in running/pending by a crash had nothing to resume it (the scheduler only wakes 'waiting'). Adds a heartbeat (workflow_runs.updated_at, stamped on every node advance + start/resume) and a recoverStaleRuns() sweep that re-enters runs whose heartbeat has gone stale (>10 min) from their persisted node. Runs on the scheduler tick AND the boot tick, so a restart catches anything stranded during downtime. Re-entry is at-least-once (the current node may re-execute) — loop counters + the late-fee math are idempotent, so the only residual risk is a duplicate reminder email. An attempts counter (migration 145, cap 5) marks a run failed instead of recovering a node that reliably crashes the process (crash-loop backstop). Flag-gated. Tests: orphan-resume + crash-loop cap. --- .../integration/workflowEngine.test.js | 36 +++++++++++ .../core/145_workflow_run_recovery.js | 33 ++++++++++ .../src/services/invoiceSchedulerService.js | 7 ++- backend/src/services/workflows/engine.js | 61 ++++++++++++++++++- 4 files changed, 133 insertions(+), 4 deletions(-) create mode 100644 backend/migrations/core/145_workflow_run_recovery.js diff --git a/backend/__tests__/integration/workflowEngine.test.js b/backend/__tests__/integration/workflowEngine.test.js index dd796db1..34bbab6b 100644 --- a/backend/__tests__/integration/workflowEngine.test.js +++ b/backend/__tests__/integration/workflowEngine.test.js @@ -279,4 +279,40 @@ describe('workflow engine', () => { const after = await db('workflows').where({ id: wf.id }).first(); expect(after.version).toBe(before.version); // unchanged }); + + test('recoverStaleRuns resumes a run orphaned mid-flow (crash recovery)', async () => { + const wfId = await makeWorkflow({ + trigger: 'recover.event', + nodes: [{ key: 'r1', type: 'trigger' }, { key: 'r2', type: 'action', config: { action: 'noop' } }], + edges: [{ from: 'r1', to: 'r2' }], + }); + // Simulate a run left 'running' at r2 with a stale heartbeat (crash mid-flow). + await db('workflow_runs').insert({ + workflow_id: wfId, version: 1, trigger_event: 'recover.event', status: 'running', current_node: 'r2', + context: JSON.stringify({ vars: {} }), dedup_key: 'recover-1', + updated_at: new Date(Date.now() - 3600000).toISOString(), + }); + const run0 = await db('workflow_runs').where({ dedup_key: 'recover-1' }).first(); + const n = await engine.recoverStaleRuns({ staleMs: 1000 }); + expect(n).toBeGreaterThanOrEqual(1); + const run = await db('workflow_runs').where({ id: run0.id }).first(); + expect(run.status).toBe('done'); + }); + + test('recoverStaleRuns abandons a crash-looping run after the attempts cap', async () => { + const wfId = await makeWorkflow({ + trigger: 'crashloop.event', + nodes: [{ key: 'c1', type: 'trigger' }, { key: 'c2', type: 'action', config: { action: 'noop' } }], + edges: [{ from: 'c1', to: 'c2' }], + }); + await db('workflow_runs').insert({ + workflow_id: wfId, version: 1, trigger_event: 'crashloop.event', status: 'running', current_node: 'c2', + context: JSON.stringify({ vars: {} }), dedup_key: 'crash-1', attempts: 5, + updated_at: new Date(Date.now() - 3600000).toISOString(), + }); + const run0 = await db('workflow_runs').where({ dedup_key: 'crash-1' }).first(); + await engine.recoverStaleRuns({ staleMs: 1000 }); + const run = await db('workflow_runs').where({ id: run0.id }).first(); + expect(run.status).toBe('failed'); + }); }); diff --git a/backend/migrations/core/145_workflow_run_recovery.js b/backend/migrations/core/145_workflow_run_recovery.js new file mode 100644 index 00000000..d44adec0 --- /dev/null +++ b/backend/migrations/core/145_workflow_run_recovery.js @@ -0,0 +1,33 @@ +/** + * Migration 145: crash-recovery fields for workflow runs. + * + * A run left in 'running'/'pending' by a crash has nothing to resume it (the + * scheduler only wakes 'waiting' runs). Add a heartbeat (`updated_at`, stamped + * on every step) so a recovery sweep can detect stale runs, plus an `attempts` + * counter so a node that reliably crashes the process can't be recovered + * forever (crash-loop backstop → marked failed after a cap). + */ +exports.up = async function (knex) { + if (!(await knex.schema.hasTable('workflow_runs'))) return; + const hasUpdated = await knex.schema.hasColumn('workflow_runs', 'updated_at'); + const hasAttempts = await knex.schema.hasColumn('workflow_runs', 'attempts'); + await knex.schema.alterTable('workflow_runs', (t) => { + if (!hasUpdated) t.timestamp('updated_at').defaultTo(knex.fn.now()); + if (!hasAttempts) t.integer('attempts').notNullable().defaultTo(0); + }); + // Recovery sweep queries by (status, updated_at). + if (!hasUpdated) { + try { await knex.schema.alterTable('workflow_runs', (t) => t.index(['status', 'updated_at'], 'workflow_runs_recovery_index')); } catch (_) {} + } +}; + +exports.down = async function (knex) { + if (!(await knex.schema.hasTable('workflow_runs'))) return; + try { await knex.schema.alterTable('workflow_runs', (t) => t.dropIndex(['status', 'updated_at'], 'workflow_runs_recovery_index')); } catch (_) {} + if (await knex.schema.hasColumn('workflow_runs', 'updated_at')) { + await knex.schema.alterTable('workflow_runs', (t) => t.dropColumn('updated_at')); + } + if (await knex.schema.hasColumn('workflow_runs', 'attempts')) { + await knex.schema.alterTable('workflow_runs', (t) => t.dropColumn('attempts')); + } +}; diff --git a/backend/src/services/invoiceSchedulerService.js b/backend/src/services/invoiceSchedulerService.js index 614349b1..49fe983d 100644 --- a/backend/src/services/invoiceSchedulerService.js +++ b/backend/src/services/invoiceSchedulerService.js @@ -46,8 +46,13 @@ async function runTick() { // Resume workflow runs whose wait has elapsed. No-op (fails closed) when // the `workflows` feature flag is off. Independent try/catch so a workflow // failure never suppresses the invoice/reminder jobs above. - const resumed = await require('./workflows').runDueWaits(); + const wf = require('./workflows'); + const resumed = await wf.runDueWaits(); if (resumed) logger.info('Workflow scheduler: resumed waiting runs', { resumed }); + // Recover runs orphaned by a crash (stuck in running/pending). Runs on the + // boot tick too, so a restart catches anything stranded during downtime. + const recovered = await wf.recoverStaleRuns(); + if (recovered) logger.warn('Workflow scheduler: recovered orphaned runs', { recovered }); } catch (err) { logger.error('Workflow resume pass failed', { err: err.message }); } diff --git a/backend/src/services/workflows/engine.js b/backend/src/services/workflows/engine.js index b4bb401f..584bb29f 100644 --- a/backend/src/services/workflows/engine.js +++ b/backend/src/services/workflows/engine.js @@ -186,7 +186,7 @@ async function advanceRun(runId) { } currentKey = nextKey; - await db('workflow_runs').where({ id: runId }).update({ current_node: currentKey || null, context: JSON.stringify(context) }); + await db('workflow_runs').where({ id: runId }).update({ current_node: currentKey || null, context: JSON.stringify(context), updated_at: db.fn.now() }); } await finishRun(runId); @@ -200,7 +200,7 @@ async function startRun(runId) { let entry = null; for (const n of nodeByKey.values()) { if (n.type === 'trigger') { entry = n; break; } } if (!entry) { await failRun(runId, 'no trigger node'); return; } - await db('workflow_runs').where({ id: runId }).update({ status: 'running', current_node: entry.node_key }); + await db('workflow_runs').where({ id: runId }).update({ status: 'running', current_node: entry.node_key, updated_at: db.fn.now() }); await advanceRun(runId); } @@ -214,7 +214,7 @@ async function resumeRun(runId, { decisionHandle = null } = {}) { const { edges } = await loadGraph(run.workflow_id, run.version); const e = outEdge(edges, run.current_node, decisionHandle); const nextKey = e ? e.to_node : null; - await db('workflow_runs').where({ id: runId }).update({ status: 'running', current_node: nextKey, wake_at: null }); + await db('workflow_runs').where({ id: runId }).update({ status: 'running', current_node: nextKey, wake_at: null, updated_at: db.fn.now() }); if (!nextKey) { await finishRun(runId); return; } await advanceRun(runId); } @@ -310,9 +310,64 @@ async function runDueWaits(limit = 100) { } } +const RECOVERY_STALE_MS = 10 * 60 * 1000; // a 'running' run idle this long = orphaned by a crash +const MAX_RECOVERY_ATTEMPTS = 5; + +/** + * Resume runs orphaned by a crash. A run left in 'running'/'pending' has nothing + * to resume it (the scheduler only wakes 'waiting'), so this sweep picks up ones + * whose heartbeat (updated_at) has gone stale and re-enters them from their + * persisted node. Re-entry is at-least-once: the current node may re-execute — + * loop counters + the late-fee math are idempotent, so the only residual risk is + * a duplicate reminder email. `attempts` caps recovery so a node that reliably + * crashes the process is marked failed instead of looping forever. Flag-gated + * (fails closed when workflows is off). Called from the scheduler tick + boot. + */ +async function recoverStaleRuns({ staleMs = RECOVERY_STALE_MS, limit = 50 } = {}) { + try { + const { isFeatureEnabled } = require('../../middleware/requireFeatureFlag'); + let enabled = false; + try { enabled = await isFeatureEnabled('workflows'); } catch (e) { return 0; } + if (!enabled) return 0; + if (!(await db.schema.hasColumn('workflow_runs', 'updated_at'))) return 0; + + const cutoff = new Date(Date.now() - staleMs).toISOString(); + const stale = await db('workflow_runs') + .whereIn('status', ['running', 'pending']) + .where('updated_at', '<=', cutoff) + .limit(limit); + + let recovered = 0; + for (const run of stale) { + try { + const attempts = Number(run.attempts) || 0; + if (attempts >= MAX_RECOVERY_ATTEMPTS) { + await failRun(run.id, `abandoned after ${attempts} recovery attempts (suspected crash loop)`); + continue; + } + await db('workflow_runs').where({ id: run.id }).update({ attempts: attempts + 1, updated_at: db.fn.now() }); + if (!run.current_node) { + await startRun(run.id); + } else { + await db('workflow_runs').where({ id: run.id }).update({ status: 'running', updated_at: db.fn.now() }); + await advanceRun(run.id); + } + recovered += 1; + } catch (err) { + logger.error('[workflow] recovery failed', { runId: run.id, error: err.message }); + } + } + return recovered; + } catch (e) { + logger.error('[workflow] recoverStaleRuns failed', { error: e.message }); + return 0; + } +} + module.exports = { emitWorkflowEvent, runDueWaits, + recoverStaleRuns, startRun, advanceRun, resumeRun,