Pipeline entry — the per-mode Sink validation and the child-spec wiring for one
supervised pipeline (a Capstan.CheckpointStore process, a Capstan.AssemblerServer,
and a Capstan.Connection).
Capstan.Supervisor starts these children in order and threads the pids, because the
Connection's :receiver must be the AssemblerServer's PID. The
Connection forwards frames with a plain send/2, so the receiver is a PID: a
send to a terminated AssemblerServer is a silent no-op, never a raise that could
restart a fail-closed pipeline into a livelock. The Connection also monitors that
PID, so the AssemblerServer's fail-closed halt (which stops it without messaging back)
is not silent — the Connection stops fail-closed on the :DOWN instead of streaming
into the dead pid. Monitoring detects the death without linking, so neither :temporary
child restarts into the livelock the plain-send avoids.
Per-mode Sink callback required-ness
Capstan.Sink declares every callback @optional_callbacks; which are REQUIRED
depends on the checkpoint mode, enforced here at start-up via function_exported?/3 so
a sink missing a required callback is refused before a first-call crash:
- lib-owned (
:checkpoint_storeconfigured):handle_transaction/1. - sink-owned (no
:checkpoint_store):checkpoint/0andhandle_transaction/1. handle_schema_change/2whenever DDL delivery is enabled — in C1 the pipeline ALWAYS delivers self-committing DDL (ADR-0003), so it is required in both modes.
Note: C1's AssemblerServer implements only lib-owned checkpoint mode. Sink-owned
required-ness is still validated here so a bad
sink-owned sink is refused; Capstan.start_link/1 refuses to run a sink-owned
pipeline in C1 (:sink_owned_mode_unsupported).
Summary
Types
The bootstrap-provided coordinator wiring (Task 10's supervisor supplies these AFTER the
assembler + snapshot-store children are up): the started AssemblerServer pid the coordinator
sends {:capstan_halt, _} to and observes the watermark from, the started {impl, handle}
snapshot store, the initial %Capstan.Snapshot.State{}, the per-table opened ChunkReader
handles, and the initial processed-watermark string (default "").
A value-free sink-validation refusal.
Functions
The AssemblerServer child spec, wired to the started checkpoint store {impl, handle}.
The lib-owned checkpoint store's implementation module and its start_link/1 options,
drawn from checkpoint_store: [module: impl, options: keyword()].
The Connection child spec, wired so its :receiver is the AssemblerServer PID, its
dump resumes from start_position, and — given the started checkpoint store
{impl, handle} — it re-reads the durable resume position on every establish (so a
reconnect resumes from the current watermark, not the frozen start-up position; design
Q7 / F1). checkpoint_store defaults to nil so a Connection wired without a store
(a unit test) keeps the injected start_position.
The Capstan.Snapshot.Coordinator child spec (snapshot mode only), mirroring assembler_spec/2.
Is this a lib-owned pipeline? True iff a non-nil :checkpoint_store is configured.
Resolve the public :start_position (default :checkpoint) against the position the
checkpoint store resumed with.
Is this an initial-snapshot pipeline? True iff a non-nil :snapshot block is configured
(mirrors lib_mode?/1). An absent (or nil) :snapshot key ⇒ pure C1.
The durable Capstan.SnapshotStore child spec (snapshot mode only), mirroring store_spec/2.
The checkpoint-store child spec (a :temporary child — a store fault halts, never restarts).
Validate the configured :sink against its checkpoint mode's required callbacks.
Refuse a snapshot table outside the capture allowlist (:snapshot_table_not_captured).
Types
@type coordinator_wiring() :: [ assembler: pid() | atom(), snapshot_store: {module(), term()}, snapshot_state: Capstan.Snapshot.State.t(), readers: %{optional({String.t(), String.t()}) => term()}, processed_set: String.t() ]
The bootstrap-provided coordinator wiring (Task 10's supervisor supplies these AFTER the
assembler + snapshot-store children are up): the started AssemblerServer pid the coordinator
sends {:capstan_halt, _} to and observes the watermark from, the started {impl, handle}
snapshot store, the initial %Capstan.Snapshot.State{}, the per-table opened ChunkReader
handles, and the initial processed-watermark string (default "").
@type sink_error() ::
:invalid_sink
| :sink_missing_handle_transaction
| :sink_missing_checkpoint
| :sink_missing_handle_schema_change
| :sink_missing_handle_snapshot
A value-free sink-validation refusal.
Functions
@spec assembler_spec( keyword(), Capstan.AssemblerServer.checkpoint_store() ) :: Supervisor.child_spec()
The AssemblerServer child spec, wired to the started checkpoint store {impl, handle}.
The lib-owned checkpoint store's implementation module and its start_link/1 options,
drawn from checkpoint_store: [module: impl, options: keyword()].
@spec connection_spec( keyword(), pid(), Capstan.Position.t() | nil, Capstan.AssemblerServer.checkpoint_store() | nil ) :: Supervisor.child_spec()
The Connection child spec, wired so its :receiver is the AssemblerServer PID, its
dump resumes from start_position, and — given the started checkpoint store
{impl, handle} — it re-reads the durable resume position on every establish (so a
reconnect resumes from the current watermark, not the frozen start-up position; design
Q7 / F1). checkpoint_store defaults to nil so a Connection wired without a store
(a unit test) keeps the injected start_position.
@spec coordinator_spec( keyword(), coordinator_wiring() ) :: Supervisor.child_spec()
The Capstan.Snapshot.Coordinator child spec (snapshot mode only), mirroring assembler_spec/2.
The real (downstream) sink + the retry budget come from the pipeline opts; wiring carries
the bootstrap-produced pieces (see coordinator_wiring/0) — Task 10's supervisor builds it
once the assembler + snapshot-store children have started. A :temporary child: a fail-closed
snapshot halt never restarts.
Is this a lib-owned pipeline? True iff a non-nil :checkpoint_store is configured.
@spec resolve_start_position( keyword(), Capstan.Position.t() | nil ) :: {:ok, Capstan.Position.t() | nil} | {:error, :start_position_override_unsupported | :start_position_current_unsupported | :config_invalid}
Resolve the public :start_position (default :checkpoint) against the position the
checkpoint store resumed with.
:checkpoint— the resumed position (a%Position{}ornilfor a fresh start);- a
%Capstan.Position{}—{:error, :start_position_override_unsupported}in C1 (an explicit override would resume theConnection's dump but NOT theAssemblerServer's watermark, which seeds from the store alone — a silent hole; so it is refused fail-closed, matchingCapstan.start_link/1's pre-flight check); :current—{:error, :start_position_current_unsupported}(C1 does not implement "start from the server's current position"; it needs a live pre-connect query the spine does not yet wire);- anything else —
{:error, :config_invalid}.
Is this an initial-snapshot pipeline? True iff a non-nil :snapshot block is configured
(mirrors lib_mode?/1). An absent (or nil) :snapshot key ⇒ pure C1.
@spec snapshot_store_spec( module(), keyword() ) :: Supervisor.child_spec()
The durable Capstan.SnapshotStore child spec (snapshot mode only), mirroring store_spec/2.
A :temporary child — a snapshot-store fault halts fail-closed, never restarts into a
re-scan-from-zero livelock.
@spec store_spec( module(), keyword() ) :: Supervisor.child_spec()
The checkpoint-store child spec (a :temporary child — a store fault halts, never restarts).
@spec validate_sink(keyword()) :: :ok | {:error, sink_error()}
Validate the configured :sink against its checkpoint mode's required callbacks.
Returns :ok, or a distinct value-free refusal naming the missing callback. Every
check is a runtime function_exported?/3 (the callbacks are @optional_callbacks).
@spec validate_snapshot_tables( keyword(), Capstan.Config.snapshot_config() | nil ) :: :ok | {:error, :snapshot_table_not_captured}
Refuse a snapshot table outside the capture allowlist (:snapshot_table_not_captured).
Every snapshot table MUST be captured by the stream: the stream is what delivers the
gtid > G changes the cursor-gate suppresses from the chunk, so a snapshot table the stream
does not carry would silently gap. :all capture trivially includes every snapshot table.
snapshot is the normalised Capstan.Config.snapshot_config() (or nil in pure C1 — a
no-op :ok).