Source Freshness Monitoring

Detect stale data sources before they degrade match quality.

Overview

Identity resolution depends on up-to-date source data. If a CRM feed stops updating, you risk matching against stale records or missing new entities entirely. Kanoniv's freshness monitoring lets you set maximum age thresholds per source and surface warnings or hard failures before reconciliation runs.

Cloud Feature

Freshness checks require the Cloud tier.

Spec Configuration

Add a freshness block to any source definition:

sources:
  - name: crm
    adapter: postgres
    location: public.contacts
    primary_key: contact_id
    freshness:
      max_age: "24h"
      warn_after: "12h"
    attributes:
      email: email_address
      name: full_name

Fields

Field	Type	Required	Description
max_age	string	Yes	Maximum acceptable age before the source is considered stale
warn_after	string	No	Emit a warning when source age exceeds this threshold

Duration Format

Durations are specified as a number followed by a unit suffix:

Example	Duration
1h	1 hour
12h	12 hours
24h	24 hours
7d	7 days
30d	30 days

How Freshness Checks Work

At the start of each reconciliation pipeline run, the engine queries the most recent data timestamp for each source:

┌───────────────┐     ┌───────────────────┐     ┌─────────────┐
│ Pipeline Start│────▶│ Query MAX(        │────▶│ Compare to  │
│               │     │   updated_at)     │     │ max_age /   │
│               │     │ per source        │     │ warn_after  │
└───────────────┘     └───────────────────┘     └──────┬──────┘
                                                       │
                                         ┌─────────────┼─────────────┐
                                         ▼             ▼             ▼
                                    ┌─────────┐  ┌──────────┐  ┌─────────┐
                                    │  Fresh  │  │  Warning │  │  Stale  │
                                    │ (pass)  │  │ (log +   │  │ (fail   │
                                    │         │  │  continue)│  │  run)   │
                                    └─────────┘  └──────────┘  └─────────┘

1. Fresh. Source age is below warn_after (or max_age if no warning threshold). Pipeline proceeds normally.
2. Warning. Source age exceeds warn_after but is below max_age. A warning is logged and the pipeline continues.
3. Stale. Source age exceeds max_age. The source is flagged as stale in the run health report. If governance requires freshness, the pipeline is blocked.

Governance Integration

Freshness checks can be made mandatory via the governance block:

governance:
  require_freshness: true

When require_freshness is true, any source that exceeds its max_age threshold will cause the pipeline to abort with a clear error message identifying the stale source.

Without require_freshness, stale sources are flagged in the health report but reconciliation continues.

Full Example

api_version: kanoniv/v2
identity_version: customer_v4

entity:
  name: customer

sources:
  - name: crm
    adapter: postgres
    location: public.contacts
    primary_key: contact_id
    freshness:
      max_age: "24h"
      warn_after: "12h"
    attributes:
      email: email_address
      name: full_name

  - name: billing
    adapter: snowflake
    location: analytics.billing.accounts
    primary_key: account_id
    freshness:
      max_age: "48h"
      warn_after: "24h"
    attributes:
      email: billing_email
      name: account_name

  - name: support
    adapter: postgres
    location: support.tickets
    primary_key: ticket_id
    freshness:
      max_age: "7d"
    attributes:
      email: requester_email

rules:
  - type: exact
    name: email_match
    field: email
    weight: 1.0

decision:
  thresholds:
    match: 0.9

governance:
  require_freshness: true

Checking Source Health

Python

import httpx

# Check health for the latest batch run
resp = httpx.get(
    "https://api.kanoniv.com/v1/batch-runs/latest",
    headers={"X-API-Key": "kn_..."},
)
run = resp.json()

for source_health in run["health"]["sources"]:
    status = source_health["freshness_status"]  # "fresh", "warning", "stale"
    age = source_health["age_seconds"]
    print(f"{source_health['name']}: {status} (age: {age}s)")

curl

# Get the latest batch run with health info
curl "https://api.kanoniv.com/v1/batch-runs/latest" \
  -H "X-API-Key: kn_..."

# Response includes:
# {
#   "health": {
#     "sources": [
#       {
#         "name": "crm",
#         "freshness_status": "fresh",
#         "age_seconds": 3600,
#         "max_age_seconds": 86400,
#         "warn_after_seconds": 43200
#       }
#     ]
#   }
# }

Alerting

Freshness events are published to Redis and included in the audit stream. If you have SIEM Integration configured, stale source events are automatically forwarded to your monitoring tools.

Event types:

Event	Description
source.freshness.warning	Source age exceeded warn_after
source.freshness.stale	Source age exceeded max_age
source.freshness.blocked	Pipeline blocked due to stale source (governance enforced)

Next Steps

• Spec Governance: Enforce freshness checks across all sources
• SIEM Integration: Route freshness alerts to monitoring tools
• Audit Logs: Track freshness events in the audit trail