Data Engineering with DataStateMachine

The Workflow as a Data Pipeline

Traditional data pipelines need separate deduplication logic, lineage tracking, quality gates, and audit tools. DataStateMachine makes each workflow node a data-aware component: it knows what it processed, when, and with what result.

The state records that drive idempotency also serve as:

• A deduplication registry — has this record been processed by this node?
• A lineage log — which node processed which entity, in which execution?
• A quality gate — what was the processing result and data?
• An event log — status transitions as events in chronological order

Pipeline Deduplication at Node Level

Standard pipeline deduplication happens at the pipeline boundary (e.g. "already imported this file?"). DataStateMachine pushes dedup down to the individual node level, where duplicates actually originate.

Without DataStateMachine:
    Source System → [Extract Node] → [Transform Node] → [Load Node]
                    ^ one dedup check here only
                    (race conditions, retries still cause duplicates)

With DataStateMachine:
    Source System → [Extract Node] → [Transform Node] → [Load Node]
                    ^ ExistsAsync    ^ ExistsAsync       ^ ExistsAsync
                    each node independently deduplicates its own work

// In each pipeline node — consistent deduplication pattern
var recordId = ctx.Input.GetString("sourceRecordId");

if (await DataStateMachine.ExistsAsync(recordId, entityKey: "etl-record", ct: ct))
    return NodeExecutionResult.Skip($"Record {recordId} already processed by this node");

bool locked = await DataStateMachine.TryAcquireLockAsync(recordId, entityKey: "etl-record", ct: ct);
if (!locked)
    return NodeExecutionResult.Skip("Processing in another thread");

// ... transform or load the record ...
await DataStateMachine.SetAsync(recordId, "etl-record", IdempotencyStatus.Processed, data: result, ct: ct);

Data Lineage via State Records

Every Process_StateLogs row is a lineage record. It captures: which entity (EntityID), which execution (EntityExecutionID), which scope (StateLogScopeID), which item (ItemKey + ItemEntityKey), when (Created/Modified), and what result (Status + Data).

A lineage query for a single entity shows every node that touched it, in every execution, with full data:

SELECT
    s.ScopeCode,
    sl.ItemEntityKey,
    sl.ItemKey,
    sl.Status,
    sl.Created       AS ProcessedAt,
    sl.Data,
    sl.SupersededByID
FROM Process_StateLogs sl
JOIN Process_StateLogScopes s ON sl.StateLogScopeID = s.ID
WHERE sl.ItemEntityKey = 'INV-2024-00892'   -- the entity
  AND sl.TenantID      = 42
  AND sl.Deleted       = 0
ORDER BY sl.Created ASC;
-- Returns: full lineage chain for this invoice across all nodes and executions

Data Quality Gates

A node can check the prior state of an upstream entity before processing it. If the upstream node failed, the downstream node can refuse to proceed — preventing bad data from flowing through the pipeline.

// Downstream node — checks upstream state before accepting input
var upstreamState = await DataStateMachine.GetAsync(
    itemKey:       customerId,
    itemEntityKey: "customer-enrichment",
    ct:            ct);

if (upstreamState?.Status == IdempotencyStatus.Failed)
    return NodeExecutionResult.Block("Customer enrichment failed — cannot proceed with scoring");

if (upstreamState is null || upstreamState.Status != IdempotencyStatus.Processed)
    return NodeExecutionResult.Skip("Waiting for upstream enrichment to complete");

// Safe to proceed
var enrichedData = JsonSerializer.Deserialize<CustomerData>(upstreamState.Data!);

Append-Only Records as Event Log

UpsertAsync creates a new row and marks the old row Superseded via SupersededByID. The table is therefore an ordered chain of events for each entity — a native event log.

This event log can be replayed. To reconstruct the state of an entity at any point in time, query all rows for that entity ordered by Created and walk the chain.

-- All events for a customer, in order (event sourcing query)
WITH RECURSIVE EventChain AS (
    SELECT * FROM Process_StateLogs
    WHERE ItemEntityKey = 'CUST-1001'
      AND SupersededByID IS NULL     -- start at the latest
      AND Deleted = 0

    UNION ALL

    SELECT sl.* FROM Process_StateLogs sl
    JOIN EventChain ec ON sl.ID = ec.SupersededByID
)
SELECT * FROM EventChain ORDER BY Created ASC;

Recovery Patterns

Because state is durable, pipeline recovery is straightforward:

Sub-Item Tracking for Enrichment Pipelines

The SubItemKey field was designed for enrichment pipelines where one entity gets multiple independent enrichments.

// One entity — three enrichment sub-operations tracked independently
// Address validation
await DataStateMachine.SetAsync(
    itemKey: customerId, itemEntityKey: "customer",
    subItemKey: "address-validation",
    status: IdempotencyStatus.Processed, data: addressResult, ct: ct);

// Credit score pull
await DataStateMachine.SetAsync(
    itemKey: customerId, itemEntityKey: "customer",
    subItemKey: "credit-score",
    status: IdempotencyStatus.Processed, data: creditResult, ct: ct);

// Social graph
await DataStateMachine.SetAsync(
    itemKey: customerId, itemEntityKey: "customer",
    subItemKey: "social-graph",
    status: IdempotencyStatus.Processed, data: socialResult, ct: ct);

// Query: which enrichments are complete for this customer?
var enrichments = await DataStateMachine.QueryAsync(
    itemKey: customerId, itemEntityKey: "customer",
    query: new StateQuery { StatusFilter = IdempotencyStatus.Processed }, ct: ct);