Framework-agnostic Elixir CDC consumer for MySQL row-based binary-log replication,
delivering committed transactions to a pluggable sink with fail-closed,
transaction-granularity delivery: the checkpoint advances only after the sink has
durably applied a transaction (ADR-0003). In C1's lib-owned checkpoint mode this
is at-least-once — a crash between the sink write and the checkpoint re-delivers the
transaction on restart (a bounded duplicate window). Effect-once delivery is the
sink-owned atomic path (the sink persisting its write and the position together), which
C1 validates but does not yet run — see Capstan.Sink and ADR-0004.
Starting a pipeline
Configure each pipeline explicitly at start_link/1 (no global mutable state):
Capstan.start_link(
connection: [
host: "replica.internal", port: 3306, username: "capstan",
password: "…", database: "orders",
ssl: true, ssl_opts: [cacertfile: "/etc/ssl/mysql-ca.pem"] # TLS on by default (ADR-0002)
],
server_id: 1001, # replica identity; MUST be unique in the topology
sink: MyApp.OrdersSink, # implements Capstan.Sink
checkpoint_store: [module: MyApp.Store], # lib-owned mode; see below
start_position: :checkpoint, # C1: :checkpoint (default) only — see below
max_command_retries: 5
)start_link/1 validates the options through Capstan.Config, enforces the per-mode
Capstan.Sink callback required-ness (Capstan.Pipeline), and — for a valid
configuration — starts a supervised Capstan.Supervisor wiring a checkpoint store, a
Capstan.AssemblerServer, and a Capstan.Connection. A bad substrate is refused
before any socket is opened (a bad config fails start-up, not first delivery).
Checkpoint mode
C1 implements lib-owned checkpoint mode. Pass
checkpoint_store: [module: impl, options: keyword()], where impl is a
Capstan.CheckpointStore implementation exporting start_link/1 (e.g.
Capstan.CheckpointStore.InMemory for tests and ephemeral pipelines). The store is
supervised as part of the pipeline. Sink-owned checkpoint mode (the sink persisting
the position atomically with its own write) is validated but not yet run by C1's
AssemblerServer; a sink-owned configuration is refused with
:sink_owned_mode_unsupported.
Start position
C1 resumes only from the durable checkpoint (start_position: :checkpoint, the
default). A populated store resumes from the persisted watermark. A fresh
(never-written) store sends the dump an empty GTID set, which requests the
server's full retained history — and a server that has already purged its earliest
logs refuses that dump, halting :data_gap. To start "from now" today, seed the
checkpoint store with the server's current @@global.gtid_executed before first
start (see usage-rules.md). An explicit %Capstan.Position{}
override and :current are refused fail-closed (:start_position_override_unsupported
/ :start_position_current_unsupported) — honoring an explicit resume position
end-to-end (the AssemblerServer seeds its watermark from the store alone today) is a
C1-completeness follow-up. Refusing avoids a silent checkpoint hole.
Return values
{:ok, supervisor_pid} on success, or {:error, reason} — a value-free atom from
Capstan.Config (:server_id_required, :config_invalid,
:tls_verification_unspecified), from Capstan.Pipeline (:invalid_sink,
:sink_missing_handle_transaction, :sink_missing_checkpoint,
:sink_missing_handle_schema_change), or from this module
(:sink_owned_mode_unsupported, :checkpoint_store_required,
:start_position_override_unsupported, :start_position_current_unsupported).
Summary
Functions
A supervisor child spec, so a pipeline can be embedded in an application's supervision
tree: {Capstan, connection: [...], server_id: ..., sink: ..., checkpoint_store: [...]}.
Validate opts and start a supervised pipeline. See the moduledoc for the option shape
and the value-free {:error, reason} set. Returns {:ok, supervisor_pid} on success.
Stop a running pipeline supervisor started by start_link/1.
Functions
@spec child_spec(keyword()) :: Supervisor.child_spec()
A supervisor child spec, so a pipeline can be embedded in an application's supervision
tree: {Capstan, connection: [...], server_id: ..., sink: ..., checkpoint_store: [...]}.
Validate opts and start a supervised pipeline. See the moduledoc for the option shape
and the value-free {:error, reason} set. Returns {:ok, supervisor_pid} on success.
@spec stop(pid()) :: :ok
Stop a running pipeline supervisor started by start_link/1.