Capstan.Snapshot (Capstan v0.2.0)

Copy Markdown View Source

Initial-snapshot public config validation + the bootstrap orchestration (C2 Task 10 — the P0 pre-seed that avoids C1b, design Q10 / Ch5).

What the bootstrap does

In snapshot mode Capstan.Supervisor calls bootstrap/4 after starting the checkpoint + snapshot stores and BEFORE it reads the checkpoint position, so the pre-seed lands first:

  1. Read the durable %Capstan.Snapshot.State{} + reconcile its table set against config. The stored state binds the tables introspected at the fresh start; the configured snapshot.tables must still match it (reconcile_tables/2) or the bootstrap halts :snapshot_config_drifted — a table ADDED to config after a :complete/mid-snapshot state would otherwise be silently never backfilled (both the :complete short-circuit and open_resume/3 key off the STORED set). status: :complete (with a matching set) ⇒ the backfill is already done — signal :complete so the supervisor wires the real sink directly (pure C1, no coordinator). No source connection is opened in this case (the reconcile is a pure key comparison).
  2. Establish the transient query connection, source-identity pinned (Ch8). The stream connection's @@server_uuid is read via Capstan.Config.read_server_uuid/1 over an independent authenticated socket, and the Capstan.Query connection is established pinned to it (:expected_server_uuid). A query connection that lands on a DIFFERENT replica than the stream — a VIP failover between the two connects — is caught :snapshot_source_mismatch (the cross-connection check, Task-9-F1 / Task-4).
  3. Resolve P0. A FRESH start reads P0 = @@global.gtid_executed over the query connection (a read fault halts :snapshot_bootstrap_gtid_read_failed). A mid-snapshot RESUME uses the STORED p0 from the durable %State{} — immune to @@gtid_executed drift across a bootstrap crash.
  4. Seed the checkpoint store. P0 is written ONLY when the checkpoint is empty or already equals p0 (seed_checkpoint/2) — it NEVER regresses a watermark the stream already advanced. This makes the dump AND the assembler watermark both resolve to P0 through the sanctioned start_position: :checkpoint path (no C1b, no ADR-0004 supersession).
  5. Seed the durable %State{} + open the readers. A Capstan.Snapshot.ChunkReader is opened per not-done snapshot table (introspecting the order-faithful PK + baseline fingerprint), and the %State{} (per-table pk_cursor: :start, p0) is persisted. On resume the durable per-table cursors are kept and each reader reopened with the stored fingerprint (a schema drift across the resume is caught on chunk 1).

The supervisor then starts the Capstan.Snapshot.Coordinator with processed_set = the live watermark (P0) so a chunk whose G ≤ P0 emits immediately (Task-8-F2), wires it as the assembler's sink by NAME, and injects the observer + monitor via Capstan.AssemblerServer.attach_coordinator/2.

Retention purge racing the bootstrap (tripwire 11)

The seed NEVER masks a retention gap. Across a bootstrap crash-window (the checkpoint seeded but the %State{} not yet written), a re-bootstrap reads a fresh gtid_executed = p0' yet seed_checkpoint/2 LEAVES the earlier P0 untouched (p0' ≠ P0). If the source has since PURGED past that P0, the stream resumes from it and the EXISTING C1 gap gate fires :data_gap — the bootstrap does not map error 1236 to :ok, nor overwrite the gapped checkpoint with the current position (which WOULD mask the loss).

Rule 1

P0 and the %State{} p0 are GTID-set STRINGS (Capstan.Gtid form) — structural, not row values. The per-table pk_cursor/fingerprint are user data; they live only in the durable %State{} (whose Inspect elides the tables map) and are never logged or telemetered here. Every failure is a value-free atom; the transient query connection's password never leaves the Capstan.Query handle.

Summary

Types

A checkpoint store handle as {callback_module, store_handle}.

The per-table opened ChunkReader handles for the coordinator's readers map.

The bootstrap outcome

A snapshot store handle as {callback_module, store_handle}.

Functions

Runs the snapshot bootstrap against the already-started checkpoint_store + snapshot_store.

Seeds the checkpoint store with p0 ONLY when the checkpoint is empty or already equals p0.

Validates the snapshot configuration surface: normalises the :snapshot block (Capstan.Config.validate_snapshot/1) and enforces snapshot_tables ⊆ captured (Capstan.Pipeline.validate_snapshot_tables/2).

Types

checkpoint_store()

@type checkpoint_store() :: {module(), Capstan.CheckpointStore.store()}

A checkpoint store handle as {callback_module, store_handle}.

readers()

@type readers() :: %{
  optional({String.t(), String.t()}) => Capstan.Snapshot.ChunkReader.t()
}

The per-table opened ChunkReader handles for the coordinator's readers map.

result()

@type result() ::
  :complete
  | {:snapshot, Capstan.Snapshot.State.t(), readers(), String.t()}
  | {:error, atom()}

The bootstrap outcome:

  • :complete — the durable %State{} is status: :complete; wire pure C1 (no coordinator).
  • {:snapshot, state, readers, processed_set} — a fresh/mid-snapshot start: the seeded %State{}, the per-table readers, and the initial processed-watermark string (P0) the coordinator seeds its advance gate with.
  • {:error, reason} — a value-free bootstrap halt.

snapshot_store()

@type snapshot_store() :: {module(), Capstan.SnapshotStore.store()}

A snapshot store handle as {callback_module, store_handle}.

Functions

bootstrap(opts, snapshot, checkpoint_store, snapshot_store)

Runs the snapshot bootstrap against the already-started checkpoint_store + snapshot_store.

opts is the pipeline wiring (:connection, optional :connect_fun / :max_command_retries test seams); snapshot is the normalised Capstan.Config.snapshot_config(). Returns a result/0. Reads the durable %State{} FIRST (short-circuiting to :complete without opening any source connection); otherwise establishes the source-pinned query connection, resolves + seeds P0, and opens the per-table readers. See the moduledoc for the full sequence and the fail-closed halts.

seed_checkpoint(arg, p0)

@spec seed_checkpoint(checkpoint_store(), String.t()) :: :ok | {:error, term()}

Seeds the checkpoint store with p0 ONLY when the checkpoint is empty or already equals p0.

A checkpoint the stream has already advanced past p0 (a resume, or a bootstrap crash-window re-read) is LEFT untouched — the seed never regresses a watermark, and never masks a retention gap by overwriting a stale-but-gapped checkpoint with the current position (tripwire 11). Equality is a GTID-set comparison (Capstan.Gtid), not a string compare. Returns :ok or a value-free store error.

validate(opts)

@spec validate(keyword()) ::
  {:ok, Capstan.Config.snapshot_config() | nil}
  | {:error, :config_invalid | :snapshot_table_not_captured}

Validates the snapshot configuration surface: normalises the :snapshot block (Capstan.Config.validate_snapshot/1) and enforces snapshot_tables ⊆ captured (Capstan.Pipeline.validate_snapshot_tables/2).

Returns {:ok, snapshot_config | nil} (nil when :snapshot is absent — pure C1) or a value-free {:error, reason} (:config_invalid, :snapshot_table_not_captured). Composes the two existing validators so the snapshot config surface has one documented entry point.