Capstan.Snapshot.CursorGate (Capstan v0.2.0)

Copy Markdown View Source

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.

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

cursor()

@type cursor() :: Capstan.Snapshot.PrimaryKey.canonical_pk() | :start

The per-table backfill cursor: a canonical PK, or :start before the first chunk.

table_spec()

@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

advance?(g, processed_set)

@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.

classify(change, cursor, table)

@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).