Change feeds

A change feed tells downstream systems what changed in your published data, as business events such as "company address changed". Consumers do not have to compare snapshots themselves. You can receive events by webhook, in Kafka, or by polling the API.

Define change events

When an asset uses merge(..., track_changes=True), Synchronoid compares each new snapshot with the current published records. You describe which differences matter and how to name them.

@sy.asset(layer="published", deps=[company_profiles])
def companies_current(ctx, company_profiles):
    return company_profiles.merge(
        key="org_id",
        track_changes=True,
        events={
            "company.address_changed": sy.OnChange("address"),
            "company.renamed": sy.OnChange("name"),
            "company.created": sy.OnCreate(),
            "company.removed": sy.OnDelete(),
        },
    )

Each event contains the record key, the fields before and after, the provenance of the new value and a unique id. Fields that change in the same run are grouped into one event per name. If an asset has no events map, Synchronoid emits the generic types record.created, record.changed and record.deleted.

{
  "id": "evt_01J9ZQ4K7M",
  "type": "company.address_changed",
  "asset": "companies_current",
  "key": {"org_id": "c-104392"},
  "before": {"address": "12 Harbour Street, Leeds"},
  "after": {"address": "48 Mill Lane, Leeds"},
  "provenance": {"source": "company_registry", "run_id": "run_8f3a21", "fetched_at": "2026-10-28T02:03:11Z"},
  "occurred_at": "2026-10-28T02:04:40Z"
}

Subscribe with a webhook

Create a subscription in the control room under Change feeds, or with the CLI. Synchronoid sends an HTTP POST with a JSON body to your URL.

$ syn feeds create crm-sync \
    --asset companies_current \
    --webhook https://example.com/hooks/synchronoid
Signing secret: whsec_4d1f...  (shown once)

Every request has the headers Synchronoid-Signature (t=timestamp,v1=hex digest) and Synchronoid-Event-Id. The digest is an HMAC-SHA256 of timestamp + "." + raw body with your signing secret. Always verify it against the raw bytes, before parsing the JSON.

import hmac, hashlib, time

def verify(secret: str, header: str, body: bytes, tolerance=300) -> bool:
    parts = dict(p.split("=", 1) for p in header.split(","))
    ts, sig = parts["t"], parts["v1"]
    if abs(time.time() - int(ts)) > tolerance:
        return False                      # too old: possible replay
    expected = hmac.new(secret.encode(), f"{ts}.".encode() + body,
                        hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, sig)

Reply with any 2xx status within 10 seconds. Other responses and timeouts are retried with exponential backoff for up to 24 hours. After that, the event stays available in the pull API.

Subscribe with Kafka

Synchronoid can publish events to a topic in your Kafka cluster. Add the cluster as a sink and choose the topic. Events are keyed by record key, so all changes to one record stay in order within a partition.

feeds:
  crm-sync:
    events: ["company.*"]
    kafka:
      brokers: ["kafka-1.internal:9093"]
      topic: "synchronoid.company-changes"
      sasl_password: ${secret:KAFKA_PASSWORD}

See the Kafka integration for network setup.

Pull with cursors

If you cannot receive requests, poll the API. Each response contains a next_cursor. Store it after you process a page and send it in the next call.

$ curl -s -H "Authorization: Bearer $SYN_TOKEN" \
    "https://api.synchronoid.com/v1/feeds/crm-sync/events?cursor=cur_7Hq2&limit=100"

Cursors do not expire for 30 days. If a cursor is older, start from ?since=2026-10-01T00:00:00Z. Details are in the REST API reference.

Delivery guarantees

Because delivery is at least once, a consumer can see the same event twice. Use the event id to ignore duplicates, and make your handler idempotent.