Durable Workflows

Durable workflows allow executions to sleep, suspend, wait for external human approvals or webhooks, survive process restarts, and resume safely days or months later.

1. Waiting for External Events with `waitForEvent()`

Use waitForEvent() to halt workflow execution until an external signal or webhook is received:

import { Step, flow, waitForEvent } from '@siglo/core';

export const expenseApprovalWorkflow = flow(
    Step.create('submitExpense', async (param: { amount: number; expenseId: string }) => ({
        submittedAt: Date.now(),
        status: 'PENDING_APPROVAL',
    })),

    // Suspends until manager approval webhook arrives
    waitForEvent('managerApproval', {
        correlationKey: param => param.expenseId,
        timeout: '7d', // Auto-expire after 7 days
    }),

    Step.create('payout', async (param: { approved: boolean; approverId: string }) => {
        if (!param.approved) {
            return { payoutStatus: 'REJECTED' };
        }
        return { payoutStatus: 'PROCESSED', approvedBy: param.approverId };
    }),
);

When run with { durable: true }, the workflow halts at waitForEvent and returns a suspension descriptor:

const pending = await expenseApprovalWorkflow.run({ expenseId: 'exp_100', amount: 500 }, { durable: true });
console.log(pending.status); // 'waiting'
console.log(pending.correlationKey); // 'exp_100'

2. Sleeping with `wait()`

Pause workflow execution for a deterministic duration without blocking CPU threads:

import { Step, flow, wait } from '@siglo/core';

export const delayedNotificationWorkflow = flow(
    Step.create('sendWelcome', () => ({ sent: true })),
    wait('24h'), // Suspends execution for 24 hours
    Step.create('sendFollowUp', () => ({ followUpSent: true })),
);

3. Resuming Suspended Workflows

When an external event occurs (e.g., an incoming webhook), resume the workflow using resumeWorkflow() or the flow instance’s flow.resume() method:

import { resumeWorkflow } from '@siglo/core';
import { expenseApprovalWorkflow } from './workflows';

// Webhook endpoint handler:
app.post('/api/webhooks/approval', async (req, res) => {
    const { expenseId, approved, approverId } = req.body;

    const finalResult = await resumeWorkflow(
        expenseApprovalWorkflow,
        expenseId, // correlationKey matching the wait node
        { approved, approverId }, // Event payload merged into context
    );

    res.json({ success: true, finalResult });
});

4. Workflow Versioning and State Migrations

Long-running workflows that span days or weeks may have their code updated while executions are sleeping. Siglo allows declarative state migrations:

import { resumeWorkflow } from '@siglo/core';

const updatedWorkflow = flow(...newSteps).version('2.0.0');

await resumeWorkflow(updatedWorkflow, correlationKey, eventPayload, {
    targetVersion: '2.0.0',
    migrations: {
        '1.0.0->2.0.0': (fromVersion: string, toVersion: string, oldState: any) => ({
            ...oldState,
            schemaVersion: 2,
            userEmail: oldState.email, // Field rename migration
        }),
    },
});

Minimal Example

import { Step, flow, waitForEvent, resumeWorkflow } from '@siglo/core';

const wf = flow(
    Step.create('step1', () => ({ orderId: 'ord_123' })),
    waitForEvent('paymentReceived', { correlationKey: p => p.orderId }),
    Step.create('step2', (p: any) => ({ completed: true, amount: p.amount })),
);

// 1. Initial run suspends
const suspended = await wf.run({ orderId: 'ord_123' }, { durable: true });
console.log(suspended.status); // 'waiting'

// 2. Webhook arrives and resumes
const finalResult = await resumeWorkflow(wf, 'ord_123', { amount: 150 });
console.log(finalResult.completed); // true
console.log(finalResult.amount); // 150

Failure Example: Expired Waits and Duplicate Resumptions

import { Step, flow, waitForEvent, resumeWorkflow, DuplicateEventException } from '@siglo/core';

const wf = flow(waitForEvent('approval', { correlationKey: 'app-99', timeout: '1h' }));

await wf.run({}, { durable: true, stackId: 'stack-99' });

// First resume succeeds
await resumeWorkflow(wf, 'app-99', { approved: true });

try {
    // Second resume throws DuplicateEventException!
    await resumeWorkflow(wf, 'app-99', { approved: true });
} catch (err: any) {
    if (err instanceof DuplicateEventException) {
        console.error('Duplicate webhook rejected:', err.message);
    }
}

Testing Example

import { expect } from 'chai';
import { createWorkflowHarness } from '@siglo/testing';
import { Step, flow, waitForEvent } from '@siglo/core';

const durableWf = flow(
    Step.create('start', () => ({ ticketId: 't-1' })),
    waitForEvent('userReply', { correlationKey: p => p.ticketId }),
    Step.create('finish', (p: any) => ({ resolvedWith: p.message })),
);

describe('Durable Workflows Suite', () => {
    it('suspends and resumes deterministically with injected events', async () => {
        const harness = createWorkflowHarness(durableWf);

        const initial = await harness.run({ ticketId: 't-1' }, { durable: true });
        expect(initial.status).to.equal('waiting');

        const finalResult = await harness.injectEvent('userReply', 't-1', {
            message: 'Problem solved',
        });

        expect(finalResult.resolvedWith).to.equal('Problem solved');
    });
});

Operational Caveats

Use Persistent Storage in Production: In production, register @siglo/postgres or @siglo/redis via useStorage(). The in-memory default storage is strictly for testing and single-process development.
Correlation Key Uniqueness: Correlation keys must be unique per waiting workflow instance. Using natural business identifiers (e.g. orderId, ticketId) avoids collision bugs.
Execution Options: { durable: true } enables suspension and wait record persistence. When omitted, waitForEvent runs in non-durable in-memory fallback mode.

Overview

Workflow Patterns

Operations & Testing

Tutorials

Developer Tools

Durable Workflows

1. Waiting for External Events with `waitForEvent()`

2. Sleeping with `wait()`

3. Resuming Suspended Workflows

4. Workflow Versioning and State Migrations

Minimal Example

Failure Example: Expired Waits and Duplicate Resumptions

Testing Example

Operational Caveats

API Reference Links

​1. Waiting for External Events with waitForEvent()

​2. Sleeping with wait()

​3. Resuming Suspended Workflows

​4. Workflow Versioning and State Migrations

​Minimal Example

​Failure Example: Expired Waits and Duplicate Resumptions

​Testing Example

​Operational Caveats

​API Reference Links

1. Waiting for External Events with `waitForEvent()`

2. Sleeping with `wait()`

3. Resuming Suspended Workflows

4. Workflow Versioning and State Migrations

Minimal Example

Failure Example: Expired Waits and Duplicate Resumptions

Testing Example

Operational Caveats

API Reference Links