Capstan.CheckpointStore behaviour (Capstan v0.1.0)

Copy Markdown View Source

The lib-owned checkpoint persistence contract.

A checkpoint store persists exactly one value per pipeline identity: the processed gtid_set STRING. file/pos are never persisted — they are diagnostic only and are derived on read (Capstan.Position.from_persisted/1 restores them as nil). Because the callback surface only ever sees the persisted string, an implementation cannot accidentally persist file/pos; there is exactly one durable value and no two-representation divergence to occur on.

The checkpoint is a processed watermark (ADR-0003): it records every committed GTID the pipeline has processed, delivered or filtered — see Capstan.Sink.

Behaviour

An implementation persists and reads back the gtid_set string for one pipeline:

  • read/1{:ok, gtid_set | nil} (nil = never written) or a value-free {:error, term()}.

  • write/2 — durably store the gtid_set string (idempotent: writing the same value twice is indistinguishable from writing it once) or {:error, term()}.

Capstan.CheckpointStore.InMemory is a process-lifetime reference implementation for tests and ephemeral pipelines — it is NOT durable across a restart.

Position boundary

read_position/2 and write_position/3 bridge Capstan.Position and the persisted string through Capstan.Position.to_persisted/1 and from_persisted/1. They are the one place the persist boundary is applied, so a caller advancing the checkpoint never hand-rolls it and can never leak file/pos.

Retry budget

default_max_retries/0, retry_decision/2, and permanent_reason?/1 mirror replicant/lib/replicant/checkpoint_store.ex so a store fault is retried a bounded number of times and then halts fail-closed, with the same counter semantics the connect-read and mid-stream write sites share (they cannot drift). A permanent reason halts immediately without spending the budget.

Summary

Types

A per-pipeline store handle (e.g. the pid/name of a store process).

Callbacks

Read the durable checkpoint for this pipeline: {:ok, gtid_set | nil} (nil = never written) or a value-free error.

Durably persist gtid_set for this pipeline. Idempotent — re-writing the same value is a no-op-equivalent. Returns :ok or a value-free error.

Functions

Default max_retries for a checkpoint-store fault when none is configured.

True when a store-fault reason is PERMANENT (retrying cannot fix it) and must halt immediately without spending the retry budget.

Read the checkpoint as a Capstan.Position (or nil), applying the persist boundary.

The shared retry decision: :retry while attempt < max_retries, else :halt.

Persist position's gtid_set alone, applying the persist boundary.

Types

store()

@type store() :: term()

A per-pipeline store handle (e.g. the pid/name of a store process).

Callbacks

read(store)

@callback read(store()) :: {:ok, String.t() | nil} | {:error, term()}

Read the durable checkpoint for this pipeline: {:ok, gtid_set | nil} (nil = never written) or a value-free error.

write(store, gtid_set)

@callback write(store(), gtid_set :: String.t()) :: :ok | {:error, term()}

Durably persist gtid_set for this pipeline. Idempotent — re-writing the same value is a no-op-equivalent. Returns :ok or a value-free error.

Functions

default_max_retries()

@spec default_max_retries() :: non_neg_integer()

Default max_retries for a checkpoint-store fault when none is configured.

permanent_reason?(arg1)

@spec permanent_reason?(atom()) :: boolean()

True when a store-fault reason is PERMANENT (retrying cannot fix it) and must halt immediately without spending the retry budget.

:config_invalid — a mis-shaped store configuration — is permanent. Every other reason is transient at the value-free boundary (a momentary blip and a wrong-host misconfig are indistinguishable here), so it is retried, then halted.

read_position(impl, store)

@spec read_position(module(), store()) ::
  {:ok, Capstan.Position.t() | nil} | {:error, term()}

Read the checkpoint as a Capstan.Position (or nil), applying the persist boundary.

Restores file/pos as nil via Capstan.Position.from_persisted/1. impl is the callback module; store is its handle.

retry_decision(attempt, max_retries)

@spec retry_decision(non_neg_integer(), non_neg_integer()) :: :retry | :halt

The shared retry decision: :retry while attempt < max_retries, else :halt.

attempt is the count of retries ALREADY made (0 on the first fault). max_retries: 0 ⇒ always :halt (halt-now opt-out). Mirrors replicant/lib/replicant/checkpoint_store.ex:319-320.

write_position(impl, store, position)

@spec write_position(module(), store(), Capstan.Position.t()) ::
  :ok | {:error, term()}

Persist position's gtid_set alone, applying the persist boundary.

file/pos are dropped by Capstan.Position.to_persisted/1 before the write, so they never reach the store. impl is the callback module; store is its handle.