Slotting Architecture · 11 min read

How to Implement Watermark Cursors for Delta Sync

A full-table pull of every SKU on every sync is the brute-force answer to keeping velocity data current, and it stops working the moment your catalog outgrows the sync window. The efficient answer is a delta sync: pull only the rows that changed since last time, tracked by a high-watermark cursor over a monotonic column like updated_at or a sequence id. The trap is that the naive version — WHERE updated_at > last_watermark — silently loses rows. A transaction that started before your sync but committed after it can carry an updated_at earlier than the watermark you just advanced, and you will never see it again. This page implements a watermark cursor that pulls changed rows since the last sync, applies a safety overlap window so those late-committing rows are re-read instead of lost, and persists the cursor only after the batch commits. It is the delta-fetch technique behind the WMS & ERP Polling Strategies guide, part of the wider Velocity Data Ingestion & WMS Sync Pipelines architecture.

Watermark cursor timeline with a safety overlap window A horizontal time axis measured by the source updated_at clock. A shaded fetch band spans the query range, whose lower bound is the last watermark minus the overlap seconds and whose upper bound is the new watermark. The leftmost slice of that band, from watermark minus overlap up to the last watermark, is highlighted as the overlap re-read zone. Changed rows appear as dots on the axis: three new rows sit to the right of the last watermark and are picked up normally, while one late-committing row sits inside the overlap zone with an updated_at earlier than the last watermark and is recovered only because the overlap window re-reads it. A note states that the cursor is persisted only after the batch commits, and the new watermark is the maximum updated_at observed. Watermark cursor with safety overlap window persist cursor ONLY after batch commit overlap WHERE updated_at ≥ watermark − overlap_seconds source updated_at → watermark − overlap last watermark (t0) new watermark = max(updated_at) late row: updated_at < t0, recovered by overlap late-committing row re-read by overlap changed rows since last sync

Prerequisites

Confirm each before pointing a delta sync at a production feed:

  • Python 3.10+ — the implementation uses X | None unions, timezone-aware datetime, and a dataclass config.
  • A monotonic change column — an indexed updated_at (set by a database trigger, not the application) or a monotonic sequence id. Without a reliably advancing column there is no watermark to track.
  • Idempotent apply logic — the overlap window deliberately re-reads rows, so your write path must be an upsert keyed on the primary key. If re-applying a row double-counts anything, the overlap will corrupt your data.
  • A durable cursor store — a small file on a persistent volume or a control-table row. It must be written transactionally with respect to the batch, or after it.
  • A validated payload contract — reuse the schema gate from Schema Validation for Inventory Feeds so a malformed row cannot advance the watermark past good data.

Configuration Block

Three parameters define the cursor: which column it tracks, how far back the overlap re-reads, and where it is stored. overlap_seconds is the safety margin — set it wider than your longest expected write-commit lag.

# watermark_sync.yaml — one profile per source table or endpoint
watermark_sync:
  cursor_field: "updated_at"          # monotonic column the watermark tracks
  overlap_seconds: 120.0              # re-read window; wider than the longest commit lag
  cursor_store: "wms_cursor.json"     # durable location for the persisted watermark
# Equivalent Python config dict consumed by the sync driver
from pathlib import Path

WATERMARK_SYNC = {
    "cursor_field": "updated_at",
    "overlap_seconds": 120.0,
    "cursor_store": Path("wms_cursor.json"),
}

Implementation

The function loads the last watermark, subtracts the overlap to form the query’s lower bound, fetches and applies the changed rows, then advances the cursor to the maximum observed cursor_field — and writes it only after the apply succeeds. The max(new_watermark, watermark) guard ensures the cursor never moves backwards even if a batch happens to contain only rows inside the overlap zone.

from __future__ import annotations

import json
import logging
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any, Callable, Sequence

logger = logging.getLogger("velocity.cursor")


@dataclass(frozen=True)
class WatermarkConfig:
    cursor_field: str = "updated_at"
    overlap_seconds: float = 120.0
    cursor_store: Path = Path("wms_cursor.json")


def _load_watermark(path: Path) -> datetime:
    if path.exists():
        return datetime.fromisoformat(json.loads(path.read_text())["watermark"])
    return datetime(1970, 1, 1, tzinfo=timezone.utc)  # epoch = full initial load


def sync_delta(
    fetch_since: Callable[[datetime], Sequence[dict[str, Any]]],
    apply_rows: Callable[[Sequence[dict[str, Any]]], None],
    cfg: WatermarkConfig,
) -> datetime:
    """Pull rows changed since the last watermark, with a safety overlap for late updates.

    The query lower bound is ``watermark - overlap_seconds``, so rows that committed slightly
    out of order (an ``updated_at`` earlier than a value already seen) are re-read instead of
    lost. ``apply_rows`` must be an idempotent upsert because the overlap re-applies rows. The
    cursor advances to the max observed ``cursor_field`` and is persisted ONLY after the apply
    succeeds, so a crash mid-batch re-processes rather than skips.
    """
    watermark = _load_watermark(cfg.cursor_store)
    lower_bound = watermark - timedelta(seconds=cfg.overlap_seconds)
    rows = fetch_since(lower_bound)
    logger.info("fetch since %s (watermark=%s, overlap=%.0fs) -> %d rows",
                lower_bound.isoformat(), watermark.isoformat(), cfg.overlap_seconds, len(rows))
    if not rows:
        return watermark

    apply_rows(rows)  # idempotent upsert: overlap guarantees some rows are re-applied

    observed = max(datetime.fromisoformat(str(r[cfg.cursor_field])) for r in rows)
    new_watermark = max(observed, watermark)  # never move the cursor backwards
    cfg.cursor_store.write_text(json.dumps({"watermark": new_watermark.isoformat()}))
    logger.info("advanced watermark -> %s", new_watermark.isoformat())
    return new_watermark

Step-by-Step Walkthrough

  1. Load the durable watermark. _load_watermark reads the persisted ISO timestamp from cursor_store, or returns the Unix epoch on first run so the initial sync is a full load. Keeping the value timezone-aware (timezone.utc) is what prevents the naive-vs-aware datetime comparison error that plagues cursor code.
  2. Subtract the overlap to form the lower bound. lower_bound = watermark - timedelta(seconds=overlap_seconds) is the whole trick. The query asks for everything at or after watermark - overlap_seconds, not strictly after the watermark, so any row that committed late with a slightly earlier updated_at still falls inside the range.
  3. Fetch and apply. fetch_since(lower_bound) runs the delta query; apply_rows upserts the result. Because the overlap guarantees repeats, the apply path must be idempotent — an upsert keyed on the primary key, never a blind insert or an increment.
  4. Advance to the observed maximum. The new watermark is the largest cursor_field value in the batch, floored at the previous watermark by max(observed, watermark) so a batch of only overlap-zone rows cannot regress the cursor.
  5. Persist last, log the move. cursor_store.write_text happens only after apply_rows returns. If the process dies before this line, the old watermark survives and the next run simply re-reads the same window — at-least-once delivery, which the idempotent apply makes safe.

Verification

Drive the sync with an in-memory row set and a list-appending apply. The first run loads from epoch and applies both rows; the persisted watermark lands on the later timestamp.

import json
import logging
from datetime import datetime
from pathlib import Path

logging.basicConfig(level=logging.INFO)

store = Path("wms_cursor.json")
store.unlink(missing_ok=True)

ROWS = [
    {"sku_id": "A", "updated_at": "2025-06-01T10:00:00+00:00"},
    {"sku_id": "B", "updated_at": "2025-06-01T10:02:00+00:00"},
]
applied: list[str] = []
cfg = WatermarkConfig(overlap_seconds=120.0, cursor_store=store)

wm = sync_delta(
    fetch_since=lambda lb: [r for r in ROWS
                            if datetime.fromisoformat(r["updated_at"]) >= lb],
    apply_rows=lambda rows: applied.extend(r["sku_id"] for r in rows),
    cfg=cfg,
)
print(wm.isoformat(), applied, json.loads(store.read_text()))
assert wm.isoformat() == "2025-06-01T10:02:00+00:00"
assert applied == ["A", "B"]

Sample expected output:

INFO:velocity.cursor:fetch since 1969-12-31T23:58:00+00:00 (watermark=1970-01-01T00:00:00+00:00, overlap=120s) -> 2 rows
INFO:velocity.cursor:advanced watermark -> 2025-06-01T10:02:00+00:00
2025-06-01T10:02:00+00:00 ['A', 'B'] {'watermark': '2025-06-01T10:02:00+00:00'}

A second run against the same rows re-reads them (they fall inside the 120-second overlap), the idempotent apply absorbs the repeat, and the watermark holds at 10:02 — exactly the at-least-once behaviour you want.

Common Pitfalls

  • Clock skew between the source and your sync. If the updated_at clock and the machine forming the query disagree, a strict > boundary drops rows that landed in the gap. Always drive the watermark from the source’s own cursor_field values (the max you actually observed), never from datetime.now() on the consumer, and size overlap_seconds to cover the worst-case skew.
  • A non-monotonic timestamp column. An application-set updated_at, a column edited by bulk backfills, or one that resets on restore is not monotonic, and the watermark will step over rows. Prefer a database-trigger-maintained updated_at or a monotonic sequence id; validate the assumption before trusting it via Schema Validation for Inventory Feeds.
  • No overlap, so late commits vanish. With overlap_seconds = 0 the query is WHERE updated_at > watermark, and any transaction that committed after you advanced the cursor but carries an earlier updated_at is lost permanently — the single most common delta-sync data-loss bug. Keep the overlap wider than your longest observed write-to-commit lag.
  • Persisting the cursor before the commit. Writing the watermark before apply_rows durably commits turns a crash into silent data loss: the rows were never stored, but the cursor says they were. Always persist last. If you need lower latency than nightly batch, feed these deltas into Real-Time Streaming Velocity Updates rather than shrinking the poll interval.

FAQ

Should I use a timestamp or a monotonic sequence id as the cursor?

A monotonic sequence id (an auto-increment or log offset) is strictly better when you have one: it is gap-free, immune to clock skew, and unambiguous about ordering, so the overlap can be a small fixed number of ids. Fall back to updated_at when no such column exists, but only if it is maintained by the database rather than the application, and always pair it with a time-based overlap to absorb commit-order surprises.

How wide should the overlap window be?

Wide enough to cover the longest realistic lag between a row’s updated_at and its transaction commit, plus any clock skew between systems. Measure it: sample the delay between commit time and updated_at under load and set overlap_seconds above the P99. A window that is too wide only costs a few redundant idempotent upserts; one that is too narrow loses data, so err generous.