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,
  heartbeat_period_ms: 15_000,           # liveness: the window MUST exceed the heartbeat
  stream_timeout_ms: 60_000              # (defaults shown; see "Streaming liveness")
)

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

Lib-owned (default): 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 (C1a): omit checkpoint_store: and implement Capstan.Sink.checkpoint/0 — the sink persists its data and the position atomically together and returns the position from handle_transaction/1; capstan reads the resume position from checkpoint/0 and advances in-memory after each durable sink write. Effect-once by construction (see usage-rules "Sink-owned checkpoint mode").

Start position

start_position: accepts three forms. :checkpoint (default) resumes from the position authority — the checkpoint store (lib-owned) or Capstan.Sink.checkpoint/0 (sink-owned); a fresh authority sends the dump an empty GTID set, which requests the server's full retained history — a server that has already purged its earliest logs refuses that dump, halting :data_gap (seed the authority with the server's current @@global.gtid_executed for a "from now" start; see usage-rules.md). An explicit %Capstan.Position{} resumes BOTH the dump and the assembler watermark from that position — the operator asserts it is a safe resume point (transactions it already covers are NOT re-delivered; anything after it streams). :current reads the server's live @@global.gtid_executed pre-dump and resumes from it — "start from now" without pre-seeding.

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, :invalid_liveness_config, :unknown_option), from Capstan.Pipeline (:invalid_sink, :sink_missing_handle_transaction, :sink_missing_checkpoint, :sink_missing_handle_schema_change, :sink_missing_handle_snapshot, :snapshot_table_not_captured), or from this module (:checkpoint_store_required).

Initial snapshot (C2)

An optional snapshot: [tables: […], store: [module: …], chunk_size: 4096] block enables a consistent backfill of pre-existing rows woven into the running stream. It is additive: with no :snapshot key the pipeline is pure C1, byte-for-byte unchanged. In snapshot mode the sink must implement Capstan.Sink.handle_snapshot/2 (else :sink_missing_handle_snapshot), the snapshot tables must be a subset of the capture allowlist (else :snapshot_table_not_captured), and a missing/mis-shaped store is refused :config_invalid. :current / %Capstan.Position{} start positions stay refused fail-closed (no C1b).