A COM_QUERY-only connection for the initial-snapshot path (C2 Task 4).
Capstan.Connection is binlog-dump-only — once it issues COM_BINLOG_DUMP_GTID the
socket carries the replication stream and cannot run a SELECT. The snapshot needs a
SEPARATE connection to read information_schema, capture each chunk's exact GTID under a
brief LOCK TABLES … READ, and page the chunk rows. Capstan.Query is that connection: it
reuses the C1 handshake / auth / TLS stack (Capstan.Protocol.Handshake) and the C1
text-resultset decoder (Capstan.Protocol.Command), and exposes the query primitives the
ChunkReader, PrimaryKey introspection, and the bootstrap P0 read all build on.
It owns no process. A %Capstan.Query{} is a connection HANDLE — an authenticated passive
socket plus the pinned source identity — that a caller (the ChunkReader, the bootstrap)
drives synchronously, exactly as Capstan.Config.check_preconditions/1 drives a socket.
Source identity (design Ch8)
A query connection that silently (re)connects to a DIFFERENT replica mid-backfill corrupts
the exact-G pairing the whole snapshot rests on (the chunk's captured GTID would name a
position on a server the chunk rows did not come from). So on EVERY (re)establish the
@@server_uuid is read and compared to the pinned value; a mismatch fails closed
:snapshot_source_mismatch. The endpoint + @@server_uuid are pinned for the connection's
life: the first establish/1 sets the pin (or verifies it against an
:expected_server_uuid — the STREAM connection's identity, supplied by the bootstrap), and
every reestablish/1 re-verifies against it. A reconnect that lands on another replica — a
VIP failover, a moved DNS record — is caught even though the endpoint string is unchanged.
Preconditions (ADR-0002, extended)
establish/1 reuses Capstan.Config.check_preconditions/1 on the query socket: the five
binlog variables gate the query connection exactly as they gate the stream, because
binlog_row_image = FULL is a HARD reconciliation dependency (a partial after-image would
make a stream-delivered change unable to stand in for a suppressed chunk row). A precondition
VIOLATION is permanent (reconnecting cannot cure a mis-configured server), so it surfaces its
distinct reason without spending the retry budget — the same posture C1 takes.
Retry budget
A connect/auth failure fails closed :snapshot_query_connect_failed. Transient faults are
budgeted through Capstan.CheckpointStore.retry_decision/2 + permanent_reason?/1 — the
shared counter the checkpoint-store and connect-read sites use, so the semantics cannot
drift. A permanent connect reason (:config_invalid) halts immediately.
Rule 1
The connection password appears in NO returned term, error, or log emitted here. The struct
carries the connection keyword (which holds the password) only to reconnect, and the
derived Inspect renders ONLY the structural identity (endpoint, server_uuid), never
connection. Every error is a value-free atom, scrubbed through Capstan.Error.from/1; a
raised transport error is discarded (it could reference the SQL or a value) in favour of a
bare reason. No SQL is ever logged.
Summary
Types
The socket-open + auth function: mirrors Capstan.Connection's :connect_fun shape so a
test can inject a scripted transport. Default: default_connect/1 (gen_tcp + Handshake).
The pinned {host, port} the connection resolves to for its life.
Options for establish/1: :connection (the C1 connection keyword), optional :connect_fun
(default default_connect/1), optional :max_command_retries (default
CheckpointStore.default_max_retries/0), and optional :expected_server_uuid (the stream
connection's identity to verify against at the first connect, design Ch8).
Functions
Closes the pinned socket. Idempotent.
The default socket-open + auth function (gen_tcp + Handshake.connect/2), mirroring the C1
Capstan.Connection connect shape. Returns {:ok, socket, handshake_info} or
{:error, reason}. Public so a live test can drive the real transport.
The pinned {host, port} endpoint.
Opens an authenticated COM_QUERY connection, checks the ADR-0002 preconditions, reads
@@server_uuid, and pins the endpoint + identity for the connection's life.
Runs sql as a COM_QUERY on the pinned socket.
Reconnects, RE-verifying @@server_uuid against the pinned value (design Ch8).
The pinned @@server_uuid (structural identity — safe to surface, never a row value).
The pure source-identity check: :ok when there is no pin yet (the first establish) or the
observed uuid matches the pin, else {:error, :snapshot_source_mismatch} (design Ch8).
Types
@type connect_fun() :: (keyword() -> {:ok, Capstan.Protocol.Packet.socket(), map()} | {:error, term()})
The socket-open + auth function: mirrors Capstan.Connection's :connect_fun shape so a
test can inject a scripted transport. Default: default_connect/1 (gen_tcp + Handshake).
@type endpoint() :: {charlist(), :inet.port_number()}
The pinned {host, port} the connection resolves to for its life.
@type establish_opt() :: {:connection, keyword()} | {:connect_fun, connect_fun()} | {:max_command_retries, non_neg_integer()} | {:expected_server_uuid, String.t()}
Options for establish/1: :connection (the C1 connection keyword), optional :connect_fun
(default default_connect/1), optional :max_command_retries (default
CheckpointStore.default_max_retries/0), and optional :expected_server_uuid (the stream
connection's identity to verify against at the first connect, design Ch8).
@type t() :: %Capstan.Query{ connect_fun: connect_fun(), connection: keyword(), endpoint: endpoint(), max_retries: non_neg_integer(), server_uuid: String.t(), socket: Capstan.Protocol.Packet.socket() }
Functions
@spec close(t()) :: :ok
Closes the pinned socket. Idempotent.
@spec default_connect(keyword()) :: {:ok, Capstan.Protocol.Packet.socket(), map()} | {:error, term()}
The default socket-open + auth function (gen_tcp + Handshake.connect/2), mirroring the C1
Capstan.Connection connect shape. Returns {:ok, socket, handshake_info} or
{:error, reason}. Public so a live test can drive the real transport.
The pinned {host, port} endpoint.
@spec establish([establish_opt()]) :: {:ok, t()} | {:error, atom()}
Opens an authenticated COM_QUERY connection, checks the ADR-0002 preconditions, reads
@@server_uuid, and pins the endpoint + identity for the connection's life.
With :expected_server_uuid set (the stream connection's identity), the observed uuid is
verified against it at connect (design Ch8); otherwise the first read SETS the pin. Returns
{:ok, t()} or a value-free {:error, reason} — :snapshot_source_mismatch on an identity
mismatch, a distinct precondition reason on a bad substrate, or :snapshot_query_connect_failed
when the connect/auth budget is exhausted.
@spec query(t(), binary()) :: {:ok, [Capstan.Protocol.Command.row()]} | {:error, atom()}
Runs sql as a COM_QUERY on the pinned socket.
Returns {:ok, rows} for a text resultset (each row a list of column values, nil for a
NULL column), {:ok, []} for a statement with no resultset (SET, LOCK TABLES,
UNLOCK TABLES), or a value-free {:error, reason} (scrubbed through Capstan.Error.from/1;
a raised transport error becomes :transport). The SQL is never logged — a chunk WHERE
literal can carry a row value (Rule 1).
Reconnects, RE-verifying @@server_uuid against the pinned value (design Ch8).
A reconnect that lands on a different replica halts :snapshot_source_mismatch; otherwise a
fresh authenticated handle to the SAME source is returned.
The pinned @@server_uuid (structural identity — safe to surface, never a row value).
@spec verify_source_identity(String.t() | nil, String.t()) :: :ok | {:error, :snapshot_source_mismatch}
The pure source-identity check: :ok when there is no pin yet (the first establish) or the
observed uuid matches the pin, else {:error, :snapshot_source_mismatch} (design Ch8).
Exposed so the reconnect re-check is provably RED-capable in isolation.