# `Capstan.Snapshot.Coordinator`
[🔗](https://github.com/baselabs/capstan/blob/v1.3.1/lib/capstan/snapshot/coordinator.ex#L1)

The initial-snapshot orchestration core (C2): a GenServer that interposes as the
`Capstan.AssemblerServer`'s sink and its `:watermark_observer`, runs the cursor-gate over
the live stream, drives the `Capstan.Snapshot.ChunkReader`, and emits each backfill chunk
through the real sink's `c:Capstan.Sink.handle_snapshot/2` once the stream's processed
watermark covers the chunk's exact GTID position `G`.

## Two roles, two message paths

  * **As the AssemblerServer's sink.** The coordinator implements the `Capstan.Sink`
    behaviour and is name-registered under `Capstan.Snapshot.Coordinator`. The AssemblerServer's
    `state.sink` is this MODULE, so its `handle_transaction/1` and `handle_schema_change/2`
    dispatch here; each is a thin module function that `GenServer.call`s the registered
    coordinator (resolved at call time, not start time — this is what breaks the start-order
    cycle, design § Pinned decisions #4). The coordinator does NOT edit `assembler_server.ex`; the
    `attach_coordinator/2` injection + the silent-death monitor are the AssemblerServer's own.

  * **As the `:watermark_observer`.** After a deferred `attach_coordinator/2`, the
    AssemblerServer sends `{:capstan_watermark, gtid_set_string}` on EVERY checkpoint advance
    (delivered, filtered, AND self-committing DDL — the Ch6 choke-point guarantee). The
    coordinator feeds that string into the advance gate.

## The cursor-gate (`handle_transaction/1`)

For each streamed `%Capstan.Change{}` on a snapshot-active table (in the snapshot set and
not yet done), the coordinator applies `Capstan.Snapshot.CursorGate.classify/3` per
row-image with the per-table cursor + PK shape, and forwards the surviving subset to the
real sink as a transaction carrying only those changes. A change on a NON-snapshot table —
or on a table whose backfill is already done (the stream is authoritative) — passes through
unchanged.

When ALL changes are suppressed, the coordinator returns `{:ok, position}` WITHOUT calling
the real sink: a fully-suppressed transaction is exactly a *filtered* transaction from the
real sink's view (ADR-0003), and the AssemblerServer still advances the watermark for it
(`assembler_server.ex:246`/`:257`). Returning the txn's position keeps a fully-suppressed
run from stalling the advance gate → deadlock.

## The advance gate

The coordinator holds AT MOST ONE buffered chunk (its `rows` + its exact `G`). When the
processed watermark covers `G` (`CursorGate.advance?/2`), the emit runs in THREE ordered steps:

  1. **Advance `delivered_pk` to `chunk.max_pk` and persist** — BEFORE the sink emit.
  2. **Emit** the chunk as a list of `%Change{op: :snapshot}` through the real sink's
     `handle_snapshot/2` with a value-free `%Capstan.Snapshot.Meta{}`.
  3. **Advance `pk_cursor` to `chunk.max_pk`, persist, telemetry, drive the next chunk.**

The `pk_cursor` (re-read floor) still advances AFTER the emit — the at-least-once boundary
(tripwire 16): a crash between the sink's `{:ok}` and the `pk_cursor` persist re-emits the one
chunk (bounded dup, C1's posture; an upsert-by-PK sink converges). The delivered-high-water persist comes FIRST, so in that same crash window `delivered_pk` survives AHEAD of the
rolled-back `pk_cursor`; on restart the cursor-gate forwards a streamed delete of an
already-delivered key (`k ≤ delivered_pk`) instead of suppressing it, sweeping what would
otherwise be a permanent crash-window phantom (`Capstan.Snapshot.CursorGate` and
`Capstan.Snapshot.State`). A fault persisting the delivered high-water halts fail-closed before any emit.

## Fail-closed halts (symmetric with C1)

A coordinator fault sends `{:capstan_halt, reason}` to the AssemblerServer (the LOUD path)
and stops with `{:shutdown, {:halt, reason}}` (`restart: :temporary`, never restarted). The
SILENT-death path — a coordinator that dies without messaging — is the AssemblerServer's
`Process.monitor` (`:snapshot_coordinator_down`). A raise in the emit/reconcile path
is scrubbed value-free to `{:snapshot_processing_crashed, Capstan.Error.from(exc)}` exactly
as the AssemblerServer scrubs a delivery-path raise (`assembler_server.ex:148-158`); a
`handle_snapshot/2` `{:error, _}` halts `{:snapshot_sink_error, _}` (the outer atom only in
telemetry). A DDL on a snapshot-active table halts `:snapshot_schema_drifted`.

## Rule 1

Row values (chunk rows, canonical PKs, the per-table cursor) travel ONLY in the delivered
`%Change{op: :snapshot}` list and in in-memory state; they are never logged or telemetered.
Telemetry (`[:capstan, :snapshot, :started | :chunk_completed | :completed | :halt]`) carries
counts in MEASUREMENTS and only structural identity (schema/table/reason) in METADATA, gated
by `Capstan.Telemetry`'s value-free allowlist (a stray PK/cursor raises). The coordinator's
own struct holds `%Capstan.Snapshot.Chunk{}` / `%Capstan.Snapshot.State{}`, each of which
derives a value-eliding `Inspect`, so even an incidental inspect of the state never surfaces
a value; and every halt is a graceful `{:shutdown, _}` exit (never an abnormal crash), so the
in-flight message is never dumped to a crash report.

# `store`

```elixir
@type store() :: {module(), term()}
```

A durable snapshot store as `{callback_module, store_handle}`.

# `table_key`

```elixir
@type table_key() :: {String.t(), String.t()}
```

The `{schema, table}` identity of a snapshot table.

# `child_spec`

Returns a specification to start this module under a supervisor.

See `Supervisor`.

# `handle_schema_change`

```elixir
@spec handle_schema_change(Capstan.SchemaChange.t(), Capstan.Position.t()) ::
  :ok | {:error, term()}
```

The AssemblerServer's `handle_schema_change/2` sink hook: a DDL on a snapshot-active table
halts `:snapshot_schema_drifted`; any other DDL forwards to the real sink.

# `handle_transaction`

```elixir
@spec handle_transaction(Capstan.Transaction.t()) ::
  {:ok, Capstan.Position.t()} | {:error, term()}
```

The AssemblerServer's `handle_transaction/1` sink hook: routes the call to the registered
coordinator, which applies the cursor-gate and forwards the surviving subset. Returns
`{:ok, position}` (the txn's position) — even when every change is suppressed.

# `start_link`

```elixir
@spec start_link(keyword()) :: GenServer.on_start()
```

Starts the coordinator, registered under `Capstan.Snapshot.Coordinator` so the AssemblerServer's
module-sink dispatch (`state.sink.handle_transaction/1` etc.) resolves it by name.

Options:

  * `:sink` (required) — the real (downstream) `Capstan.Sink` module.
  * `:assembler` (required) — the AssemblerServer pid/name for `{:capstan_halt, _}`.
  * `:snapshot_store` (required) — `{impl_module, store_handle}`.
  * `:snapshot_state` (required) — the initial `%Capstan.Snapshot.State{}`.
  * `:readers` (required) — `%{{schema, table} => reader_handle}` (opened `ChunkReader`s).
  * `:chunk_reader` — the reader module (default `Capstan.Snapshot.ChunkReader`).
  * `:processed_set` — the initial processed watermark string (default `""`).
  * `:table_order` — the backfill order (default: not-done tables, sorted).

---

*Consult [api-reference.md](api-reference.md) for complete listing*