Incremental Sync and Change Data Capture for Royalty Data

Reloading a Digital Service Provider’s entire historical catalog of stream reports on every run is rarely viable once territory-level granularity and multi-year retroactive adjustments enter the picture: a single monthly Spotify or Apple Music export can carry tens of millions of rows, and reprocessing all of them to catch the handful that changed wastes compute, delays payout close, and risks double-counting revenue that was already settled. Incremental synchronization solves this by tracking exactly which records are new, updated, or retracted since the last successful run, feeding only that delta into the Data Ingestion & Streaming Sync Pipelines framework that ultimately reconciles royalties. Getting the mechanics right — durable cursors, log-based capture, and deterministic merge semantics — separates a pipeline that can be trusted with financial statements from one that quietly drifts out of balance.

Full Reloads vs. Incremental Extraction

A full reload re-fetches and re-transforms every row a DSP has ever reported, then replaces the downstream table wholesale. It is simple and self-correcting, but its cost grows linearly with catalog size while the useful signal — what actually changed — usually shrinks over time as older reporting periods stabilize. Incremental extraction inverts that relationship: each run asks the source “what changed since checkpoint X” and applies only that answer. The checkpoint, commonly called a high-watermark cursor, is the load-bearing piece of state in the whole architecture. It might be a reporting_period string (2026-05), a monotonically increasing offset or sequence ID, an opaque etag returned by the DSP’s reporting endpoint, or an updated_at timestamp watermark. Each has different failure characteristics — a timestamp cursor is vulnerable to clock skew and boundary duplicates, while a sequence ID is only as trustworthy as the source’s guarantee that IDs are strictly ordered and never reused.

Change Data Capture Patterns

Change data capture (CDC) is the general term for detecting and propagating row-level changes rather than diffing entire tables. Three patterns cover nearly every DSP integration encountered in royalty ETL work. updated_at-based CDC queries the source for rows where a mutation timestamp exceeds the last cursor value; it is the easiest pattern to implement against a REST reporting API but depends entirely on the DSP populating that field correctly and monotonically. Log-based CDC reads a durable change stream — a database write-ahead log, a Kafka topic, or a DSP-provided delta-export feed — and replays inserts, updates, and deletes in the order they were committed; it is strictly more accurate than timestamp polling but requires the source to expose that log at all, which most consumer-facing DSP APIs do not. Tombstone records close the third gap: when a DSP retracts a previously reported stream (a fraud takedown, a territory reassignment, a duplicate submission correction), it emits a tombstone rather than simply omitting the row, so the reconciliation layer can explicitly null out or reverse the prior financial entry instead of leaving a stale, unretracted figure in the ledger.

Python ETL Architecture

Cursor state belongs in a transactional store, not a flat file or an in-memory variable — a crash between fetching a delta and committing the cursor advance must never leave the two out of sync. A small Postgres table or a DuckDB table backed by durable storage both work well; the requirement is that the cursor write and the data write commit atomically, or that the merge itself is idempotent enough to tolerate a replayed batch. For payload shapes, Pydantic v2 models validate each incoming row against the expected royalty schema (isrc, territory_code, reporting_period, streams, revenue_usd) before it reaches the merge step, and Polars or DuckDB perform the actual diff-and-merge because both handle anti-joins and window functions over multi-million-row deltas without materializing the full historical table. Streaming ingestion, as covered under Async Batch Processing for High-Volume Streams, still matters here: a restatement event from a DSP can retroactively touch a much larger row count than a routine daily delta, so the merge path needs the same chunked, memory-bounded execution as the primary ingestion path, not a separate unbounded code path reserved for “small” incremental jobs.

Step 1: Persist a Durable High-Watermark Cursor

The cursor must be readable and writable inside the same transaction as the data it gates, so a partial failure cannot advance one without the other.

from datetime import datetime
from pydantic import BaseModel, Field

class SyncCursor(BaseModel):
    source_id: str
    reporting_period: str = Field(pattern=r"^\d{4}-\d{2}$")
    last_offset: int = Field(ge=0)
    etag: str | None = None
    committed_at: datetime

def advance_cursor(conn, cursor: SyncCursor) -> None:
    with conn.transaction():
        conn.execute(
            """
            UPDATE sync_cursors
            SET last_offset = %s, etag = %s, committed_at = %s
            WHERE source_id = %s AND reporting_period = %s
            """,
            (cursor.last_offset, cursor.etag, cursor.committed_at,
             cursor.source_id, cursor.reporting_period),
        )

Step 2: Detect Changes via updated_at and Log-Based CDC

Poll the DSP endpoint using the persisted cursor, and treat records flagged as retracted as tombstones rather than silent drops.

import httpx

async def fetch_delta(client: httpx.AsyncClient, cursor: SyncCursor) -> list[dict]:
    resp = await client.get(
        "/reporting/stream-events",
        params={"updated_since": cursor.committed_at.isoformat(),
                "offset": cursor.last_offset},
    )
    resp.raise_for_status()
    payload = resp.json()
    for row in payload["records"]:
        row["is_tombstone"] = row.get("event_type") == "RETRACTED"
    return payload["records"]

Step 3: Idempotent UPSERT/Merge into the Lake

A natural key of isrc + territory_code + reporting_period + payee_id drives the merge; tombstones reverse the prior amount instead of deleting history, preserving the audit trail.

import duckdb

def merge_delta(con: duckdb.DuckDBPyConnection, batch_id: str) -> None:
    con.execute("""
        MERGE INTO royalty_facts AS target
        USING staged_delta AS src
        ON target.isrc = src.isrc
           AND target.territory_code = src.territory_code
           AND target.reporting_period = src.reporting_period
           AND target.payee_id = src.payee_id
        WHEN MATCHED AND src.is_tombstone THEN
            UPDATE SET revenue_usd = 0, streams = 0, status = 'RETRACTED', batch_id = src.batch_id
        WHEN MATCHED THEN
            UPDATE SET revenue_usd = src.revenue_usd, streams = src.streams, batch_id = src.batch_id
        WHEN NOT MATCHED AND NOT src.is_tombstone THEN
            INSERT (isrc, territory_code, reporting_period, payee_id, revenue_usd, streams, batch_id)
            VALUES (src.isrc, src.territory_code, src.reporting_period, src.payee_id,
                     src.revenue_usd, src.streams, src.batch_id)
    """)

Reconciliation Gates & Validation

Before a delta is eligible for merge, every row passes schema validation against the Pydantic model, and the batch as a whole is checked for ISRC-to-ISWC resolution against the catalog registry — an unresolvable ISRC routes to an exception queue rather than blocking the rest of the batch. A split-sum assertion confirms that publishing and master shares for each affected work still total 100% after the merge, catching cases where a restatement silently altered a split without a corresponding correction record. Records that fail any gate are quarantined with the original payload, the failing rule, and the batch_id, so royalty managers can adjudicate without re-polling the DSP.

Failure Modes & Rollback

A circuit breaker should trip when a delta is anomalously large relative to the historical average for that source and reporting period — a common symptom of a DSP-side backfill or a cursor that reset to zero — halting the merge until an operator confirms the volume is expected. Because the merge is idempotent on the natural key, replaying a batch after a crash never double-applies it, and a versioned parquet snapshot of the affected partitions taken immediately before each merge gives a fast rollback path if a bad batch slips past the gates.

Security & Audit Trail

Cursor tables and staged deltas are encrypted at rest, and write access is restricted through RBAC to the ingestion service identity rather than individual engineers. Every merge operation appends an immutable log entry — batch_id, row counts by operation type, and a SHA-256 hash of the staged delta — so that a later audit can reconstruct exactly which rows changed a given royalty statement and why.

Related

Incremental extraction depends on the polling cadence and pagination discipline described in DSP API Polling Strategies, scales through the same execution model as Async Batch Processing for High-Volume Streams, and lands its merged output in the partitioned tables described under Data Lake Architecture for Streaming Metrics. For a deeper treatment of cursor design trade-offs specifically for DSP reporting endpoints, see Cursor-Based Incremental Sync for DSP Reports, which extends the durable-cursor pattern introduced above with a comparison of timestamp, sequence-ID, etag, and reporting-period strategies. Together these patterns keep the broader Data Ingestion & Streaming Sync Pipelines framework financially accurate without re-processing catalog history on every run.