Capstan.Pipeline (Capstan v0.1.0)

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.

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

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

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.

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 checkpoint store resumed with.

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.

Types

sink_error()

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

A value-free sink-validation refusal.

Functions

assembler_spec(opts, checkpoint_store)

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

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

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.

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}
  | {: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{} or nil for a fresh start);
  • a %Capstan.Position{} — {:error, :start_position_override_unsupported} in C1 (an explicit override would resume the Connection's dump but NOT the AssemblerServer's watermark, which seeds from the store alone — a silent hole; so it is refused fail-closed, matching Capstan.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}.

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