Workflows — Hangfire Batches en Job Chaining

Dit document beschrijft hoe Hangfire Batches worden gebruikt voor de uitvoering van workflow executions, hoe de graph van nodes wordt doorlopen en hoe job chaining en vertragingen werken.

Architecturale beslissing: waarom Hangfire Batches

Elke workflow execution wordt als één Hangfire Batch aangemaakt. Dit geeft drie concrete voordelen:

Zichtbaarheid — alle jobs van een execution zijn in het Hangfire Dashboard zichtbaar als één groep met de naam Workflow Execution: {naam}.
Cancelability — bij annulering kan de hele batch worden teruggezocht via HangfireBatchId.
Contextloze scheduling — individuele activity jobs kennen hun batch ID niet. Ze vragen het op uit de database en hangen zichzelf aan de batch via BatchJob.Attach.

Aanmaken van een execution — BatchJob.StartNew

De batch wordt aangemaakt in WorkflowTriggerService, vóórdat de execution-rij in de database wordt aangelegd. De WorkflowDispatcherJob is de eerste job in de batch:

// WorkflowTriggerService.cs
var executionId = Guid.NewGuid();
var hangfireBatchId = BatchJob.StartNew(b =>
{
    b.Enqueue<WorkflowDispatcherJob>(
        job => job.RunAsync(request.TenantId, executionId, CancellationToken.None));
}, $"Workflow Execution: {workflow.Name}");

await workflowsRepo.CreateExecutionAsync(
    executionId, request.TenantId, workflow.Id, workflow.ActiveVersionId.Value,
    hangfireBatchId, request.TriggerType, triggerDataJson, ct);

Het patroon is bewust: de HangfireBatchId wordt opgeslagen in Workflows_Executions zodat latere jobs — die al draaien en hun context niet kennen — de batch ID kunnen ophalen.

WorkflowDispatcherJob — van batch naar trigger

De WorkflowDispatcherJob is de centrale orchestrator die de workflow definition laadt, de trigger node identificeert en de eerste activity job inschrijft.

Stappen

Laad de WorkflowExecution (bevat gelocked VersionId en HangfireBatchId).
Laad de WorkflowVersion en deserialiseer de WorkflowDefinitionDto (nodes + edges).
Zoek de trigger node (Type == "trigger").
Evalueer of de workflow moet wachten op een hire date (GetHireDateDelay).
Maak een ExecutionJob-rij aan in de database voor de trigger node.
Voeg de WorkflowTriggerActivityJob toe aan de batch via BatchJob.Attach.

// WorkflowDispatcherJob.cs — direct enqueue (geen vertraging)
lock (BatchLock)
{
    BatchJob.Attach(execution.HangfireBatchId, b =>
    {
        hangfireJobId = b.Enqueue<WorkflowTriggerActivityJob>(
            job => job.RunAsync(triggerJobId, executionId, ct));
    });
}
await workflowsRepo.UpdateExecutionStatusAsync(executionId, "Running", false, null, ct);

Hire date scheduling

Wanneer de trigger node onboardingDateType == "StartDate" heeft én de employeeHireDate in de toekomst ligt, wordt de trigger job niet direct ge-enqueued maar gepland op de hire datum:

// WorkflowDispatcherJob.cs — scheduled enqueue voor hire date
lock (BatchLock)
{
    BatchJob.Attach(execution.HangfireBatchId, b =>
    {
        hangfireJobId = b.Schedule<WorkflowTriggerActivityJob>(
            job => job.RunAsync(triggerJobId, executionId, ct),
            hireDelay.Value);
    });
}
await workflowsRepo.UpdateExecutionStatusAsync(executionId, "Scheduled", false, null, ct);

De execution krijgt status Scheduled. Wanneer de WorkflowTriggerActivityJob na de vertraging start, zet die de status om naar Running.

Graph traversal — hoe de volgende node wordt bepaald

Na elke voltooide job bepaalt WorkflowActivityJobBase.ScheduleNextNodesAsync welke nodes als volgende worden ingepland. Dit werkt op basis van de edges in de WorkflowDefinitionDto.

Stap-voor-stap

// WorkflowActivityJobBase.cs
var outgoingEdges = workflowDef.ActualEdges
    .Where(e => e.ActualSource == currentNodeId.ToString())
    .ToList();

Voor elke outgoing edge:

Zoek de target node op in workflowDef.Nodes.
Controleer of de target node meerdere inkomende edges heeft (merge point).
Als het een delay node is: roep HandleDelayNodeAsync aan.
Anders: roep EnqueueActivityNodeAsync aan.

Merge point handling

Een merge point is een node met meer dan één inkomende edge. Zonder speciale behandeling zou de node meerdere keren worden geenqueued — eén keer door elke voltooide bovenliggende job.

De check in CanScheduleMergeNodeAsync voorkomt dit: de target node wordt alleen gepland als:

alle parent nodes de status Completed hebben, én
de target node nog geen job-rij heeft in de database.

// WorkflowActivityJobBase.cs
var incomingEdges = workflowDef.ActualEdges
    .Where(e => e.ActualTarget == targetNode.Id)
    .ToList();

if (incomingEdges.Count > 1)
{
    var canSchedule = await CanScheduleMergeNodeAsync(
        executionId, targetNode.Id, incomingEdges, workflowDef, ct);
    if (!canSchedule) continue;
}

BatchJob.Attach — jobs toevoegen vanuit lopende jobs

Alle activity jobs voegen vervolgstappen toe aan de batch via BatchJob.Attach. Ze halen de HangfireBatchId op uit de database:

// WorkflowActivityJobBase.cs — EnqueueActivityNodeAsync
var execution = await _workflowsRepo.GetExecutionAsync(executionId, ct);

lock (BatchLock)
{
    BatchJob.Attach(execution.HangfireBatchId, b =>
    {
        hangfireJobId = node.Type switch
        {
            "action"    => b.Enqueue<WorkflowActionActivityJob>(
                               x => x.RunAsync(jobId, executionId, ct)),
            "condition" => b.Enqueue<WorkflowConditionActivityJob>(
                               x => x.RunAsync(jobId, executionId, ct)),
            "trigger"   => b.Enqueue<WorkflowTriggerActivityJob>(
                               x => x.RunAsync(jobId, executionId, ct)),
            _ => throw new InvalidOperationException($"Unknown node type: {node.Type}")
        };
    });
}

De lock (BatchLock) is een static object lock die thread-safe toegang tot BatchJob.Attach garandeert.

Delay nodes — vertragingen in de workflow

Een delay node wordt niet als een echte Hangfire job uitgevoerd maar vertegenwoordigt een wachttijd. De verwerking verloopt als volgt:

Bereken de vertraging via CalculateDelay (zie hieronder).
Maak een ExecutionJob-rij aan voor de delay node met status Running.
Zoek alle outgoing edges van de delay node (SPLIT-patroon is mogelijk).
Plan voor elke volgende node een Hangfire scheduled job met de berekende vertraging.

// WorkflowActivityJobBase.cs — ScheduleDelayedJobAsync
lock (BatchLock)
{
    BatchJob.Attach(execution.HangfireBatchId, b =>
    {
        hangfireJobId = b.Schedule(
            () => ExecuteScheduledNodeAsync(jobId, executionId, node.Type, delayJobId, isLastNode, ct),
            delay);
    });
}

De isLastNode vlag zorgt ervoor dat de delay job pas als Completed wordt gemarkeerd nadat de laatste vervolgnode is gestart — dit voorkomt race conditions bij een SPLIT na een delay.

Ondersteunde delay types

De vertraging wordt berekend in CalculateDelay op basis van de delayType property in de node data:

delayType	Werking
`Duration`	Vaste TimeSpan, bijv. `"1.00:00:00"` voor 1 dag
`UntilDateTime`	Wacht tot een specifiek tijdstip (`TargetDateTime`)
`NextDay`	Volgende dag om een opgegeven tijd
`NextBusinessDay`	Volgende werkdag (slaat za/zo over) om een opgegeven tijd
`SpecificDayOfWeek`	Eerstvolgende specifieke weekdag om een opgegeven tijd

Fallback bij ongeldig formaat: 1 uur.

Condition nodes — conditionele vertakking

De WorkflowConditionActivityJob wijkt af van de basis job lifecycle: hij beheert zelf de scheduling van vervolgstappen (via ShouldScheduleNextNodesAutomatically() => false).

Na evaluatie van de regels wordt de juiste tak gevolgd op basis van de SourceHandle van de edges ("true" of "false"):

// WorkflowConditionActivityJob.cs
var branchToExecute = conditionResult ? "true" : "false";

foreach (var edge in outgoingEdges)
{
    var isMatchingBranch = string.Equals(
        edge.SourceHandle, branchToExecute, StringComparison.OrdinalIgnoreCase);

    if (isMatchingBranch)
        // Enqueue of delay voor de actieve tak
    else
        // SkipBranchRecursivelyAsync voor de niet-genomen tak
}

Nodes op de niet-genomen tak worden recursief aangemaakt als skipped jobs (IsSkipped = true). Dit is nodig zodat CheckAndCompleteExecutionAsync correct kan vaststellen of de workflow klaar is — skipped jobs worden genegeerd bij de afronding.

Execution afronding

Na elke voltooide job, als er geen outgoing edges zijn, wordt CheckAndCompleteExecutionAsync aangeroepen:

// WorkflowActivityJobBase.cs
var activeJobs = jobs.Where(j =>
    j.JobStatus == "Pending" ||
    j.JobStatus == "Running" ||
    j.JobStatus == "Scheduled").ToList();

if (!activeJobs.Any())
{
    var failedJobs = jobs.Where(j => j.JobStatus == "Failed").ToList();
    // Skipped jobs tellen niet mee als fout

    if (failedJobs.Any())
        await _workflowsRepo.UpdateExecutionStatusAsync(executionId, "Failed", true, ...);
    else
        await _workflowsRepo.UpdateExecutionStatusAsync(executionId, "Completed", true, null, ct);
}

De execution wordt als Completed gemarkeerd wanneer alle niet-skipped jobs zijn voltooid. De aanwezigheid van skipped jobs (conditionele tak niet genomen) is geen reden voor een fout.

Volledige batch/job flow

Error handling

Wanneer een activity job een exception gooit:

De ExecuteJobAsync methode in WorkflowActivityJobBase vangt de exception op.
De job status wordt op Failed gezet in Workflows_ExecutionJobs (inclusief ErrorMessage en ErrorStackTrace).
De exception wordt opnieuw gegooid (throw), waardoor Hangfire de job als mislukt registreert en eventueel retry-pogingen uitvoert.
Hangfire's ingebouwde retry mechanisme is actief — na het maximum aantal pogingen eindigt de job in de Failed staat in het Hangfire Dashboard.
CheckAndCompleteExecutionAsync markeert de execution uiteindelijk als Failed zodra alle actieve jobs zijn afgerond.

De WorkflowDispatcherJob heeft eigen foutafhandeling die de execution direct op Failed zet als de workflow definitie niet geladen kan worden.

Last modified: 24 March 2026