Capstan.Snapshot.ChunkReader (Capstan v0.2.0)

Copy Markdown View Source

The brief-lock exact-G capture + PK-range paging + structural fingerprint (C2 Task 5, design Ch1 — THE linchpin).

A ChunkReader owns a Capstan.Query connection and, per chunk of a snapshot table T, captures the chunk rows paired with an exact GTID position G that is a provable lower bound on the chunk's read view for T. That pairing is the correctness property the whole snapshot rests on: a lock-free @@gtid_executed read LEADS InnoDB row-visibility under concurrent commit (probe-confirmed: 300000/300000 under 3 writers), so a bare read would let a stale chunk row overwrite a fresh streamed row — silent corruption. A brief LOCK TABLES … READ quiesces T's writes while G and a START TRANSACTION WITH CONSISTENT SNAPSHOT view are captured together, making G exact.

The pinned capture sequence (design § Pinned decisions #2)

Per chunk of T from cursor, EXACTLY:

(i)   SET SESSION lock_wait_timeout = <bounded>   -- MDL default ~1yr; a bound turns a
                                                  --   contended lock into a HALT, not a wedge
(ii)  LOCK TABLES `T` READ                        -- waits for in-flight T-writes, blocks new ones
(iii) G = SELECT @@global.gtid_executed           -- EXACT: every T-txn <= G is committed AND visible
(iv)  START TRANSACTION WITH CONSISTENT SNAPSHOT   -- pins the read view V; for T, V == state as-of G
(v)   UNLOCK TABLES                                -- release; the pinned view V survives the unlock
(vi)  SELECT <projected cols> FROM `T`             -- as-of V, lock-free
        WHERE pk > cursor ORDER BY pk LIMIT chunk_size + 1   -- +1 = finality look-ahead
(vii) COMMIT

Step (iii) uses @@global.gtid_executed; gtid_executed is a global-scoped variable, so @@global.gtid_executed and the pinned decision's shorthand @@gtid_executed read the same value. The projection is EXPLICIT (never SELECT *) so the chunk carries the full row image and the per-chunk fingerprint is well-defined.

Halts (value-free atoms; design § Preconditions)

Faults are classified POSITIONALLY, by which phase of the pinned sequence failed — this is how the reader honours the Task-4 forward-finding that Capstan.Query.query/2 scrubs the MySQL error CODE (so a lock-wait timeout 1205 and a chunk-read fault both return the same value-free atom and cannot be told apart from the return value alone). The statement that failed is the distinct signal instead:

  • :snapshot_lock_unavailable — a fault while ACQUIRING the lock: the bounding SET SESSION lock_wait_timeout or the LOCK TABLES … READ itself. A lock-wait timeout (bounded by step (i)) and a missing LOCK TABLES privilege both surface here — both are "could not obtain the lock". Budgeted, then halts.
  • :snapshot_chunk_read_failed — a fault ONCE THE LOCK IS HELD or the view is being read: SELECT @@global.gtid_executed, START TRANSACTION WITH CONSISTENT SNAPSHOT, UNLOCK TABLES, the chunk SELECT, or COMMIT. The chunk SELECT is the canonical case; the surrounding view statements share its fate (the chunk could not be read consistently). Budgeted, then halts.
  • :snapshot_schema_drifted — the per-chunk structural fingerprint changed from the baseline (an in-flight DDL). PERMANENT (a drifted schema does not un-drift on retry), so it halts immediately without spending the budget. (The other drift arm — an observed %Capstan.SchemaChange{} on the table — is the coordinator's, Task 8; the reader owns the fingerprint arm.)

Budgeted faults reuse Capstan.CheckpointStore.retry_decision/2 — the shared counter, never re-derived — so retry semantics cannot drift from C1. Permanence is classified POSITIONALLY, not via CheckpointStore.permanent_reason?/1: that helper only recognizes a store-config fault (:config_invalid) and would misclassify a code-scrubbed reader fault, so schema drift is marked permanent by construction (check_fingerprint/1 returns {:permanent, _} directly) and every lock/read fault is budgeted. The reader retries the whole capture on its OWN connection; it does NOT reconnect (a reconnect re-verifies source identity, Ch8, and is the coordinator's lifecycle concern) — a dead connection burns the budget and halts fail-closed, which the coordinator then propagates.

Rule 1 (design § Rule 1, tripwire 13)

The cursor is a row VALUE embedded as a SQL literal in WHERE pk > <cursor>. On a :snapshot_chunk_read_failed the failing SQL is NEVER logged or returned: Capstan.Query never logs SQL and returns a value-free atom, and this module maps faults to bare atoms and discards every raw fault term — so the cursor appears in no log, error, or crash report. The produced %Capstan.Snapshot.Chunk{} derives a value-eliding Inspect (rows/max_pk are never rendered). No row value or cursor is ever logged or telemetered here.

Summary

Types

The paging cursor: a canonical PK (the previous chunk's max_pk) or :start.

The result of read_chunk/2

t()

Functions

Opens a reader for {schema, table} over an established Capstan.Query handle.

Reads the next chunk of rows with pk > cursor (or from the start when cursor is :start), paired with the chunk's exact GTID position G, under the pinned brief-lock capture sequence.

Types

cursor()

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

The paging cursor: a canonical PK (the previous chunk's max_pk) or :start.

read_result()

@type read_result() ::
  {:ok, Capstan.Snapshot.Chunk.t(), boolean(), t()}
  | {:done, t()}
  | {:error, atom()}

The result of read_chunk/2:

  • {:ok, chunk, final?, reader} — a chunk of 1..chunk_size rows; final? is true when the one-row look-ahead (the chunk SELECT reads LIMIT chunk_size + 1) proved no further chunk exists — i.e. the read returned at most chunk_size rows. final? is derived from the look-ahead, NOT a length < chunk_size count, so a table whose row count is an exact multiple of chunk_size still gets a final?: true last chunk (closeout F6). reader carries the incremented sequence number.
  • {:done, reader} — the page from cursor is empty; the table is fully paged. Under the look-ahead a NON-empty table's last chunk is already final?: true, so this arises only for an empty table (a :start read that returns no rows).
  • {:error, reason} — a value-free halt atom.

t()

@type t() :: %Capstan.Snapshot.ChunkReader{
  chunk_size: pos_integer(),
  fingerprint: String.t(),
  lock_wait_timeout: pos_integer(),
  max_retries: non_neg_integer(),
  pk_columns: [String.t()],
  pk_indices: [non_neg_integer()],
  pk_types: [Capstan.Snapshot.PrimaryKey.pk_type()],
  projection: [String.t()],
  query: Capstan.Query.t(),
  schema: String.t(),
  seq: non_neg_integer(),
  table: String.t()
}

Functions

open(query, arg, opts \\ [])

@spec open(Capstan.Query.t(), {String.t(), String.t()}, keyword()) ::
  {:ok, t()} | {:error, atom()}

Opens a reader for {schema, table} over an established Capstan.Query handle.

Introspects the order-faithful PK (via Capstan.Snapshot.PrimaryKey.introspect/2) and reads the table's ordinal column list to fix the EXPLICIT projection and the baseline structural fingerprint. Options:

  • :chunk_size — rows per chunk (default 4096).
  • :lock_wait_timeout — the bounded MDL wait, in seconds (default 5).
  • :max_retries — the budgeted-fault retry ceiling (default CheckpointStore.default_max_retries/0).
  • :seq — the resume chunk sequence number (default 0).
  • :fingerprint — a baseline fingerprint from durable state; when set, the first chunk's fingerprint is compared against IT (so a schema drift across a resume is caught on chunk 1) rather than against the freshly-read columns.

Returns {:ok, t()} or a value-free {:error, reason} — an introspection halt (:snapshot_table_no_primary_key, :snapshot_pk_unsupported_type) surfaces unchanged; a transient fault reading information_schema (the PK introspection or the column list) surfaces :snapshot_chunk_read_failed.

read_chunk(reader, cursor)

@spec read_chunk(t(), cursor()) :: read_result()

Reads the next chunk of rows with pk > cursor (or from the start when cursor is :start), paired with the chunk's exact GTID position G, under the pinned brief-lock capture sequence.

Returns a read_result/0. The produced %Capstan.Snapshot.Chunk{} carries g (the exact lower bound), rows (the full row images as column => value maps), and max_pk (the canonical PK of the last row — the value the cursor advances to). Faults are classified positionally to :snapshot_lock_unavailable / :snapshot_chunk_read_failed; a fingerprint change halts :snapshot_schema_drifted.