Automating sales outreach, onboarding drips, and recurring data pipelines requires a system that can trigger workflows on a schedule, execute them as directed acyclic graphs (DAGs), pause when a node requires a delay or human input, and resume later from exactly where it left off. A cron-driven workflow scheduler combines cron expression parsing with a DAG execution engine that persists run state to a database, enabling workflows that span hours or days without holding server resources open for the duration.
The scheduler component parses cron expressions attached to workflow definitions, evaluates them against the current time on a heartbeat interval, and fans out triggers across the applicable set of contacts or entities. For a workflow scheduled at “0 9 * * 1-5” (9am weekdays) that targets 500 active contacts, the scheduler creates 500 individual workflow runs — one per contact — each initialized with the contact’s data as the seed context. This fan-out pattern is essential for outreach automation where the same workflow template is personalized and executed independently for each recipient.
The DAG engine executes each run by traversing nodes in topological order, dispatching each node to its type-specific handler, and persisting the result to a node_results table before advancing. When the engine encounters a delay node (e.g., “wait 3 days”) or a human approval node, it records the pause point — the current node ID, the accumulated context, and the resume timestamp — and yields. A separate resume worker polls for runs whose resume time has passed and restarts them from the paused node, skipping all previously completed nodes. This persistence-first design means the server can crash, restart, or scale horizontally without losing workflow state.
Each workflow definition includes a cron schedule ("0 9 * * 1-5") and a target audience query (e.g., all contacts with status “active”). The scheduler evaluates cron expressions on a heartbeat (every 60 seconds), determines which workflows are due, resolves their target audiences, and creates individual run records — one per target entity. Each run is initialized with the entity’s data merged into the workflow’s default context. The scheduler uses a last_triggered_at timestamp per workflow to prevent duplicate triggers if the heartbeat fires multiple times within the same cron window.
The workflow definition is a DAG where nodes represent operations (send email, AI analysis, filter, delay, human approval) and edges define execution order. The engine computes a topological sort of the nodes, respecting edge dependencies, and executes them sequentially. Each node handler receives the accumulated context from prior nodes and returns an updated context plus an optional routing signal (for conditional branches). The engine records each node’s result — status, output data, execution duration, and any error — to a node_results table, creating a complete audit trail of the run’s execution history.
A delay node specifies a duration (e.g., "3d", "12h", "30m") that the workflow should wait before advancing to the next node. Instead of sleeping a thread or holding a connection open, the engine computes the absolute resume timestamp (now + duration), persists it to the run record along with the current node ID and context snapshot, and marks the run as paused. A resume worker periodically queries for paused runs whose resume time has passed, loads their persisted state, and continues execution from the delay node’s successor.
The run context is a JSON object that accumulates outputs from each node. After every node execution, the engine writes the updated context to the workflow_runs table and the node-specific result to node_results. This dual persistence serves two purposes: the run context enables resumption (the engine can reconstruct the exact state at any point), and the node results enable auditing (operators can inspect what each node produced, how long it took, and whether it succeeded or failed). Together, they provide both operational reliability and compliance visibility.
When a paused run resumes, the engine must not re-execute nodes that already completed. It loads the list of completed node IDs from node_results, filters them from the topological order, and begins execution at the first incomplete node. This skip logic must be idempotent: if the resume worker crashes mid-execution and retries, previously completed nodes are skipped again based on the persisted results, not in-memory state. Idempotent resumption is the foundation that makes the entire system crash-safe.
The scheduler runs as a long-lived process with a 60-second heartbeat. On each tick, it loads all active workflow definitions with cron schedules, evaluates each cron expression against the current time, and identifies workflows that are due. For each due workflow, it resolves the target audience (querying the contacts table with the workflow’s audience filter), creates a batch of run records, and updates the workflow’s last_triggered_at timestamp. The entire trigger-and-fan-out operation runs within a database transaction to ensure atomicity — either all runs are created and the timestamp is updated, or none are.
The execution engine receives a run ID, loads the workflow definition and current run context from the database, computes the topological order of nodes, and iterates. For each node, it checks if a result already exists in node_results (skip if so), dispatches to the node handler, persists the result, and updates the run context. If a node is a delay type, the engine computes the resume time, persists the pause state, and returns — the run will be picked up later by the resume worker. If a node is a filter type, the engine follows only the matching outbound edge based on the handler’s routing signal.
The resume worker is a separate process (or the same scheduler with a different heartbeat) that polls for runs in paused status whose resume_at timestamp is in the past. For each eligible run, it calls the execution engine with the run ID. The engine’s skip logic handles the rest — it loads completed nodes, starts from where the run left off, and continues until completion or another pause point. The resume worker uses a SELECT FOR UPDATE with SKIP LOCKED pattern to prevent multiple worker instances from picking up the same run.
Parse cron expressions, evaluate schedules, and fan out workflow triggers across target contacts.
// scheduler/cron-scheduler.ts
import { parseExpression } from "cron-parser";
interface WorkflowSchedule {
workflowId: string;
cron: string;
audienceFilter: Record<string, unknown>;
lastTriggeredAt: Date | null;
}
interface Contact {
id: string;
data: Record<string, unknown>;
}
interface SchedulerDeps {
getActiveSchedules(): Promise<WorkflowSchedule[]>;
getContacts(filter: Record<string, unknown>): Promise<Contact[]>;
createRuns(
workflowId: string,
contacts: Contact[]
): Promise<string[]>; // returns run IDs
updateLastTriggered(workflowId: string, at: Date): Promise<void>;
}
export async function schedulerTick(
deps: SchedulerDeps,
now: Date = new Date()
): Promise<{ triggered: number; runsCreated: number }> {
const schedules = await deps.getActiveSchedules();
let triggered = 0;
let runsCreated = 0;
for (const schedule of schedules) {
const interval = parseExpression(schedule.cron, {
currentDate: schedule.lastTriggeredAt ?? new Date(0),
});
// Check if the next occurrence is at or before now
const next = interval.next().toDate();
if (next > now) continue;
// Prevent double-trigger: skip if last triggered within
// the same cron window
if (
schedule.lastTriggeredAt &&
schedule.lastTriggeredAt.getTime() >= next.getTime()
) {
continue;
}
const contacts = await deps.getContacts(schedule.audienceFilter);
if (contacts.length === 0) continue;
const runIds = await deps.createRuns(
schedule.workflowId,
contacts
);
await deps.updateLastTriggered(schedule.workflowId, now);
triggered += 1;
runsCreated += runIds.length;
}
return { triggered, runsCreated };
}
// Run the scheduler on a 60-second heartbeat
export function startScheduler(deps: SchedulerDeps): NodeJS.Timer {
return setInterval(async () => {
try {
const result = await schedulerTick(deps);
if (result.triggered > 0) {
console.log(
`Scheduler: triggered ${result.triggered} workflows, ` +
`created ${result.runsCreated} runs`
);
}
} catch (err) {
console.error("Scheduler tick failed:", err);
}
}, 60_000);
}Execute workflow nodes in topological order, persist results after each node, and support pause/resume for delay nodes.
// engine/dag-executor.ts
type NodeStatus = "pending" | "completed" | "failed" | "skipped";
type RunStatus = "running" | "paused" | "completed" | "failed";
interface NodeResult {
nodeId: string;
status: NodeStatus;
output: Record<string, unknown>;
durationMs: number;
error?: string;
}
interface WorkflowRun {
id: string;
workflowId: string;
status: RunStatus;
context: Record<string, unknown>;
pausedAtNodeId?: string;
resumeAt?: Date;
nodeResults: NodeResult[];
}
interface DAGNode {
id: string;
type: string;
data: Record<string, unknown>;
dependencies: string[]; // node IDs that must complete first
}
interface DAGDefinition {
nodes: DAGNode[];
}
interface ExecutorDeps {
loadRun(runId: string): Promise<WorkflowRun>;
loadDefinition(workflowId: string): Promise<DAGDefinition>;
saveNodeResult(runId: string, result: NodeResult): Promise<void>;
updateRun(
runId: string,
fields: Partial<WorkflowRun>
): Promise<void>;
}
type NodeHandler = (
node: DAGNode,
context: Record<string, unknown>
) => Promise<{
output: Record<string, unknown>;
nextHandle?: string;
pauseDuration?: number; // ms — if set, pause the run
}>;
const handlers = new Map<string, NodeHandler>();
export function registerNodeHandler(
type: string,
handler: NodeHandler
): void {
handlers.set(type, handler);
}
// --- Built-in delay handler ---
registerNodeHandler("delay", async (node, _ctx) => {
const durationMs =
parseDuration(node.data.duration as string) ?? 0;
return { output: {}, pauseDuration: durationMs };
});
function parseDuration(s: string): number | null {
const match = s.match(/^(\d+)(ms|s|m|h|d)$/);
if (!match) return null;
const n = parseInt(match[1], 10);
const unit = match[2];
const multipliers: Record<string, number> = {
ms: 1,
s: 1000,
m: 60_000,
h: 3_600_000,
d: 86_400_000,
};
return n * (multipliers[unit] ?? 0);
}
// --- Topological sort ---
function topoSort(nodes: DAGNode[]): DAGNode[] {
const sorted: DAGNode[] = [];
const visited = new Set<string>();
const nodeMap = new Map(nodes.map((n) => [n.id, n]));
function visit(id: string) {
if (visited.has(id)) return;
visited.add(id);
const node = nodeMap.get(id)!;
for (const dep of node.dependencies) {
visit(dep);
}
sorted.push(node);
}
for (const node of nodes) visit(node.id);
return sorted;
}
// --- Main execution loop ---
export async function executeRun(
runId: string,
deps: ExecutorDeps
): Promise<void> {
const run = await deps.loadRun(runId);
const definition = await deps.loadDefinition(run.workflowId);
const completedIds = new Set(
run.nodeResults
.filter((r) => r.status === "completed")
.map((r) => r.nodeId)
);
const ordered = topoSort(definition.nodes);
let ctx = { ...run.context };
await deps.updateRun(runId, { status: "running" });
for (const node of ordered) {
// Skip already-completed nodes (idempotent resumption)
if (completedIds.has(node.id)) continue;
const handler = handlers.get(node.type);
if (!handler) {
await deps.saveNodeResult(runId, {
nodeId: node.id,
status: "failed",
output: {},
durationMs: 0,
error: `No handler for type: ${node.type}`,
});
await deps.updateRun(runId, { status: "failed" });
return;
}
const start = Date.now();
try {
const result = await handler(node, ctx);
const durationMs = Date.now() - start;
await deps.saveNodeResult(runId, {
nodeId: node.id,
status: "completed",
output: result.output,
durationMs,
});
ctx = { ...ctx, ...result.output };
await deps.updateRun(runId, { context: ctx });
// Handle pause (delay node)
if (result.pauseDuration && result.pauseDuration > 0) {
const resumeAt = new Date(
Date.now() + result.pauseDuration
);
await deps.updateRun(runId, {
status: "paused",
pausedAtNodeId: node.id,
resumeAt,
context: ctx,
});
return; // Yield — resume worker picks up later
}
} catch (err) {
const durationMs = Date.now() - start;
await deps.saveNodeResult(runId, {
nodeId: node.id,
status: "failed",
output: {},
durationMs,
error: String(err),
});
await deps.updateRun(runId, { status: "failed" });
return;
}
}
await deps.updateRun(runId, { status: "completed" });
}In the merchant dashboard platform, the cron-driven scheduler powers multi-step outreach campaigns that span days or weeks. A “new prospect nurture” workflow, scheduled at 9am on weekdays, fans out across all contacts that entered the pipeline in the last 24 hours. The DAG starts with an AI analysis node that summarizes the prospect’s company and recent activity, passes through a delay node (“wait 2 days”), then sends a personalized follow-up email. If the prospect doesn’t reply within 5 days (tracked by a webhook listener node), a second delay and breakup email follow. Each node’s result is persisted, so ops engineers can query the node_results table to see exactly which step each prospect is on and why any particular run stalled.
The resumable execution design also handles operational resilience. When the server was restarted during a deployment, 147 in-flight workflow runs were paused at various delay nodes. After the restart, the resume worker picked them all up within its next polling interval and continued each from its persisted pause point — no runs were lost or duplicated. This crash-safe property is what makes the system viable for production outreach where a lost or duplicated email damages the sender’s reputation.
last_triggered_at guard prevents this, but only if it is checked and updated atomically within a transaction. Without transactional protection, a race between two scheduler instances can still produce duplicates.SELECT FOR UPDATE SKIP LOCKED to ensure exclusive ownership, and mark the run as running before beginning execution.