Replicant.SnapshotProgress (Replicant v0.2.0)

Copy Markdown View Source

The incremental-backfill progress token (spec §6.2): a pure struct tracking the table queue, the in-progress table and its last delivered PK bound, and completion.

The wire form is version-tagged ({:replicant_snapshot_progress, @version, map}) self-identifying encoding — and decoded with binary_to_term(bin, [:safe]). It contains PK-bound ROW VALUES: it is data-plane (persisted by the sink or the checkpoint store) and must NEVER reach an error, log, or telemetry event (Critical Rule 1). A decode/shape fault is scrubbed to the bare :snapshot_progress_invalid — the raised exception is discarded, never inspected.

Summary

Types

t()
table_ref()

Functions

advance(sp, bound)

Record the last delivered chunk's upper PK bound for the in-progress table.

complete?(snapshot_progress)

Whether the whole backfill is complete (terminal).

decode(bin)

Decode a persisted token. EVERY failure — truncation, tamper, a foreign term, an unknown version, an atom-forging binary — returns the bare value-free {:error, :snapshot_progress_invalid} (spec §6.2/§9; Error.decode_failure/1 precedent: the exception is discarded).

encode(sp)

Version-tagged wire encoding.

finish_table(sp)

The in-progress table is fully delivered; move on.

mark_complete(sp)

Mark the whole backfill complete (terminal; written only after the completion call, spec §6.3).

new(tables, floor_lsn)

A fresh token over the discovered publication tables (spec §6.2).

next(sp)

The next unit of work: the in-progress table at its bound, the next queued table, or :complete.

redo_table(sp)

Reset the in-progress table's bound to re-read it from the start (PK-less redo, spec §6.4).

Types

t()

@type t() :: %Replicant.SnapshotProgress{
  bound: [term()] | nil,
  complete?: boolean(),
  current: table_ref() | nil,
  done: [String.t()],
  floor_lsn: Replicant.lsn(),
  pending: [table_ref()]
}

table_ref()

@type table_ref() :: %{
  schema: String.t(),
  table: String.t(),
  qualified: String.t(),
  pk_raw: [String.t()],
  pk_quoted: [String.t()]
}

Functions

advance(sp, bound)

@spec advance(t(), [term()]) :: t()

Record the last delivered chunk's upper PK bound for the in-progress table.

REQUIRES an in-progress table (the caller establishes one via next/1 first). Calling this with no current table is a caller bug, not a supported path — it raises FunctionClauseError rather than silently masking the ordering error.

complete?(snapshot_progress)

@spec complete?(t()) :: boolean()

Whether the whole backfill is complete (terminal).

decode(bin)

@spec decode(term()) :: {:ok, t()} | {:error, :snapshot_progress_invalid}

Decode a persisted token. EVERY failure — truncation, tamper, a foreign term, an unknown version, an atom-forging binary — returns the bare value-free {:error, :snapshot_progress_invalid} (spec §6.2/§9; Error.decode_failure/1 precedent: the exception is discarded).

encode(sp)

@spec encode(t()) :: binary()

Version-tagged wire encoding.

finish_table(sp)

@spec finish_table(t()) :: t()

The in-progress table is fully delivered; move on.

REQUIRES an in-progress table (the caller establishes one via next/1 first). Calling this with no current table is a caller bug, not a supported path — it raises FunctionClauseError rather than silently masking the ordering error.

mark_complete(sp)

@spec mark_complete(t()) :: t()

Mark the whole backfill complete (terminal; written only after the completion call, spec §6.3).

new(tables, floor_lsn)

@spec new([table_ref()], Replicant.lsn()) :: t()

A fresh token over the discovered publication tables (spec §6.2).

next(sp)

@spec next(t()) :: {:table, table_ref(), [term()] | nil, t()} | :complete

The next unit of work: the in-progress table at its bound, the next queued table, or :complete.

redo_table(sp)

@spec redo_table(t()) :: t()

Reset the in-progress table's bound to re-read it from the start (PK-less redo, spec §6.4).