Reconciliation Run Lifecycle Events

The reconciliation run lifecycle separates infrastructure events (emitted the moment a run’s workflow finishes) from platform events (emitted after the database is updated). This separation eliminates duplicate notifications and guarantees that downstream consumers always see DB-consistent state.

Key Concepts

Infrastructure vs Platform Events

Infrastructure events (run.completed, run.errored) are emitted the moment a run’s workflow finishes. They reflect the raw execution outcome and may arrive before the database is updated.
Platform events (run.finalised) are emitted by Core-API’s RunCompletionController after the database has been updated, the job event has been recorded, and the WebSocket broadcast has been sent. This guarantees DB consistency for all downstream consumers.

`run.finalised` — The Authoritative Completion Event

run.finalised is the single, DB-backed event to act on when you want one notification per run. Built-in email notifications fire from run.finalised only — never from the raw infrastructure run.completed / run.errored events — which is why a finished run produces exactly one notification rather than a duplicate per stage. If you subscribe a webhook to run completion, prefer run.finalised for the same reason. run.finalised is not emitted for cancelled runs (see Cancellation).

Errored vs Unmatched

Errored means the workflow could not run to completion due to an infrastructure issue (crash, timeout, resource failure). Represented by run.errored.
Unmatched means the reconciliation comparison ran successfully but found data discrepancies outside tolerances. This is a business outcome, not an error. Represented by run.completed with result: UNMATCHED.

RunResult

The result field on run.completed and run.finalised events uses the RunResult enum:

Value	Meaning
`MATCHED`	100% join match across all sources AND all tolerances passed
`UNMATCHED`	Any unmatched groups across sources OR any tolerance failures

100% join match is required for MATCHED. Any non-zero values in unmatched_by_source result in UNMATCHED.

Event Flow

As a run progresses, DataRecs emits the events below in order. Any event type you’ve subscribed a webhook to is delivered to your endpoint as a signed CloudEvent; the same events drive built-in email notifications. The diagram shows the sequence from your perspective as a subscriber — when each event fires and what carries the authoritative outcome.

sequenceDiagram
    participant User
    participant DataRecs as DataRecs Platform
    participant You as Your Webhook Endpoint

    User->>DataRecs: Trigger run
    DataRecs->>You: run.triggered
    DataRecs->>You: run.queued

    Note over DataRecs: Workflow executes...
    DataRecs->>You: run.extraction.started / completed / errored
    DataRecs->>You: run.comparison.started / completed / errored
    DataRecs->>You: run.stage.started / completed / errored

    Note over DataRecs: Workflow finishes (infrastructure events)
    DataRecs->>You: run.completed (result: MATCHED | UNMATCHED)<br/>OR run.errored (error: {code, message})

    Note over DataRecs: Database updated, status confirmed
    DataRecs->>You: run.finalised (DB-backed, authoritative)

run.completed / run.errored reflect the raw workflow outcome the moment it finishes and may arrive before the run’s status is fully persisted. run.finalised fires once the database is updated and is the authoritative, single completion event — subscribe to it when you want exactly one notification per run.

You only receive the event types you subscribe to. To receive every lifecycle event, subscribe with ["*"]; to act only on outcomes, subscribe to run.finalised (or run.completed / run.errored).

Cancellation Flow

Cancellation uses a cancel_requested flag rather than an immediate status change. This handles the race condition where the run may have already finished by the time the cancel request is processed.

sequenceDiagram
    participant User
    participant DataRecs as DataRecs Platform
    participant You as Your Webhook Endpoint

    User->>DataRecs: Cancel run
    DataRecs->>You: run.cancel_requested (audit trail)

    alt Run had not started yet
        DataRecs->>You: run.cancelled
    else Run was in progress
        Note over DataRecs: Workflow is stopped...
        DataRecs->>You: run.cancelled (NOT run.finalised)
    end

    alt Race: run finished before the cancel took effect
        Note over DataRecs: Cancel was too late — run completed normally
        DataRecs->>You: run.finalised (normal completion)
    end

A confirmed-cancelled run emits run.cancelled, not run.finalised. Because built-in email notifications fire only on run.finalised, cancelled runs never generate a completion email — and a webhook subscribed only to run.finalised likewise won’t see them. Subscribe to run.cancelled if you need to act on cancellations. If the run actually finished before the cancel landed (a stale cancel), you get run.finalised for the normal completion instead.

Event Type Reference

Every event is delivered as a CloudEvent. The CloudEvent type field is the event-type string shown below — there is no extra prefix (the same string is the NATS subject suffix internally). The DataRecsEventType enum in data-models-events is the single source of truth.

Run Lifecycle Events

CloudEvent `type`	Source	Emitting Service	Description
`reconciliation.run.triggered`	`/datarecs/core-api`	Core-API	Run was triggered by a user or schedule
`reconciliation.run.queued`	`/datarecs/core-api`	Core-API	Workflow was submitted successfully
`reconciliation.run.queue_errored`	`/datarecs/core-api`	Core-API	Workflow submission failed
`reconciliation.run.completed`	`/datarecs/reconciliation-worker`	Workflow exit handler	Workflow succeeded (infrastructure event)
`reconciliation.run.errored`	`/datarecs/reconciliation-worker`	Workflow exit handler	Workflow failed (infrastructure event)
`reconciliation.run.finalised`	`/datarecs/core-api`	Core-API	DB updated, authoritative status (platform event)
`reconciliation.run.cancel_requested`	`/datarecs/core-api`	Core-API	User requested cancellation (audit trail)
`reconciliation.run.cancelled`	`/datarecs/core-api`	Core-API	Run confirmed cancelled
`reconciliation.run.rows_processed`	`/datarecs/reconciliation-worker`	Reconciliation Worker	Progress event

Extraction Lifecycle Events

CloudEvent `type`	Source	Emitting Service	Description
`reconciliation.run.extraction.started`	`/datarecs/reconciliation-worker`	Reconciliation Worker	Extractor began execution
`reconciliation.run.extraction.completed`	`/datarecs/reconciliation-worker`	Reconciliation Worker	Extractor finished successfully
`reconciliation.run.extraction.errored`	`/datarecs/reconciliation-worker`	Reconciliation Worker	Extractor encountered an error

Comparison Lifecycle Events

CloudEvent `type`	Source	Emitting Service	Description
`reconciliation.run.comparison.started`	`/datarecs/reconciliation-worker`	Reconciliation Worker	Comparator began execution
`reconciliation.run.comparison.completed`	`/datarecs/reconciliation-worker`	Reconciliation Worker	Comparator finished (rich payload with join stats)
`reconciliation.run.comparison.errored`	`/datarecs/reconciliation-worker`	Reconciliation Worker	Comparator encountered an error

Stage Lifecycle Events

CloudEvent `type`	Source	Emitting Service	Description
`reconciliation.run.stage.started`	`/datarecs/reconciliation-worker`	Reconciliation Worker	Stage began execution
`reconciliation.run.stage.completed`	`/datarecs/reconciliation-worker`	Reconciliation Worker	Stage finished (includes result and tolerances)
`reconciliation.run.stage.errored`	`/datarecs/reconciliation-worker`	Reconciliation Worker	Stage encountered an error

Payload Schemas

`run.triggered`

Emitted by Core-API when a user or schedule triggers a run.

{
  "run_id": "uuid",
  "job_id": "uuid",
  "mode": "IMMEDIATE | SCHEDULED",
  "triggered_by": "user-id | scheduled"
}

The mode field records the trigger origin: IMMEDIATE (user clicked “Run Now”) or SCHEDULED (Airflow cron schedule fired).

`run.queued`

Emitted by Core-API after successful Argo workflow submission.

{
  "run_id": "uuid",
  "job_id": "uuid",
  "argo_workflow_name": "recon-<job_id>-<run_id>"
}

`run.queue_errored`

Emitted by Core-API when Argo workflow submission fails.

{
  "run_id": "uuid",
  "job_id": "uuid",
  "reason": "Platform-Agent rejected the workflow: resource quota exceeded"
}

A resource quota exceeded reason means the run hit a platform limit — see Limits & Quotas. Note that triggering a run while you are already at your concurrent-run limit is rejected synchronously with a 403 QUOTA_EXCEEDED at the API, before any event is emitted.

`run.completed`

Emitted by the Argo exit handler when the workflow succeeds. This is an infrastructure event.

{
  "run_id": "uuid",
  "job_id": "uuid",
  "tenant_id": "uuid",
  "workflow_name": "recon-<job_id>-<run_id>",
  "result": "MATCHED | UNMATCHED"
}

The result field is read from /tmp/comparator-result.json, written by the comparator worker before it exits. If the file is missing or unreadable, the exit handler falls through to emitting run.errored with code UNKNOWN.

`run.errored`

Emitted by the Argo exit handler when the workflow fails. This is an infrastructure event.

{
  "run_id": "uuid",
  "job_id": "uuid",
  "tenant_id": "uuid",
  "workflow_name": "recon-<job_id>-<run_id>",
  "error": {
    "code": "TIMED_OUT | WORKER_OOM | QUERY_FAILED | QUOTA_EXCEEDED_STORAGE | UNKNOWN",
    "message": "Human-readable description of the failure"
  }
}

`run.finalised`

Emitted by Core-API after the database has been updated — the authoritative, single completion event per run, and the one built-in email notifications fire from. NOT emitted for cancelled runs.

Completed Run
Errored Run

{
  "run_id": "uuid",
  "job_id": "uuid",
  "tenant_id": "uuid",
  "result": "MATCHED | UNMATCHED",
  "rows_processed": 1005,
  "rows_matched": 975,
  "rows_unmatched": 30
}

{
  "run_id": "uuid",
  "job_id": "uuid",
  "tenant_id": "uuid",
  "error": {
    "code": "TIMED_OUT | WORKER_OOM | QUERY_FAILED | QUOTA_EXCEEDED_STORAGE | UNKNOWN",
    "message": "Human-readable description of the failure"
  }
}

Exactly one of result or error is present in the payload.

`run.cancel_requested`

Emitted by Core-API when a user requests cancellation. This is an audit trail event only — it does not change the run status or trigger downstream processing.

{
  "run_id": "uuid",
  "job_id": "uuid",
  "tenant_id": "uuid",
  "cancelled_by": "user-id",
  "reason": "User requested cancellation"
}

`run.cancelled`

Emitted by Core-API when a run is confirmed cancelled.

{
  "run_id": "uuid",
  "job_id": "uuid",
  "tenant_id": "uuid",
  "cancelled_by": "user-id",
  "reason": "User requested cancellation"
}

`run.extraction.started`

{
  "run_id": "uuid",
  "job_id": "uuid",
  "tenant_id": "uuid",
  "extraction_index": 0,
  "connection_type": "postgres"
}

`run.extraction.completed`

{
  "run_id": "uuid",
  "job_id": "uuid",
  "tenant_id": "uuid",
  "extraction_index": 0,
  "rows_extracted": 15000
}

`run.extraction.errored`

{
  "run_id": "uuid",
  "job_id": "uuid",
  "tenant_id": "uuid",
  "extraction_index": 0,
  "error": {
    "code": "QUERY_FAILED",
    "message": "relation \"orders\" does not exist"
  }
}

`run.comparison.started`

{
  "run_id": "uuid",
  "job_id": "uuid",
  "tenant_id": "uuid",
  "input_source_count": 2
}

`run.comparison.completed`

The richest payload in the lifecycle. Contains full join statistics, tolerance results, and the overall comparison outcome.

{
  "run_id": "uuid",
  "job_id": "uuid",
  "tenant_id": "uuid",
  "source_row_counts": {
    "source_0": 1000,
    "source_1": 1005
  },
  "join_stats": {
    "matched_groups": 980,
    "unmatched_by_source": {
      "source_0": 5,
      "source_1": 20
    }
  },
  "tolerances": [
    {
      "measure_name": "total_amount",
      "tolerance_type": "ABSOLUTE",
      "tolerance_value": 0.01,
      "within_tolerance_count": 975,
      "outside_tolerance_count": 5,
      "passed": false
    }
  ],
  "result": "UNMATCHED",
  "rows_compared": 1005,
  "rows_matched": 975,
  "rows_unmatched": 30
}

The join_stats.unmatched_by_source field is a Record<string, number> that scales to N data sources, rather than a two-source left_only/right_only model.

`run.comparison.errored`

{
  "run_id": "uuid",
  "job_id": "uuid",
  "tenant_id": "uuid",
  "error": {
    "code": "COMPARISON_FAILED",
    "message": "DuckDB out of memory during full outer join"
  }
}

`run.stage.started`

{
  "run_id": "uuid",
  "job_id": "uuid",
  "tenant_id": "uuid",
  "stage_name": "balance-check"
}

`run.stage.completed`

Each stage runs one comparison across all sources for that stage’s dimensions and tolerances. The stage is the unit of pass/fail.

{
  "run_id": "uuid",
  "job_id": "uuid",
  "tenant_id": "uuid",
  "stage_name": "balance-check",
  "result": "MATCHED | UNMATCHED",
  "tolerances": [
    {
      "measure_name": "total_amount",
      "tolerance_type": "ABSOLUTE",
      "tolerance_value": 0.01,
      "within_tolerance_count": 975,
      "outside_tolerance_count": 5,
      "passed": false
    }
  ]
}

`run.stage.errored`

{
  "run_id": "uuid",
  "job_id": "uuid",
  "tenant_id": "uuid",
  "stage_name": "balance-check",
  "error": {
    "code": "COMPARISON_FAILED",
    "message": "Stage failed during comparison phase"
  }
}

`run.rows_processed`

Progress event emitted during execution. Unchanged from the previous design.

{
  "run_id": "uuid",
  "job_id": "uuid",
  "stage_name": "balance-check",
  "row_count": 5000,
  "mismatch_count": 12
}

Error Code Taxonomy

When a workflow fails, the workflow exit handler inspects the failure metadata and maps it to a well-known error code. These codes appear in the error.code field of run.errored and run.finalised payloads.

Error Code	Failure Reason	Description
`TIMED_OUT`	Workflow exceeded `activeDeadlineSeconds`	The workflow ran longer than the configured deadline (default 3600s). The failure message contains `activeDeadlineSeconds` or `exceeded deadline`.
`WORKER_OOM`	Pod OOMKilled or unexpected termination	A worker pod was killed by the kernel OOM killer or terminated unexpectedly. The failure message contains `OOMKilled` or `memory`.
`QUERY_FAILED`	Extractor step failed	An extractor step failed, typically due to a SQL query error or connection failure. Detected when the failed step name starts with `extract-`.
`QUOTA_EXCEEDED_STORAGE`	Artifact storage quota exhausted	The run failed because its tenant’s artifact storage was full. Core-API upgrades an otherwise-generic failure to this code when the underlying error carries the storage-quota signal — see Limits & Quotas.
`UNKNOWN`	Any other failure	The failure reason could not be classified into a known category. This is the default fallback.

The exit handler reads the workflow failure metadata and status to classify the failure, and Core-API may upgrade the code (for example to QUOTA_EXCEEDED_STORAGE) when finalising. The mapping is deterministic — each failure maps to exactly one code.

Cancellation State Machine

The cancellation flow uses a two-phase approach to handle the race condition between a user’s cancel request and the Argo workflow’s natural completion.

User requests cancellation via the API.
Core-API sets cancel_requested = true on the run record (without changing status).
Core-API emits run.cancel_requested (audit trail only).
If the run is QUEUED: Core-API immediately sets status to CANCELLED and emits run.cancelled. If argo_workflow_name is set, Core-API also calls Platform-Agent to clean up the workflow.
If the run is RUNNING: Core-API calls Platform-Agent to stop the Argo workflow. The final status is determined later by RunCompletionController when the Argo completion event arrives.
When the Argo event arrives, RunCompletionController checks the cancel_requested flag:
- cancel_requested = true + run.errored → workflow was stopped by the cancel. Final status: CANCELLED. Emits run.cancelled. Does NOT emit run.finalised.
- cancel_requested = true + run.completed → workflow finished before cancel took effect (stale cancel). Final status: COMPLETED. Emits run.finalised normally.
- cancel_requested = false → normal processing. Emits run.finalised.

Subscribing to these events

You receive these events by creating a webhook subscription that selects the event types you care about. See Using Webhooks for the full setup; the patterns below are the common choices for run-lifecycle events.

One notification per run (recommended). Subscribe to reconciliation.run.finalised only. It is DB-backed and authoritative, fires exactly once per run, and is not emitted for cancelled runs — so you get a single, consistent “this run is done” signal.
Outcome with the raw infrastructure timing. Subscribe to reconciliation.run.completed and reconciliation.run.errored if you want the result the instant the workflow finishes. These can arrive slightly before the run’s status is fully persisted; for state you’ll read back from the API, prefer run.finalised.
Progress / live updates. Subscribe to the extraction, comparison, and stage events (e.g. reconciliation.run.stage.completed) to drive a progress UI or per-stage alerting.
Cancellations. Subscribe to reconciliation.run.cancelled to act on cancelled runs (they don’t emit run.finalised).
Everything. Subscribe with ["*"] to receive the full lifecycle, then filter in your handler on the CloudEvent type.

Deliveries are at-least-once and not ordered — deduplicate on the webhook-id header and reconcile state from the payload rather than arrival order. See Building a reliable consumer.

Reconciliation Run Lifecycle Events

Key Concepts

Infrastructure vs Platform Events

run.finalised — The Authoritative Completion Event

Errored vs Unmatched

RunResult

Event Flow

Cancellation Flow

Event Type Reference

Run Lifecycle Events

Extraction Lifecycle Events

Comparison Lifecycle Events

Stage Lifecycle Events

Payload Schemas

run.triggered

run.queued

run.queue_errored

run.completed

run.errored

run.finalised

run.cancel_requested

run.cancelled

run.extraction.started

run.extraction.completed

run.extraction.errored

run.comparison.started

run.comparison.completed

run.comparison.errored

run.stage.started

run.stage.completed

run.stage.errored

run.rows_processed