Cursor-Based Incremental Sync for DSP Reports

Choosing what a cursor is built from — a timestamp, a sequence ID, an etag, or a reporting-period string — determines whether a DSP integration silently loses rows during a restatement or reliably resumes exactly where it left off. This decision sits one level below the general durable-cursor pattern described under Incremental Sync and Change Data Capture for Royalty Data, and it is the part of the pipeline where an implementation detail — how the cursor is persisted, and what happens when a later-arriving row lands behind it — has a direct, measurable effect on payout accuracy. Late-arriving corrections and DSP-side restatements are not edge cases in royalty reporting; they are a routine monthly occurrence, and a cursor design that cannot represent “this reporting period changed after I already synced it” will eventually understate or overstate an artist’s statement.

Cursor Strategy Trade-offs

Cursor Type	Ordering Guarantee	Handles Restatements?	Failure Mode	Royalty Impact
Timestamp (`updated_at`)	Weak — depends on source clock and write-time accuracy	Only if the DSP bumps `updated_at` on correction	Clock skew or bulk backfills can duplicate or skip rows at the boundary	Boundary duplication double-counts a stream; skew-induced skips understate revenue for the affected window
Sequence ID / offset	Strong, if the source guarantees monotonic, non-reused IDs	No — a correction to an already-synced row has no new ID	Gaps from deleted or renumbered IDs are invisible to the consumer	Cannot detect retroactive corrections at all; must be paired with a separate restatement feed or it silently ships stale figures
Etag / version token	Strong per-resource, weak across resources	Yes — a changed etag on a resource signals it must be re-fetched	Etag churn from unrelated metadata changes triggers unnecessary re-syncs	Safest against silent staleness, but over-triggering wastes compute and can flap `reconciliation_status` if not debounced
Reporting-period string (`YYYY-MM`)	Coarse — no intra-period ordering	Yes — reprocess the whole period when the DSP reopens it	Cannot tell which rows within a reopened period actually changed without a full diff	Guarantees correctness for the whole period at the cost of re-validating unchanged rows every time the period reopens

In practice, DSP integrations that must be audit-defensible pair a reporting-period cursor for restatement detection with a per-row etag or content hash for change detection within that period — the period cursor answers “do I need to look at May again,” and the etag answers “which rows in May actually changed.”

Prerequisites & Assumptions

This implementation assumes Python 3.11+, httpx>=0.27 for async HTTP, pydantic>=2.6 for validation, and a Postgres or DuckDB-backed cursor table reachable from the ingestion worker. It assumes the DSP reporting endpoint supports at minimum a reporting_period filter and returns either an etag header or a stable per-row content_hash field; if neither is available, a hash of the row’s own payload must be computed client-side as a substitute. right_share_pct and payee_id are assumed present on every line item, since the merge step needs them to detect a changed split independent of a changed revenue figure.

Step 1: Model the Cursor and Detect Reopened Periods

The cursor tracks the last reporting period synced and a per-period completion flag, so a period that the DSP reopens for correction can be distinguished from one still in progress.

python

from datetime import date
from pydantic import BaseModel, Field

class PeriodCursor(BaseModel):
    source_id: str
    reporting_period: str = Field(pattern=r"^\d{4}-\d{2}$")
    period_closed: bool
    last_synced_at: date

def needs_resync(cursor: PeriodCursor, dsp_period_status: str) -> bool:
    """DSP marks a period 'REOPENED' when it issues corrections after close."""
    return dsp_period_status == "REOPENED" or not cursor.period_closed

Step 2: Fetch and Hash Rows for Change Detection

Within a reporting period flagged for resync, compare each row’s content_hash (or etag) against the value stored from the prior sync to isolate only the rows that actually changed.

python

import hashlib
import httpx

def row_hash(row: dict) -> str:
    key = f"{row['isrc']}|{row['territory_code']}|{row['payee_id']}|{row['revenue_usd']}|{row['right_share_pct']}"
    return hashlib.sha256(key.encode()).hexdigest()

async def fetch_period_rows(client: httpx.AsyncClient, reporting_period: str) -> list[dict]:
    resp = await client.get("/reporting/period", params={"reporting_period": reporting_period})
    resp.raise_for_status()
    rows = resp.json()["records"]
    for row in rows:
        row["content_hash"] = row.get("etag") or row_hash(row)
    return rows

Step 3: Exactly-Once Merge with Transactional Cursor Advance

The merge and the cursor update commit in the same transaction, so a crash between them cannot leave the pipeline believing a period synced when it did not, or vice versa.

python

import duckdb

def merge_and_advance(con: duckdb.DuckDBPyConnection, reporting_period: str,
                       source_id: str, period_closed: bool) -> None:
    con.execute("BEGIN")
    try:
        con.execute("""
            MERGE INTO royalty_facts AS target
            USING staged_period_rows AS src
            ON target.isrc = src.isrc
               AND target.territory_code = src.territory_code
               AND target.payee_id = src.payee_id
               AND target.reporting_period = src.reporting_period
            WHEN MATCHED AND target.content_hash != src.content_hash THEN
                UPDATE SET revenue_usd = src.revenue_usd,
                           right_share_pct = src.right_share_pct,
                           content_hash = src.content_hash
            WHEN NOT MATCHED THEN
                INSERT (isrc, territory_code, payee_id, reporting_period,
                        revenue_usd, right_share_pct, content_hash)
                VALUES (src.isrc, src.territory_code, src.payee_id, src.reporting_period,
                        src.revenue_usd, src.right_share_pct, src.content_hash)
        """)
        con.execute("""
            UPDATE sync_cursors SET period_closed = ?, last_synced_at = CURRENT_DATE
            WHERE source_id = ? AND reporting_period = ?
        """, (period_closed, source_id, reporting_period))
        con.execute("COMMIT")
    except Exception:
        con.execute("ROLLBACK")
        raise

Verification & Validation

After each merge, a split-sum assertion confirms right_share_pct still totals 100% per isrc and territory — a changed-hash row that altered a split without a matching counter-adjustment on another payee’s share fails this check and routes to quarantine rather than posting. A row-count checksum comparing len(staged_period_rows) against the delta reported in the DSP’s own period-status metadata catches silent truncation from pagination bugs. The ISRC resolution rate — the percentage of synced rows that matched an existing catalog entry rather than falling to the unclaimed queue — should be logged per period and alerted on if it drops more than a few points period-over-period, since a sudden drop usually indicates an upstream ISRC formatting change rather than genuinely new, unmatched content.

Edge Cases & Gotchas

A reporting period can be reopened by the DSP weeks or months after it was first marked closed, so period_closed must never be treated as permanent — always re-check the DSP’s own period-status field rather than trusting a local flag indefinitely. Etags are sometimes recycled across unrelated resources by DSPs with careless cache implementations, so pair etag comparison with a payee_id and isrc scope rather than trusting a bare etag match. Territory-code mismatches (UK vs GB) between a reopened period’s corrected rows and the original rows can cause the merge key to miss, inserting a duplicate rather than updating in place — normalize territory codes to ISO 3166-1 alpha-2 before hashing. Finally, watch for NULL right_share_pct values on corrected rows where a DSP’s correction payload omits fields it considers unchanged; treating a NULL as zero rather than “carry forward the prior value” will silently zero out a payee’s share on the next statement.

The exactly-once merge semantics here rely on the same idempotency guarantees covered in Building idempotent ingestion pipelines in Python, and the underlying cursor concepts are introduced more broadly under Incremental Sync and Change Data Capture for Royalty Data.