Snapshot, then stream

1def initialSnapshot():
2    if snapshot.mode == 'never':
3        return streamFromLsn(currentLsn())  # skip history
4    tx = db.begin('REPEATABLE READ')
5    snapshotLsn = tx.currentLsn()
6    if snapshot.mode != 'no_data':
7        for table in captured.tables:
8            for row in tx.select_all(table):
9                emit(op='r', after=row, lsn=snapshotLsn)
10    tx.commit()
11    streamFromLsn(snapshotLsn)  # seam

1def streamFromLsn(lsn):
2    cursor = replicationSlot.startReplication(
3        startLsn = lsn,
4    )
5    for change in cursor:
6        # change.lsn >= snapshotLsn — concurrent
7        # writes during snapshot replay here too
8        emit(
9            op = change.op,  # c | u | d
10            before = change.before,
11            after = change.after,
12            lsn = change.lsn,
13        )

1# A row updated WHILE the snapshot was running surfaces twice:
2#   1. op=r tagged at snapshotLsn (from the bulk read)
3#   2. op=u tagged at change.lsn > snapshotLsn (from the WAL)
4# The connector cannot suppress (2) without risking a dropped
5# write, so dedupe is the consumer's job.
6def onEvent(evt):
7    key = (evt.table, evt.pk)
8    if evt.lsn <= seen.get(key, -1):
9        return  # already applied — drop
10    apply(evt)
11    seen[key] = evt.lsn