Skip to content

Sync-cursor and historic orchestration

What this is

In DataProduct mode, production single-instance validation combines:

  1. Historic partitions — range-filtered workers validate the existing estate (per DataProduct version change or manual re-run).
  2. Steady-state sync-cursor — a DMS modeling trigger signals the handler; it pulls forward-only changes via instances.sync() since the last saved cursor.

These paths are not strictly sequential end-to-end. In one orchestrator run, after timestamp collection / partition discovery, the orchestrator creates the sync trigger, fires edge baseline jobs, and triggers partition workers without waiting for each to finish.

Local run_validation() and trigger-based validation_type: instance are useful for development. Most estates run sync-cursor for steady state after historic orchestration has been started. When to iterate locally versus enqueue a full historic scan (including force=True without a version bump): Confirm that rules work.

This page is instance SHACL sync + historic orchestration. Steady-state edge existence uses separate deploy-time DMS modeling triggers ({dp_wf}-edge-existence, node + edge per rule) — not the instance sync trigger from Step 3. Orchestrator Step 3b only fires a one-time edge_existence_baseline estate scan that shares the historic job_run_id; it does not create edge triggers. See Edge existence.

User mental model

DataProduct version change (or manual deploy_validation_pipeline)
        │
        ▼
Historic orchestrator (validation_type: orchestrator)
        │  Timestamp collection / partition discovery (Steps 1–2)
        │  Step 3:   create DMS sync trigger (orchestrator-only — deploy does not)
        │  Step 3b:  edge_existence_baseline jobs (async, wait=False)
        │  Step 4–5: trigger N partition workers (async, wait=False) — parallel with 3b
        ▼
Partition workers (validation_type: partitioned) — cursor-resumable, monitored by schedule
        │
        ▼
Sync-cursor handler (validation_type: instance_sync_cursor)
        │  DMS trigger (no_backfill) → handler instances.sync() forward-only
        ▼
Records API (DataQualityValidationRecord)

Orchestrator completion does not mean edge baseline finished. Step 3b baseline jobs are fire-and-forget; the orchestrator does not monitor or re-trigger them.

All paths in a historic run share one job_run_id (canonical format job_{view_external_id}_{epoch_seconds}) so dashboards and triage can correlate partitions, sync-cursor, and edge baseline.

Orchestrator-only sync triggers

Deploy does not create or update DMS instance sync triggers. Only the historic orchestrator creates them (Step 3), after timestamp collection / partition discovery, before partition workers are triggered, with:

  • Correct job_run_id and initial_sync_cutoff (historic upper bound from collection)
  • Full trigger input (ruleset refs, records config, instance spaces)
  • Sync lock initialization

If you deploy workflows but never run historic orchestration for a view, no sync trigger exists and incremental validation will not start.

See Migration 0.4.1 for the regression that briefly re-created triggers at deploy time.

Sync trigger: no_backfill + initial_sync_cutoff

The DMS sync trigger uses sync_mode="no_backfill" so trigger payloads contain only modeling changes from trigger creation onward — not the full historic backlog.

The sync-cursor handler also calls instances.sync() with no_backfill. On a fresh cursor (including after orchestrator cursor reset), that opens a tail cursor at now. It does not automatically drain instances updated during historic partition work.

initial_sync_cutoff is set by the orchestrator from the max lastUpdatedTime (or partition max) seen during timestamp collection. On the first sync-cursor run when no cursor exists, the handler adds lastUpdatedTime >= initial_sync_cutoff so instances already covered by historic partitions are skipped. After the first run saves a cursor, subsequent runs continue forward-only from that cursor without re-applying the cutoff filter.

The cutoff aligns steady-state sync with historic bounds; it is not a guarantee of gap-free coverage if timestamp collection was partial or instances changed during historic runs.

Edge views (used_for: edge)

When a Data Product view is used_for: edge, historic partitions and the sync-cursor path query the edges DMS channel: HasData on that edge view plus instance-space membership (In(["edge","space"], …)). This is ordinary SHACL on the edge instance (property constraints on the edge). It is not typed-edge cardinality (dqs:edgeExistenceConstraint), which stays node-focus and uses a separate {dp_wf}-edge-existence workflow with Equals(["edge","type"], …) filters — see Edge existence.

Validation records for edge-view SHACL use the edge URI as focusNode and omit focusNodeInstance (that field is a node direct relation). See Records output.

Cursor reset on historic handoff

When the orchestrator creates the sync trigger, it resets the sync cursor so a stale cursor from an earlier deploy-time trigger does not bypass initial_sync_cutoff.

Symptom before fix: sync-cursor drained a large backlog with steady "Sync lock held" noise because an old cursor position was preserved across handoff.

Expected after handoff: once the tail cursor is established, sync-cursor runs should be incremental — not a full estate replay through the sync path.

Sync trigger creation failures

Sync trigger upsert can fail (query timeout, missing secrets, permissions). The orchestrator logs a WARNING, records sync_trigger_error in orchestration state, and continues with partition triggering. Historic validation may complete successfully even when no sync trigger exists — fix incremental sync manually (deploy_validation_pipeline() or re-run historic orchestration).

job_run_id contract

Rule Detail
Canonical format job_{view_external_id}_{epoch_seconds} (view external ID lowercased)
Shared across Historic partitions, sync-cursor, edge baseline Step 3b
Orchestrator / queue build_historic_job_run_id() — same id from deploy_validation_pipeline() and HistoricJobQueue
Sync-cursor Resolved from sync cursor state or explicit payload — required (no silent timestamp fallback)
Deploy trigger input No deploy-time job_run_id baked into sync trigger — handlers resolve from cursor / orchestrator handoff

Legacy hq_* ids from older queue paths will not match new job_* records. Re-run historic after upgrade if dashboard filters use jobRunId.

See Invoke → job_run_id and Troubleshooting.

Historic payload scope (HistoricJobQueue)

HistoricJobQueue rebuilds orchestrator input from the live DataProduct API at job start (instanceSpaces.read on the published DataProduct view) — the same path as deploy_validation_pipeline(). Both paths then overlay stored ViewConfigState from DMS.

When view YAML declared instance_space / instance_spaces, that narrower list is persisted at deploy and wins over DataProduct instanceSpaces.read for historic scans and the orchestrator sync-trigger filter. When no override is stored, the live DataProduct value remains in effect.

Compare orchestrator logs for instance_spaces against both the DataProduct view definition and the ViewConfigState node ({view_space}__{view_external_id}) when counts diverge.

Historic job queue enqueue (0.4.4+)

Besides deploy_validation_pipeline() with historic_mode="orchestrator", operators can enqueue historic jobs without firing parallel partition workers immediately:

from cognite_data_quality import enqueue_historic_validation

enqueue_historic_validation(
    client,
    data_quality_space="sp_data_quality",
    trigger_queue_manager=True,
)

The dq-historic-queue-manager workflow processes jobs one at a time. Each job runs the full orchestrator path (timestamp collection, sync trigger, partitions) for one view.

Enqueue complements data_product_sync: sync enqueues only when it deploys workflows for a new version. If workflows already exist with the correct [dp:{version}] tag, sync skips without enqueueing — call enqueue_historic_validation() after deploy_validation_infrastructure() in CI or brownfield estates.

Idempotency: enqueue skips jobs already completed for the same DataProduct version; active jobs for the same view are superseded. Pass force=True to re-queue a completed historic run without a version bump, including views with historic_reprocess_on_update: false. Already queued or running jobs are still skipped. If batch retrieve of existing queue nodes fails, the enqueue result includes warnings — duplicate jobs are possible until the warning is resolved.

data_product_sync payload force: true redeploys workflows only; it does not re-queue completed historic jobs for the same version.

Operator checklist

After historic orchestration for a view:

  1. Sync trigger exists — dq-shacl-trigger-{view} (or consolidated DP-scoped id) is active in Workflows (check orchestrator warnings if missing).
  2. Cursor state — sync cursor node in dataQuality space shows DMS properties syncCursor and processedCount (handler logs may use snake_case processed_count).
  3. Logs — Dimension lookup: N named shapes, M property paths when rules include dq:dimension (Dimensions).
  4. Records — new DataQualityValidationRecord rows with expected jobRunId and ruleSetVersion (Records output).
  5. Edge baseline — if rules include dqs:edgeExistenceConstraint, Step 3b jobs were triggered (completion is separate — see Edge existence).

Previous section

Next section