Skip to content

Stream retention guard

What this is

CDF Records streams delete data after settings.lifecycle.dataDeletedAfter. Some validation surfaces only write a complete run on DataProduct/RuleSet version change (instance historic, edge existence baseline) or only rewrite the current SHACL window (timeseries). The stream retention guard re-triggers those complete runs before the last qualifying jobRunId expires.

When to use it

Leave the guard on (default) when operators query Records for a full historic snapshot, an estate-wide edge baseline, or a complete current-window timeseries run — and the stream is not unlimited.

Turn a surface off with ensure_complete_run_in_stream: false on the view YAML, timeseries YAML, or (for edge rules) the stored EdgeExistenceRuleState flag.

Uniqueness is not in this guard: the dedicated uniqueness cron already posts a complete aggregate pass. RAW, sync-cursor instance runs, and incremental edge_existence_event records are also out of scope.

User mental model

  1. Deploy writes stored payloads and ensure_complete_run_in_stream (default true) onto DMS state nodes.
  2. dq-stream-retention-guard runs once per night in 01:00–04:59 UTC, with the minute/hour jittered by sha256(COGNITE_PROJECT) % 240 so tenants on a shared cluster are not all refreshing at :00.
  3. Each cycle: GET /streams/{id} for dataDeletedAfter (never YAML), then Records uniqueValues on jobRunId, then refresh surfaces that would expire within stream_retention_lead_time (default P1D).
  4. Instance historic is re-queued on the same DataProduct/RuleSet version (hq_{view}_retention). Timeseries and edge baseline are functions.call with a fresh job_run_id.

Runtime behavior

What is refreshed

Surface Refresh Notes
Instance historic HistoricJobQueue hq_{view}_retention Same version must not block. historic_reprocess_on_update: false does not apply. Skip only if that view already has a queued or running HQ job.
Edge existence baseline edge_existence_baseline Modeling triggers are incremental (edge existence). Completeness requires a finished baseline (jobRunId plus pagination cleared), not recent event records.
Timeseries Normal-mode validation_type: timeseries One complete current window for the config. Does not set backfill.enabled. A P7D stream cannot keep a 30-day backfill corpus.

Detection

Retention period is always GET /streams/{id} → settings.lifecycle.dataDeletedAfter. Missing value → unlimited → skip. Streams API errors → print a warning and skip that cycle.

Latest run start comes from Records aggregate uniqueValues on jobRunId (filter ruleSetId + ruleSetVersion, size 2000), then prefix-filtered per surface. Instance historic completeness uses job_{view}_{epoch} from the historic orchestrator (build_historic_job_run_id). hq_* strings are HistoricJobQueue node ids and are not Records jobRunIds. Uniqueness ids (job_{view}_uniqueness_{epoch}) are excluded from instance historic matching. See Records output.

uniqueValues is typically count top-N, not recency. Distinct jobRunIds that share the same ruleSetId/ruleSetVersion (uniqueness, incidental windows, many historic ids) compete for those 2000 buckets. A recent low-count historic id can fall out of the window and look like “no qualifying run.” v1 does not page beyond this cap.

Refresh when now >= run_start + retention - lead_time, or when there is no qualifying complete run. Timeseries and edge baseline also skip when the last retention functions.call is still Running, or when a refresh was triggered within lead_time. Edge completeness prefers a finished edge_{rule}_{epoch} over a newer sibling job_{view}_{epoch} that has no pagination row for that rule. An existing pagination node with a cleared cursor (including processed_count == 0) is complete; a missing node is not.

Schedule

  • Night window: 01:00–04:59 UTC (240 minute slots).
  • Default cron: {minute} {hour} * * * from sha256(project) % 240.
  • YAML stream_retention_guard_cron wins when set.
  • Missing project name → 14 2 * * * (not :00).
  • Deploy prints the resolved expression and persists it on DataQualitySettings.

This is not a sleep inside the function. CDF workflow crons are fixed at deploy.

Settings (YAML)

# settings.yaml
stream_retention_lead_time: P1D          # optional; default P1D
stream_retention_guard_cron: "30 3 * * *"  # optional override

# views/my_view.yaml or timeseries/my_sensors.yaml
ensure_complete_run_in_stream: true   # default; set false to opt out

Timeseries backfill

v1 re-triggers normal mode only. Older backfill windows still expire with the stream. Full backfill replay is out of scope.

Limitations

  • Does not make history longer than stream retention (records streams).
  • Writing the same results to a second Records stream is not available yet.
  • Size stream_retention_lead_time for HistoricJobQueue wait p99 plus historic run p99, or a same-version refresh can miss the expiry window.
  • Multi-DataProduct views share one ViewConfigState node ({view_space}__{view_external_id}); last deploy/sync payload wins.
  • Records uniqueValues on jobRunId uses size: 2000 (count top-N). Very large numbers of distinct ids for one RuleSet can hide a recent historic run.