Implementing Watermark Cursors for EDC Delta Sync in Python

A delta-sync worker keyed only on LastUpdatedDate runs cleanly for weeks, then a scheduled cycle either drops a coordinator’s form save or re-ingests one it already had — and the reconciliation report shows a count that no longer matches the source. That intermittent, hard-to-reproduce symptom is what this page resolves. Clinical data managers, biotech and pharma ETL engineers, and regulatory reviewers hit it whenever a single timestamp column is asked to serve as a sync cursor, because three failure modes collide at the query boundary: many records share an identical LastUpdatedDate written in one transaction, an EDC tenant on site-local time drifts the timestamp across a daylight-saving edge, and the API’s inclusive-versus-exclusive treatment of the boundary is ambiguous. This page is the cursor-mechanics deep-dive inside Incremental Sync and Change Data Capture for EDC Pipelines, which in turn sits within the broader Automated EDC Ingestion & Sync Pipelines discipline. Engineered correctly, the cursor becomes a durable, monotonic checkpoint whose every advance is defensible under 21 CFR Part 11 when an inspector reconstructs exactly how far the pipeline had reconciled the source of record.

Composite Cursor Advance at a Glance #

The worker persists a (watermark_ts, last_seen_id) pair, re-scans a small overlap window, filters the tail with the composite comparison, deduplicates on the ODM natural key, and only commits the new cursor after the load succeeds.

Root Cause: Why a Single-Timestamp Watermark Leaks #

A watermark keyed only on LastUpdatedDate assumes the timestamp is unique and strictly increasing per record. Neither holds in an EDC system, and each broken assumption produces one of the two symptoms.

Equal timestamps cause both loss and duplication. When a coordinator saves a CRF, the EDC writes every ItemData value in that form with a single transaction timestamp, so a page can contain dozens of records sharing an identical LastUpdatedDate. If the cursor stores only that timestamp and the next query filters strictly greater-than, every record equal to the watermark is skipped — silent loss. If instead the query filters greater-than-or-equal to avoid that loss, the boundary records are re-fetched every run — duplication, harmless only if the sink happens to be idempotent. A single timestamp cannot express “I have processed some of the records at this instant but not all of them”; a composite (timestamp, id) cursor can.

Timezone drift breaks monotonicity. If the EDC tenant reports LastUpdatedDate in a site’s local wall-clock time, the value is not monotonic across a daylight-saving fall-back: 01:30 occurs twice, so a record edited in the second 01:30 carries a timestamp that appears earlier than one edited in the first. A watermark that advanced into the first pass will skip the second. Normalizing every timestamp to UTC at ingress is the only durable fix.

Boundary inclusivity is ambiguous per vendor. Some EDC APIs treat a changedSince parameter as inclusive, others as exclusive, and the documentation is frequently silent. Building the cursor to be correct under an inclusive lower bound — and relying on the natural-key dedup to absorb the redundant boundary row — makes the pipeline robust regardless of which convention the vendor actually implements.

Step-by-Step Implementation #

Each step owns a single responsibility and produces a runnable building block. Compose them into one delta-sync worker per study.

1. Persist a composite (watermark_ts, last_seen_id) cursor #

Store both halves of the cursor in a durable, transactional table, always in UTC. The last_seen_id is the tie-breaker that resolves records sharing a timestamp — it must be a stable, monotonic identifier such as the source row’s ODM audit sequence or transaction id.

# ALCOA+ requirement: Consistent — the cursor is the pipeline's reconciliation
# point; both halves are stored in UTC so the checkpoint is unambiguous and the
# advance is strictly monotonic across timezone and DST boundaries.
from datetime import datetime, timezone
import sqlite3


def load_cursor(db: sqlite3.Connection, study_oid: str, site_ref: str) -> tuple[datetime, str]:
    row = db.execute(
        "SELECT watermark_ts, last_seen_id FROM sync_cursor WHERE study=? AND site=?",
        (study_oid, site_ref),
    ).fetchone()
    if row is None:                                  # first run: epoch start, no id
        return datetime(1970, 1, 1, tzinfo=timezone.utc), ""
    ts = datetime.fromisoformat(row[0]).astimezone(timezone.utc)
    return ts, row[1]

2. Re-scan a small overlap window each run #

Subtract a bounded overlap from the stored watermark to build the query’s lower bound, so any edit that landed during the previous run’s in-flight window is re-read. Order the query by (timestamp, id) so the composite comparison in the next step is meaningful and pagination is stable.

# ALCOA+ requirement: Complete — the overlap re-scan guarantees a back-dated or
# boundary edit is never skipped; ordering by (ts, id) makes the delta window
# deterministic and safely resumable mid-page.
from datetime import timedelta
import httpx


def fetch_window(client: httpx.Client, url: str, study_oid: str,
                 watermark_ts: datetime, overlap: timedelta) -> list[dict]:
    lower_bound = (watermark_ts - overlap).astimezone(timezone.utc)
    resp = client.get(url, params={
        "study": study_oid,
        "changedSince": lower_bound.isoformat(),     # inclusive lower bound
        "orderBy": "LastUpdatedDate,AuditSeq",       # deterministic tie-break order
    })
    resp.raise_for_status()
    return resp.json()["records"]

3. Filter the tail with the composite tuple comparison #

Discard everything at or below the exact cursor using Python’s native tuple ordering, which compares the timestamp first and falls back to the id only on a tie. This keeps records strictly after the cursor while correctly retaining the unprocessed remainder of a shared-timestamp batch.

# 21 CFR Part 11 relevance: the composite filter is the control that prevents
# both silent record loss and duplication at equal-timestamp boundaries — the
# defect an inspector reconstructs from a count mismatch.
def strictly_after(records: list[dict], watermark_ts: datetime,
                   last_seen_id: str) -> list[dict]:
    cursor = (watermark_ts, last_seen_id)
    kept = []
    for r in records:
        ts = datetime.fromisoformat(r["LastUpdatedDate"]).astimezone(timezone.utc)
        if (ts, r["AuditSeq"]) > cursor:             # tuple compare: ts then id
            kept.append(r)
    return kept

4. Deduplicate on the ODM natural key and upsert #

The overlap re-scan deliberately re-reads records, so dedup on the natural-key tuple before loading, then apply an idempotent upsert so any residual duplicate is a no-op rather than a second row.

# ALCOA+ requirement: Accurate — dedup on the ODM natural key collapses the
# intentional overlap re-reads, and the upsert makes the load idempotent so a
# replay updates in place instead of inserting a duplicate clinical record.
NATURAL_KEY = ("StudyOID", "SiteRef", "SubjectKey", "EventOID",
               "FormOID", "ItemGroupOID", "ItemOID")


def dedup_and_upsert(db: sqlite3.Connection, records: list[dict]) -> int:
    seen, rows = set(), []
    for r in records:
        key = tuple(r[k] for k in NATURAL_KEY)
        if key in seen:
            continue                                 # collapse overlap re-reads
        seen.add(key)
        rows.append(r)
    db.executemany(
        "INSERT INTO item_data (study, site, subject, event, form, item_group, item, value) "
        "VALUES (:StudyOID,:SiteRef,:SubjectKey,:EventOID,:FormOID,:ItemGroupOID,:ItemOID,:Value) "
        "ON CONFLICT(study, site, subject, event, form, item_group, item) "
        "DO UPDATE SET value=excluded.value",        # idempotent on the natural key
        rows,
    )
    return len(rows)

5. Commit the cursor atomically after the load succeeds #

Advance the cursor to the maximum (timestamp, id) observed in the loaded batch, and write it inside the same transaction that commits the rows. If the load fails, the cursor never moves, so the next run re-reads the window and no record is lost.

# ALCOA+ requirement: Complete — the cursor advances only inside the committed
# load transaction; a crash before commit re-reads the window rather than
# skipping records, so the watermark can never regress past unwritten data.
def commit_cursor(db: sqlite3.Connection, study_oid: str, site_ref: str,
                  records: list[dict]) -> None:
    if not records:
        return                                       # empty batch: do not advance
    new_ts = max(datetime.fromisoformat(r["LastUpdatedDate"]).astimezone(timezone.utc)
                 for r in records)
    new_id = max(r["AuditSeq"] for r in records
                 if datetime.fromisoformat(r["LastUpdatedDate"]).astimezone(timezone.utc) == new_ts)
    db.execute(
        "INSERT INTO sync_cursor(study, site, watermark_ts, last_seen_id) "
        "VALUES(?,?,?,?) ON CONFLICT(study, site) DO UPDATE SET "
        "watermark_ts=excluded.watermark_ts, last_seen_id=excluded.last_seen_id",
        (study_oid, site_ref, new_ts.isoformat(), new_id),
    )
    db.commit()                                      # single atomic transaction

Verification and Audit Trail #

Confirming the fix means proving two things from the log, not asserting them: that a replay of an identical window inserts zero new rows, and that the reconciled count matches the source. Capture, per cycle, a structured record whose fields let an inspector re-derive the cursor’s behavior.

To confirm the fix end to end, assert three properties against a mocked EDC endpoint: a page of records sharing one LastUpdatedDate is fully ingested with none skipped and none duplicated; replaying the identical window advances nothing and inserts zero rows; and a cursor written under a simulated crash-before-commit re-reads its window on the next run. Genuine discrepancies the reconciliation surfaces feed Automated Clinical Query Generation rather than blocking the cycle.

Edge Cases and Vendor-Specific Gotchas #

No stable tie-breaker id. Some EDC exports expose only LastUpdatedDate and no monotonic audit sequence. When that happens, synthesize a deterministic tie-breaker from the ODM natural-key hash so the composite comparison still has a stable second element — never fall back to row position, which reorders between pages and reintroduces the loss.

Overlap window shorter than late-arrival latency. The overlap only catches edits that land within its span below the watermark. If back-dated corrections routinely arrive later than that — a common pattern when sites reconcile paper source days afterward — the re-scan misses them. Size the window from observed late-arrival latency, and treat a reconciliation count mismatch as the alarm that it is set too narrow, mirroring the discipline in Incremental Sync and Change Data Capture for EDC Pipelines.

Cursor stored in local time. A cursor persisted in anything but UTC will silently regress at a daylight-saving transition and skip an hour of edits. Store watermark_ts in UTC only, normalize every incoming timestamp before comparison, and reject any naive datetime at the boundary rather than guessing its zone.

Frequently Asked Questions #

Related #

• Incremental Sync and Change Data Capture for EDC Pipelines — the parent discipline this cursor mechanism implements.
• Async Polling Strategies for EDC Updates — the polling cadence that drives each delta cycle this cursor checkpoints.
• Automating Medidata Rave Data Pulls with Python — a vendor-specific extraction that persists a LastUpdatedDate cursor per site.
• Handling API Rate Limits in Clinical Sync — keeping the repeated overlap-window queries inside vendor quota.
• Automated EDC Ingestion & Sync Pipelines — the parent reference for this discipline.