feat(workflows): scheduler resumes elapsed wait nodes
Adds engine.runDueWaits() — polls waiting runs whose wake_at has passed and resumes the ones parked on a wait node (gate timeouts handled later by the approvals layer). Flag-gated (fails closed when workflows is off). Wired into the existing hourly invoiceScheduler tick in its own try/catch so a workflow failure never suppresses the invoice/reminder jobs. Test covers not-due vs elapsed resume.
This commit is contained in:
@@ -134,4 +134,28 @@ describe('workflow engine', () => {
|
|||||||
run = await db('workflow_runs').where({ id: run0.id }).first();
|
run = await db('workflow_runs').where({ id: run0.id }).first();
|
||||||
expect(run.status).toBe('done');
|
expect(run.status).toBe('done');
|
||||||
});
|
});
|
||||||
|
|
||||||
|
test('runDueWaits resumes only elapsed wait nodes', async () => {
|
||||||
|
await makeWorkflow({
|
||||||
|
trigger: 'wait.event',
|
||||||
|
nodes: [
|
||||||
|
{ key: 'w1', type: 'trigger' },
|
||||||
|
{ key: 'w2', type: 'wait', config: { delayMinutes: 60 } },
|
||||||
|
{ key: 'w3', type: 'action', config: { action: 'noop' } },
|
||||||
|
],
|
||||||
|
edges: [{ from: 'w1', to: 'w2' }, { from: 'w2', to: 'w3' }],
|
||||||
|
});
|
||||||
|
const runIds = await engine.emitWorkflowEvent('wait.event', { entityType: 'e', entityId: 7 });
|
||||||
|
const runId = runIds[0];
|
||||||
|
let run = await db('workflow_runs').where({ id: runId }).first();
|
||||||
|
expect(run.status).toBe('waiting');
|
||||||
|
|
||||||
|
expect(await engine.runDueWaits()).toBe(0); // wake_at ~60min out → not due
|
||||||
|
|
||||||
|
await db('workflow_runs').where({ id: runId }).update({ wake_at: new Date(Date.now() - 1000).toISOString() });
|
||||||
|
const resumed = await engine.runDueWaits();
|
||||||
|
expect(resumed).toBeGreaterThanOrEqual(1);
|
||||||
|
run = await db('workflow_runs').where({ id: runId }).first();
|
||||||
|
expect(run.status).toBe('done');
|
||||||
|
});
|
||||||
});
|
});
|
||||||
|
|||||||
@@ -42,6 +42,15 @@ async function runTick() {
|
|||||||
} catch (err) {
|
} catch (err) {
|
||||||
logger.error('Event reminder pass failed', { err: err.message });
|
logger.error('Event reminder pass failed', { err: err.message });
|
||||||
}
|
}
|
||||||
|
try {
|
||||||
|
// 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();
|
||||||
|
if (resumed) logger.info('Workflow scheduler: resumed waiting runs', { resumed });
|
||||||
|
} catch (err) {
|
||||||
|
logger.error('Workflow resume pass failed', { err: err.message });
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
function startInvoiceScheduler() {
|
function startInvoiceScheduler() {
|
||||||
|
|||||||
@@ -270,8 +270,49 @@ async function emitWorkflowEvent(triggerType, { entityType = null, entityId = nu
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Resume runs whose wait has elapsed. Called from the cron scheduler tick.
|
||||||
|
* Only advances `wait` nodes — gate timeouts are handled by the approvals
|
||||||
|
* layer. Fails CLOSED if the workflows flag is off (master kill-switch).
|
||||||
|
*/
|
||||||
|
async function runDueWaits(limit = 100) {
|
||||||
|
try {
|
||||||
|
const { isFeatureEnabled } = require('../../middleware/requireFeatureFlag');
|
||||||
|
let enabled = false;
|
||||||
|
try { enabled = await isFeatureEnabled('workflows'); } catch (e) { return 0; }
|
||||||
|
if (!enabled) return 0;
|
||||||
|
|
||||||
|
const nowIso = new Date().toISOString();
|
||||||
|
const due = await db('workflow_runs')
|
||||||
|
.where({ status: 'waiting' })
|
||||||
|
.whereNotNull('wake_at')
|
||||||
|
.where('wake_at', '<=', nowIso)
|
||||||
|
.limit(limit);
|
||||||
|
|
||||||
|
let resumed = 0;
|
||||||
|
for (const run of due) {
|
||||||
|
try {
|
||||||
|
const node = await db('workflow_nodes')
|
||||||
|
.where({ workflow_id: run.workflow_id, version: run.version, node_key: run.current_node })
|
||||||
|
.first();
|
||||||
|
if (node && node.type === 'wait') {
|
||||||
|
await resumeRun(run.id);
|
||||||
|
resumed += 1;
|
||||||
|
}
|
||||||
|
} catch (err) {
|
||||||
|
logger.error('[workflow] runDueWaits item failed', { runId: run.id, error: err.message });
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return resumed;
|
||||||
|
} catch (e) {
|
||||||
|
logger.error('[workflow] runDueWaits failed', { error: e.message });
|
||||||
|
return 0;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
module.exports = {
|
module.exports = {
|
||||||
emitWorkflowEvent,
|
emitWorkflowEvent,
|
||||||
|
runDueWaits,
|
||||||
startRun,
|
startRun,
|
||||||
advanceRun,
|
advanceRun,
|
||||||
resumeRun,
|
resumeRun,
|
||||||
|
|||||||
Reference in New Issue
Block a user