The pure strict-once classification core of the initial snapshot (C2, design §Architecture).

While a table backfills, the C1 stream keeps running. Every streamed %Change{} on a snapshot-active table must be routed so each pre-existing key is delivered EXACTLY ONCE — once by the stream (for keys already backfilled) or once by a chunk (for keys not yet backfilled), never both, never neither. This module is the pure decision function that routes them; it holds no state and does no I/O.

The gate (classify/3)

Given a streamed change, the table's per-table cursor (the canonical PK of the last backfilled row, or :start before the first chunk), and the table's pk_columns / pk_types / complete?, each row-image is classified per its canonical PK k:

• forward iff k ≤ cursor (already backfilled) OR the table is complete? — the stream is authoritative for these keys;
• suppress otherwise (k > cursor) — a not-yet-backfilled key whose future chunk will deliver it. Forwarding it here would double-deliver (stream + chunk).

A :delete is the ONE asymmetric case: it gates on delivered_pk (the high-water the sink has RECEIVED) rather than pk_cursor (the re-read floor). The two are equal in steady state, so suppression is unchanged; they diverge only across the emit→cursor-persist crash window, where delivered_pk sits ahead of the rolled-back pk_cursor. Forwarding a delete of an already-delivered key (k ≤ delivered_pk) then sweeps the row rather than leaving a permanent phantom (closeout F1) — the re-read chunk, taken as-of a fresh G with the row already gone, would otherwise omit it while the suppressed delete never reached the sink. See the table_spec doc; inserts/updates keep gating on pk_cursor so strict-once is preserved.

k is derived from the change's record via Capstan.Snapshot.PrimaryKey.canonical/2 (the raw column values in PK ordinal order) and compared with PrimaryKey.compare/2 — the only ordering the cursor-gate is allowed to use, restricted to order-faithful PK types so the Elixir comparison provably matches MySQL ORDER BY (Ch4).

classify/3 returns the list of surviving forward-images (0, 1, or 2), which the coordinator (Task 8) folds into the transaction's changes and forwards to the real sink. A suppressed change yields [].

PK-changing UPDATE split (Ch2, tripwire 17)

A binlog %Change{op: :update} that moves a row's PK carries old_record (old PK k_old) and record (new PK k_new) with k_old ≠ k_new. Gating such an update on a single key is wrong when the two straddle the cursor, so it is split into delete(k_old) + upsert(k_new), each gated on its own key — the canonical DELETE(old) + INSERT(new) decomposition of a PK move. Both images carry the FULL after/before-image (ADR-0002), so each is self-sufficient:

• delete(k_old) — %Change{op: :delete, old_record: old_record, record: nil}.
• upsert(k_new) — %Change{op: :insert, record: record, old_record: nil}. It is an insert because a PK-changing UPDATE makes k_new a brand-new key; the sink applies it as upsert-by-PK (the HARD C2 sink precondition), so it converges even if k_new's slot was already backfilled.

Straddle (a) k_old ≤ cursor < k_new → forward the delete, suppress the upsert (its future chunk delivers k_new) → no phantom. Straddle (b) k_new ≤ cursor < k_old → suppress the delete (k_old's chunk never ran, nothing was emitted there), forward the upsert → no gap. A non-PK-mutating update (k_old = k_new) is ONE image, unchanged.

The advance-gate predicate (advance?/2)

A buffered chunk read as-of an exact GTID position G may only be emitted (and its cursor advanced) once the stream's processed watermark covers G: Gtid.subset?(Gtid.parse(g), processed_set). This is the ordering that makes suppression correct — the cursor cannot advance to include a key until every gtid ≤ G has been processed by the stream. All GTID-set math routes through Capstan.Gtid (ADR-0001); no hand-rolled interval arithmetic lives here.

Rule 1

Pure functions, no I/O: nothing here logs or telemeters a PK value. Canonical PKs and row values live only in the returned %Change{} images (whose Inspect already elides the value maps) and in the caller's in-memory state.