Capstan.Pipeline (Capstan v1.2.1)

Copy Markdown View Source

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_store configured): handle_transaction/1.
  • sink-owned (no :checkpoint_store): checkpoint/0 and handle_transaction/1.
  • handle_schema_change/2 whenever DDL delivery is enabled — in C1 the pipeline ALWAYS delivers self-committing DDL (ADR-0003), so it is required in both modes.

Both modes run (C1a): lib-owned wires a store child; sink-owned validates checkpoint/0 here and the AssemblerServer seeds from it — no store child.

Summary

Types

coordinator_wiring()

The bootstrap-provided coordinator wiring (the 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 "").

sink_error()

A value-free sink-validation refusal.

Functions

assembler_spec(opts, checkpoint_store)

The AssemblerServer child spec, wired to the started checkpoint store {impl, handle}.

batch_mode?(opts)

The configured batch mode: nil (no batching), :lib_owned, or :sink_owned.

checkpoint_store(opts)

The lib-owned checkpoint store's implementation module and its start_link/1 options, drawn from checkpoint_store: [module: impl, options: keyword()].

connection_spec(opts, receiver, start_position, checkpoint_store \\ nil)

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.

coordinator_spec(opts, wiring)

The Capstan.Snapshot.Coordinator child spec (snapshot mode only), mirroring assembler_spec/2.

lib_mode?(opts)

Is this a lib-owned pipeline? True iff a non-nil :checkpoint_store is configured.

resolve_start_position(opts, resumed)

Resolve the public :start_position (default :checkpoint) against the position the authority (checkpoint store / sink checkpoint) resumed with.

snapshot_mode?(opts)

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.

snapshot_store_spec(impl, store_options)

The durable Capstan.SnapshotStore child spec (snapshot mode only), mirroring store_spec/2.

store_spec(impl, store_options)

The checkpoint-store child spec (a :temporary child — a store fault halts, never restarts).

validate_sink(opts)

Validate the configured :sink against its checkpoint mode's required callbacks.

validate_snapshot_tables(opts, arg2)

Refuse a snapshot table outside the capture allowlist (:snapshot_table_not_captured).

Types

coordinator_wiring()

@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 (the 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 "").

sink_error()

@type sink_error() ::
  :invalid_sink
  | :sink_missing_handle_transaction
  | :sink_missing_checkpoint
  | :sink_missing_handle_schema_change
  | :sink_missing_handle_snapshot
  | :sink_missing_handle_batch

A value-free sink-validation refusal.

Functions

assembler_spec(opts, checkpoint_store)

@spec assembler_spec(
  keyword(),
  Capstan.AssemblerServer.checkpoint_store() | nil
) :: Supervisor.child_spec()

The AssemblerServer child spec, wired to the started checkpoint store {impl, handle}.

batch_mode?(opts)

@spec batch_mode?(keyword()) :: nil | :lib_owned | :sink_owned

The configured batch mode: nil (no batching), :lib_owned, or :sink_owned.

checkpoint_store(opts)

@spec checkpoint_store(keyword()) :: {module(), keyword()}

The lib-owned checkpoint store's implementation module and its start_link/1 options, drawn from checkpoint_store: [module: impl, options: keyword()].

connection_spec(opts, receiver, start_position, checkpoint_store \\ nil)

@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.

coordinator_spec(opts, wiring)

@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) — the supervisor builds it once the assembler + snapshot-store children have started. A :temporary child: a fail-closed snapshot halt never restarts.

lib_mode?(opts)

@spec lib_mode?(keyword()) :: boolean()

Is this a lib-owned pipeline? True iff a non-nil :checkpoint_store is configured.

resolve_start_position(opts, resumed)

@spec resolve_start_position(
  keyword(),
  Capstan.Position.t() | nil
) ::
  {:ok, Capstan.Position.t() | nil}
  | {:ok, :current, nil}
  | {:error, :config_invalid}

Resolve the public :start_position (default :checkpoint) against the position the authority (checkpoint store / sink checkpoint) resumed with.

  • :checkpoint — the resumed position (a %Position{} or nil for a fresh start);
  • a %Capstan.Position{} — the override itself (C1b): BOTH the dump and the assembler watermark seed from it, so the two can never disagree;
  • :current — {:ok, :current, nil} marker (C1b): the caller resolves it against the server's live @@global.gtid_executed over an authenticated socket, pre-dump;
  • anything else — {:error, :config_invalid}.

snapshot_mode?(opts)

@spec snapshot_mode?(keyword()) :: boolean()

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.

snapshot_store_spec(impl, store_options)

@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.

store_spec(impl, store_options)

@spec store_spec(
  module(),
  keyword()
) :: Supervisor.child_spec()

The checkpoint-store child spec (a :temporary child — a store fault halts, never restarts).

validate_sink(opts)

@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).

validate_snapshot_tables(opts, arg2)

@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).