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 iscomplete?— 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 makesk_newa brand-new key; the sink applies it as upsert-by-PK (the HARD C2 sink precondition), so it converges even ifk_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.
Summary
Types
The per-table backfill cursor: a canonical PK, or :start before the first chunk.
The per-table gate context: the introspected PK shape (pk_columns in ordinal order and
their order-faithful pk_types), whether the table's backfill is complete?, and the
optional delivered_pk — the high-water the sink has RECEIVED, used as the DELETE threshold
(see the module doc's crash-window section). When delivered_pk is absent, deletes gate on
the cursor argument (pre-F1 behaviour). Extra keys are ignored, so the coordinator may pass
a richer per-table state map.
Functions
The advance-gate predicate: is the chunk's exact GTID position g covered by
processed_set? True iff Gtid.subset?(Gtid.parse(g), processed_set).
Classifies one streamed %Change{} against the table's cursor and table_spec,
returning the list of surviving forward-images ([], one, or — for a straddling
PK-changing UPDATE that forwards both halves — two).
Types
@type cursor() :: Capstan.Snapshot.PrimaryKey.canonical_pk() | :start
The per-table backfill cursor: a canonical PK, or :start before the first chunk.
@type table_spec() :: %{ :pk_columns => [String.t()], :pk_types => [Capstan.Snapshot.PrimaryKey.pk_type()], :complete? => boolean(), optional(:delivered_pk) => cursor(), optional(atom()) => term() }
The per-table gate context: the introspected PK shape (pk_columns in ordinal order and
their order-faithful pk_types), whether the table's backfill is complete?, and the
optional delivered_pk — the high-water the sink has RECEIVED, used as the DELETE threshold
(see the module doc's crash-window section). When delivered_pk is absent, deletes gate on
the cursor argument (pre-F1 behaviour). Extra keys are ignored, so the coordinator may pass
a richer per-table state map.
Functions
@spec advance?(String.t(), Capstan.Gtid.t() | String.t()) :: boolean()
The advance-gate predicate: is the chunk's exact GTID position g covered by
processed_set? True iff Gtid.subset?(Gtid.parse(g), processed_set).
g is always a canonical GTID-set string. processed_set may be either an already-parsed
Capstan.Gtid.t() (the watermark-observer feed) or a string (parsed here). No interval
math is hand-rolled — the containment decision is Capstan.Gtid's.
@spec classify(Capstan.Change.t(), cursor(), table_spec()) :: [Capstan.Change.t()]
Classifies one streamed %Change{} against the table's cursor and table_spec,
returning the list of surviving forward-images ([], one, or — for a straddling
PK-changing UPDATE that forwards both halves — two).
An :insert is gated on its record key, a :delete on its old_record key, and an
:update on its (equal) key unless it moves the PK, in which case it is split into
delete(k_old) + upsert(k_new) and each half is gated on its own key (see the module
doc). A :snapshot change never reaches the gate (the gate classifies streamed images
only) and raises a value-free ArgumentError — a loud, fail-closed misuse signal. It is an
EXPLICIT clause rather than a FunctionClauseError on purpose (Rule 1): a no-clause-match
error captures the call args — including the cursor, a canonical PK / user data — in its
stacktrace frame, whereas the explicit clause leaves the frame at arity 3 (no arg values).