capstan usage rules

capstan streams committed MySQL row changes from the binary log to a sink in your supervision tree. This file is the consumer-facing contract; the permanent decision record is in docs/adr/.

C1 scope (ADR-0004): C1 ships lib-owned checkpoint mode and resume-from-durable-checkpoint (start_position: :checkpoint) only. Sink-owned checkpoint mode and explicit start positions are deferred and refused fail-closed today.

Starting a pipeline

Configure each pipeline explicitly at Capstan.start_link/1 — there is no global mutable state.

Capstan.start_link(
  connection: [
    host: "replica.internal",
    port: 3306,
    username: "capstan",
    password: "…",
    database: "orders",
    ssl: true,                                  # default true (ADR-0002)
    ssl_opts: [cacertfile: "/etc/ssl/mysql-ca.pem", server_name_indication: :disable],
    auth_plugins: [:caching_sha2_password]      # default; name :mysql_native_password to allow it
  ],
  server_id: 1001,                              # replica identity; MUST be unique in the topology
  sink: MyApp.OrdersSink,                       # implements Capstan.Sink
  checkpoint_store: [module: MyApp.Store, options: []],   # lib-owned mode; see below
  start_position: :checkpoint,                  # C1: :checkpoint (default) only
  tables: [{"orders", "line_items"}],           # or :all (default); filter applied before decode
  max_command_retries: 5,                       # default 5; pre-establish failures only
  reconnect_backoff: 1_000,                     # default; ms between reconnect attempts
  heartbeat_period_ms: 15_000,                  # default; server heartbeat on a quiet stream
  stream_timeout_ms: 60_000                     # default; MUST be > heartbeat_period_ms
)

• Returns {:ok, supervisor_pid}, or {:error, reason} where reason is a value-free atom. A bad substrate or config is refused before any socket opens.
• Embed it with the child spec: {Capstan, connection: [...], server_id: ..., sink: ..., checkpoint_store: [...]}.

TLS (ADR-0002)

ssl defaults true. With TLS on you MUST make an explicit peer-verification choice, or start fails closed with :tls_verification_unspecified:

• authenticated: pass cacertfile: (or cacerts:). MySQL's auto-generated cert is self-signed with no SAN, so also pass server_name_indication: :disable (VERIFY_CA — chain verified, hostname not); or
• unauthenticated confidentiality only: pass verify: :verify_none explicitly; or
• opt out entirely with ssl: false.

Streaming liveness

A silent half-open partition cannot hang the pipeline: capstan asks the server for a replication heartbeat every heartbeat_period_ms (default 15 000) of stream idleness, and a liveness timer declares the stream dead when no frame — event or heartbeat — arrives within stream_timeout_ms (default 60 000, tolerating three missed heartbeats). On fire it emits [:capstan, :connection, :stream_timeout], drops the connection, and reconnects; a persistent partition halts :stream_stalled. stream_timeout_ms MUST exceed heartbeat_period_ms, or the pipeline refuses to start (:invalid_liveness_config) — a window at or below the heartbeat interval would false-drop a healthy idle stream.

First start — an empty checkpoint requests FULL retained history

A never-written store sends the binlog dump an empty GTID set, which means "everything the server still retains". Two consequences:

• On a server that has already purged its earliest logs (@@global.gtid_purged non-empty), the server refuses the dump and the pipeline halts :data_gap.
• On a server with complete retained history, the sink replays all of it from the top.

To start "from now" instead, seed the checkpoint store before first start with the server's current executed set — the store is the position authority, so a pre-seeded value is simply resumed from:

# 1. Read the server's current position with any MySQL client:
#      SELECT @@global.gtid_executed;
# 2. Write that string into your store before starting the pipeline:
:ok = MyApp.Store.write(store, gtid_executed)
# Capstan.start_link/1 now resumes from it — nothing before it is delivered.

Value-free start-up refusals

Capstan.start_link/1 returns {:error, reason} with one of: :server_id_required, :config_invalid, :tls_verification_unspecified, :invalid_sink, :sink_missing_handle_transaction, :sink_missing_checkpoint, :sink_missing_handle_schema_change, :checkpoint_store_required, :sink_owned_mode_unsupported, :start_position_override_unsupported, :start_position_current_unsupported. In snapshot mode also: :sink_missing_handle_snapshot, :snapshot_table_not_captured, and :config_invalid for a mis-shaped snapshot: block or a tables: :all snapshot set (see Initial snapshot). No row value, column value, or password ever appears in an error, log line, or telemetry payload (Rule 1).

The Capstan.Sink behaviour

Three callbacks; every one is @optional_callbacks — which are required depends on the mode. For C1's lib-owned mode you implement handle_transaction/1 and handle_schema_change/2.

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

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

@callback checkpoint() :: {:ok, Capstan.Position.t() | nil} | {:error, term()}
# checkpoint/0 is consulted only in the (deferred) sink-owned mode.

Three load-bearing rules — each guards a silent-loss class the type signature alone does not:

1. changes is Enumerable.t(), not a list. Never call length/1 / Enum.count/1 on it. C1 delivers a list, but the type is the day-one contract: a later row makes changes a lazy, single-pass, disk-backed enumerable valid only during the delivery call. Enumerate exactly once, streaming (Enum.reduce/3, Enum.each/2, for, or a single-pass Stream).

2. Dedup with Capstan.Gtid.member?/2 — never an ordinal comparison. A GTID set is legitimately non-contiguous (failover, purge, multi-source), so commit_lsn <= checkpoint-style comparison silently skips or re-applies. Skip a transaction when Capstan.Gtid.member?(checkpoint_set, {uuid, gno}) is true.

3. The checkpoint is a PROCESSED watermark, not a delivery log. It records every committed GTID processed — delivered and filtered. It does not mean "delivered".

handle_schema_change/2 receives only structured schema/table/kind — the raw DDL statement text is redacted before it reaches the sink (Rule 1). Return {:error, term()} from either delivery callback to halt the pipeline fail-closed without advancing the checkpoint.

Checkpoint store (lib-owned mode)

Capstan.CheckpointStore persists exactly one value per pipeline: the processed gtid_set string (file/pos are diagnostic and never persisted). Implement two callbacks:

@callback read(store()) :: {:ok, String.t() | nil} | {:error, term()}   # nil = never written
@callback write(store(), gtid_set :: String.t()) :: :ok | {:error, term()}  # idempotent

Capstan.CheckpointStore.InMemory is a process-lifetime reference implementation (start_link/1, read/1, write/2) for tests and ephemeral pipelines — not durable across a restart. For production, implement the two callbacks over storage you already trust (your application database, a file written atomically) — write/2 must be durable and idempotent. Bridge Capstan.Position through read_position/2 / write_position/3, which apply the persist boundary so you never leak file/pos.

Wiring: checkpoint_store: [module: impl, options: keyword()] — the pipeline supervises the store, calling impl.start_link(options) (options defaults to []). A store fault at read or write is retried up to 5 times (not configurable in C1; :config_invalid is treated as permanent and halts immediately), then the pipeline halts fail-closed — {:checkpoint_read_failed, reason} / {:checkpoint_write_failed, reason}. This budget is separate from max_command_retries, which governs only pre-establish connection failures.

Initial snapshot (C2)

Backfill the rows that pre-exist the pipeline's start, woven into the running stream so there is no gap and no duplicate at the snapshot→stream handoff (ADR-0005). Absent the :snapshot block, a pipeline is pure C1 — this is purely additive.

Capstan.start_link(
  connection: [...], server_id: 1001, sink: MyApp.Sink,
  checkpoint_store: [module: MyApp.GtidStore],
  tables: [{"app", "orders"}],                 # capture allowlist (must be a concrete list to snapshot :all)
  snapshot: [
    tables: [{"app", "orders"}],               # ⊆ captured; DEFAULTS to the captured list
    store:  [module: MyApp.SnapshotStore],      # durable snapshot progress (SEPARATE from the gtid store)
    chunk_size: 4096                            # rows per chunk (bounds the brief lock's hold + memory)
  ]
)

Source privilege. The query account needs LOCK TABLES on the snapshot tables (in addition to SELECT) — capstan takes a brief per-chunk LOCK TABLES … READ to capture an exact GTID position (the read-only-replica-safe substitute for a written watermark; ADR-0005). No RELOAD / FTWRL. The preflight (scripts/capstan-preflight.sql) checks it.

The sink implements handle_snapshot/2 (optional in general; required in snapshot mode):

@callback handle_snapshot([Capstan.Change.t()], Capstan.Snapshot.Meta.t()) :: :ok | {:error, term()}

• The first argument is a concrete list of %Capstan.Change{op: :snapshot} (record = the full row, old_record = nil) — a bounded, fully-materialized chunk, NOT the lazy Enumerable.t() of handle_transaction/1. Capstan.Snapshot.Meta carries only value-free structural identity (schema/table/chunk_seq/g/final_chunk?).
• Apply each chunk row as an UPSERT keyed on its primary key. This is a HARD precondition: a snapshot row may be superseded by a later streamed change, and the bounded crash-window re-emit relies on upsert-by-PK to converge. Return {:error, term()} to halt fail-closed.

Guarantees. Strict-once in normal operation. The only duplicate window is a crash between the chunk's {:ok} and the durable pk_cursor persist — the one in-flight chunk re-emits, exactly C1's bounded posture; an upsert-by-PK sink converges. A DELETE landing in that same window is swept, not left as a phantom: a delivered_pk high-water (persisted before each emit) lets the cursor-gate forward a streamed delete of an already-delivered key on restart. Resumable mid-backfill: the per-table PK cursor lives in MyApp.SnapshotStore (implement read/1+write/2 over a %Capstan.Snapshot.State{}, same durable/idempotent contract as the checkpoint store), and the backfill resumes with WHERE pk > cursor — never a re-scan from zero. A store whose status: :complete never re-snapshots.

The snapshot table set is fixed once a durable state exists. Adding or removing a table in snapshot.tables after a :complete/mid-snapshot state is a config/state divergence and halts :snapshot_config_drifted — a newly-added table would otherwise be silently never backfilled. To re-backfill a changed set, drop the durable snapshot state (a fresh start re-introspects config).

Supported primary keys (order-faithful only). Chunking pages by MySQL ORDER BY pk while the gate compares in Elixir, so the PK type's Elixir order must provably match MySQL's: integer (signed/unsigned, incl. BIGINT UNSIGNED), BINARY/VARBINARY, and composites of those. A collation-ordered STRING PK (CHAR/VARCHAR/TEXT) is refused :snapshot_pk_unsupported_type (a named follow-up, ROADMAP C2a). A tables: :all snapshot — which arises when the capture allowlist is itself :all — is refused :config_invalid: specify an explicit snapshot table list ("snapshot every table on the server" is ambiguous and dangerous; ROADMAP C2b).

Runtime halts (fail-closed)

Any condition that could otherwise lose or corrupt data silently halts the pipeline: the processes exit {:shutdown, {:halt, reason}} and — because every child is restart: :temporary — the pipeline is not restarted. Your application observes the halt via [:capstan, :connection, :halt] / [:capstan, :assembler, :halt] telemetry (metadata reason) and decides whether to restart, reprovision, or page. Every reason is value-free.

Server preconditions (checked at every connect; a violation halts without retrying): :binlog_format_not_row, :binlog_row_image_not_full, :binlog_row_metadata_not_full, :binlog_row_value_options_not_empty, :gtid_mode_not_on, :precondition_query_failed.

Connection halts:

• :data_gap — the server purged binlogs the pipeline still needs (also the fresh-start case above). Re-seed or reprovision; resuming would silently skip transactions.
• :source_identity_mismatch — the checkpoint carries GTIDs this server never executed; almost always a checkpoint pointed at the wrong server.
• :server_id_conflict — another replica with the same server_id is attached.
• :stream_stalled — a persistent network partition outlived the liveness reconnect budget.
• :command_retries_exhausted — pre-establish failures (connect/auth/query) exceeded max_command_retries.
• :invalid_liveness_config — stream_timeout_ms <= heartbeat_period_ms.
• :checksum_negotiation_failed, :unrecognized_dump_error, {:dump_failed, code}, :unexpected_stream_packet, :receiver_down — protocol-level refusals; the reason names the site.

Delivery halts (tagged tuples; telemetry reports the outer atom):

• {:sink_error, reason} — your sink returned {:error, reason}; the checkpoint is held, so the transaction is re-delivered after a restart (at-least-once).
• {:checkpoint_read_failed, reason} / {:checkpoint_write_failed, reason} — store fault after the retry budget.
• {:assembler_error, reason} — a stream desync (:begin_without_gtid, :rows_without_transaction, …) or a row the pipeline refuses to decode: :unmapped_table_id, {:unsupported_column_type, detail} (spatial, SET, pre-5.6 temporals), a compressed transaction payload, an unknown event type.
• :unsupported_transaction_shape — an XA transaction (XA PREPARE); the buffered rows are discarded, never delivered (ADR-0003).
• {:event_parse_failed, reason} — CRC mismatch or truncated event.
• {:event_decode_crashed, %Capstan.Error{}} / {:event_processing_crashed, %Capstan.Error{}} — an unexpected raise, captured value-free.

Snapshot halts (snapshot mode; each a distinct value-free reason per silent-loss condition):

• :snapshot_table_no_primary_key — a snapshot table has no usable key (no PK and no NOT NULL-complete unique key).
• :snapshot_pk_unsupported_type — a PK column outside the order-faithful allowlist (a collation-ordered string, DECIMAL/DOUBLE/temporal).
• :snapshot_lock_unavailable — the brief LOCK TABLES … READ failed (missing LOCK TABLES privilege, or a lock-wait timeout) beyond the retry budget.
• :snapshot_schema_drifted — a snapshot table's structure changed mid-backfill (a per-chunk fingerprint change, or an observed DDL on the table).
• :snapshot_source_mismatch — the query connection's @@server_uuid differs from the stream's (a reconnect to a different replica, at connect or on any query-conn reconnect).
• :snapshot_config_drifted — the configured snapshot.tables no longer match the durable %Capstan.Snapshot.State{}'s table set (a table added/removed after a :complete/mid-snapshot state). Halts rather than silently never backfilling the added table; drop the durable snapshot state to re-backfill a changed set.
• :snapshot_chunk_read_failed / :snapshot_query_connect_failed / :snapshot_bootstrap_gtid_read_failed — a chunk read, query-connect, or P0 read fault beyond the budget.
• :snapshot_state_read_failed / :snapshot_state_write_failed — a SnapshotStore fault (budgeted, then halts).
• :snapshot_coordinator_down — the snapshot coordinator died silently (a stranded backfill is a gap, so it halts loud).
• {:snapshot_sink_error, reason} / {:snapshot_processing_crashed, %Capstan.Error{}} — your handle_snapshot/2 returned {:error, _}, or the snapshot path raised (captured value-free).

Telemetry

All events are emitted through a value-free metadata allowlist (a payload with a key outside it raises rather than ships). Measurements maps are empty in C1 — attach to the event names:

The only log line emitted in normal operation is a Logger warning when authenticating with the deprecated mysql_native_password plugin.

Substrate requirements (ADR-0002)

capstan verifies these at connect and refuses to start (a distinct reason per violation) if any is wrong:

binlog_format            = ROW
binlog_row_image         = FULL
binlog_row_metadata      = FULL
binlog_row_value_options = ''        # PARTIAL_JSON is refused — it emits JSON diffs, not values
gtid_mode                = ON

enforce_gtid_consistency = ON is recommended (implied by gtid_mode = ON) but is not separately checked. Multi-source replication is supported — a GTID set expresses multiple source UUIDs natively. The replication account authenticates via caching_sha2_password by default and needs REPLICATION SLAVE, REPLICATION CLIENT, and SELECT.