The DataStateMachine Life Cycle

Phase Overview

The DataStateMachine lifecycle spans five phases. Each phase has a clear owner and defined responsibilities.

Full Lifecycle Diagram

Workflow Engine
    │  node execution triggered
    ▼
BaseNodeExecutor.OnPreProcess
    │
    ├── InitialiseDataStateMachineAsync()
    │       │
    │       ├── read node config → DataStateMachineConfig (handler key, scope, options)
    │       │
    │       ├── DataStateMachineFactory.CreateAsync(elem, thread, config)
    │       │       │
    │       │       ├── resolve IDataStateMachineHandler by HandlerKey
    │       │       │     └── if key not found → throw (misconfiguration)
    │       │       │
    │       │       ├── resolve ScopeID from scope name (e.g. "Element" → 4)
    │       │       │
    │       │       └── build StateHandlerContext {
    │       │               StateLogScopeID, ScopeCode,
    │       │               EntityID, EntityExecutionID,
    │       │               TenantID, SourceAppID,
    │       │               HandlerConfig
    │       │               // ItemKey fields left at "" — set per-call
    │       │             }
    │       │
    │       └── assign to _dataStateMachine = new ContextualDataStateMachine(handler, baseCtx)
    │             OR NullDataStateMachine.Instance (if no config block)
    │
    └── node proceeds to Execute phase

Node.ExecuteAsync(ctx, settings, ct)
    │
    ├── await DataStateMachine.ExistsAsync(itemKey, entityKey)
    │     └── ContextualDataStateMachine: builds callCtx = baseCtx with { ItemKey=itemKey, ItemEntityKey=entityKey }
    │         → handler.ExistsAsync(callCtx, ct)
    │
    ├── [if not exists] await DataStateMachine.TryAcquireLockAsync(itemKey, entityKey)
    │     └── handler writes Processing record with LockedUntil = UtcNow + 30 min
    │         → returns true (lock acquired) or false (already locked by another thread)
    │
    ├── [if locked] perform business operation
    │
    ├── await DataStateMachine.SetAsync(itemKey, entityKey, Processed, data)
    │     └── handler updates record to Processed; clears lock
    │
    └── [on exception] await DataStateMachine.ReleaseLockAsFailedAsync(itemKey, entityKey, error)
          └── handler updates record to Failed; clears lock


RECOVERY PATH (background)
CatchupWorkerService (scheduled)
    │
    └── ProcessStateLogRepository.ResetExpiredLocksAsync()
          │  WHERE Status = Processing AND LockedUntil < UtcNow
          └── UPDATE Status = Failed, LockedUntil = NULL


STATUS TRANSITION DIAGRAM
None ──→ Processing ──→ Processed  (happy path)
                   ├──→ Failed     (exception / releaseLockAsFailed)
                   └──→ [timeout]  → Failed (via CatchupWorker)

Processed ──→ Superseded  (when UpsertAsync adds a newer record for same key)
Any ──→ Skipped           (node explicitly calls SetAsync(Skipped))
Any ──→ Expired           (record exceeds retention policy)

Annotated Sample — Invoice Processing Node

Walk through a concrete example: an InvoiceDispatchExecutor that uses the built-in handler at Element scope.

Step 1 — Node Configuration

// Node settings in Flow Studio
{
  "invoiceApiEndpoint": "https://api.billing.internal/dispatch",
  "dataStateMachine": {
    "handler": "built-in",
    "scope":   "Element",
    "options": {}
  }
}

Step 2 — Initialisation (automatic, BaseNodeExecutor)

// Called automatically in OnPreProcess — developer writes none of this
var config = Settings.DataStateMachineConfig;  // parsed from JSON above
_dataStateMachine = await _factory.CreateAsync(element, thread, config);
// → ContextualDataStateMachine wrapping BuiltInDataStateMachineHandler
// → base context: ScopeID=4(Element), EntityID=invoiceElementId, TenantID=tenant

Step 3 — Node Execute (developer writes this)

protected override async Task<NodeExecutionResult> ExecuteAsync(
    NodeExecutionContext ctx, InvoiceSettings settings, CancellationToken ct)
{
    var invoiceId = ctx.Input.GetString("invoiceId");   // e.g. "INV-2024-00892"

    // Guard: has this invoice already been dispatched?
    if (await DataStateMachine.ExistsAsync(invoiceId, entityKey: "invoice", ct: ct))
        return NodeExecutionResult.Skip($"Invoice {invoiceId} already dispatched");

    // Acquire processing lock — prevents concurrent duplicate runs
    bool locked = await DataStateMachine.TryAcquireLockAsync(invoiceId, entityKey: "invoice", ct: ct);
    if (!locked)
        return NodeExecutionResult.Skip($"Invoice {invoiceId} is being processed by another thread");

    try
    {
        // Perform the actual work
        var response = await _billingApi.DispatchAsync(invoiceId, settings.ApiEndpoint, ct);

        // Mark complete — store response envelope
        await DataStateMachine.SetAsync(
            itemKey:       invoiceId,
            itemEntityKey: "invoice",
            status:        IdempotencyStatus.Processed,
            data:          JsonSerializer.Serialize(new { dispatchedAt = DateTime.UtcNow, response }),
            ct:            ct);

        return NodeExecutionResult.Success(new { invoiceId, dispatched = true });
    }
    catch (Exception ex)
    {
        // Release lock so CatchupWorker (or manual retry) can re-process
        await DataStateMachine.ReleaseLockAsFailedAsync(invoiceId, entityKey: "invoice",
            errorDetail: ex.Message, ct: ct);
        throw;
    }
}

Step 4 — What Gets Written to Process_StateLogs

Duplicate Run — What Happens

If the same workflow runs again with the same invoiceId (e.g. from a retry trigger or duplicate message), ExistsAsync returns true immediately. The node returns Skip without ever calling the billing API. No duplicate charge. No race condition. One line of guard code.