Tracking Last-Sync Timestamps for Incremental LMS Pulls

An incremental LMS pull is only as trustworthy as the timestamp it resumes from. Store that watermark carelessly — in a variable that dies with the process, in a file rewritten before the data commits, in a naive local-time string — and the pipeline either re-scrapes everything or, worse, skips edits and never notices. This guide implements a durable last-sync store the correct way: a persistent checkpoint keyed per entity, advanced only after the batch it describes has committed, offset by an overlap window, and stored in UTC. It is the timestamp-persistence half of incremental sync state management; the deduplication that overlap re-reads make necessary is covered separately in deduplicating incrementally synced grade records.

Prerequisites

Confirm each before running the script — it assumes all of them.

• Python 3.10+ (the code uses str | None unions and datetime.fromisoformat UTC handling).
• The standard-library sqlite3 module (bundled with CPython) — no external database server is required for this store, though the same schema ports directly to Postgres.
• A callable pull(since: datetime) -> list[dict] for your platform that accepts a lower-bound timestamp and returns records carrying an updated_at field. Wire it to your Canvas submitted_since/graded_since, Moodle timemodified, or Blackboard modified reader.
• A decision on grain: one watermark row per (source, entity) you resume independently (for example canvas/submissions, moodle/attendance). Do not share a single watermark across entities that change at different rates.
• An overlap value in seconds, sized above your measured worst-case clock skew plus read-replica lag. Start at 120 seconds and tune from observation.
• Records and timestamps handled in UTC end to end; if your source emits local time, convert on ingestion, never at query time.

Step-by-step implementation

1. Persist the watermark only after the batch commits. The write order is the whole ballgame: commit the pulled data, then write the new watermark, ideally in the same transaction. If the process dies before the watermark write, the transaction rolls back and the next run re-reads the same window — no gap. Advance the watermark first and a crash strands the un-committed window permanently.

2. Subtract an overlap window from the stored watermark on read. Query the source with since = watermark − overlap, not the bare watermark. The overlap re-fetches a small sliver of already-seen rows so an edit written at the exact boundary second, or made visible late by replica lag, is never stranded on the wrong side of the cut.

3. Keep one watermark per entity. Gradebook submissions, attendance, and rosters change at different cadences; a shared watermark forces the slowest stream’s boundary onto the fastest and re-reads far more than necessary. A composite (source, entity) key lets each advance on its own schedule.

4. Store timestamps in UTC, always. Persist and compare in UTC so a daylight-saving transition or a worker in a different timezone cannot shift the boundary by an hour. Use timezone-aware datetime objects and serialize with an explicit offset; parse them back the same way.

5. Anchor the new watermark to observed record timestamps. Set the next watermark to the maximum updated_at actually seen in the batch, taken as max(old_watermark, batch_max) so it never regresses. Never set it to datetime.now() — wall-clock time assumes the source’s clock is yours and that every earlier edit was already returned, both false under lag.

6. Guard against concurrent runs. Claim the checkpoint row before reading it (a status lease) so two scheduled workers cannot both read the old watermark and race to advance it, silently discarding one run’s progress.

Complete runnable code block

"""Durable last-sync watermark store for incremental LMS pulls (SQLite)."""
from __future__ import annotations

import sqlite3
from datetime import datetime, timedelta, timezone
from typing import Callable

UTC = timezone.utc
EPOCH = datetime(1970, 1, 1, tzinfo=UTC)  # safe "never synced" default


def connect(path: str = "sync_state.db") -> sqlite3.Connection:
    conn = sqlite3.connect(path, isolation_level=None)  # explicit txn control
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS sync_checkpoint (
            source     TEXT NOT NULL,
            entity     TEXT NOT NULL,
            watermark  TEXT NOT NULL,          -- ISO-8601 UTC string
            status     TEXT NOT NULL DEFAULT 'idle',
            updated_at TEXT,                   -- when the row last advanced
            PRIMARY KEY (source, entity)
        )
        """
    )
    return conn


def read_watermark(conn: sqlite3.Connection, source: str, entity: str) -> datetime:
    row = conn.execute(
        "SELECT watermark FROM sync_checkpoint WHERE source=? AND entity=?",
        (source, entity),
    ).fetchone()
    if row is None:
        return EPOCH
    return datetime.fromisoformat(row[0])  # aware datetime round-trips exactly


def claim(conn: sqlite3.Connection, source: str, entity: str) -> bool:
    """Lease the row so a second worker cannot race the watermark."""
    conn.execute("BEGIN IMMEDIATE")
    conn.execute(
        """INSERT INTO sync_checkpoint (source, entity, watermark, status)
           VALUES (?, ?, ?, 'running')
           ON CONFLICT(source, entity) DO UPDATE SET status='running'
           WHERE sync_checkpoint.status != 'running'""",
        (source, entity, EPOCH.isoformat()),
    )
    changed = conn.total_changes
    conn.execute("COMMIT")
    return changed > 0


def sync_once(
    conn: sqlite3.Connection,
    source: str,
    entity: str,
    pull: Callable[[datetime], list[dict]],
    persist: Callable[[sqlite3.Connection, list[dict]], None],
    overlap_seconds: int = 120,
) -> datetime:
    """Read watermark, pull the delta, persist data + watermark atomically."""
    if not claim(conn, source, entity):
        raise RuntimeError(f"{source}/{entity} already running — refusing to race")

    watermark = read_watermark(conn, source, entity)
    since = watermark - timedelta(seconds=overlap_seconds)   # step 2: overlap
    records = pull(since)                                     # updated_since delta

    if not records:
        conn.execute(
            "UPDATE sync_checkpoint SET status='idle' WHERE source=? AND entity=?",
            (source, entity),
        )
        return watermark

    # Anchor to observed timestamps and never regress (step 5).
    batch_max = max(datetime.fromisoformat(r["updated_at"]) for r in records)
    new_watermark = max(watermark, batch_max)

    conn.execute("BEGIN IMMEDIATE")
    try:
        persist(conn, records)                               # commit DATA first
        # Persist the watermark LAST, same transaction (step 1).
        conn.execute(
            """UPDATE sync_checkpoint
                  SET watermark=?, updated_at=?, status='idle'
                WHERE source=? AND entity=?""",
            (new_watermark.isoformat(), datetime.now(UTC).isoformat(), source, entity),
        )
        conn.execute("COMMIT")
    except Exception:
        conn.execute("ROLLBACK")   # data AND watermark roll back together
        conn.execute(
            "UPDATE sync_checkpoint SET status='failed' WHERE source=? AND entity=?",
            (source, entity),
        )
        raise
    return new_watermark


if __name__ == "__main__":
    conn = connect()

    def fake_pull(since: datetime) -> list[dict]:
        now = datetime.now(UTC)
        return [{"id": 1, "score": 88.0, "updated_at": now.isoformat()}]

    def fake_persist(c: sqlite3.Connection, rows: list[dict]) -> None:
        c.executemany("CREATE TABLE IF NOT EXISTS grades (id INT PRIMARY KEY, score REAL, updated_at TEXT)"
                      if False else "INSERT OR REPLACE INTO grades VALUES (?,?,?)",
                      [(r["id"], r["score"], r["updated_at"]) for r in rows])

    conn.execute("CREATE TABLE IF NOT EXISTS grades (id INT PRIMARY KEY, score REAL, updated_at TEXT)")
    wm = sync_once(conn, "canvas", "submissions", fake_pull, fake_persist)
    print("advanced watermark to", wm.isoformat())

Verification and output validation

• Watermark advances, never regresses. Run sync_once twice; assert the second returned watermark is >= the first. A regression means you anchored to now() on a machine with a slow clock instead of to batch_max.
• Empty pull leaves the watermark untouched. Point pull at a since in the future so it returns []; assert the stored watermark is unchanged and status returns to idle.
• Round-trip is UTC and exact. assert read_watermark(conn, "canvas", "submissions").tzinfo is not None — a naive datetime here means the ISO string lost its offset and future comparisons are ambiguous.
• Rollback leaves no partial state. Make persist raise on purpose; assert the watermark did not advance and status='failed', proving data and watermark rolled back together.
• Overlap is actually applied. Log the computed since; assert it equals watermark - overlap, not the bare watermark.
• The lease works. Call claim twice without releasing; the second returns False, proving a concurrent run is refused rather than allowed to race.

Troubleshooting

• Records reappear every run. The watermark is not persisting. Confirm the UPDATE runs inside the committed transaction and that isolation_level=None plus explicit BEGIN/COMMIT are used — the default sqlite3 autocommit behavior can leave the write in an uncommitted implicit transaction.
• A window went missing after a crash. You advanced the watermark before the data committed. Move the watermark UPDATE after persist() inside the same BEGIN … COMMIT, as the script does, so a crash rolls both back.
• TypeError: can't compare offset-naive and offset-aware datetimes. A source timestamp was parsed without a timezone. Normalize every incoming updated_at to UTC-aware on ingestion; datetime.fromisoformat preserves an offset only if the string carries one.
• Boundary edits are silently skipped. The overlap is too small for your replica lag. Measure the p99 gap between a record’s updated_at and its first appearance in a pull, then set overlap_seconds above it — widening overlap only costs idempotent re-reads.
• Two schedulers corrupt the watermark. Both ran without leasing. Ensure claim runs first and that a stale running row is cleared by a timeout so a dead worker does not lock the stream forever.
• Watermark jumped far into the future, skipping data. A single skewed record carried a future updated_at and you trusted it. Cap batch_max at now() + tolerance before advancing, and investigate the source node’s clock.

Related

• Incremental Sync State Management — the full checkpoint model, per-platform delta endpoints, and the loop this store plugs into.
• Deduplicating incrementally synced grade records — collapsing the duplicates the overlap window deliberately re-reads.
• Error retry logic for sync jobs — retrying a failed pull so the un-advanced watermark is simply retried on the next attempt.
• Handling Canvas API rate limits — pacing the updated_since delta so a tight window also stays inside the cost budget.

Part of: Incremental Sync State Management